gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
67 lines
3.5 KiB
Python
67 lines
3.5 KiB
Python
"""Missão 2 / Bloco 1 — dry-run fiscal read-only sobre um banco arbitrário (Agent 2 = GTI.FDB incluso).
|
|
|
|
Reaproveita o motor puro (central.app.field_contract) e as funções analyze()/summarize()
|
|
já existentes em scripts/local_completion.py (mesma taxonomia ALREADY_CORRECT/CHANGE_REQUIRED/
|
|
NOT_APPLICABLE/MANUAL_INPUT_REQUIRED). NUNCA chama apply()/execute_guarded_product_update —
|
|
este script não importa firebird_write_executor e não abre conexão rw=True em nenhum ponto.
|
|
|
|
python scripts/dry_run_universe.py --name GTI --path "D:\\PROJETOS\\GTISISTEMAS\\Banco\\GTI.FDB" [--company-code C] [--crt-source AUTO|EMISSOR|C000004]
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
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"))
|
|
import run_fiscal_decision_v1 as base # noqa: E402
|
|
import local_completion as lc # noqa: E402
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--name", required=True)
|
|
ap.add_argument("--path", required=True)
|
|
ap.add_argument("--company-code")
|
|
ap.add_argument("--crt-source")
|
|
ap.add_argument("--out")
|
|
a = ap.parse_args()
|
|
base.load_dotenv(ROOT / ".env")
|
|
token = os.environ["AGENT_DEV_TOKEN"]
|
|
if a.company_code or a.crt_source:
|
|
lc.COMPANY_CFG[a.name] = {k: v for k, v in (("company_code", a.company_code), ("crt_source", a.crt_source)) if v}
|
|
an = lc.analyze(a.name, a.path, token)
|
|
summary = lc.summarize(an)
|
|
from collections import Counter
|
|
scope_by_key = Counter()
|
|
for res in an["results"].values():
|
|
for fld, r in res.items():
|
|
if fld.startswith("_"):
|
|
continue
|
|
if r.get("state") == "MANUAL_INPUT_REQUIRED":
|
|
scope_by_key[(fld, r.get("input_key"), r.get("input_scope"))] += 1
|
|
summary["manual_inputs_by_scope"] = {f"{f}|{k}|{s}": n for (f, k, s), n in sorted(scope_by_key.items(), key=lambda x: -x[1])}
|
|
out_path = Path(a.out) if a.out else (ROOT / "docs" / "runs" / f"{a.name}_dryrun_universe_summary.json")
|
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
|
out_path.write_text(json.dumps(summary, ensure_ascii=False, indent=1, default=str), encoding="utf-8")
|
|
print(json.dumps({"name": a.name, "total_products": summary["total_products"], "company": summary["company"],
|
|
"change_required_cells": summary["change_required_cells"], "write_eligible_cells": summary["write_eligible_cells"]}, indent=1, default=str))
|
|
ndjson_path = ROOT / "docs" / "runs" / f"{a.name}_dryrun_universe.ndjson.gz"
|
|
import gzip
|
|
with gzip.open(ndjson_path, "wt", encoding="utf-8") as f:
|
|
from central.app.field_contract import WRITE_FIELDS, CONTRACT_VERSION
|
|
for p in an["products"]:
|
|
code = p["_rk"]; res = an["results"][code]
|
|
for fld in WRITE_FIELDS:
|
|
r = res[fld]
|
|
f.write(json.dumps({"db": a.name, "codigo": p["CODIGO"], "produto": p["PRODUTO"], "field": fld, "current": r["current"], "proposed": r["proposed"],
|
|
"status": r["state"], "reason": r["reason"], "input_key": r.get("input_key"),
|
|
"provenance": f"{CONTRACT_VERSION}|" + ",".join(f"{k}:{v}" for k, v in res["_ds"].items()) + f"|role={res['_facts']['role']}"}, ensure_ascii=False, default=str) + "\n")
|
|
print(f"summary -> {out_path}")
|
|
print(f"cells -> {ndjson_path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|