GTI_GERENCIADOR_FISCAL/fiscal/scripts/rtc_resolution_dryrun.py

"""READ-ONLY dry-run of the deterministic cClassTrib resolver over real Firebird databases. Writes NOTHING to any database.

    python scripts/rtc_resolution_dryrun.py --db TESTESOFICIAIS CONSTRUCAO ... [--out docs/runs/rtc_resolution_dryrun.json]

For each database, three passes over the same read-only snapshot:
  AS_FOUND            the ERP's own RTC switch and the company's real CRT (what the Agent would report today)
  WHATIF_RTC_ON       RTC forced ON (pure evaluator; PARAMETROS is never touched), real CRT
  WHATIF_NORMAL_ON    RTC forced ON and regime normal (CRT 3): what the rules WOULD resolve once an official orientation exists for Simples
NCM validity = the code exists in the active official NCM table exported from the Hub (docs/runs/ncm_valid_codes.txt).
"""
import argparse
import csv
import json
import sys
from collections import Counter
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT)); sys.path.insert(0, str(ROOT / "scripts")); sys.path.insert(0, str(ROOT / "agent" / "src"))
from dotenv import load_dotenv  # noqa: E402

load_dotenv(ROOT / ".env")
import local_completion as lc  # noqa: E402
from gti_fiscal_agent import firebird_reader, unified_fiscal  # noqa: E402
from unified_coverage_e2e import resolved_crt  # noqa: E402

NCM_OK = set((ROOT / "docs" / "runs" / "ncm_valid_codes.txt").read_text().split())


def digits(value):
    return "".join(ch for ch in str(value or "") if ch.isdigit())


def one_pass(env, company, override, ncm_valid):
    conn = firebird_reader._connect(env)                      # READ COMMITTED READ ONLY
    try:
        return unified_fiscal.assess_from_connection(conn, company, rtc_override=override, ncm_valid=ncm_valid)
    finally:
        conn.close()


def main():
    ap = argparse.ArgumentParser(); ap.add_argument("--db", nargs="+", required=True); ap.add_argument("--out", default="docs/runs/rtc_resolution_dryrun.json"); ap.add_argument("--sample", type=int, default=100)
    a = ap.parse_args()
    report, sample = {}, []
    for name in a.db:
        path = lc.allowlist()[name]; env = lc.env_for(path)
        crt = resolved_crt(name, path)
        products = firebird_reader.read_products(env)
        ncm_valid = {str(p["CODIGO"]): digits(p.get("CLASSIFICACAO_FISCAL")) in NCM_OK for p in products}
        passes = {}
        for label, override, company in (("AS_FOUND", None, {"crt": crt}), ("WHATIF_RTC_ON", True, {"crt": crt}), ("WHATIF_NORMAL_ON", True, {"crt": "3"})):
            out = one_pass(env, company, override, ncm_valid)
            summ = out["context"].get("rtc_resolution")
            findings = out["findings"]
            by_status = Counter((f.get("rtc_resolution") or {}).get("status", "NOT_EVALUATED") for f in findings.values())
            conflicts = Counter(b["type"] for f in findings.values() for b in f["blockers"] if b["category"] == "RTC")
            passes[label] = {"crt": company["crt"], "rtc_enabled": out["context"]["rtc_enabled"], "products": len(findings), "resolution_summary": summ, "by_status": dict(by_status), "rtc_blocker_types": dict(conflicts),
                             "AUTO_APPLY_CANDIDATES": sum(1 for f in findings.values() if (f.get("rtc_resolution") or {}).get("auto_apply_candidate") and not [b for b in f["blockers"] if b["type"] in ("CCLASSTRIB_CONFLICT",)])}
            if label == "WHATIF_NORMAL_ON" and len(sample) < a.sample:
                by_code = {str(p["CODIGO"]): p for p in products}
                for code, f in findings.items():
                    res = f.get("rtc_resolution")
                    if res and len(sample) < a.sample and (res["auto_apply_candidate"] or res["status"] == "AMBIGUOUS") and (len([s for s in sample if s["database"] == name]) < max(1, a.sample // len(a.db))):
                        p = by_code.get(code, {})
                        sample.append({"database": name, "product": code, "ncm": digits(p.get("CLASSIFICACAO_FISCAL")), "current_cclasstrib": (p.get("CCLASSTRIB") or "").strip() if p.get("CCLASSTRIB") else "", "proposed_cclasstrib": res["resolved_cclasstrib"] or "",
                                       "status": res["status"], "rule_id": res["rule_id"], "reason": res["reason"], "confidence_type": res["confidence_type"], "scenario": "WHATIF_NORMAL_ON (regime normal, RTC on)"})
            passes[label]["context"] = {k: out["context"][k] for k in ("rtc_param", "rtc_source", "rtc_observed_at", "local_catalog_hash", "official_catalog", "catalog_integrity")}
        report[name] = {"crt": crt, "total_products": len(products), "passes": passes}
        print(name, "crt", crt, {k: (v["rtc_enabled"], v["by_status"], v["AUTO_APPLY_CANDIDATES"]) for k, v in passes.items()}, flush=True)
    doc = {"mode": "READ_ONLY_DRY_RUN", "FIREBIRD_WRITES": 0, "WRITE_ATTEMPTS": 0, "WRITE_SUCCESSES": 0, "databases": report}
    (ROOT / a.out).write_text(json.dumps(doc, ensure_ascii=False, indent=1, default=str), encoding="utf-8")
    with open(ROOT / "docs" / "runs" / "rtc_resolution_audit_sample.csv", "w", newline="", encoding="utf-8-sig") as fh:
        w = csv.DictWriter(fh, fieldnames=list(sample[0]) if sample else ["database"]); w.writeheader(); w.writerows(sample)
    print("sample rows:", len(sample))


if __name__ == "__main__":
    main()