File size: 2,768 Bytes
c0cb280 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | import logging
from typing import Any, Dict, List
from accounting.models import Account, AccountType, EntryType, JournalEntry, Transaction
from sqlalchemy.orm import Session
logger = logging.getLogger(__name__)
class IntercompanyManager:
"""
Manager for handling multi-entity transactions and intercompany eliminations.
"""
def __init__(self, db: Session):
self.db = db
def get_intercompany_transactions(self, workspace_id: str) -> List[Transaction]:
"""Fetch all transactions involving other workspaces"""
return self.db.query(Transaction).filter(
Transaction.workspace_id == workspace_id,
Transaction.is_intercompany == True
).all()
def find_unmatched_intercompany(self, workspace_id: str) -> List[Dict[str, Any]]:
"""
Identify intercompany transactions that don't have a matching
entry in the counterparty workspace.
"""
txs = self.get_intercompany_transactions(workspace_id)
unmatched = []
for tx in txs:
if not tx.counterparty_workspace_id:
continue
# Look for a transaction in the counterparty workspace with same external_id or matching amount
# This is a simplified check
matching = self.db.query(Transaction).filter(
Transaction.workspace_id == tx.counterparty_workspace_id,
Transaction.is_intercompany == True,
Transaction.counterparty_workspace_id == workspace_id
).first()
if not matching:
unmatched.append({
"transaction_id": tx.id,
"target_workspace": tx.counterparty_workspace_id,
"date": tx.transaction_date,
"description": tx.description
})
return unmatched
def generate_elimination_report(self, workspace_id: str) -> Dict[str, Any]:
"""
Calculate total intercompany volume to be eliminated for consolidation.
"""
txs = self.get_intercompany_transactions(workspace_id)
total_volume = 0.0
by_counterparty = {}
for tx in txs:
# We determine volume by summing journal entry amounts (one side)
amount = sum(je.amount for je in tx.journal_entries if je.type == EntryType.DEBIT)
total_volume += amount
cp = tx.counterparty_workspace_id or "Unknown"
by_counterparty[cp] = by_counterparty.get(cp, 0.0) + amount
return {
"total_elimination_volume": total_volume,
"breakdown_by_counterparty": by_counterparty,
"transaction_count": len(txs)
}
|