Spaces:
Sleeping
Sleeping
| from datetime import datetime | |
| import logging | |
| from typing import Any, Dict, List, Optional | |
| from accounting.models import Account, JournalEntry, Transaction | |
| from sqlalchemy.orm import Session | |
| from core.cross_system_reasoning import get_reasoning_engine | |
| from integrations.asana_service import AsanaService | |
| from integrations.slack_service_unified import SlackUnifiedService | |
| logger = logging.getLogger(__name__) | |
| class FinancialWorkflowService: | |
| """ | |
| Automates cross-system workflows triggered by financial events. | |
| Bridges Finance (Zoho/Xero/QBO) with Operations (Asana/Slack/HubSpot). | |
| """ | |
| def __init__(self, db: Session): | |
| self.db = db | |
| self.reasoning = get_reasoning_engine() | |
| self.asana = AsanaService() | |
| self.slack = SlackUnifiedService() | |
| async def handle_transaction_event(self, transaction_id: str): | |
| """ | |
| Triggered when a new transaction is ingested or its status changes. | |
| """ | |
| tx = self.db.query(Transaction).filter(Transaction.id == transaction_id).first() | |
| if not tx: | |
| return | |
| # 1. Check for Task Completion | |
| # If the transaction metadata links to a task (e.g., from Knowledge Graph extraction) | |
| task_id = tx.metadata_json.get("task_id") | |
| if task_id: | |
| logger.info(f"Financial Event: Transaction {tx.id} matches Task {task_id}") | |
| # Workflow: Mark task as completed if it was a payment for a service | |
| await self._handle_payment_task_completion(tx, task_id) | |
| # 2. Check for Budget Alerts | |
| alerts = await self.reasoning.check_financial_integrity(self.db, tx.workspace_id) | |
| for alert in alerts: | |
| if alert["type"] == "FINANCIAL_BUDGET_OVERRUN": | |
| # Workflow: Notify Slack about budget overrun | |
| # Note: In real scenarios, use slack.post_message with a valid token | |
| logger.info(f"Workflow Triggered: Slack alert for budget overrun in {tx.workspace_id}") | |
| async def _handle_payment_task_completion(self, tx: Transaction, task_id: str): | |
| """ | |
| Handle task completion when a payment is received. | |
| This method checks if the transaction represents an Accounts Receivable (AR) payment | |
| and marks the associated task as completed. | |
| Args: | |
| tx: The transaction that triggered this workflow | |
| task_id: The ID of the linked task | |
| """ | |
| try: | |
| # Check transaction type from metadata | |
| tx_type = tx.metadata_json.get("transaction_type", "").lower() | |
| is_payment_received = ( | |
| tx_type == "ar_payment" or | |
| tx_type == "payment_received" or | |
| tx.metadata_json.get("is_ar_payment", False) or | |
| (tx.amount > 0 and tx.description and any( | |
| keyword in tx.description.lower() for keyword in | |
| ["payment received", "invoice payment", "customer payment"] | |
| )) | |
| ) | |
| if not is_payment_received: | |
| logger.info(f"Transaction {tx.id} is not an AR payment, skipping task completion") | |
| return | |
| logger.info(f"Processing AR payment {tx.id} for task {task_id} completion") | |
| # Get task details from Asana | |
| try: | |
| task_result = await self.asana.get_task(task_id) | |
| if not task_result or task_result.get("completed"): | |
| logger.info(f"Task {task_id} already completed or not found") | |
| return | |
| except Exception as e: | |
| logger.warning(f"Could not fetch task {task_id} from Asana: {e}") | |
| # Continue anyway - try to mark as completed | |
| # Mark task as completed in Asana | |
| completion_result = await self.asana.complete_task( | |
| task_id=task_id, | |
| completed_at=datetime.now().isoformat() | |
| ) | |
| if completion_result.get("success"): | |
| logger.info(f"Successfully marked task {task_id} as completed due to payment {tx.id}") | |
| # Update transaction metadata with completion audit trail | |
| if not tx.metadata_json: | |
| tx.metadata_json = {} | |
| tx.metadata_json.update({ | |
| "task_completion": { | |
| "task_id": task_id, | |
| "completed_at": datetime.now().isoformat(), | |
| "completed_by": "workflow_automation", | |
| "trigger_transaction_id": tx.id, | |
| "completion_reason": "payment_received" | |
| } | |
| }) | |
| self.db.commit() | |
| # Optionally notify in Slack | |
| workspace_id = tx.workspace_id | |
| message = ( | |
| f"✅ Task {task_id} automatically marked as completed\n" | |
| f"Payment: {tx.amount} ({tx.description or 'No description'})\n" | |
| f"Transaction ID: {tx.id}" | |
| ) | |
| logger.info(f"Workflow completion: {message}") | |
| else: | |
| logger.warning(f"Failed to mark task {task_id} as completed: {completion_result}") | |
| except Exception as e: | |
| logger.error(f"Error handling payment task completion for transaction {tx.id}, task {task_id}: {e}") | |
| # Don't raise - we don't want to fail the transaction processing | |
| # due to workflow automation issues | |
| async def automate_invoice_to_task(self, workspace_id: str, invoice_data: Dict[str, Any]): | |
| """ | |
| Example Workflow: When an invoice is created in Zoho, create a reminder task in Asana. | |
| """ | |
| invoice_no = invoice_data.get("invoice_number") | |
| amount = invoice_data.get("total") | |
| # Create task in Asana | |
| result = await self.asana.create_task( | |
| workspace_id=workspace_id, | |
| name=f"Follow up on Invoice {invoice_no}", | |
| notes=f"Payment of ${amount} expected. Linked to Zoho Books invoice." | |
| ) | |
| return result | |