62 lines
3.9 KiB
Python
62 lines
3.9 KiB
Python
"""Read-only extras: (1) history agreement restricted to identity-consistent joins (item NCM == product NCM),
|
|
(2) CEST historical support for the CEST-candidate products against the Hub's official candidates."""
|
|
import json, os, sys, 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 _connect
|
|
for l in (ROOT / ".env").read_text(encoding="utf-8").splitlines():
|
|
if "=" in l and not l.startswith("#"):
|
|
k, _, v = l.partition("="); os.environ.setdefault(k.strip(), v.strip())
|
|
s = lambda v: None if v is None else str(v).strip()
|
|
dg = lambda v: "".join(c for c in str(v or "") if c.isdigit())
|
|
c = _connect(os.environ).cursor()
|
|
c.execute("select CODIGO, PRODUTO, CLASSIFICACAO_FISCAL, CFOP, CST, CSOSN, CEST, COD_SAIDA_PIS from C000025")
|
|
P = {s(r[0]): dict(zip(("D", "NCM", "CFOP", "CST", "CSOSN", "CEST", "PIS"), map(s, r[1:]))) for r in c.fetchall()}
|
|
c.execute("select CODPRODUTO, NCM, CFOP, CST, CSOSN, CEST, CST_PIS_COFINS, GRUPO_ICMS from NFCE_ITENS where NCM is not null")
|
|
consistent = Counter(); inconsistent = 0; agree = {f: Counter() for f in ("CFOP", "CST", "CSOSN", "CEST", "PIS")}
|
|
hist_cest = {}
|
|
for cod, ncm, cfop, cst, csosn, cest, pis, gi in c.fetchall():
|
|
p = P.get(s(cod))
|
|
if not p: continue
|
|
if s(ncm) != p["NCM"]:
|
|
inconsistent += 1
|
|
else:
|
|
consistent["items"] += 1
|
|
for f, iv in (("CFOP", s(cfop)), ("CST", s(cst)), ("CSOSN", s(csosn)), ("CEST", s(cest)), ("PIS", s(pis))):
|
|
pv = p[f]
|
|
if iv is None: continue
|
|
eq = dg(iv) == dg(pv) if f == "CEST" else (iv.lstrip("0") == (pv or "").lstrip("0") if f == "PIS" else iv == pv)
|
|
agree[f]["MATCH" if eq else "MISMATCH"] += 1
|
|
if dg(cest):
|
|
hist_cest.setdefault(s(cod), Counter())[dg(cest)] += 1
|
|
out = {"ncm_identity_consistent_items": consistent["items"], "ncm_identity_inconsistent_items": inconsistent,
|
|
"agreement_consistent_joins_only": {f: dict(v) for f, v in agree.items()}}
|
|
|
|
# also all-time (any NCM) CEST history per product
|
|
c.execute("select CODPRODUTO, CEST from NFCE_ITENS where CEST is not null and trim(CEST) <> ''")
|
|
allc = {}
|
|
for cod, cest in c.fetchall():
|
|
allc.setdefault(s(cod), Counter())[dg(cest)] += 1
|
|
|
|
tok = [l.split("=", 1)[1].strip() for l in open(ROOT / ".env", encoding="utf-8") if l.startswith("AGENT_DEV_TOKEN=")][0]
|
|
matrix = json.loads((ROOT / "docs" / "FISCAL_DECISION_MATRIX_V1.json").read_text(encoding="utf-8"))
|
|
cands = [m for m in matrix["products"] if m["fields"]["CEST"]["status"] == "RULE_CANDIDATE"]
|
|
req = urllib.request.Request("http://localhost:8000/v1/classify/batch", data=json.dumps({"products": [{"current_ncm": m["NCM"], "description": m["DESCRIPTION"] or ""} for m in cands]}).encode(),
|
|
headers={"Content-Type": "application/json", "Authorization": "Bearer " + tok})
|
|
decs = json.load(urllib.request.urlopen(req, timeout=600))["results"]
|
|
rows = []
|
|
for m, d in zip(cands, decs):
|
|
code = s(m["CODIGO"]); off = {dg(x["code"]): x for x in d["classification"].get("cest_candidates", [])}
|
|
h = dict(allc.get(code, {}))
|
|
supported = [k for k in h if k in off]
|
|
mva_ceps = {dg(x.get("cest")) for x in d["classification"].get("ba_st_mva_candidates", [])}
|
|
rows.append({"CODIGO": code, "DESCRIPTION": m["DESCRIPTION"], "NCM": m["NCM"], "official_candidates": len(off), "historical_cest": h,
|
|
"historical_in_official_candidates": supported, "historical_in_ba_st_mva": [k for k in h if k in mva_ceps],
|
|
"product_ncm_matches_item_ncm": None,
|
|
"verdict": "CEST_HISTORICALLY_SUPPORTED" if len(h) == 1 and len(supported) == 1 else ("HISTORY_MULTIPLE_OR_UNSUPPORTED" if h else "NO_HISTORY")})
|
|
out["cest_candidates"] = rows
|
|
out["cest_historically_supported"] = sum(1 for r in rows if r["verdict"] == "CEST_HISTORICALLY_SUPPORTED")
|
|
print(json.dumps(out, ensure_ascii=True, indent=1))
|