feat: close final incomplete product analysis

2026-09-22 14:34:15 -03:00
parent e3ac68c70e
commit dca4847c21
5 changed files with 5799 additions and 0 deletions
@@ -0,0 +1,116 @@
+"""Inventory the exact final incomplete products using read-only Firebird queries."""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from collections import Counter
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+from central.app.fiscal_completeness import evaluate_product, load_requiredness_matrix  # noqa: E402
+from central.app.ai.fingerprint import product_fingerprint, validate_gtin  # noqa: E402
+
+
+def load_env():
+    path = ROOT / ".env"
+    if path.exists():
+        for line in path.read_text(encoding="utf-8").splitlines():
+            if "=" in line and not line.lstrip().startswith("#"):
+                key, _, value = line.partition("=")
+                os.environ.setdefault(key.strip(), value.strip())
+
+
+def _context(cur):
+    cur.execute("SELECT UF, CRT FROM C000004")
+    rows = {(str(uf or '').strip().upper(), str(crt or '').strip()) for uf, crt in cur.fetchall()}
+    return next(iter(rows)) if len(rows) == 1 else (None, None)
+
+
+def _blocker(field: str, company_context: bool) -> str:
+    if not company_context:
+        return "MISSING_COMPANY_CONTEXT"
+    if field == "SITUACAO_TRIBUTARIA":
+        return "MISSING_ST_FACT"
+    return "OTHER_PROVEN_BLOCKER"
+
+
+def inventory_database(path: Path, matrix: dict) -> list[dict]:
+    import fdb
+
+    conn = fdb.connect(database=str(path), user=os.environ["FB_USER"], password=os.environ["FB_PASSWORD"], fb_library_name=os.environ.get("FIREBIRD_CLIENT_LIBRARY"), isolation_level=fdb.ISOLATION_LEVEL_READ_COMMITED_RO)
+    try:
+        cur = conn.cursor()
+        cur.execute("SELECT TRIM(RDB$FIELD_NAME) FROM RDB$RELATION_FIELDS WHERE RDB$RELATION_NAME = 'C000025'")
+        available = {row[0] for row in cur.fetchall()}
+        selected = [field for field in matrix["fields"] if field in available]
+        identity_fields = [field for field in ("CODIGO", "PRODUTO", "GTIN", "CODBARRA", "UNIDADE", "CLASSIFICACAO_FISCAL", "NCM", "CODGRUPO", "CODSUBGRUPO") if field in available]
+        cur.execute(f"SELECT {','.join(identity_fields + selected)} FROM C000025")
+        names = identity_fields + selected
+        product_rows = cur.fetchall()
+        uf, crt = _context(cur)
+        company_context = uf is not None and crt in {"1", "2", "3"}
+        result = []
+        for row in product_rows:
+            product = dict(zip(names, row))
+            context = {
+                "uf": uf, "crt": crt, "regime": "SIMPLES" if crt == "1" else "NORMAL" if crt in {"2", "3"} else None,
+                "operation": "SALE", "contribuinte_ipi": "N", "has_numeric_tipi": False, "has_supported_ex": False,
+                "product_matches_ex": False, "has_official_cest": bool(str(product.get("CEST") or "").strip()),
+                "st_subject": str(product.get("SITUACAO_TRIBUTARIA") or "").strip() == "S", "erp_consumes_entry_fields": False,
+                "icms_taxed": str(product.get("SITUACAO_TRIBUTARIA") or "").strip() == "T", "origin_proven": False,
+                "erp_consumes_ipi_cst": False, "piscofins_profile_proven": False, "piscofins_rate_applies": False,
+            }
+            evaluation = evaluate_product(product, context, matrix)
+            missing = []
+            for field, field_result in evaluation["fields"].items():
+                if field_result["status"] != "REQUIRED_MISSING":
+                    continue
+                missing.append({
+                    "FIELD": field,
+                    "WHY_REQUIRED": matrix["fields"][field]["when"],
+                    "CURRENT_VALUE": product.get(field),
+                    "WHY_UNRESOLVED": "required value is absent or invalid and no deterministic source proved a replacement",
+                    "EVIDENCE_AVAILABLE": ["C000004 company context" if company_context else "company context unavailable", "C000025 product master", "read-only inbound evidence map"],
+                    "BLOCKER_CLASS": _blocker(field, company_context),
+                })
+            if missing:
+                result.append({
+                    "DATABASE": path.stem,
+                    "PRODUCT_CODE": str(product.get("CODIGO") or "").strip(),
+                    "DESCRIPTION": str(product.get("PRODUTO") or "").strip(),
+                    "GTIN": validate_gtin(product.get("GTIN") or product.get("CODBARRA")),
+                    "GTIN_RAW": product.get("GTIN") or product.get("CODBARRA"),
+                    "FINGERPRINT": product_fingerprint(product.get("PRODUTO"), None, product.get("UNIDADE")),
+                    "NCM_CURRENT": product.get("CLASSIFICACAO_FISCAL") or product.get("NCM"),
+                    "GROUP": product.get("CODGRUPO"),
+                    "SUBGROUP": product.get("CODSUBGRUPO"),
+                    "REQUIRED_MISSING": missing,
+                    "PRIORITY_CLASS": "HUMAN_FACT_REQUIRED" if not company_context else "RESOLVABLE_NOW_DETERMINISTICALLY_CANDIDATE",
+                })
+        return result
+    finally:
+        conn.close()
+
+
+def main():
+    load_env()
+    matrix = load_requiredness_matrix(ROOT / "docs/FISCAL_REQUIREDNESS_MATRIX_V1.json")
+    paths = [ROOT / "TESTESOFICIAIS.FDB"] + sorted((ROOT / "Banco").glob("*.FDB"))
+    products = []
+    for path in paths:
+        products.extend(inventory_database(path, matrix))
+    blockers = Counter(item["BLOCKER_CLASS"] for product in products for item in product["REQUIRED_MISSING"])
+    artifact = {
+        "artifact": "FINAL_INCOMPLETE_PRODUCTS_V1", "mode": "READ_ONLY", "total_incomplete_products": len(products),
+        "unique_product_keys": len({(p["DATABASE"], p["PRODUCT_CODE"]) for p in products}), "blockers_by_type": dict(sorted(blockers.items())),
+        "products": products, "firebird_writes": 0, "external_api_calls": 0,
+    }
+    (ROOT / "docs/FINAL_INCOMPLETE_PRODUCTS_V1.json").write_text(json.dumps(artifact, ensure_ascii=False, indent=2, default=str) + "\n", encoding="utf-8")
+    print(json.dumps({"artifact": artifact["artifact"], "total_incomplete_products": len(products), "blockers_by_type": artifact["blockers_by_type"]}))
+
+
+if __name__ == "__main__":
+    main()