"""Detector: QuickBooks overdue invoices, bucketed by age (30/60/90 days)."""
from __future__ import annotations
from datetime import timedelta
from typing import List
from .base import BaseDetector, LeakCandidate
from ._common import connector_available, days_since, utcnow
BUCKETS = [
# (bucket_id, label, min_days, max_days (exclusive), severity)
("30-60d", "30-60 days", 30, 60, "medium"),
("60-90d", "60-90 days", 60, 90, "high"),
("90d+", "90+ days", 90, None, "critical"),
]
class QbOverdueDetector(BaseDetector):
id = "qb_overdue_30d"
name = "QuickBooks Overdue Invoices"
connector = "quickbooks"
enabled_by_default = True
description = (
"Flags unpaid QuickBooks invoices more than 30 days past due, "
"grouped into 30-60, 60-90, and 90+ day buckets."
)
default_params = {"min_days_overdue": 30}
def check(self, company_id: str) -> List[LeakCandidate]:
if not connector_available(company_id, self.connector):
return []
from app.models import QuickbooksInvoice
cutoff = utcnow() - timedelta(days=30)
invoices = (
QuickbooksInvoice.query.filter(
QuickbooksInvoice.company_id == company_id,
QuickbooksInvoice.status.notin_(["Paid", "Void"]),
QuickbooksInvoice.due_date.isnot(None),
QuickbooksInvoice.due_date < cutoff,
).all()
)
# Group into age buckets
grouped = {b[0]: [] for b in BUCKETS}
for inv in invoices:
age = days_since(inv.due_date)
for bucket_id, _label, lo, hi, _sev in BUCKETS:
if age >= lo and (hi is None or age < hi):
grouped[bucket_id].append(inv)
break
candidates: List[LeakCandidate] = []
for bucket_id, label, _lo, _hi, severity in BUCKETS:
bucket_invoices = grouped[bucket_id]
if not bucket_invoices:
continue
total = round(sum(inv.total_amount or 0.0 for inv in bucket_invoices), 2)
count = len(bucket_invoices)
candidates.append(
LeakCandidate(
detector_id=self.id,
source=f"QuickBooks: Overdue Invoices ({label})",
description=(
f"{count} invoice{'s' if count != 1 else ''} totaling "
f"${total:,.2f} are {label} past due"
),
estimated_loss=total,
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:{bucket_id}",
"invoice_ids": [inv.id for inv in bucket_invoices],
"bucket": bucket_id,
},
source_connector="quickbooks",
rule_params={"min_days_overdue": 30, "bucket": bucket_id},
)
)
return candidates