Import ClientFlow production v4928.1.5.132.4

app/invoice_evidence.py

@@ -0,0 +1,170 @@
+"""Shared invoice-sent evidence rules.
+
+Keep the central next-action engine, the workflow guard and audits aligned when
+legacy SEND_INVOICE tasks were created from a quotation and only later enriched
+with the actual invoice in ``metadata.invoice_delivery_context``.
+"""
+from __future__ import annotations
+
+import json
+from typing import Any, Iterable
+
+
+_COMPLETED_STATUSES = {
+    "done",
+    "completed",
+    "complete",
+    "closed",
+    "resolved",
+    "concluida",
+    "concluido",
+    "concluída",
+    "concluído",
+}
+
+
+def _dict(value: Any) -> dict[str, Any]:
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, str) and value:
+        try:
+            parsed = json.loads(value)
+            return parsed if isinstance(parsed, dict) else {}
+        except Exception:
+            return {}
+    return {}
+
+
+def _norm(value: Any) -> str:
+    return str(value or "").strip()
+
+
+def _upper(value: Any) -> str:
+    return _norm(value).upper()
+
+
+def task_is_completed(task: dict[str, Any]) -> bool:
+    return _norm(task.get("status")).lower() in _COMPLETED_STATUSES or bool(
+        task.get("completed_at")
+    )
+
+
+def task_looks_like_invoice_sent(task: dict[str, Any]) -> bool:
+    code = _upper(task.get("action_code"))
+    if code in {"SEND_INVOICE", "SEND_FISCAL_INVOICE", "INVOICE_SENT"} or code.startswith("SEND_INVOICE"):
+        return True
+
+    searchable = " ".join(
+        _norm(task.get(key))
+        for key in ("action", "note", "description", "subject")
+    ).upper()
+    compact = searchable.translate(
+        str.maketrans(
+            {
+                "Á": "A", "À": "A", "Ã": "A", "Â": "A",
+                "É": "E", "Ê": "E", "Í": "I",
+                "Ó": "O", "Õ": "O", "Ô": "O",
+                "Ú": "U", "Ç": "C",
+            }
+        )
+    )
+    return bool(
+        "SEND INVOICE" in compact
+        or "INVOICE SENT" in compact
+        or "ENVIAR FATURA" in compact
+        or "FATURA ENVIADA" in compact
+        or ("FATURA" in compact and ("ENVIAD" in compact or "ANEX" in compact or "PDF" in compact))
+    )
+
+
+def _context_references(context: dict[str, Any]) -> tuple[set[str], set[str]]:
+    ids = {
+        _norm(context.get(key))
+        for key in ("document_id", "invoice_id", "external_id")
+        if _norm(context.get(key))
+    }
+    numbers = {
+        _upper(context.get(key))
+        for key in ("document_number", "invoice_number", "number")
+        if _upper(context.get(key))
+    }
+    return ids, numbers
+
+
+def task_invoice_references(task: dict[str, Any]) -> tuple[set[str], set[str]]:
+    """Return authoritative invoice identifiers declared by a task.
+
+    ``invoice_delivery_context`` is populated after the real invoice exists and
+    therefore supersedes the original next-action context, which may still name
+    the quotation used to create the task.  We intentionally do not merge both
+    contexts because a stale quotation identifier must not contradict the later
+    invoice-specific evidence.
+    """
+    metadata = _dict(task.get("metadata"))
+    delivery_context = _dict(metadata.get("invoice_delivery_context"))
+    delivery_ids, delivery_numbers = _context_references(delivery_context)
+    if delivery_ids or delivery_numbers:
+        return delivery_ids, delivery_numbers
+
+    direct_ids, direct_numbers = _context_references(metadata)
+    next_action = _dict(metadata.get("next_action"))
+    next_ids, next_numbers = _context_references(next_action)
+    return direct_ids | next_ids, direct_numbers | next_numbers
+
+
+def invoice_references(invoices: Iterable[dict[str, Any]]) -> tuple[set[str], set[str]]:
+    ids: set[str] = set()
+    numbers: set[str] = set()
+    for invoice in invoices or []:
+        ids.update(
+            _norm(invoice.get(key))
+            for key in ("id", "external_id")
+            if _norm(invoice.get(key))
+        )
+        numbers.update(
+            _upper(invoice.get(key))
+            for key in ("document_number", "number")
+            if _upper(invoice.get(key))
+        )
+    return ids, numbers
+
+
+def completed_send_invoice_task_evidence(
+    tasks: Iterable[dict[str, Any]],
+    invoices: Iterable[dict[str, Any]],
+) -> bool:
+    """True when a completed invoice-send task matches a current invoice.
+
+    Declared invoice identifiers are conservative: when present they must match
+    the current invoice set.  Legacy tasks with no identifiers can still count
+    when their text clearly represents sending an invoice for the same
+    opportunity.
+    """
+    invoice_rows = list(invoices or [])
+    invoice_ids, invoice_numbers = invoice_references(invoice_rows)
+
+    for raw_task in tasks or []:
+        task = dict(raw_task)
+        if not task_looks_like_invoice_sent(task) or not task_is_completed(task):
+            continue
+
+        task_ids, task_numbers = task_invoice_references(task)
+        if task_ids and invoice_ids and task_ids.isdisjoint(invoice_ids):
+            continue
+        if task_numbers and invoice_numbers and task_numbers.isdisjoint(invoice_numbers):
+            continue
+
+        if not task_ids and not task_numbers and invoice_numbers:
+            searchable = " ".join(
+                _norm(task.get(key))
+                for key in ("action", "note", "description", "subject")
+            ).upper()
+            # Exact invoice mention is strongest.  Preserve compatibility for
+            # old same-opportunity tasks that only say "Enviar fatura".
+            if searchable and not any(number in searchable for number in invoice_numbers):
+                if not task_looks_like_invoice_sent(task):
+                    continue
+
+        return True
+
+    return False