59 lines
6.0 KiB
Python
59 lines
6.0 KiB
Python
"""Questionnaire V2: aggregates every remaining MANUAL_INPUT_REQUIRED cell (docs/runs/local_completion/*_second.ndjson.gz) into the fewest operator questions.
|
|
Only categories C (company configuration), D (product-group/NCM configuration) and E (product fact) may become questions."""
|
|
import csv, gzip, json, pickle, sys
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]; ST = ROOT / "docs" / "runs" / "local_completion"
|
|
sys.path.insert(0, str(ROOT))
|
|
from central.app.context_keys import ENUMS # noqa: E402
|
|
|
|
DBS = ["CONSTRUCAO", "MMMODAS", "RINCAO", "UNIAO", "TESTESOFICIAIS"]
|
|
CAT = {"origin": ("E", "COMPANY"), "pis_cofins_profile": ("C", "COMPANY"), "entry_scope": ("C", "COMPANY"), "icms_rate": ("C", "COMPANY"), "cest": ("D", "NCM"), "ex_tipi": ("D", "NCM"),
|
|
"st_withheld": ("D", "GROUP"), "st_status": ("D", "GROUP"), "icms_treatment": ("D", "GROUP"), "product_role": ("D", "GROUP"), "ncm": ("E", "PRODUCT"), "duplicate_code": ("E", "PRODUCT")}
|
|
QUESTION = {"origin": "Qual a origem da mercadoria (0-8) dos produtos {scope}? (nenhuma evidencia no ERP; o digito do CST e apenas preenchimento)",
|
|
"pis_cofins_profile": "Qual o perfil PIS/COFINS da empresa (CST de entrada/saida e aliquotas)? JSON com pis_in_cst, pis_out_cst, pis_rate, cofins_in_cst, cofins_out_cst, cofins_rate",
|
|
"entry_scope": "As compras de mercadoria desta empresa sao predominantemente INTRA (dentro da UF) ou INTER (outra UF)?",
|
|
"icms_rate": "Qual a aliquota de ICMS padrao desta empresa (regime normal)?",
|
|
"cest": "Qual o CEST correto para o NCM {scope} (ou NONE quando o produto nao esta sujeito)?",
|
|
"ex_tipi": "Qual o EX da TIPI aplicavel ao NCM {scope} (ou NONE)?",
|
|
"st_withheld": "Os produtos {scope} sujeitos a ST tem o ICMS-ST ja retido na compra (Y/N)?",
|
|
"st_status": "Os produtos {scope} estao sujeitos a ST (SUBJECT/NOT_SUBJECT)? O ERP declara ST retida mas a tabela oficial nao lista o NCM",
|
|
"icms_treatment": "Qual o tratamento de ICMS dos produtos {scope}?",
|
|
"product_role": "Os produtos {scope} sao revenda, producao propria, insumo ou servico?",
|
|
"ncm": "Qual o NCM correto do produto {scope} (o atual nao existe na tabela oficial)?",
|
|
"duplicate_code": "O CODIGO {scope} aparece em 2 linhas fisicas (violacao de PK na copia); corrigir a integridade do banco (gfix/validate) antes de classificar"}
|
|
WHY = {"origin": "CST/CSOSN e a origem do NF-e dependem dela; nao pode ser presumida 0", "pis_cofins_profile": "PIS/COFINS fora dos regimes especiais das tabelas SPED 4.3.x", "entry_scope": "CFOP de entrada (1xxx vs 2xxx)",
|
|
"icms_rate": "aliquota interna", "cest": "CEST oficial para o NCM tem varios candidatos ou o CEST do ERP nao consta da lista oficial", "ex_tipi": "varias linhas TIPI com aliquotas diferentes para o NCM",
|
|
"st_withheld": "ST comprovada pela tabela BA mas retencao desconhecida", "st_status": "conflito entre declaracao do ERP e tabela oficial", "icms_treatment": "CST de ICMS nao derivavel", "product_role": "revenda/producao define CFOP e IPPT",
|
|
"ncm": "NCM invalido bloqueia IPI/EX/CEST/ST/CFOP/CSOSN", "duplicate_code": "UPDATE por CODIGO exige rowcount 1"}
|
|
agg = defaultdict(lambda: {"products": set(), "cells": 0, "fields": defaultdict(int), "examples": []})
|
|
for n in DBS:
|
|
snap = pickle.load(gzip.open(ST / f"{n}_before.pkl.gz", "rb")); cols = snap["columns"]; rows = snap["rows"]
|
|
for l in gzip.open(ST / f"{n}_second.ndjson.gz", "rt", encoding="utf-8"):
|
|
r = json.loads(l)
|
|
if r["status"] != "MANUAL_INPUT_REQUIRED":
|
|
continue
|
|
k = r["input_key"]; cat, scope = CAT[k]; row = rows.get(r["codigo"]); g = (row[cols.index("CODGRUPO")] if row else None); ncm = (row[cols.index("CLASSIFICACAO_FISCAL")] if row else None)
|
|
sk = {"COMPANY": "", "GROUP": g or "", "NCM": ncm or "", "PRODUCT": r["codigo"]}[scope]
|
|
a = agg[(n, k, scope, sk)]; a["products"].add(r["row_key"]); a["cells"] += 1; a["fields"][r["field"]] += 1
|
|
if len(a["examples"]) < 3 and row: a["examples"].append(row[cols.index("PRODUTO")])
|
|
out = []
|
|
for (n, k, scope, sk), a in agg.items():
|
|
cat = CAT[k][0]
|
|
opts = sorted(ENUMS[k]) if k in ENUMS else ("JSON" if k == "pis_cofins_profile" else "7 digits or NONE" if k == "cest" else "8 digits" if k == "ncm" else "free")
|
|
who = "toda a empresa" if scope == "COMPANY" else f"{scope} {sk}"
|
|
out.append({"DATABASE": n, "COMPANY": n, "SCOPE": scope, "GROUP": sk if scope == "GROUP" else "", "NCM": sk if scope == "NCM" else "", "FIELD": "|".join(sorted(a["fields"])), "KEY": k, "CATEGORY": cat,
|
|
"QUESTION": QUESTION[k].format(scope=who), "VALID_OPTIONS": opts, "PRODUCTS_AFFECTED": len(a["products"]), "WHY_NEEDED": WHY[k], "WHAT_IT_WILL_RESOLVE": f"{a['cells']} celulas", "CELLS": a["cells"],
|
|
"EXAMPLES": "; ".join(a["examples"])})
|
|
out.sort(key=lambda x: -x["CELLS"])
|
|
for i, q in enumerate(out, 1): q["QUESTION_ID"] = f"Q{i:04d}"
|
|
(ROOT / "docs" / "LOCAL_FISCAL_CONTEXT_QUESTIONS_V2.json").write_text(json.dumps({"artifact": "LOCAL_FISCAL_CONTEXT_QUESTIONS_V2", "total_questions": len(out), "total_cells": sum(q["CELLS"] for q in out), "questions": out}, ensure_ascii=False, indent=1), encoding="utf-8")
|
|
hdr = ["QUESTION_ID", "DATABASE", "COMPANY", "SCOPE", "GROUP", "NCM", "FIELD", "KEY", "CATEGORY", "QUESTION", "VALID_OPTIONS", "PRODUCTS_AFFECTED", "WHY_NEEDED", "WHAT_IT_WILL_RESOLVE", "CELLS", "EXAMPLES", "VALUE"]
|
|
with open(ROOT / "docs" / "LOCAL_FISCAL_CONTEXT_QUESTIONS_V2.csv", "w", newline="", encoding="utf-8-sig") as f:
|
|
w = csv.DictWriter(f, fieldnames=hdr, extrasaction="ignore"); w.writeheader()
|
|
for q in out: w.writerow({**q, "VALID_OPTIONS": q["VALID_OPTIONS"] if isinstance(q["VALID_OPTIONS"], str) else "|".join(q["VALID_OPTIONS"]), "VALUE": ""})
|
|
from collections import Counter
|
|
print(len(out), sum(q["CELLS"] for q in out), dict(Counter(q["CATEGORY"] for q in out)), dict(Counter(q["KEY"] for q in out)))
|
|
print([(q["QUESTION_ID"], q["DATABASE"], q["KEY"], q["CELLS"]) for q in out[:8]])
|