gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
70 lines
5.7 KiB
Python
70 lines
5.7 KiB
Python
"""Compose the LOCAL COMPLETION V1 artifacts from docs/runs/local_completion/* (no database access):
|
|
docs/LOCAL_FULL_FISCAL_DRY_RUN_V1.json, docs/LOCAL_FULL_FISCAL_WRITE_AUDIT_V1.json, docs/LOCAL_DATABASE_VALIDATION_REPORT_V1.md, docs/local_completion_totals.json
|
|
"""
|
|
import gzip
|
|
import json
|
|
import sys
|
|
from collections import Counter, defaultdict
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
from central.app.field_contract import WRITE_FIELDS # noqa: E402
|
|
|
|
ST = ROOT / "docs" / "runs" / "local_completion"
|
|
NAMES = [d["name"] for d in json.loads((ROOT / "docs" / "LOCAL_WRITE_ALLOWLIST_V1.json").read_text(encoding="utf-8"))["databases"]]
|
|
load = lambda n, suffix: json.loads((ST / f"{n}_{suffix}.json").read_text(encoding="utf-8")) if (ST / f"{n}_{suffix}.json").exists() else None # noqa: E731
|
|
|
|
dry, audit_doc, per_db = {}, {}, {}
|
|
totals = Counter()
|
|
report = ["# LOCAL_DATABASE_VALIDATION_REPORT_V1", "",
|
|
"Local completion mission over the five authorized disposable copies. States per (product, field): ALREADY_CORRECT (incl. `DECLARED_VALID_NOT_CONTRADICTED`), CORRECTED (written and independently re-read), NOT_APPLICABLE, MANUAL_INPUT_REQUIRED(<input>), FAILED (write refused/failed). No cell is left in a generic state.", ""]
|
|
for n in NAMES:
|
|
d, ap, vf, sp = load(n, "dryrun_summary"), load(n, "apply_summary"), load(n, "verify_summary"), load(n, "second_pass")
|
|
if not d:
|
|
continue
|
|
failed_by_field = Counter()
|
|
ok_by_field = Counter()
|
|
p = ST / f"{n}_write_audit.ndjson.gz"
|
|
if p.exists():
|
|
with gzip.open(p, "rt", encoding="utf-8") as f:
|
|
for line in f:
|
|
a = json.loads(line)
|
|
if a.get("outcome") == "SUCCEEDED":
|
|
for fld in a["fields"]:
|
|
ok_by_field[fld] += 1
|
|
else:
|
|
for fld in a.get("fields", []):
|
|
failed_by_field[fld] += 1
|
|
rows = {}
|
|
for f in WRITE_FIELDS:
|
|
c = d["fields"][f]
|
|
corrected = ok_by_field.get(f, 0)
|
|
failed = failed_by_field.get(f, 0)
|
|
rows[f] = {"ALREADY_CORRECT": c.get("ALREADY_CORRECT", 0), "CORRECTED": corrected, "NOT_APPLICABLE": c.get("NOT_APPLICABLE", 0), "MANUAL_INPUT_REQUIRED": c.get("MANUAL_INPUT_REQUIRED", 0), "FAILED": failed,
|
|
"CHANGE_REQUIRED_planned": c.get("CHANGE_REQUIRED", 0)}
|
|
per_db[n] = {"total_products": d["total_products"], "fields": rows, "company": d["company"], "manual_inputs": d["manual_inputs"], "roles": d["roles"], "apply": ap, "verify": vf, "second_pass": sp}
|
|
for f in WRITE_FIELDS:
|
|
for k in ("ALREADY_CORRECT", "CORRECTED", "NOT_APPLICABLE", "MANUAL_INPUT_REQUIRED", "FAILED", "CHANGE_REQUIRED_planned"):
|
|
totals[k] += rows[f][k]
|
|
totals[f"{f}:{k}"] += rows[f][k]
|
|
dry[n] = {"total_products": d["total_products"], "company": d["company"], "states_before_writes": d["fields"], "top_reasons": d["reasons"], "manual_inputs": d["manual_inputs"], "roles": d["roles"],
|
|
"detail_file": f"docs/runs/local_completion/{n}_dryrun.ndjson.gz (git-ignored; one JSON line per product/field: BEFORE, PROPOSED, STATUS, REASON, PROVENANCE, WRITE_ELIGIBLE)"}
|
|
audit_doc[n] = {"path": "exact allowlist path", "apply": ap, "verify": vf, "second_pass": {k: sp[k] for k in ("SECOND_PASS_CHANGE_REQUIRED", "SECOND_PASS_WRITES")} if sp else None,
|
|
"cells_by_field": {f: {"attempted": rows[f]["CORRECTED"] + rows[f]["FAILED"], "succeeded": rows[f]["CORRECTED"], "failed": rows[f]["FAILED"]} for f in WRITE_FIELDS if rows[f]["CORRECTED"] + rows[f]["FAILED"]},
|
|
"detail_file": f"docs/runs/local_completion/{n}_write_audit.ndjson.gz"}
|
|
report += [f"## {n}", f"Produtos: {d['total_products']} | empresa: UF {d['company']['uf']} / CRT {d['company']['crt']} (código {d['company']['company_code']})", ""]
|
|
if ap:
|
|
report.append(f"Escrita: {ap['cells_succeeded']} células corrigidas em {ap['products_updated']} produtos ({ap['cells_failed']} falharam); verificação: {vf['post_write_verified']}/{vf['planned_cells']} conferem, {vf['unplanned_changes']} mudanças não planejadas. Segunda passagem: CHANGE_REQUIRED={sp['SECOND_PASS_CHANGE_REQUIRED']}, writes={sp['SECOND_PASS_WRITES']}." if sp else f"Escrita: {ap['cells_succeeded']} células corrigidas; verificação {vf['post_write_verified']}/{vf['planned_cells']}.")
|
|
report += ["", "| campo | ALREADY_CORRECT | CORRECTED | NOT_APPLICABLE | MANUAL_INPUT_REQUIRED | FAILED |", "|---|---|---|---|---|---|"]
|
|
for f in WRITE_FIELDS:
|
|
r = rows[f]
|
|
report.append(f"| {f} | {r['ALREADY_CORRECT']} | {r['CORRECTED']} | {r['NOT_APPLICABLE']} | {r['MANUAL_INPUT_REQUIRED']} | {r['FAILED']} |")
|
|
report += ["", f"Células fiscais corrigidas: **{sum(rows[f]['CORRECTED'] for f in WRITE_FIELDS)}**. Entradas manuais pedidas (campo|chave): " + ", ".join(f"{k}={v}" for k, v in list(d["manual_inputs"].items())[:8]) + " ...", ""]
|
|
|
|
(ROOT / "docs" / "LOCAL_FULL_FISCAL_DRY_RUN_V1.json").write_text(json.dumps({"artifact": "LOCAL_FULL_FISCAL_DRY_RUN_V1", "contract": "field-contract-v1", "databases": dry}, ensure_ascii=False, indent=1), encoding="utf-8")
|
|
(ROOT / "docs" / "LOCAL_FULL_FISCAL_WRITE_AUDIT_V1.json").write_text(json.dumps({"artifact": "LOCAL_FULL_FISCAL_WRITE_AUDIT_V1", "databases": audit_doc}, ensure_ascii=False, indent=1, default=str), encoding="utf-8")
|
|
(ROOT / "docs" / "LOCAL_DATABASE_VALIDATION_REPORT_V1.md").write_text("\n".join(report) + "\n", encoding="utf-8")
|
|
(ROOT / "docs" / "runs" / "local_completion" / "totals.json").write_text(json.dumps({"totals": dict(totals), "per_db": per_db}, ensure_ascii=False, indent=1, default=str), encoding="utf-8")
|
|
print(json.dumps({k: v for k, v in totals.items() if ":" not in k}))
|