clientflow_backend/app/invoice_evidence.py

"""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