gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
107 lines
4.9 KiB
Python
107 lines
4.9 KiB
Python
"""Measure required-field completeness from live read-only product masters."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
from central.app.fiscal_completeness import evaluate_product, load_requiredness_matrix # noqa: E402
|
|
|
|
|
|
def _env() -> None:
|
|
path = ROOT / ".env"
|
|
if path.exists():
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
if "=" in line and not line.lstrip().startswith("#"):
|
|
key, _, value = line.partition("=")
|
|
os.environ.setdefault(key.strip(), value.strip())
|
|
|
|
|
|
def measure_database(path: Path, matrix: dict) -> dict:
|
|
import fdb
|
|
|
|
conn = fdb.connect(database=str(path), user=os.environ["FB_USER"], password=os.environ["FB_PASSWORD"], fb_library_name=os.environ.get("FIREBIRD_CLIENT_LIBRARY"), isolation_level=fdb.ISOLATION_LEVEL_READ_COMMITED_RO)
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT TRIM(RDB$FIELD_NAME) FROM RDB$RELATION_FIELDS WHERE RDB$RELATION_NAME = 'C000025'")
|
|
available = {row[0] for row in cur.fetchall()}
|
|
fields = [field for field in matrix["fields"] if field in available]
|
|
cur.execute("SELECT UF, CRT FROM C000004")
|
|
company_rows = {(str(uf or '').strip().upper(), str(crt or '').strip()) for uf, crt in cur.fetchall()}
|
|
company = next(iter(company_rows)) if len(company_rows) == 1 else (None, None)
|
|
select_fields = ",".join(["CODIGO"] + fields)
|
|
cur.execute(f"SELECT {select_fields} FROM C000025")
|
|
complete = incomplete = 0
|
|
reason_counts: dict[str, int] = {}
|
|
required_valid = required_missing = 0
|
|
for row in cur.fetchall():
|
|
product = dict(zip(["CODIGO"] + fields, row))
|
|
context = {
|
|
"uf": company[0],
|
|
"crt": company[1],
|
|
"regime": "SIMPLES" if company[1] == "1" else "NORMAL" if company[1] in {"2", "3"} else None,
|
|
"operation": "SALE",
|
|
"contribuinte_ipi": "S" if str(product.get("IPI") or "").strip() else "N",
|
|
"has_numeric_tipi": False,
|
|
"has_supported_ex": False,
|
|
"product_matches_ex": False,
|
|
"has_official_cest": bool(str(product.get("CEST") or "").strip()),
|
|
"st_subject": str(product.get("SITUACAO_TRIBUTARIA") or "").strip() == "S",
|
|
"erp_consumes_entry_fields": False,
|
|
"icms_taxed": str(product.get("SITUACAO_TRIBUTARIA") or "").strip() == "T",
|
|
"origin_proven": False,
|
|
"erp_consumes_ipi_cst": False,
|
|
"piscofins_profile_proven": False,
|
|
"piscofins_rate_applies": False,
|
|
}
|
|
result = evaluate_product(product, context, matrix)
|
|
for field_result in result["fields"].values():
|
|
if field_result["status"] == "REQUIRED_AND_VALID":
|
|
required_valid += 1
|
|
elif field_result["status"] == "REQUIRED_MISSING":
|
|
required_missing += 1
|
|
if result["FISCALLY_COMPLETE"]:
|
|
complete += 1
|
|
else:
|
|
incomplete += 1
|
|
for field_result in result["fields"].values():
|
|
if field_result["status"] == "REQUIRED_MISSING":
|
|
reason_counts[field_result["field"]] = reason_counts.get(field_result["field"], 0) + 1
|
|
return {
|
|
"database": path.name,
|
|
"TOTAL_PRODUCTS": complete + incomplete,
|
|
"FISCALLY_COMPLETE": complete,
|
|
"FISCALLY_INCOMPLETE": incomplete,
|
|
"required_valid": required_valid,
|
|
"required_missing": required_missing,
|
|
"missing_reasons": reason_counts,
|
|
"company_context_observed": company[0] is not None,
|
|
"manual_context_not_answered": True,
|
|
"mode": "READ_ONLY",
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def main() -> None:
|
|
_env()
|
|
matrix = load_requiredness_matrix(ROOT / "docs/FISCAL_REQUIREDNESS_MATRIX_V1.json")
|
|
paths = [ROOT / "TESTESOFICIAIS.FDB"] + sorted((ROOT / "Banco").glob("*.FDB"))
|
|
databases = []
|
|
for path in paths:
|
|
try:
|
|
databases.append(measure_database(path, matrix))
|
|
except Exception as exc:
|
|
databases.append({"database": path.name, "status": "UNAVAILABLE", "error_type": type(exc).__name__, "mode": "READ_ONLY"})
|
|
artifact = {"artifact": "AUTONOMOUS_FISCAL_COMPLETION_V1", "requiredness_matrix": "FISCAL_REQUIREDNESS_MATRIX_V1", "firebird_writes": 0, "external_api_calls": 0, "databases": databases}
|
|
(ROOT / "docs/AUTONOMOUS_FISCAL_COMPLETION_V1.json").write_text(json.dumps(artifact, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
print(json.dumps({"artifact": artifact["artifact"], "databases": len(databases), "firebird_writes": 0}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|