from datetime import datetime import logging from typing import Any, Dict, List, Optional from accounting.categorizer import AICategorizer from accounting.models import Account, EntryType, JournalEntry, Transaction from sqlalchemy.orm import Session from integrations.atom_communication_ingestion_pipeline import ( CommunicationAppType, ingestion_pipeline, ) from integrations.quickbooks_service import QuickBooksService from integrations.xero_service import XeroService from integrations.zoho_books_service import ZohoBooksService logger = logging.getLogger(__name__) class AccountingSyncManager: """ Unified manager for synchronizing data across multiple accounting ledgers (Zoho, Xero, QuickBooks, Stripe/Plaid). """ def __init__(self, db: Session): self.db = db self.zoho = ZohoBooksService() self.xero = XeroService() self.qbo = QuickBooksService() self.categorizer = AICategorizer(db) async def sync_external_transactions( self, workspace_id: str, platform: str, credentials: Dict[str, Any] ) -> Dict[str, Any]: """ Pull transactions from an external platform and ingest into ATOM's ledger. """ raw_transactions = [] if platform == "zoho": raw_transactions = await self.zoho.get_bank_transactions( credentials["access_token"], credentials["organization_id"], credentials.get("account_id") ) mapped_txs = self._map_zoho_transactions(raw_transactions, workspace_id) elif platform == "xero": raw_transactions = await self.xero.get_invoices( credentials["access_token"], credentials["tenant_id"] ) mapped_txs = self._map_xero_transactions(raw_transactions, workspace_id) elif platform == "quickbooks": raw_transactions = await self.qbo.get_expenses( credentials["realm_id"], credentials["access_token"] ) mapped_txs = self._map_qbo_transactions(raw_transactions, workspace_id) else: raise ValueError(f"Unsupported platform: {platform}") ingested_count = 0 for tx_data in mapped_txs: # Check for existing exists = self.db.query(Transaction).filter( Transaction.workspace_id == workspace_id, Transaction.metadata_json.contains(tx_data["external_id"]) ).first() if not exists: tx = Transaction( workspace_id=workspace_id, description=tx_data["description"], amount=tx_data["amount"], transaction_date=tx_data["date"], metadata_json={"external_id": tx_data["external_id"], "platform": platform} ) self.db.add(tx) self.db.flush() # Auto-categorize self.categorizer.categorize_transaction(tx.id) ingested_count += 1 # Ingest into semantic memory (LanceDB + Knowledge Graph) try: ingestion_pipeline.ingest_message( app_type=platform if platform != "quickbooks" else "quickbooks", message_data={ "id": f"tx_{tx.id}", "timestamp": tx.transaction_date.isoformat(), "sender": platform, "content": f"Financial Transaction: {tx.description}. Amount: {tx.amount}. Merchant: {tx.metadata_json.get('merchant', 'Unknown')}", "metadata": { "transaction_id": tx.id, "workspace_id": workspace_id, "amount": tx.amount, "external_id": tx_data["external_id"] } } ) except Exception as ex: logger.error(f"Failed to ingest transaction {tx.id} into semantic memory: {ex}") self.db.commit() return {"status": "success", "ingested": ingested_count, "platform": platform} def _map_zoho_transactions(self, raw: List[Dict], ws_id: str) -> List[Dict]: return [{ "description": t.get("description", "Zoho Transaction"), "amount": float(t.get("amount", 0)), "date": datetime.strptime(t["date"], "%Y-%m-%d") if "date" in t else datetime.now(), "external_id": str(t.get("transaction_id", "")) } for t in raw] def _map_xero_transactions(self, raw: List[Dict], ws_id: str) -> List[Dict]: return [{ "description": f"Xero Invoice: {t.get('InvoiceNumber','')}", "amount": float(t.get("Total", 0)), "date": datetime.strptime(t["DateString"], "%Y-%m-%dT%H:%M:%S") if "DateString" in t else datetime.now(), "external_id": str(t.get("InvoiceID", "")) } for t in raw] def _map_qbo_transactions(self, raw: List[Dict], ws_id: str) -> List[Dict]: return [{ "description": t.get("PrivateNote", "QBO Expense"), "amount": float(t.get("TotalAmt", 0)), "date": datetime.strptime(t["TxnDate"], "%Y-%m-%d") if "TxnDate" in t else datetime.now(), "external_id": str(t.get("Id", "")) } for t in raw]