gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
168 lines
8.5 KiB
Python
168 lines
8.5 KiB
Python
"""Read-only fiscal coverage matrix (V2) 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
|
|
|
|
coverage = {
|
|
"artifact": "FISCAL_COVERAGE_FULL_V2",
|
|
"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": {
|
|
"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": "MISSING_BOOTSTRAP_FILE"},
|
|
"anp": {"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_match": 0, "cclasstrib_available": False, "cclasstrib_status": "MISSING_BOOTSTRAP_FILE",
|
|
"ccredpres_match": 0, "ccredpres_available": "RTC_CCREDPRES" in active_kinds, "ccredpres_record_count": record_counts.get("RTC_CCREDPRES", 0),
|
|
"cbs_rate_match": 0, "cbs_rate_available": False, "cbs_rate_status": "MISSING_BOOTSTRAP_FILE",
|
|
"biofuel_match": 0, "biofuel_available": "RTC_BIOFUEL_INDEX" in active_kinds, "biofuel_record_count": record_counts.get("RTC_BIOFUEL_INDEX", 0),
|
|
"note": "cCredPres and biofuel index datasets are ACTIVE but have no join key present in C000025 (no cProdANP/cCredPres field on this ERP's grocery/restaurant catalog), so match_found=0 is a data-shape fact, not a gap.",
|
|
},
|
|
"active_dataset_kinds": sorted(active_kinds),
|
|
}
|
|
|
|
output_path = ROOT / "docs" / "FISCAL_COVERAGE_FULL_V2.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()
|