"""Detector: QuickBooks draft invoices older than 14 days (unbilled work)."""
from __future__ import annotations
from datetime import timedelta
from typing import List
from .base import BaseDetector, LeakCandidate
from ._common import connector_available, utcnow
class QbDraftDetector(BaseDetector):
id = "qb_draft_invoices"
name = "QuickBooks Draft Invoices (Unbilled Work)"
connector = "quickbooks"
enabled_by_default = True
description = (
"Flags QuickBooks invoices sitting in Draft status for more than "
"14 days — likely completed work that has never been billed."
)
default_params = {"min_days_draft": 14}
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=14)
drafts = (
QuickbooksInvoice.query.filter(
QuickbooksInvoice.company_id == company_id,
QuickbooksInvoice.status == "Draft",
QuickbooksInvoice.tx_date.isnot(None),
QuickbooksInvoice.tx_date < cutoff,
).all()
)
if not drafts:
return []
total = round(sum(inv.total_amount or 0.0 for inv in drafts), 2)
count = len(drafts)
return [
LeakCandidate(
detector_id=self.id,
source="QuickBooks: Draft Invoices (>14 days)",
description=(
f"{count} draft invoice{'s' if count != 1 else ''} totaling "
f"${total:,.2f} have been unbilled for 14+ days"
),
estimated_loss=total,
severity="medium",
metadata_json={
"dedupe_key": f"{self.id}:aggregate",
"invoice_ids": [inv.id for inv in drafts],
},
source_connector="quickbooks",
rule_params={"min_days_draft": 14},
)
]