File size: 3,515 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 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | from datetime import datetime, timedelta
import logging
from typing import Any, Dict, List
from accounting.models import Entity, Invoice, InvoiceStatus
from sqlalchemy.orm import Session
from core.websockets import manager
logger = logging.getLogger(__name__)
class CollectionAgent:
"""
Automated agent for monitoring Accounts Receivable and sending follow-ups.
"""
def __init__(self, db: Session):
self.db = db
async def check_overdue_invoices(self, workspace_id: str) -> List[Dict[str, Any]]:
"""
Identify invoices that are past their due date and trigger follow-ups.
"""
now = datetime.utcnow()
overdue_invoices = self.db.query(Invoice).filter(
Invoice.workspace_id == workspace_id,
Invoice.status == InvoiceStatus.OPEN,
Invoice.due_date < now
).all()
reminders_sent = []
for invoice in overdue_invoices:
# 1. Update status to OVERDUE
invoice.status = InvoiceStatus.OVERDUE
# 2. Generate Reminder
reminder = self._generate_reminder_message(invoice)
# 3. "Send" Reminder (Mock: log and broadcast to UI)
logger.info(f"Sending reminder for Invoice {invoice.invoice_number} to {invoice.customer.name}")
# Internal notification for the user
await manager.broadcast(f"workspace:{workspace_id}", {
"type": "accounting.reminder_sent",
"data": {
"invoice_id": invoice.id,
"customer": invoice.customer.name,
"amount": invoice.amount,
"reminder": reminder
}
})
reminders_sent.append({
"invoice_id": invoice.id,
"customer": invoice.customer.name,
"amount": invoice.amount
})
self.db.commit()
return reminders_sent
def _generate_reminder_message(self, invoice: Invoice) -> str:
"""AI-assisted (template for now) reminder generation"""
days_overdue = (datetime.utcnow() - invoice.due_date).days
return (
f"Hello {invoice.customer.name}, this is a reminder that Invoice {invoice.invoice_number} "
f"for ${invoice.amount:,.2f} is now {days_overdue} days overdue. "
"Please process the payment at your earliest convenience."
)
def generate_aging_report(self, workspace_id: str) -> Dict[str, Any]:
"""Generate a summary of AR aging"""
invoices = self.db.query(Invoice).filter(
Invoice.workspace_id == workspace_id,
Invoice.status.in_([InvoiceStatus.OPEN, InvoiceStatus.OVERDUE])
).all()
now = datetime.utcnow()
report = {
"current": 0.0, # 0-30 days
"overdue_30": 0.0, # 31-60 days
"overdue_60": 0.0, # 61-90 days
"overdue_90": 0.0, # 90+ days
"total_ar": 0.0
}
for inv in invoices:
days = (now - inv.due_date).days
report["total_ar"] += inv.amount
if days <= 0:
report["current"] += inv.amount
elif days <= 30:
report["overdue_30"] += inv.amount
elif days <= 60:
report["overdue_60"] += inv.amount
else:
report["overdue_90"] += inv.amount
return report
|