gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
222 lines
12 KiB
Python
222 lines
12 KiB
Python
"""Fiscal Decision Engine V1 runner.
|
|
|
|
Reads all C000025 products (read-only), classifies via the Hub, applies
|
|
central.app.fiscal_decision per domain, writes the matrix/coverage/priorities
|
|
artifacts, and with --apply executes ONLY AUTO_APPLY_SAFE proposals against
|
|
TESTESOFICIAIS.FDB (path guard enforced by firebird_write_executor), followed
|
|
by an independent re-read.
|
|
"""
|
|
import argparse
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.request
|
|
from collections import Counter, defaultdict
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
sys.path.insert(0, str(ROOT / "agent" / "src"))
|
|
|
|
from central.app.fiscal_decision import DECISION_ENGINE_VERSION, ERP_FIELD_MAP, NATURE, decide_product # noqa: E402
|
|
from gti_fiscal_agent.firebird_reader import _connect, parse_company_context # noqa: E402
|
|
|
|
DOMAIN_FIELDS = ("NCM", "IPI", "EX_TIPI", "CEST", "MVA", "CFOP", "CSOSN", "CST_ICMS", "ORIGEM", "PIS_CST", "COFINS_CST",
|
|
"PIS_COFINS_RATE", "FCP", "RTC_CCLASSTRIB", "RTC_CBS_RATE", "RTC_CCREDPRES")
|
|
NEXT_ACTION = {
|
|
"NCM": "Review LEGACY/INVALID NCMs manually (no unambiguous successor); never auto-change.",
|
|
"IPI": "Keep auto-apply; NON_NUMERIC/blank TIPI rows need a business decision on how ERP represents NT.",
|
|
"EX_TIPI": "Propose only when a single concrete EX applies; needs product-description evidence.",
|
|
"CEST": "Improve description evidence per candidate group (do not lower threshold); manual review queue for candidates.",
|
|
"MVA": "Confirm CEST first; then identify the ERP destination field for MVA (none proven).",
|
|
"CFOP": "Requires operation context (direction, destination UF, ST, purpose) — capture it at document level.",
|
|
"CSOSN": "Requires sale scenario/ST retention context; consider per-product default review for ST-annex NCMs.",
|
|
"CST_ICMS": "Not used under CRT=1; only relevant if the client changes regime.",
|
|
"ORIGEM": "Obtain objective product origin (supplier/import flag) before any write; ERP CST prefix is not proof.",
|
|
"PIS_CST": "Needs regime+incidence rule table (deferred 4.3.x) — measure need from client data first.",
|
|
"COFINS_CST": "Same as PIS_CST.",
|
|
"PIS_COFINS_RATE": "Needs source tables (deferred); never assume 0%.",
|
|
"FCP": "Needs destination UF + operation + BA FCP item list per NCM.",
|
|
"RTC_CCLASSTRIB": "Needs official NCM->annex/cClassTrib mapping (LC 214 annexes) — not in acquired datasets.",
|
|
"RTC_CBS_RATE": "Blocked by cClassTrib; apply 0.009 only where the official rule applies.",
|
|
"RTC_CCREDPRES": "Official conditions not derivable from product data.",
|
|
}
|
|
|
|
|
|
def load_dotenv(path):
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
k, _, v = line.partition("=")
|
|
os.environ.setdefault(k.strip(), v.strip())
|
|
|
|
|
|
def http(url, token, payload=None):
|
|
data = json.dumps(payload, ensure_ascii=True).encode() if payload is not None else None
|
|
req = urllib.request.Request(url, data=data, headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, method="POST" if payload is not None else "GET")
|
|
with urllib.request.urlopen(req, timeout=600) as r:
|
|
return json.load(r)
|
|
|
|
|
|
def read_all(env):
|
|
conn = _connect(env)
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute("""SELECT CODIGO,PRODUTO,CLASSIFICACAO_FISCAL,CFOP,CST,CSOSN,CEST,IPI,EX_TIPI,ORIGEM,COD_SAIDA_PIS,COD_SAIDA_COFINS,ALIQ_PIS FROM C000025 ORDER BY CODIGO""")
|
|
names = [c[0].strip() for c in cur.description]
|
|
rows = [dict(zip(names, r)) for r in cur.fetchall()]
|
|
cur.execute("SELECT UF, CRT FROM C000004")
|
|
return rows, parse_company_context(cur.fetchall())
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def build_refs(hub, token, products, company_uf):
|
|
refs = {}
|
|
for key, dom in (("csosn", "CSOSN"), ("cst_icms", "CST_ICMS"), ("cst_pis", "CST_PIS"), ("cst_cofins", "CST_COFINS"), ("origem", "ORIGEM_MERCADORIA")):
|
|
refs[key] = set(http(f"{hub}/v1/reference/fiscal_codes/{dom}", token)["codes"])
|
|
valid = set()
|
|
for code in {str(p["CFOP"]).strip() for p in products if p.get("CFOP")}:
|
|
if http(f"{hub}/v1/reference/cfop/{code}", token)["found"]:
|
|
valid.add(code)
|
|
refs["cfop_valid"] = valid
|
|
refs["fcp_uf"] = {}
|
|
r = http(f"{hub}/v1/reference/fcp/{company_uf}", token)
|
|
if r["found"]:
|
|
refs["fcp_uf"][company_uf] = r
|
|
refs["cbs_rate_2026"] = http(f"{hub}/v1/reference/rtc_cbs_rate/2026", token).get("rate")
|
|
status = http(f"{hub}/v1/datasets/status", token)
|
|
refs["cclasstrib_count"] = sum(d["record_count"] for d in status if d["kind"] == "RTC_CCLASSTRIB" and d["status"] == "ACTIVE")
|
|
return refs
|
|
|
|
|
|
def load_executor():
|
|
spec = importlib.util.spec_from_file_location("firebird_write_executor", ROOT / "scripts" / "firebird_write_executor.py")
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
def summarize(matrix):
|
|
cov = {}
|
|
for dom in DOMAIN_FIELDS:
|
|
rows = [m["fields"][dom] for m in matrix]
|
|
c = Counter(r["status"] for r in rows)
|
|
act = Counter(r["action_class"] for r in rows)
|
|
cov[dom] = {
|
|
"nature": NATURE[dom],
|
|
"DATA_AVAILABLE": sum(r["data_available"] for r in rows),
|
|
"MATCH_FOUND": sum(r["match_found"] for r in rows),
|
|
"RULE_RESOLVED": c["RULE_RESOLVED"], "RULE_CANDIDATE": c["RULE_CANDIDATE"], "RULE_UNRESOLVED": c["RULE_UNRESOLVED"],
|
|
"OPERATION_CONTEXT_REQUIRED": c["OPERATION_CONTEXT_REQUIRED"],
|
|
"CHANGE_REQUIRED": sum(1 for r in rows if r["write_eligible"]),
|
|
"action_class": dict(act),
|
|
"reason_codes": dict(Counter(r["reason_code"] for r in rows).most_common(6)),
|
|
}
|
|
return cov
|
|
|
|
|
|
def priorities_md(cov):
|
|
def key(d):
|
|
c = cov[d]
|
|
return (-c["action_class"].get("AUTO_APPLY_SAFE", 0), -c["CHANGE_REQUIRED"], -c["RULE_RESOLVED"], -c["RULE_CANDIDATE"], d)
|
|
lines = ["# NEXT_FISCAL_AUTOMATION_PRIORITIES_V1", "",
|
|
"Generated by `scripts/run_fiscal_decision_v1.py` from `docs/FISCAL_DECISION_COVERAGE_V1.json` (521 products).",
|
|
"Ordering is mechanical: AUTO_APPLY_SAFE desc, CHANGE_REQUIRED desc, RESOLVED_PRODUCTS desc, candidates desc.", "",
|
|
"| FIELD | RESOLVED_PRODUCTS | CHANGE_REQUIRED | AUTO_APPLY_SAFE | CANDIDATES | OP_CONTEXT_REQUIRED | BLOCKER | NEXT_ACTION |", "|---|---|---|---|---|---|---|---|"]
|
|
for d in sorted(cov, key=key):
|
|
c = cov[d]
|
|
blocker = next(iter(c["reason_codes"]), "-")
|
|
lines.append(f"| {d} | {c['RULE_RESOLVED']} | {c['CHANGE_REQUIRED']} | {c['action_class'].get('AUTO_APPLY_SAFE', 0)} | {c['RULE_CANDIDATE']} | {c['OPERATION_CONTEXT_REQUIRED']} | {blocker} | {NEXT_ACTION[d]} |")
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def apply_writes(matrix, products_by_code, executor):
|
|
"""Only AUTO_APPLY_SAFE + write_eligible; executor enforces path/allowlist/compare-before-write."""
|
|
import fdb
|
|
database_path = os.environ["FIREBIRD_DATABASE_PATH"]
|
|
executor.assert_authorized_write_path(database_path)
|
|
per_field = defaultdict(lambda: Counter())
|
|
audit = []
|
|
conn = fdb.connect(database=database_path, user=os.environ["FB_USER"], password=os.environ["FB_PASSWORD"], fb_library_name=os.environ.get("FIREBIRD_CLIENT_LIBRARY"))
|
|
try:
|
|
for m in matrix:
|
|
for dom, rec in m["fields"].items():
|
|
if not (rec["write_eligible"] and rec["action_class"] == "AUTO_APPLY_SAFE"):
|
|
continue
|
|
field = {"IPI": "IPI", "EX_TIPI": "EX_TIPI", "CEST": "CEST"}[dom]
|
|
per_field[field]["ATTEMPTED"] += 1
|
|
proposal = {"product_code": m["CODIGO"], "field": field, "current_value": rec["current_value"], "proposed_value": rec["proposed_value"],
|
|
"status": "CHANGE_REQUIRED", "write_eligible": True, "source_dataset": rec["provenance"]["dataset"], "source_version": rec["provenance"]["version"]}
|
|
try:
|
|
r = executor.execute_guarded_update(conn, database_path=database_path, proposal=proposal, now=datetime.now(timezone.utc).isoformat())
|
|
per_field[field]["SUCCEEDED"] += 1
|
|
audit.append({**r.as_dict(), "outcome": "SUCCEEDED"})
|
|
except executor.WriteRefused as e:
|
|
per_field[field]["FAILED"] += 1
|
|
audit.append({"product_code": m["CODIGO"], "field": field, "proposed_value": rec["proposed_value"], "outcome": "FAILED", "error": str(e)})
|
|
finally:
|
|
conn.close()
|
|
ro = _connect(os.environ)
|
|
try:
|
|
cur = ro.cursor()
|
|
for a in audit:
|
|
if a["outcome"] != "SUCCEEDED":
|
|
continue
|
|
cur.execute(f"SELECT {a['field']} FROM C000025 WHERE CODIGO = ?", (a["product_code"],))
|
|
live = cur.fetchone()[0]
|
|
if str(live).strip() == str(a["proposed_value"]).strip():
|
|
per_field[a["field"]]["POST_WRITE_VERIFIED"] += 1
|
|
finally:
|
|
ro.close()
|
|
return {f: dict(c) for f, c in per_field.items()}, audit
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--token", required=True)
|
|
ap.add_argument("--hub-url", default="http://localhost:8000")
|
|
ap.add_argument("--apply", action="store_true")
|
|
args = ap.parse_args()
|
|
load_dotenv(ROOT / ".env")
|
|
|
|
products, company = read_all(os.environ)
|
|
refs = build_refs(args.hub_url, args.token, products, company['uf'])
|
|
decisions = []
|
|
for i in range(0, len(products), 100):
|
|
chunk = products[i:i + 100]
|
|
decisions += http(f"{args.hub_url}/v1/classify/batch", args.token, {"products": [{"current_ncm": p["CLASSIFICACAO_FISCAL"], "description": p["PRODUTO"] or ""} for p in chunk]})["results"]
|
|
|
|
matrix = []
|
|
for p, d in zip(products, decisions):
|
|
matrix.append({"CODIGO": p["CODIGO"], "DESCRIPTION": p["PRODUTO"], "NCM": p["CLASSIFICACAO_FISCAL"], "fields": decide_product(p, d, company, refs)})
|
|
|
|
cov = summarize(matrix)
|
|
totals = Counter()
|
|
for m in matrix:
|
|
for rec in m["fields"].values():
|
|
totals[rec["action_class"]] += 1
|
|
(ROOT / "docs" / "FISCAL_DECISION_MATRIX_V1.json").write_text(json.dumps({
|
|
"artifact": "FISCAL_DECISION_MATRIX_V1", "engine": DECISION_ENGINE_VERSION, "company_context": company,
|
|
"total_products": len(matrix), "erp_field_map": ERP_FIELD_MAP, "products": matrix}, ensure_ascii=False, indent=1, default=str), encoding="utf-8")
|
|
coverage = {"artifact": "FISCAL_DECISION_COVERAGE_V1", "engine": DECISION_ENGINE_VERSION, "total_products": len(matrix), "company_context": company,
|
|
"erp_fields_mapped": sum(1 for f in ERP_FIELD_MAP if f["MAPPING"] == "CONFIRMED"),
|
|
"erp_fields_unconfirmed": [f["ERP_FIELD"] for f in ERP_FIELD_MAP if f["MAPPING"] != "CONFIRMED"],
|
|
"action_class_totals": dict(totals), "domains": cov, "reference_cbs_rate_2026": refs["cbs_rate_2026"]}
|
|
write_result = None
|
|
if args.apply:
|
|
write_result, audit = apply_writes(matrix, {p["CODIGO"]: p for p in products}, load_executor())
|
|
coverage["writes"] = write_result
|
|
(ROOT / "docs" / "runs").mkdir(exist_ok=True)
|
|
(ROOT / "docs" / "runs" / "fiscal_decision_v1_write_audit.json").write_text(json.dumps(audit, ensure_ascii=False, indent=1, default=str), encoding="utf-8")
|
|
(ROOT / "docs" / "FISCAL_DECISION_COVERAGE_V1.json").write_text(json.dumps(coverage, ensure_ascii=False, indent=1), encoding="utf-8")
|
|
(ROOT / "docs" / "NEXT_FISCAL_AUTOMATION_PRIORITIES_V1.md").write_text(priorities_md(cov), encoding="utf-8")
|
|
print(json.dumps({"action_class_totals": dict(totals), "writes": write_result,
|
|
"domains": {d: {k: v for k, v in c.items() if k in ("RULE_RESOLVED", "RULE_CANDIDATE", "RULE_UNRESOLVED", "OPERATION_CONTEXT_REQUIRED", "CHANGE_REQUIRED", "MATCH_FOUND")} for d, c in cov.items()}}, indent=1))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|