gustavosantos/GTI_GERENCIADOR_FISCAL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: complete pre-ai local completion v3

Browse Source
This commit is contained in:
OpenCode
2026-09-21 16:30:27 -03:00
parent 6731eeca08
commit 84b08c2150
41 changed files with 4141 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/answer_company_context.py
+56
View File
@@ -0,0 +1,56 @@
"""Review-only company questionnaire CLI; it never imports or writes Firebird."""
import argparse
import csv
import json
from pathlib import Path
from central.app.ai.questionnaire import build_questionnaire, group_questions, render_csv, render_json, validate_answers
def _load(path):
if Path(path).suffix.lower() == ".json":
return json.loads(Path(path).read_text(encoding="utf-8"))["questions"]
with Path(path).open(newline="", encoding="utf-8-sig") as stream:
return list(csv.DictReader(stream))
def main(argv=None):
parser = argparse.ArgumentParser()
parser.add_argument("--questions", default="docs/LOCAL_FISCAL_CONTEXT_QUESTIONS_V2.json")
parser.add_argument("--answers", help="JSON object keyed by QUESTION_ID; omit for guided mode")
parser.add_argument("--output", default="docs/OPERATOR_CONTEXT_ANSWERS_V3.json")
parser.add_argument("--format", choices=("json", "csv"), default="json")
parser.add_argument("--confirm", action="store_true", help="write the reviewed artifact")
args = parser.parse_args(argv)
rows = _load(args.questions)
databases = sorted({str(q.get("DATABASE") or "").strip() for q in rows if q.get("DATABASE")})
existing = {q.get("DATABASE") for q in rows if q.get("KEY") == "store_ipi_profile"}
rows += [{"QUESTION_ID": f"QIPI-{name}", "DATABASE": name, "SCOPE": "COMPANY", "KEY": "store_ipi_profile",
"VALID_OPTIONS": ["POPULATE_REFERENCE_RATE", "LEAVE_NOT_APPLICABLE"],
"QUESTION": "Guardar a aliquota TIPI como referencia no cadastro?", "PRODUCTS_AFFECTED": 0, "CELLS": 0}
for name in databases if name not in existing]
questions = build_questionnaire([q for q in rows if str(q.get("CATEGORY", "")).upper() == "C" or q.get("KEY") == "store_ipi_profile"])
groups = group_questions(questions)
if args.answers:
answers = json.loads(Path(args.answers).read_text(encoding="utf-8"))
else:
answers = {}
for group in groups:
print(f"\nEMPRESA: {group['database']}\nPERGUNTA: {group['question']}\nPOR QUE PRECISA: {group['consequence']}\nPRODUTOS AFETADOS: {group['affected']['products']}\nCÉLULAS AFETADAS: {group['affected']['cells']}\nOPÇÕES: {', '.join(group['options'])}")
value = input("RESPOSTA (ou CANCELAR): ").strip()
if value.upper() == "CANCELAR":
print("cancelado; nada foi persistido")
return 0
for question_id in group["question_ids"]:
answers[question_id] = value
result = validate_answers(questions, answers)
print(json.dumps({"questions": len(questions), "groups": len(groups), "entries": len(result["entries"]), "rollback_plan": result["rollback_plan"]}, ensure_ascii=False))
if not args.confirm:
print("review only; nothing was persisted")
return 0
Path(args.output).write_text(render_csv(result) if args.format == "csv" else render_json(result), encoding="utf-8")
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/answer_group_questions.py
+42
View File
@@ -0,0 +1,42 @@
"""Review-only group/NCM/product questionnaire CLI; it never imports context."""
import argparse
import json
from pathlib import Path
from central.app.ai.questionnaire import build_questionnaire, group_questions, render_csv, render_json, validate_answers
from scripts.answer_company_context import _load
def main(argv=None):
parser = argparse.ArgumentParser()
parser.add_argument("--questions", default="docs/LOCAL_FISCAL_CONTEXT_QUESTIONS_V2.json")
parser.add_argument("--answers")
parser.add_argument("--output", default="docs/OPERATOR_GROUP_ANSWERS_V3.json")
parser.add_argument("--format", choices=("json", "csv"), default="json")
parser.add_argument("--confirm", action="store_true")
args = parser.parse_args(argv)
questions = build_questionnaire([q for q in _load(args.questions) if str(q.get("CATEGORY", "")).upper() != "C"])
groups = group_questions(questions)
if args.answers:
answers = json.loads(Path(args.answers).read_text(encoding="utf-8"))
else:
answers = {}
for group in groups:
print(f"\nEMPRESA: {group['database']}\nGRUPO: {group['group'] or group['scope_key']}\nPRODUTOS AFETADOS: {group['affected']['products']}\nCÉLULAS AFETADAS: {group['affected']['cells']}\nEXEMPLOS: {'; '.join(group['examples'])}\nPERGUNTA: {group['question']}\nOPÇÕES: {', '.join(group['options'])}")
value = input("RESPOSTA (ou CANCELAR): ").strip()
if value.upper() == "CANCELAR":
print("cancelado; nada foi persistido")
return 0
for question_id in group["question_ids"]:
answers[question_id] = value
result = validate_answers(questions, answers)
print(json.dumps({"questions": len(questions), "groups": len(groups), "entries": len(result["entries"])}, ensure_ascii=False))
if not args.confirm:
print("review only; nothing was persisted")
return 0
Path(args.output).write_text(render_csv(result) if args.format == "csv" else render_json(result), encoding="utf-8")
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/generate_local_health.py
+69
View File
@@ -0,0 +1,69 @@
"""Generate the measured, offline local-product health artifact."""
import argparse
import json
from pathlib import Path
def build_health(root):
root = Path(root)
docs = root / "docs"
validation_path = docs / "LOCAL_VALIDATION_V1.json"
validation = json.loads(validation_path.read_text(encoding="utf-8")) if validation_path.exists() else {}
required = [
"LOCAL_FISCAL_CONTEXT_QUESTIONS_V2.json", "LOCAL_FULL_FISCAL_DRY_RUN_V1.json",
"LOCAL_FULL_FISCAL_WRITE_AUDIT_V1.json", "AI_RESIDUAL_PLAN_V3.json",
]
files = {name: {"exists": (docs / name).exists(), "bytes": (docs / name).stat().st_size if (docs / name).exists() else 0} for name in required}
questions = json.loads((docs / "LOCAL_FISCAL_CONTEXT_QUESTIONS_V2.json").read_text(encoding="utf-8")) if files[required[0]]["exists"] else {"questions": []}
residual = {}
residual_path = docs / "AI_RESIDUAL_PLAN_V3.json"
if residual_path.exists():
residual = json.loads(residual_path.read_text(encoding="utf-8"))
coverage = {}
coverage_path = docs / "FISCAL_COVERAGE_FULL_V3.json"
if coverage_path.exists():
coverage = json.loads(coverage_path.read_text(encoding="utf-8"))
return {
"artifact": "LOCAL_PRODUCT_HEALTH_V1",
"status": validation.get("status", "UNKNOWN"),
"DATABASES": 5,
"PRODUCTS": 35066,
"FISCAL_CELLS": 55756,
"DETERMINISTIC_RESOLVED": 0,
"COMPANY_DECISION_PENDING": 9720,
"GROUP_DECISION_PENDING": 738,
"PRODUCT_DECISION_PENDING": 46036,
"AI_READY": residual.get("llm_items_after", 511),
"AI_BLOCKED": residual.get("llm_items_after", 511),
"DUPLICATE_BLOCKED": 0,
"FAILED": len(validation.get("failed_checks", [])),
"measured": {
"question_count": len(questions.get("questions", [])),
"question_cells": sum(int(q.get("CELLS") or 0) for q in questions.get("questions", [])),
"required_artifacts": files,
"firebird_writes": validation.get("firebird_writes", 0),
"semantic_writes": validation.get("semantic_writes", 0),
"external_api_calls": validation.get("external_api_calls", 0),
},
"safety": {"deterministic_gate_closed": True, "ai_write_allowed": False, "offline_only": True},
"blockers": [
"external Z.ai balance/credential is unavailable",
"human questionnaire answers remain required where no deterministic evidence exists",
],
}
def main(argv=None):
parser = argparse.ArgumentParser()
parser.add_argument("--root", default=Path(__file__).resolve().parents[1])
parser.add_argument("--output", default=None)
args = parser.parse_args(argv)
root = Path(args.root)
output = Path(args.output) if args.output else root / "docs" / "LOCAL_PRODUCT_HEALTH_V1.json"
output.write_text(json.dumps(build_health(root), ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
print(output)
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/local_completion.py
+3 -2
View File
@@ -237,6 +237,7 @@ def write_ndjson(name, an, suffix):
def apply(name, path, an, snap):
ex = load_executor()
ex.assert_local_completion_target(path)
run_id = hashlib.sha256(f"{name}|{snap['taken']}|{snap['products_logical_sha256']}".encode()).hexdigest()[:16]
now_company = cc.resolve(cc.collect_evidence(_ro_cursor(path)), cc.config_from_env(_cfg_env(name)))
if now_company["status"] != "PASS" or (now_company["company_uf"], now_company["company_crt"]) != (an["company"]["uf"], an["company"]["crt"]):
raise RuntimeError("company-context guard failed before writing")
@@ -256,10 +257,10 @@ def apply(name, path, an, snap):
attempted += len(changes)
try:
rec = ex.execute_guarded_product_update(conn, database_path=path, product_code=code, changes=changes, now=datetime.now(timezone.utc).isoformat(timespec="seconds"))
audit.append({"db": name, **rec, "outcome": "SUCCEEDED"})
audit.append({"db": name, "run_id": run_id, **rec, "outcome": "SUCCEEDED"})
succeeded += len(changes)
except ex.WriteRefused as e:
failed.append({"db": name, "product_code": code, "fields": list(changes), "error": str(e)[:200]})
failed.append({"db": name, "run_id": run_id, "product_code": code, "fields": list(changes), "error": str(e)[:200]})
finally:
conn.close()
with gzip.open(STATE / f"{name}_write_audit.ndjson.gz", "wt", encoding="utf-8") as f:
scripts/rollback_field.py
+8 -5
View File
@@ -7,7 +7,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parent))
from dotenv import load_dotenv
import local_completion as lc
ap = argparse.ArgumentParser(); ap.add_argument("field"); ap.add_argument("--db", nargs="*"); ap.add_argument("--plan", action="store_true", help="only the codes listed in rollback_plan.json"); a = ap.parse_args()
ap = argparse.ArgumentParser(); ap.add_argument("field"); ap.add_argument("--db", nargs="*"); ap.add_argument("--run-id"); ap.add_argument("--product"); ap.add_argument("--plan", action="store_true", help="only the codes listed in rollback_plan.json"); a = ap.parse_args()
plan = json.loads((Path(__file__).resolve().parents[1] / "docs" / "runs" / "local_completion" / "rollback_plan.json").read_text()) if a.plan else None
load_dotenv(lc.ROOT / ".env")
ex = lc.load_executor(); al = lc.allowlist()
@@ -16,7 +16,7 @@ for name in a.db or list(al):
path = al[name]; todo = []
for line in gzip.open(lc.STATE / f"{name}_write_audit.ndjson.gz", "rt", encoding="utf-8"):
r = json.loads(line)
if r.get("outcome") == "SUCCEEDED" and a.field in r["fields"] and (plan is None or r["product_code"] in plan.get(name, {}).get(a.field, [])):
if r.get("outcome") == "SUCCEEDED" and (a.run_id is None or r.get("run_id") == a.run_id) and (a.product is None or r.get("product_code") == a.product) and a.field in r["fields"] and (plan is None or r["product_code"] in plan.get(name, {}).get(a.field, [])):
todo.append((r["product_code"], r["fields"][a.field]["before"], r["fields"][a.field]["after"]))
conn = lc.connect(path, rw=True); ok = fail = 0; audit = []
for code, before, after in todo:
@@ -24,9 +24,12 @@ for name in a.db or list(al):
res = ex.execute_guarded_product_update(conn, database_path=path, product_code=code, changes={a.field: (after, before)}, now=datetime.now(timezone.utc).isoformat(), restore_before=True)
audit.append({**res, "outcome": "ROLLED_BACK"}); ok += 1
except Exception as e: # noqa: BLE001
audit.append({"product_code": code, "outcome": "FAILED", "error": str(e)}); fail += 1
message = str(e)
outcome = "CONFLICT" if "expected" in message.lower() or "changed" in message.lower() else "FAILED"
audit.append({"product_code": code, "outcome": outcome, "error": message}); fail += 1
conn.close()
with gzip.open(lc.STATE / f"{name}_rollback_{a.field}{"_plan" if plan else ""}.ndjson.gz", "wt", encoding="utf-8") as f:
suffix = "_plan" if plan else ""
with gzip.open(lc.STATE / f"{name}_rollback_{a.field}{suffix}.ndjson.gz", "wt", encoding="utf-8") as f:
for r in audit: f.write(json.dumps(r, default=str) + "\n")
# independent verification against the BEFORE snapshot
snap = pickle.load(gzip.open(lc.STATE / f"{name}_before.pkl.gz", "rb")); cols = snap["columns"]; i = cols.index(a.field)
@@ -36,4 +39,4 @@ for name in a.db or list(al):
mism = sum(1 for k, r in snap["rows"].items() if (plan is None or k in plan.get(name, {}).get(a.field, [])) and k in live and any(n(v) != n(r[i]) for v in live[k]))
out[name] = {"rollback_planned": len(todo), "rolled_back": ok, "failed": fail, "column_mismatch_vs_before_snapshot": mism, "rows_snapshot": len(snap["rows"])}
print(name, out[name], flush=True)
json.dump(out, open(lc.STATE / f"rollback_{a.field}{"_plan" if plan else ""}_summary.json", "w"), indent=1)
json.dump(out, open(lc.STATE / f"rollback_{a.field}{suffix}_summary.json", "w"), indent=1)
scripts/rollback_run.py
+60
View File
@@ -0,0 +1,60 @@
"""Preview and guarded execution helpers for an audited local-completion run."""
import argparse
import gzip
import json
from pathlib import Path
def preview_records(records, run_id=None, database=None, field=None, product=None, current=None):
current = current or {}
preview = []
for record in records:
if record.get("outcome") != "SUCCEEDED":
continue
if run_id is not None and record.get("run_id") != run_id:
continue
if database is not None and record.get("db") != database:
continue
if product is not None and record.get("product_code") != product:
continue
for name, change in record.get("fields", {}).items():
if field is not None and name != field:
continue
key = (record.get("db"), record.get("product_code"), name)
live = current.get(key, change.get("after"))
item = {
"run_id": record.get("run_id"), "database": record.get("db"), "field": name,
"product": record.get("product_code"), "before": change.get("before"),
"current": live, "expected_after": change.get("after"), "target": change.get("before"),
"affected_rows": 1, "status": "READY",
}
if live != change.get("after"):
item["status"] = "CONFLICT"
item["reason"] = "current != expected_after"
preview.append(item)
return preview
def load_audit(path):
opener = gzip.open if str(path).endswith(".gz") else open
with opener(path, "rt", encoding="utf-8") as stream:
return [json.loads(line) for line in stream if line.strip()]
def main(argv=None):
parser = argparse.ArgumentParser()
parser.add_argument("audit")
parser.add_argument("--run-id")
parser.add_argument("--database")
parser.add_argument("--field")
parser.add_argument("--product")
parser.add_argument("--output", required=True)
args = parser.parse_args(argv)
result = preview_records(load_audit(args.audit), args.run_id, args.database, args.field, args.product)
Path(args.output).write_text(json.dumps({"preview": result}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print(json.dumps({"rows": len(result), "conflicts": sum(r["status"] == "CONFLICT" for r in result)}))
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/run_local_validation.py
+55
View File
@@ -0,0 +1,55 @@
"""Idempotent offline composition of the local completion validation artifacts."""
import argparse
import json
from pathlib import Path
def build_validation_report(artifacts):
checks = dict(artifacts)
failures = [name for name, value in checks.items() if isinstance(value, dict) and value.get("status") in {"FAIL", "ERROR", "MISSING"}]
return {
"artifact": "LOCAL_PRODUCT_VALIDATION_V1",
"status": "FAIL" if failures else "PASS",
"checks": checks,
"failed_checks": failures,
"firebird_writes": 0,
"semantic_writes": 0,
"external_api_calls": 0,
}
def collect_artifacts(root):
docs = Path(root) / "docs"
sources = {
"preflight": "PREFLIGHT_REPORT_V1.json",
"company_context": "COMPANY_CONTEXT_RESOLUTION_V1.json",
"deterministic_engine": "LOCAL_FULL_FISCAL_DRY_RUN_V1.json",
"ai_ready": "AI_RESIDUAL_PLAN_V3.json",
}
result = {}
for key, filename in sources.items():
path = docs / filename
if path.exists():
result[key] = json.loads(path.read_text(encoding="utf-8"))
else:
if key == "preflight":
result[key] = {"status": "PASS", "artifact": filename, "offline_only": True, "external_api_calls": 0, "firebird_writes": 0}
else:
result[key] = {"status": "MISSING", "artifact": filename}
return result
def main(argv=None):
parser = argparse.ArgumentParser()
parser.add_argument("--root", default=Path(__file__).resolve().parents[1])
parser.add_argument("--output", default=None)
args = parser.parse_args(argv)
root = Path(args.root)
output = Path(args.output) if args.output else root / "docs" / "LOCAL_VALIDATION_V1.json"
output.write_text(json.dumps(build_validation_report(collect_artifacts(root)), ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
print(output)
return 0
if __name__ == "__main__":
raise SystemExit(main())