gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
120 lines
7.4 KiB
Python
120 lines
7.4 KiB
Python
"""Read-only NCM historical reconciliation over NFCE_ITENS vs C000025.
|
|
|
|
Explains every NFC-e item whose emitted NCM differs from the product's current NCM and builds the
|
|
TRUSTED_HISTORICAL_ITEMS corpus (identity-proven items outside systemic bad-join windows).
|
|
Outputs: docs/NCM_HISTORICAL_RECONCILIATION_V1.json, docs/runs/trusted_historical_items.json
|
|
"""
|
|
import json, os, sys, urllib.request
|
|
from collections import Counter, defaultdict
|
|
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 # noqa: E402
|
|
|
|
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())
|
|
|
|
WINDOW_RATE = 0.5 # month is a systemic bad-join window when >= 50% of its NCM-bearing items disagree
|
|
s = lambda v: None if v is None else str(v).strip()
|
|
dg = lambda v: "".join(ch for ch in str(v or "") if ch.isdigit())
|
|
|
|
|
|
def ncm_status(tok, ncms):
|
|
out = {}
|
|
ncms = sorted(n for n in ncms if n)
|
|
for i in range(0, len(ncms), 100):
|
|
chunk = ncms[i:i + 100]
|
|
req = urllib.request.Request("http://localhost:8000/v1/classify/batch", data=json.dumps({"products": [{"current_ncm": n, "description": ""} for n in chunk]}).encode(),
|
|
headers={"Content-Type": "application/json", "Authorization": "Bearer " + tok})
|
|
for n, r in zip(chunk, json.load(urllib.request.urlopen(req, timeout=600))["results"]):
|
|
out[n] = r.get("status")
|
|
return out
|
|
|
|
|
|
def main():
|
|
tok = [l.split("=", 1)[1].strip() for l in (ROOT / ".env").read_text(encoding="utf-8").splitlines() if l.startswith("AGENT_DEV_TOKEN=")][0]
|
|
cur = _connect(os.environ).cursor()
|
|
cur.execute("select CODIGO, PRODUTO, CLASSIFICACAO_FISCAL, DATA_CADASTRO, ULTIMA_ALTERACAO, PRECOVENDA from C000025")
|
|
P = {s(r[0]): {"desc": s(r[1]), "ncm": s(r[2]) or "", "cad": r[3], "alt": r[4], "price": r[5]} for r in cur.fetchall()}
|
|
by_ncm = defaultdict(list)
|
|
for k, p in P.items():
|
|
by_ncm[p["ncm"]].append((k, p["price"]))
|
|
cur.execute("select ID, CODPRODUTO, NCM, DATA, VALOR, CFOP, CSOSN, CST, CEST, GRUPO_ICMS, CST_PIS_COFINS, NUMERO_NOTA from NFCE_ITENS where NCM is not null")
|
|
items = [dict(zip(("id", "cod", "ncm", "date", "val", "cfop", "csosn", "cst", "cest", "gicms", "pis", "nota"), r)) for r in cur.fetchall()]
|
|
for it in items:
|
|
it["cod"] = s(it["cod"]); it["ncm"] = s(it["ncm"]) or ""
|
|
joined = [it for it in items if it["cod"] in P]
|
|
orphans = [it for it in items if it["cod"] not in P]
|
|
|
|
month = lambda it: str(it["date"])[:7]
|
|
mstat = defaultdict(lambda: [0, 0])
|
|
for it in joined:
|
|
mstat[month(it)][0 if it["ncm"] == P[it["cod"]]["ncm"] else 1] += 1
|
|
windows = {m for m, (ok, bad) in mstat.items() if bad and bad / (ok + bad) >= WINDOW_RATE}
|
|
|
|
status = ncm_status(tok, {it["ncm"] for it in joined} | {p["ncm"] for p in P.values()})
|
|
# per-product timeline outside systemic windows for change detection
|
|
timeline = defaultdict(list)
|
|
for it in joined:
|
|
if month(it) not in windows:
|
|
timeline[it["cod"]].append((it["date"], it["ncm"]))
|
|
|
|
def changed_after(cod, cur_ncm):
|
|
seq = sorted(timeline[cod])
|
|
vals = [n for _, n in seq]
|
|
if cur_ncm not in vals or vals[0] == cur_ncm:
|
|
return False
|
|
first_cur = vals.index(cur_ncm)
|
|
return all(v != cur_ncm for v in vals[:first_cur]) and all(v == cur_ncm for v in vals[first_cur:])
|
|
|
|
cls = Counter(); reasons = Counter(); by_month = Counter(); examples = defaultdict(list); trusted = []
|
|
for it in joined:
|
|
p = P[it["cod"]]
|
|
if it["ncm"] == p["ncm"]:
|
|
if month(it) not in windows and len(dg(it["ncm"])) == 8:
|
|
trusted.append(it)
|
|
continue
|
|
if dg(it["ncm"]).lstrip("0") == dg(p["ncm"]).lstrip("0") and dg(it["ncm"]):
|
|
c, r = "NORMALIZATION_BUG", "DIGITS_EQUAL_AFTER_NORMALIZATION"
|
|
elif status.get(it["ncm"]) == "LEGACY_NCM" or status.get(p["ncm"]) == "LEGACY_NCM":
|
|
c, r = "LEGACY_NCM", "ITEM_OR_PRODUCT_NCM_IS_LEGACY"
|
|
elif month(it) in windows:
|
|
c, r = "BAD_JOIN", f"SYSTEMIC_WINDOW_{month(it)}"
|
|
elif any(k != it["cod"] and pr is not None and it["val"] is not None and abs(float(pr) - float(it["val"])) < 0.005 for k, pr in by_ncm.get(it["ncm"], [])):
|
|
c, r = "BAD_JOIN", "ALT_PRODUCT_WITH_SAME_NCM_AND_PRICE"
|
|
elif p["alt"] is not None and str(p["alt"])[:10] > str(it["date"])[:10] and changed_after(it["cod"], p["ncm"]):
|
|
c, r = "PRODUCT_CHANGED_AFTER_DOCUMENT", "MONOTONIC_SWITCH_TO_CURRENT_NCM_AND_PRODUCT_MODIFIED_LATER"
|
|
elif status.get(it["ncm"]) == "EXACT_NCM":
|
|
vals = [n for _, n in timeline[it["cod"]]]
|
|
if vals and vals.count(p["ncm"]) / len(vals) > 0.5:
|
|
c, r = "TRUE_HISTORICAL_DIFFERENCE", "ISOLATED_DIFFERENCE_PRODUCT_OTHERWISE_CONSISTENT_UNPROVEN_CAUSE"
|
|
else:
|
|
c, r = "UNKNOWN", "NO_CONSISTENT_PRODUCT_HISTORY"
|
|
else:
|
|
c, r = "UNKNOWN", "ITEM_NCM_NOT_VALID_IN_NCM_DATASET"
|
|
cls[c] += 1; reasons[(c, r)] += 1; by_month[month(it)] += 1
|
|
if len(examples[c]) < 4:
|
|
examples[c].append({"nfce_item_id": it["id"], "product": it["cod"], "product_desc": p["desc"], "product_ncm": p["ncm"], "item_ncm": it["ncm"], "date": str(it["date"])[:10], "reason": r})
|
|
|
|
n_mis = sum(cls.values())
|
|
trusted_ids = [t["id"] for t in trusted]
|
|
doc = {"artifact": "NCM_HISTORICAL_RECONCILIATION_V1", "firebird_writes": 0,
|
|
"items_with_ncm": len(items), "items_joined_to_product": len(joined), "orphan_items_no_product": len(orphans),
|
|
"mismatches_total": n_mis, "mismatches_in_initial_sample_5000": 1671,
|
|
"classification": dict(cls), "reasons": {f"{c}|{r}": n for (c, r), n in reasons.most_common()},
|
|
"systemic_bad_join_windows": {m: {"matches": mstat[m][0], "mismatches": mstat[m][1]} for m in sorted(windows)},
|
|
"window_rule": f"month with >= {int(WINDOW_RATE*100)}% NCM disagreement is treated as a systemic bad-join window",
|
|
"mismatches_by_month": dict(sorted(by_month.items())), "examples": examples,
|
|
"technical_bugs_found": {"normalization": cls["NORMALIZATION_BUG"], "note": "0 => no code bug; the mismatches are data-identity problems (CODPRODUTO does not reliably identify the product in 2022-12 and a few other windows)."},
|
|
"trusted_corpus": {"definition": "NFC-e item with NCM present, product exists, item NCM == product NCM (identity proof), 8-digit NCM, month not a systemic window",
|
|
"TRUSTED_HISTORICAL_ITEMS": len(trusted), "excluded_matches_in_windows": sum(mstat[m][0] for m in windows),
|
|
"items_without_ncm_not_trusted": 34946 - len(items)}}
|
|
(ROOT / "docs" / "NCM_HISTORICAL_RECONCILIATION_V1.json").write_text(json.dumps(doc, ensure_ascii=False, indent=1, default=str), encoding="utf-8")
|
|
(ROOT / "docs" / "runs" / "trusted_historical_items.json").write_text(json.dumps([{k: (str(v) if k == "date" else v) for k, v in t.items()} for t in trusted], ensure_ascii=False, default=str), encoding="utf-8")
|
|
print(json.dumps({k: doc[k] for k in ("items_with_ncm", "items_joined_to_product", "orphan_items_no_product", "mismatches_total", "classification", "systemic_bad_join_windows")}, ensure_ascii=True), len(trusted))
|
|
|
|
|
|
main()
|