gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
100 lines
8.1 KiB
Python
100 lines
8.1 KiB
Python
"""Contexto Fiscal: the ADMIN-only, human-language view of the handful of fiscal decisions field_contract.resolve_product genuinely cannot
|
|
derive from the ERP alone (see central/app/enrichment/fiscal_completion.py). Reuses FiscalContextEntry (the SAME table blocker_resolution
|
|
already writes operator answers into) - no new persistence model, no opaque JSON box in the UI.
|
|
|
|
DADO_DE_TESTE=true: Each decision here was audited on test/homologation environments (Agent 2 and lab databases) to validate the fiscal
|
|
engine mechanism. NONE of these findings represent conclusions on any real production client. Real client data will be validated
|
|
separately upon first real client onboarding. Evidence text below is a dated, one-time manual read-only measurement on test data, not a live query."""
|
|
from __future__ import annotations
|
|
|
|
DECISIONS = [
|
|
{
|
|
"key": "icms_treatment", "scope": "COMPANY", "title": "Tratamento padrão de ICMS dos produtos comuns",
|
|
"evidence": "[DADO_DE_TESTE=true: Base de homologação Agent 2] CSOSN padrão do ERP (C000004.ICMS_CSOSN): 102. 487 de 521 produtos cadastrados já usam 102. As exceções existentes "
|
|
"(500 = ICMS-ST retido, em 5 produtos; 101 = com direito a crédito, em 4 produtos; 103, em 1 produto) são preservadas "
|
|
"e nunca sobrescritas - a confirmação abaixo só afeta os produtos com CSOSN em branco (28 hoje). "
|
|
"Dado de laboratório/teste; não constitui conclusão sobre cliente real.",
|
|
"suggestion": "TAXED", "allowed_values": ["TAXED", "EXEMPT", "NOT_TAXED", "ST_WITHHELD", "SUSPENDED", "DEFERRED"],
|
|
"affected_fields": ["CST", "CSOSN"], "confidence": "Média (DADO_DE_TESTE=true: medida em banco de homologação; pendente de validação com cliente real)",
|
|
"measured_at": "2026-09-27", "is_test_data": True, "data_label": "DADO_DE_TESTE=true",
|
|
},
|
|
{
|
|
"key": "pis_cofins_profile", "scope": "COMPANY", "title": "Perfil de PIS/COFINS (código e alíquota de entrada/saída)",
|
|
"evidence": "[DADO_DE_TESTE=true: Auditoria em 5 bancos de laboratório e Agent 2] 34.946 itens de NFC-e emitidos; 3.336 (9,5%) carregam um CST de PIS/COFINS na nota, sempre o código 01 - mas os "
|
|
"mesmos produtos têm os campos de cadastro (código de saída, alíquota) sempre em branco no SGV, incluindo os poucos "
|
|
"produtos que TÊM esses campos preenchidos manualmente (a nota emitida ignora o valor cadastrado). Reprocessado em "
|
|
"2026-09-28 (Missão 2 / Bloco 2.1): os 3.336 itens estão concentrados quase inteiramente em 2022 (3.336 de 3.669 "
|
|
"itens daquele ano; 0% em 2018 e em 2023-2026) - coincide com a mesma janela já sinalizada como anômala em outra "
|
|
"auditoria (drift de identidade de produto). O mesmo padrão de cadastro nunca preenchido na emissão se repete, sem "
|
|
"exceção, nos outros 4 bancos de teste auditados (CONSTRUCAO, MMMODAS, RINCAO CRT=3, UNIAO): 0% de itens com CST de "
|
|
"PIS/COFINS em todo o histórico, em todos os anos. Amostra de laboratório; não é garantia de comportamento de cliente real em produção. "
|
|
"Ver docs/dossie_contador.md e docs/runs/BLOCO2_PIS_COFINS_CROSS_DB_V1.json.", "suggestion": None,
|
|
"affected_fields": ["COD_ENTRADA_PIS", "COD_SAIDA_PIS", "ALIQ_PIS", "COD_ENTRADA_COFINS", "COD_SAIDA_COFINS", "ALIQ_COFINS"],
|
|
"confidence": "Baixa para um perfil único da empresa (DADO_DE_TESTE=true: considerar NOT_APPLICABLE ou OPERATION_TIME; pendente de confirmação contábil no piloto real)",
|
|
"measured_at": "2026-09-28", "is_test_data": True, "data_label": "DADO_DE_TESTE=true",
|
|
"value_hint": 'JSON de perfil (ex.: {"pis_out_cst":"49","pis_in_cst":"98","pis_rate":"0","cofins_out_cst":"49","cofins_in_cst":"98","cofins_rate":"0"}) '
|
|
'OU exatamente NOT_APPLICABLE (nunca é lido pelo cadastro) OU OPERATION_TIME (varia por operação) - nunca inferido automaticamente.',
|
|
"dossier_link": "/static/docs/dossie_contador.md",
|
|
},
|
|
{
|
|
"key": "entry_scope", "scope": "COMPANY", "title": "CFOP de entrada padrão (compras dentro/fora do estado)",
|
|
"evidence": "[DADO_DE_TESTE=true: Base de homologação Agent 2] Apenas 12 documentos de compra no ERP de teste para 526 produtos; 11 sem CFOP preenchido, 1 com CFOP intra-estadual. "
|
|
"Evidência de laboratório insuficiente para qualquer sugestão - o CFOP de entrada normalmente é definido pela operação/documento "
|
|
"de compra, não pelo cadastro do produto. Ver docs/dossie_contador.md.", "suggestion": None,
|
|
"affected_fields": ["CFOP_ENTRADA"], "confidence": "Muito baixa (DADO_DE_TESTE=true: recomendação: MIXED_OR_DYNAMIC; validação definitiva apenas com cliente real)",
|
|
"measured_at": "2026-09-27", "allowed_values": ["INTRA", "INTER", "MIXED_OR_DYNAMIC"], "is_test_data": True, "data_label": "DADO_DE_TESTE=true",
|
|
"dossier_link": "/static/docs/dossie_contador.md",
|
|
},
|
|
]
|
|
|
|
|
|
VALID_KEYS = {d["key"] for d in DECISIONS}
|
|
|
|
|
|
def set_fiscal_context_decision(session, *, tenant_id: int, company_key: str, key: str, value: str | None, reason: str, admin, installation_id: int):
|
|
"""Confirm/alter/clear one Contexto Fiscal decision. value=None clears it (the product/group falls back to whatever the engine can
|
|
prove on its own - never a silent default). Reuses the SAME table and validator the blocker-resolution workflow already writes
|
|
through; flags every already-enriched product on this installation for re-evaluation so the new (or cleared) policy takes effect."""
|
|
from datetime import datetime, timezone
|
|
from sqlalchemy import select
|
|
from . import context_keys
|
|
from .audit import record_audit
|
|
from .enrichment.status import mark_for_reevaluation
|
|
from .models import FiscalContextEntry
|
|
|
|
if key not in VALID_KEYS:
|
|
raise ValueError(f"unknown Contexto Fiscal key {key!r}")
|
|
entry = session.scalar(select(FiscalContextEntry).where(FiscalContextEntry.tenant_id == tenant_id, FiscalContextEntry.company_key == company_key,
|
|
FiscalContextEntry.scope == "COMPANY", FiscalContextEntry.scope_key == "", FiscalContextEntry.key == key))
|
|
before = entry.value if entry else None
|
|
if value is None:
|
|
if entry is not None:
|
|
session.delete(entry)
|
|
else:
|
|
normalized, error = context_keys.validate_entry({"scope": "COMPANY", "key": key, "value": value, "reason": reason})
|
|
if error:
|
|
raise ValueError(error)
|
|
if entry is None:
|
|
entry = FiscalContextEntry(tenant_id=tenant_id, company_key=company_key, source="OPERATOR", updated_at=datetime.now(timezone.utc), **normalized)
|
|
session.add(entry)
|
|
else:
|
|
entry.value, entry.reason, entry.source, entry.updated_at = normalized["value"], normalized["reason"], "OPERATOR", datetime.now(timezone.utc)
|
|
flagged = mark_for_reevaluation(session, installation_id)
|
|
record_audit(session, admin_user_id=admin.id, tenant_id=tenant_id, action="FISCAL_CONTEXT_DECISION", target_type="installation", target_id=company_key,
|
|
context={"key": key, "from": before, "to": value, "reason": reason, "operator": admin.display_name or admin.username, "role": admin.external_role or admin.auth_source, "products_flagged": flagged},
|
|
severity="WARNING")
|
|
|
|
|
|
def build_fiscal_context_view(session, tenant_id: int, company_key: str) -> list[dict]:
|
|
from sqlalchemy import select
|
|
from .models import FiscalContextEntry
|
|
current = {(r.scope, r.scope_key, r.key): r for r in session.scalars(
|
|
select(FiscalContextEntry).where(FiscalContextEntry.tenant_id == tenant_id, FiscalContextEntry.company_key == company_key))}
|
|
out = []
|
|
for d in DECISIONS:
|
|
row = current.get((d["scope"], "", d["key"]))
|
|
out.append({**d, "current_value": row.value if row else None, "current_reason": row.reason if row else None,
|
|
"last_changed": row.updated_at.isoformat() if row and getattr(row, "updated_at", None) else None,
|
|
"last_changed_by": row.source if row else None})
|
|
return out
|