feat: add autonomous fiscal completion v1

central/app/fiscal_completeness.py

@@ -0,0 +1,83 @@
+"""Pure requiredness and fiscal completeness evaluation."""
+
+import json
+from pathlib import Path
+from typing import Any
+
+
+VALID_CLASSIFICATIONS = {
+    "REQUIRED", "CONDITIONAL", "OPTIONAL_REFERENCE", "NOT_APPLICABLE", "ERP_NOT_USED"
+}
+
+
+def load_requiredness_matrix(path: str | Path) -> dict[str, Any]:
+    matrix = json.loads(Path(path).read_text(encoding="utf-8"))
+    fields = matrix.get("fields", {})
+    if set(matrix.get("classifications", [])) != VALID_CLASSIFICATIONS:
+        raise ValueError("requiredness matrix classifications are incomplete")
+    if len(fields) != 23 or any(rule.get("classification") not in VALID_CLASSIFICATIONS for rule in fields.values()):
+        raise ValueError("requiredness matrix must define 23 valid field rules")
+    return matrix
+
+
+def _has_value(value: Any) -> bool:
+    return value is not None and (not isinstance(value, str) or bool(value.strip()))
+
+
+def _conditional_state(field: str, context: dict[str, Any]) -> str:
+    if field == "IPI":
+        return "REQUIRED" if context.get("contribuinte_ipi") == "S" and context.get("has_numeric_tipi") else "NOT_APPLICABLE"
+    if field == "EX_TIPI":
+        return "REQUIRED" if context.get("has_supported_ex") and context.get("product_matches_ex") else "NOT_APPLICABLE"
+    if field == "CEST":
+        return "REQUIRED" if context.get("has_official_cest") or context.get("st_subject") else "NOT_APPLICABLE"
+    if field in {"CFOP", "SITUACAO_TRIBUTARIA"}:
+        return "REQUIRED" if context.get("operation", "SALE") == "SALE" else "NOT_APPLICABLE"
+    if field in {"CFOP_ENTRADA", "CST_ENTRADA", "COD_ENTRADA_PIS", "COD_ENTRADA_COFINS"}:
+        return "REQUIRED" if context.get("operation") == "PURCHASE" and context.get("erp_consumes_entry_fields") else "NOT_APPLICABLE"
+    if field == "CST":
+        return "REQUIRED" if context.get("regime") == "NORMAL" and context.get("operation", "SALE") == "SALE" else "NOT_APPLICABLE"
+    if field == "CSOSN":
+        return "REQUIRED" if context.get("regime") == "SIMPLES" and context.get("operation", "SALE") == "SALE" else "NOT_APPLICABLE"
+    if field == "ALIQUOTA":
+        return "REQUIRED" if context.get("icms_taxed") else "NOT_APPLICABLE"
+    if field == "COD_ORIGEM_MERCADORIA":
+        return "REQUIRED" if context.get("origin_proven") else "NOT_APPLICABLE"
+    if field == "CST_IPI":
+        return "REQUIRED" if context.get("contribuinte_ipi") == "S" and context.get("erp_consumes_ipi_cst") else "NOT_APPLICABLE"
+    if field in {"COD_SAIDA_PIS", "COD_SAIDA_COFINS"}:
+        return "REQUIRED" if context.get("operation", "SALE") == "SALE" and context.get("piscofins_profile_proven") else "NOT_APPLICABLE"
+    if field in {"ALIQ_PIS", "ALIQ_COFINS"}:
+        return "REQUIRED" if context.get("piscofins_rate_applies") else "NOT_APPLICABLE"
+    if field in {"INCIDENCIA_PISCOFINS", "PISCOFINS"}:
+        return "REQUIRED" if context.get("piscofins_profile_proven") else "NOT_APPLICABLE"
+    return "NOT_APPLICABLE"
+
+
+def evaluate_field(field: str, value: Any, context: dict[str, Any], matrix: dict[str, Any]) -> dict[str, Any]:
+    rule = matrix["fields"][field]
+    classification = rule["classification"]
+    if classification == "ERP_NOT_USED":
+        status = "ERP_NOT_SUPPORTED"
+        valid = True
+    elif classification == "NOT_APPLICABLE":
+        status = "NOT_APPLICABLE"
+        valid = True
+    elif classification == "OPTIONAL_REFERENCE":
+        status = "OPTIONAL_VALID" if _has_value(value) else "OPTIONAL_EMPTY"
+        valid = True
+    elif classification == "REQUIRED":
+        status = "REQUIRED_AND_VALID" if _has_value(value) else "REQUIRED_MISSING"
+        valid = _has_value(value)
+    else:
+        conditional = _conditional_state(field, context)
+        if conditional == "NOT_APPLICABLE":
+            status, valid = "CONDITIONAL_NOT_TRIGGERED", True
+        else:
+            status, valid = ("CONDITIONAL_AND_VALID", True) if _has_value(value) else ("REQUIRED_MISSING", False)
+    return {"field": field, "classification": classification, "status": status, "valid": valid, "value_present": _has_value(value), "source": rule["source"]}
+
+
+def evaluate_product(product: dict[str, Any], context: dict[str, Any], matrix: dict[str, Any]) -> dict[str, Any]:
+    fields = {field: evaluate_field(field, product.get(field), context, matrix) for field in matrix["fields"]}
+    return {"FISCALLY_COMPLETE": all(result["valid"] for result in fields.values()), "fields": fields}