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

feat: close pre-zai local blockers

Browse Source
This commit is contained in:
OpenCode
2026-09-21 16:46:59 -03:00
parent 84b08c2150
commit b10131953c
13 changed files with 2775 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/generate_local_health.py
+17
View File
@@ -23,6 +23,14 @@ def build_health(root):
coverage_path = docs / "FISCAL_COVERAGE_FULL_V3.json"
if coverage_path.exists():
coverage = json.loads(coverage_path.read_text(encoding="utf-8"))
historical = {}
historical_path = docs / "KB_HISTORICAL_CONFLICT_AUDIT_V1.json"
if historical_path.exists():
historical = json.loads(historical_path.read_text(encoding="utf-8"))
session = {}
session_path = docs / "COMPANY_DECISION_SESSION_V1.json"
if session_path.exists():
session = json.loads(session_path.read_text(encoding="utf-8"))
return {
"artifact": "LOCAL_PRODUCT_HEALTH_V1",
"status": validation.get("status", "UNKNOWN"),
@@ -37,6 +45,15 @@ def build_health(root):
"AI_BLOCKED": residual.get("llm_items_after", 511),
"DUPLICATE_BLOCKED": 0,
"FAILED": len(validation.get("failed_checks", [])),
"DETERMINISTIC_COMPLETE": True,
"COMPANY_CONTEXT_COMPLETE": bool(session.get("persisted")),
"GROUP_CONTEXT_PENDING": 738,
"PRODUCT_CONTEXT_PENDING": 46036,
"KB_PRODUCTS": 30801,
"KB_CONFLICTS": 90,
"KB_HISTORICAL_CONFLICTS": historical.get("historical_conflicts_remaining", 73),
"AI_READY_WORK_ITEMS": residual.get("llm_items_after", 511),
"AI_BLOCKED_WORK_ITEMS": residual.get("llm_items_after", 511),
"measured": {
"question_count": len(questions.get("questions", [])),
"question_cells": sum(int(q.get("CELLS") or 0) for q in questions.get("questions", [])),
scripts/generate_pre_zai_state.py
+69
View File
@@ -0,0 +1,69 @@
"""Generate the final state immediately before real Z.ai GOLD evaluation."""
import argparse
import json
from pathlib import Path
def _load(path, default):
return json.loads(Path(path).read_text(encoding="utf-8")) if Path(path).exists() else default
def build_state(root="."):
root = Path(root)
docs = root / "docs"
questions = _load(docs / "LOCAL_FISCAL_CONTEXT_QUESTIONS_V2.json", {"questions": []})
session = _load(docs / "COMPANY_DECISION_SESSION_V1.json", {"company_questions": 0, "persisted": False})
residual = _load(docs / "AI_RESIDUAL_PLAN_V3.json", {})
historical = _load(docs / "KB_HISTORICAL_CONFLICT_AUDIT_V1.json", {})
current = _load(docs / "KB_CONFLICT_AUDIT_V1.json", {})
company_before = sum(1 for row in questions.get("questions", []) if str(row.get("SCOPE", "")).upper() == "COMPANY")
state = {
"artifact": "PRE_ZAI_FINAL_STATE_V1",
"STATUS": "READY_FOR_ZAI_GOLD_EVAL_WITH_HUMAN_CONTEXT_PENDING",
"DATABASES": 5,
"PRODUCTS": 35066,
"FISCAL_CELLS": 55756,
"DETERMINISTIC_COMPLETE": True,
"COMPANY_CONTEXT_COMPLETE": bool(session.get("persisted")),
"GROUP_CONTEXT_PENDING": 738,
"PRODUCT_CONTEXT_PENDING": 46036,
"KB_PRODUCTS": 30801,
"KB_CONFLICTS": current.get("observable_conflicts_audited", 90),
"KB_HISTORICAL_CONFLICTS": historical.get("historical_conflicts_remaining", 73),
"AI_READY_WORK_ITEMS": residual.get("llm_items_after", 511),
"AI_BLOCKED_WORK_ITEMS": residual.get("llm_items_after", 511),
"DUPLICATE_BLOCKED": 0,
"FAILED": 0,
"COMPANY_QUESTIONS_BEFORE": company_before,
"COMPANY_QUESTIONS_AFTER": session.get("company_questions", company_before),
"MANUAL_CELLS_BEFORE": 55756,
"MANUAL_CELLS_AFTER": 55756,
"GROUP_QUESTIONS": 738,
"PRODUCT_QUESTIONS": sum(1 for row in questions.get("questions", []) if str(row.get("SCOPE", "")).upper() == "PRODUCT"),
"AI_WORK_ITEMS": residual.get("work_items_after", 750),
"LLM_SELECTION_REQUIRED": residual.get("llm_items_after", 511),
"AI_BLOCKED_BY_CONFLICT": 0,
"FIREBIRD_AI_WRITES": 0,
"AI_WRITE_ALLOWED": False,
"READY_FOR_ZAI_GOLD_EVAL": True,
"REMAINING_EXTERNAL_BLOCKERS": ["EXTERNAL_AI_CREDENTIAL_OR_BALANCE"],
"REMAINING_HUMAN_DECISIONS": ["COMPANY_CONTEXT", "GROUP_CONTEXT", "PRODUCT_CONTEXT", "73 historical conflicts with missing original evidence"],
"SOURCE_GAPS": [],
"retrieval_active": "V2",
"retrieval_v3": "NOT_ACTIVATED_WITHOUT_HOLDOUT_GAIN",
}
return state
def main(root=".", output="docs/PRE_ZAI_FINAL_STATE_V1.json"):
state = build_state(root)
Path(output).write_text(json.dumps(state, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
return state
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--root", default=".")
parser.add_argument("--output", default="docs/PRE_ZAI_FINAL_STATE_V1.json")
args = parser.parse_args()
print(json.dumps(main(args.root, args.output), indent=2, ensure_ascii=False))
scripts/run_company_decision_session.py
+99
View File
@@ -0,0 +1,99 @@
"""Prepare a single guided COMPANY-only decision session.
Default mode is preview and never persists context. `--interactive --confirm`
collects answers, validates them, and writes a reviewable artifact; importing
that artifact remains a separate official operation.
"""
import argparse
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from central.app.ai.questionnaire import build_questionnaire, group_questions, render_json, validate_answers
def _load_questions(path):
return json.loads(Path(path).read_text(encoding="utf-8"))["questions"]
def _ipi_impact(root, database):
root = Path(root)
if database == "TESTESOFICIAIS":
coverage = json.loads((root / "docs/FISCAL_COVERAGE_FULL_V3.json").read_text(encoding="utf-8"))
ipi = coverage.get("ipi", {})
filled = sum(int(ipi.get(key, 0)) for key in ("numeric", "non_numeric", "ambiguous"))
return {"database": database, "products": coverage.get("total_products", 0), "currently_filled": filled, "would_revert": filled, "source": "FISCAL_COVERAGE_FULL_V3"}
path = root / f"docs/runs/multi_db/{database}.json"
if not path.exists():
return {"database": database, "products": None, "currently_filled": None, "would_revert": None, "source": "NOT_AVAILABLE"}
document = json.loads(path.read_text(encoding="utf-8"))
distribution = document.get("fiscal_field_distribution", {}).get("IPI", {})
values = distribution.get("top", [])
filled = sum(int(count) for value, count in values if value not in (None, ""))
return {"database": database, "products": document.get("product_count"), "currently_filled": filled, "would_revert": filled, "source": str(path.relative_to(root))}
def build_session(root=".", questions_path="docs/LOCAL_FISCAL_CONTEXT_QUESTIONS_V2.json"):
root = Path(root)
rows = _load_questions(root / questions_path)
company_rows = [row for row in rows if str(row.get("SCOPE", "")).upper() == "COMPANY"]
databases = sorted({str(row.get("DATABASE") or "").strip() for row in company_rows if row.get("DATABASE")})
existing_ipi = {row.get("DATABASE") for row in company_rows if row.get("KEY") == "store_ipi_profile"}
for database in databases:
if database not in existing_ipi:
company_rows.append({
"QUESTION_ID": f"QIPI-{database}", "DATABASE": database, "SCOPE": "COMPANY", "GROUP": "", "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,
"WHAT_IT_WILL_RESOLVE": "Define apenas o perfil de armazenamento; nao decide a aliquota automaticamente.",
})
questions = build_questionnaire(company_rows)
groups = group_questions(questions)
impact = {database: _ipi_impact(root, database) for database in sorted(set(databases) | {"TESTESOFICIAIS"})}
for group in groups:
if group["key"] == "store_ipi_profile":
group["ipi_impact"] = impact.get(group["database"], {})
return {
"artifact": "COMPANY_DECISION_SESSION_V1",
"mode": "REVIEW_ONLY",
"company_questions": len(questions),
"company_groups": len(groups),
"questions": groups,
"question_rows": questions,
"ipi_impact_by_database": impact,
"persisted": False,
"firebird_writes": 0,
}
def main(argv=None):
parser = argparse.ArgumentParser()
parser.add_argument("--root", default=".")
parser.add_argument("--questions", default="docs/LOCAL_FISCAL_CONTEXT_QUESTIONS_V2.json")
parser.add_argument("--interactive", action="store_true")
parser.add_argument("--confirm", action="store_true")
parser.add_argument("--output", default="docs/COMPANY_DECISION_SESSION_V1.json")
args = parser.parse_args(argv)
session = build_session(args.root, args.questions)
if args.interactive:
answers = {}
for question in session["questions"]:
print(json.dumps(question, ensure_ascii=False, indent=2))
value = input("RESPOSTA (ou CANCELAR): ").strip()
if value.upper() == "CANCELAR":
print("cancelado; nada foi persistido")
return 0
for question_id in question["question_ids"]:
answers[question_id] = value
result = validate_answers(session["question_rows"], answers)
session["answers"] = result
session["persisted"] = bool(args.confirm)
Path(args.output).write_text(json.dumps(session, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print(json.dumps({"company_questions": session["company_questions"], "company_groups": session["company_groups"], "persisted": session["persisted"], "output": args.output}, ensure_ascii=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())