gti
7e83d17533
field_contract.resolve_product was only exercised by its own unit tests. /v1/products/resolve now calls it for every product whose NCM resolves, merging CFOP/CST/CSOSN/PIS-COFINS/CST_IPI/SITUACAO_TRIBUTARIA/IAT/IPPT decisions alongside CLASSIFICACAO_FISCAL/CEST/IPI - reusing the same engine, the same FiscalContextEntry table the blocker-resolution workflow already writes to, no duplicated rules. Closes three real transport/discovery gaps found while wiring it in: - Agent heartbeat never sent company_crt (env-var UF only) or company_contribuinte_ipi, so Installation.company_crt stayed permanently empty even though company_context.resolve() already proves both from C000004/EMISSOR for every coverage job. Resolved once, cached, retried only while unresolved. Migration 0032 adds Installation.company_contribuinte_ipi (NULL=unknown, never coerced to "N"). - The Agent's enrichment payload only ever reported 3 old fields (CLASSIFICACAO_FISCAL/CEST/IPI) as "current", so the Hub could never see the product's own CFOP/CST/CSOSN/... nor the classification columns (TIPO/PROD_COZINHA/PRODUTO_SERVICO/CODGRUPO) infer_role needs - separated into a read-only CURRENT_REPORT_FIELDS list, the write allowlist (PRODUCT_ENRICHMENT_WRITE_V1) untouched. - Added a targeted, product_code-scoped evidence query (firebird_reader.read_product_evidence: recipes/composed/purchases/stock, indexed WHERE, never a full-table scan) so st_withheld/product_role can be proven for a single ad-hoc resolve, not only during a full sync job. Verified read-only against the real database (D:\PROJETOS\GTISISTEMAS\Banco\GTI.FDB): CRT=1, CONTRIBUINTE_IPI=N, company_context.resolve() reaches HIGH confidence; infer_role correctly resolved RESALE/OWN_PRODUCTION for 5/8 sample real products with zero operator questions. 1121 tests passed (tests + central/tests + agent/tests), 0 failed. Version 1.3.7 (single GTI_FISCAL_VERSION shared by Hub and Agent per central/app/version.py's own design - never independently versioned).
72 lines
3.9 KiB
Python
72 lines
3.9 KiB
Python
"""Validation of operator fiscal-context entries (facts the ERP does not know). Pure; shared by the Hub API, the importer CLI and the resolver."""
|
|
import json
|
|
|
|
SCOPES = ("COMPANY", "GROUP", "NCM", "PRODUCT")
|
|
ENUMS = {
|
|
"product_role": {"RESALE", "OWN_PRODUCTION", "INPUT", "SERVICE"},
|
|
"st_status": {"SUBJECT", "NOT_SUBJECT"},
|
|
"st_withheld": {"Y", "N"},
|
|
"icms_treatment": {"TAXED", "EXEMPT", "NOT_TAXED", "ST_WITHHELD", "SUSPENDED", "DEFERRED"},
|
|
# MIXED_OR_DYNAMIC: purchases are not predominantly intra- or interstate, so no single
|
|
# static entry CFOP is fiscally correct for all operations (section 0, AI semantic layer V1).
|
|
"entry_scope": {"INTRA", "INTER", "MIXED_OR_DYNAMIC"},
|
|
"rounding": {"A", "T"},
|
|
"piscofins": {"N", "S"},
|
|
"incidence": {"NENHUMA"},
|
|
"origin": {str(i) for i in range(9)},
|
|
"store_ipi_profile": {"POPULATE_REFERENCE_RATE", "LEAVE_NOT_APPLICABLE"},
|
|
# C000004.CONTRIBUINTE_IPI is not yet sent at heartbeat (no Agent-side reader exists for it): until that auto-discovery lands, this is the
|
|
# one-time operator answer field_contract's CST_IPI rule needs - never assumed "N" just because the Agent hasn't reported it.
|
|
"contribuinte_ipi": {"S", "N"},
|
|
}
|
|
PROFILE_KEYS = {"pis_in_cst", "pis_out_cst", "pis_rate", "cofins_in_cst", "cofins_out_cst", "cofins_rate"}
|
|
FREE_KEYS = {"ncm", "cest", "ex_tipi", "icms_rate", "credit_rate", "cst_ipi", "service_cfop", "pis_cofins_profile"}
|
|
ALLOWED_KEYS = set(ENUMS) | FREE_KEYS
|
|
PRODUCT_ONLY = {"ncm"}
|
|
COMPANY_ONLY = {"store_ipi_profile", "contribuinte_ipi"}
|
|
NCM_KEYS = {"cest", "ex_tipi"} # tied to the NCM tables: answered per NCM or per product # an NCM answer is always about one product
|
|
|
|
|
|
def validate_entry(entry):
|
|
"""Returns (normalized_entry, error). error is None when valid."""
|
|
scope = str(entry.get("scope", "")).strip().upper()
|
|
key = str(entry.get("key", "")).strip()
|
|
value = entry.get("value")
|
|
value = "" if value is None else str(value).strip()
|
|
scope_key = str(entry.get("scope_key") or "").strip()
|
|
if scope not in SCOPES:
|
|
return None, f"scope must be one of {SCOPES}"
|
|
if key not in ALLOWED_KEYS:
|
|
return None, f"unknown key {key!r}"
|
|
if scope != "COMPANY" and not scope_key:
|
|
return None, "scope_key is required for GROUP/NCM/PRODUCT scope"
|
|
if key in COMPANY_ONLY and scope != "COMPANY":
|
|
return None, f"{key} is a COMPANY-scope answer"
|
|
if key in PRODUCT_ONLY and scope != "PRODUCT":
|
|
return None, f"{key} is a PRODUCT-scope answer"
|
|
if key in NCM_KEYS and scope not in ("PRODUCT", "NCM"):
|
|
return None, f"{key} can only be answered for a PRODUCT or an NCM"
|
|
if not value:
|
|
return None, "value is empty"
|
|
if key in ENUMS and value.upper() not in ENUMS[key] and value not in ENUMS[key]:
|
|
return None, f"{key} must be one of {sorted(ENUMS[key])}"
|
|
if key in ENUMS:
|
|
value = value.upper() if key != "origin" else value
|
|
if key == "ncm" and not (value.isdigit() and len(value) == 8):
|
|
return None, "ncm must be 8 digits"
|
|
if key == "cest" and value.upper() != "NONE" and not (value.replace(".", "").isdigit() and len(value.replace(".", "")) == 7):
|
|
return None, "cest must be 7 digits or NONE"
|
|
if key in ("icms_rate", "credit_rate"):
|
|
try:
|
|
float(value.replace(",", "."))
|
|
except ValueError:
|
|
return None, f"{key} must be numeric"
|
|
if key == "pis_cofins_profile":
|
|
try:
|
|
prof = json.loads(value)
|
|
except ValueError:
|
|
return None, "pis_cofins_profile must be JSON"
|
|
if not isinstance(prof, dict) or not prof or set(prof) - PROFILE_KEYS:
|
|
return None, f"pis_cofins_profile keys must be a non-empty subset of {sorted(PROFILE_KEYS)}"
|
|
return {"scope": scope, "scope_key": scope_key if scope != "COMPANY" else "", "key": key, "value": value, "reason": (entry.get("reason") or "").strip() or None}, None
|