"""Read-only fiscal coverage matrix (V3) over all C000025 products in TESTESOFICIAIS.FDB, after importing CFOP/FCP/ANP/NCM_UTRIB/RTC_BIOFUEL_INDEX/ RTC_CCREDPRES from sources/raw/bootstrap_v1/. No Firebird writes. Separates DATA_AVAILABLE (a dataset exists) from MATCH_FOUND (this product's own field — NCM or CFOP — actually matches an official row) from RULE_RESOLVED (a fiscal determination was made) — the mission's own required distinction. Never invents a rule to inflate RULE_RESOLVED. """ import json import os import sys import urllib.request from collections import Counter from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "agent" / "src")) from gti_fiscal_agent.firebird_reader import read_products # noqa: E402 def load_dotenv(path: Path) -> None: for line in path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") os.environ.setdefault(key.strip(), value.strip()) def http_get(url, token): request = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"}) with urllib.request.urlopen(request, timeout=15) as response: return json.load(response) def classify_batch(hub_url, token, products, chunk_size=100): results = [] for start in range(0, len(products), chunk_size): chunk = products[start:start + chunk_size] payload = {"products": [{"current_ncm": p["CLASSIFICACAO_FISCAL"], "description": p["PRODUTO"] or ""} for p in chunk]} request = urllib.request.Request( f"{hub_url}/v1/classify/batch", data=json.dumps(payload, ensure_ascii=True).encode(), headers={"Content-Type": "application/json", "Authorization": f"Bearer {token}"}, method="POST", ) with urllib.request.urlopen(request, timeout=60) as response: results.extend(json.load(response)["results"]) return results def main(): token = sys.argv[1] if len(sys.argv) > 1 else os.environ.get("AGENT_DEV_TOKEN", "") hub_url = os.environ.get("GTI_HUB_URL", "http://localhost:8000") load_dotenv(ROOT / ".env") products = read_products(os.environ) total = len(products) decisions = classify_batch(hub_url, token, products) datasets = http_get(f"{hub_url}/v1/datasets/status", token) active_kinds = {d["kind"] for d in datasets if d["status"] == "ACTIVE"} record_counts = {d["kind"]: d["record_count"] for d in datasets if d["status"] == "ACTIVE"} ncm_status = Counter(d.get("status") for d in decisions) ncm_missing = sum(1 for p in products if not (p["CLASSIFICACAO_FISCAL"] or "").strip()) ncm_valid = ncm_status.get("EXACT_NCM", 0) ncm_legacy = ncm_status.get("LEGACY_NCM", 0) ncm_invalid = sum( 1 for p, d in zip(products, decisions) if (p["CLASSIFICACAO_FISCAL"] or "").strip() and d.get("status") not in ("EXACT_NCM", "LEGACY_NCM") ) ipi_buckets = Counter() cest_match = cest_confirmed = cest_candidates = 0 mva_match = mva_resolved = mva_candidates = 0 ncm_utrib_match = 0 cfop_match = 0 cfop_seen = {} ncm_utrib_seen = {} for product, decision in zip(products, decisions): ipi_status = decision.get("ipi_status") if ipi_status == "LEGACY_NCM": pass elif ipi_status == "NO_TIPI_MAPPING_FOUND": ipi_buckets["NO_MATCH"] += 1 elif ipi_status == "IPI_CANDIDATES": ipi_buckets["AMBIGUOUS"] += 1 elif ipi_status == "IPI_RESOLVED": rate_kind = (decision.get("classification", {}).get("ipi") or {}).get("rate_kind") ipi_buckets[{"NUMERIC": "NUMERIC", "NON_NUMERIC": "NON_NUMERIC", "BLANK": "SOURCE_BLANK"}.get(rate_kind, "NO_MATCH")] += 1 else: ipi_buckets["NO_MATCH"] += 1 cest_status = decision.get("cest_status") if cest_status not in (None, "NO_CEST_MAPPING_FOUND", "LEGACY_NCM"): cest_match += 1 if cest_status == "CEST_CONFIRMED": cest_confirmed += 1 elif cest_status == "CEST_CANDIDATES": cest_candidates += 1 mva_status = decision.get("ba_st_mva_status") if mva_status not in (None, "NO_MVA_MAPPING_FOUND", "LEGACY_NCM"): mva_match += 1 if mva_status == "MVA_RESOLVED": mva_resolved += 1 elif mva_status == "MVA_CANDIDATES": mva_candidates += 1 matched_ncm = decision.get("classification", {}).get("ncm") if matched_ncm: if matched_ncm not in ncm_utrib_seen: ncm_utrib_seen[matched_ncm] = http_get(f"{hub_url}/v1/reference/ncm_utrib/{matched_ncm}", token)["found"] if ncm_utrib_seen[matched_ncm]: ncm_utrib_match += 1 cfop_value = (product.get("CFOP") or "").strip() if cfop_value: if cfop_value not in cfop_seen: cfop_seen[cfop_value] = http_get(f"{hub_url}/v1/reference/cfop/{cfop_value}", token)["found"] if cfop_seen[cfop_value]: cfop_match += 1 cbs_2026 = http_get(f"{hub_url}/v1/reference/rtc_cbs_rate/2026", token) coverage = { "artifact": "FISCAL_COVERAGE_FULL_V3", "firebird_writes": 0, "total_products": total, "ncm_valid": ncm_valid, "ncm_legacy": ncm_legacy, "ncm_invalid": ncm_invalid, "ncm_missing": ncm_missing, "ipi": { "numeric": ipi_buckets["NUMERIC"], "non_numeric": ipi_buckets["NON_NUMERIC"], "source_blank": ipi_buckets["SOURCE_BLANK"], "ambiguous": ipi_buckets["AMBIGUOUS"], "no_match": ipi_buckets["NO_MATCH"], }, "cest": {"match": cest_match, "confirmed": cest_confirmed, "candidates": cest_candidates}, "ba_st_mva": {"match": mva_match, "resolved": mva_resolved, "candidates": mva_candidates}, "pis_cofins_4_3_10_17": "DEFERRED_BY_BUSINESS_SCOPE", "pis_cofins": { "data_available": True, "match_found": 0, "rule_resolved": 0, "rule_unresolved": total, "note": "CST_PIS/CST_COFINS code domain is ACTIVE (66 codes) but no NCM-linked SPED Tabela 4.3.10-4.3.17 determination table was available to acquire (MISSING_BOOTSTRAP_FILE) — no per-product PIS/COFINS rule can be resolved.", }, "cfop": {"dataset_available": "CFOP" in active_kinds, "record_count": record_counts.get("CFOP", 0), "match_found": cfop_match, "note": "match = this product's own C000025.CFOP value exists in the official table"}, "cst_icms": {"dataset_available": "NFE_MOC_TABLES" in active_kinds, "record_count": 11}, "csosn": {"dataset_available": "NFE_MOC_TABLES" in active_kinds, "record_count": 10}, "origin": {"dataset_available": "NFE_MOC_TABLES" in active_kinds, "record_count": 9}, "fcp": {"dataset_available": "FCP" in active_kinds, "record_count": record_counts.get("FCP", 0), "note": "per-UF rate table; not linked to a product-level UF field in C000025, so match_found is not meaningful per-product"}, "icms_monophasic": {"match_found": 0, "dataset_available": False, "status": "OUT_OF_SCOPE_MVP"}, "anp": {"scope": "ACTIVE_REFERENCE_ONLY", "dataset_available": "ANP" in active_kinds, "record_count": record_counts.get("ANP", 0), "match_found": 0, "note": "C000025 has no ANP/fuel product code field to join on; dataset exists but is not applicable to this ERP's product catalog"}, "ncm_utrib": {"dataset_available": "NCM_UTRIB" in active_kinds, "record_count": record_counts.get("NCM_UTRIB", 0), "match_found": ncm_utrib_match}, "rtc": { "cclasstrib_available": "RTC_CCLASSTRIB" in active_kinds, "cclasstrib_record_count": record_counts.get("RTC_CCLASSTRIB", 0), "cclasstrib_match": 0, "cbs_rate_available": "RTC_CBS_RATES" in active_kinds, "cbs_rate_record_count": record_counts.get("RTC_CBS_RATES", 0), "cbs_rate_2026": cbs_2026, "ccredpres_available": "RTC_CCREDPRES" in active_kinds, "ccredpres_record_count": record_counts.get("RTC_CCREDPRES", 0), "biofuel_available": "RTC_BIOFUEL_INDEX" in active_kinds, "biofuel_scope": "ACTIVE_REFERENCE_ONLY", "note": "cClassTrib/cCredPres are DATA_AVAILABLE only: C000025 carries no cClassTrib/cCredPres field to join on, so match_found=0 and rule_resolved=0 are data-shape facts, not gaps. No rule was invented.", }, "core_domains_active": {d: (k in active_kinds) for d, k in {"NCM": "NCM", "TIPI_IPI": "TIPI_IPI", "CEST": "CEST", "BA_ST_MVA": "BA_ST_MVA", "CFOP": "CFOP", "ORIGIN": "NFE_MOC_TABLES", "CST_ICMS": "NFE_MOC_TABLES", "CSOSN": "NFE_MOC_TABLES", "CST_PIS": "NFE_MOC_TABLES", "CST_COFINS": "NFE_MOC_TABLES", "FCP": "FCP", "RTC_CCLASSTRIB": "RTC_CCLASSTRIB", "RTC_CCREDPRES": "RTC_CCREDPRES", "RTC_CBS_RATES": "RTC_CBS_RATES"}.items()}, "active_dataset_kinds": sorted(active_kinds), } output_path = ROOT / "docs" / "FISCAL_COVERAGE_FULL_V3.json" output_path.write_text(json.dumps(coverage, indent=2, ensure_ascii=False), encoding="utf-8") print(json.dumps({k: v for k, v in coverage.items() if k not in ("pis_cofins", "rtc")}, indent=2, ensure_ascii=False)) if __name__ == "__main__": main()