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

fix: product_role never infers OWN_PRODUCTION from a single COMPOSICAO occurrence

Browse Source 
Missão 4 / Bloco A.1 (prioridade máxima): central.app.field_contract.infer_role
treated ANY C000032.TIPO=COMPOSICAO row for a product as proof of own
production, with no ratio check - a beer sold 19 times and used once as a
recipe component (Agent 2, real data) was classified OWN_PRODUCTION from
that single occurrence, proposing to change its CFOP and IPPT.

New deterministic threshold in field_contract.py: COMPOSICAO must be a
majority (>=70%) of that product's own C000032 movement history, with at
least 5 total events, to count as own-production evidence. Below 30% the
evidence is discarded entirely (falls through to other role evidence,
e.g. TIPO=VENDA -> RESALE). Between 30% and 70% (with enough events to be
meaningful) is genuinely conflicting evidence -> UNKNOWN, never guessed
either way. RECEITA_INGREDIENTES/PROD_COZINHA-based evidence (a direct
recipe-table linkage, not a statistical ratio) is untouched.

agent/src/gti_fiscal_agent/firebird_reader.py now counts COMPOSICAO vs other
C000032 rows per product (read_product_evidence: single indexed query;
read_evidence: one grouped query for the whole catalog) into a new
ev["composed_ratio"] key, alongside the existing "composed" set (kept for
other, non-role consumers like coverage.py). central/app/enrichment/api.py's
sanitize_product_evidence and fiscal_completion.py's empty-ev fallback both
carry the new key through.

New tests (central/tests/test_field_contract.py) cover the 5 real cases named
in the mission: the beer (1 vs 19, 5% - discarded), a second beer (3 vs 4,
42.9% - MIXED/UNKNOWN), the liqueur (34 vs 22, 60.7% - still MIXED/UNKNOWN,
below the 70% bar), the juice with PROD_COZINHA=1+RECEITA_INGREDIENTES
(unaffected, stays OWN_PRODUCTION), and the tilapia/corvina fish (no
composed evidence at all, stays RESALE) - plus a strong 90% majority case
and an insufficient-sample (<5 events) case. A new end-to-end test in
agent/tests/test_cfop_ippt_auto_apply.py (committed separately, see the
feature/cfop-ippt-write branch) proves the weak-evidence beer never reaches
select_candidates as a write candidate, independent of any write flag.

Re-running the dry-run over Agent 2's real database (GTI.FDB, 527 products)
after this fix: AUTO_FILL_SAFE CFOP/IPPT cells drop from 12 to 3 (the two
fish CFOP fills + the juice's IPPT, all with strong, non-statistical
evidence) - docs/runs/GTI_dryrun_universe_summary_after_A1.json. This
corrects the Missão 3 report's 12/9(downgraded)/4(sustained) manual
judgment call with a deterministic, tested engine rule.

No schema/migration change. Zero Firebird writes (read-only analysis only).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
gti
2026-09-28 10:53:10 -03:00
co-authored by Claude Sonnet 5
parent a3ced77fc5
commit 1abb7e516e
8 changed files with 379 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/src/gti_fiscal_agent/firebird_reader.py
+21 -7
View File
@@ -96,18 +96,26 @@ def read_product_evidence(conn, code):
for t, f in cur.fetchall():
cols[t].add(f)
except Exception:
return {"recipes": set(), "composed": set(), "purchases": {}, "stock_entry": {}, "stock_in": {}}
return {"recipes": set(), "composed": set(), "composed_ratio": {}, "purchases": {}, "stock_entry": {}, "stock_in": {}}
has = lambda t, *c: t in cols and all(x in cols[t] for x in c) # noqa: E731
s = lambda v: None if v is None else str(v).strip() # noqa: E731
ev = {"recipes": set(), "composed": set(), "purchases": {}, "stock_entry": {}, "stock_in": {}}
ev = {"recipes": set(), "composed": set(), "composed_ratio": {}, "purchases": {}, "stock_entry": {}, "stock_in": {}}
code_padded = str(code).zfill(6)
if has("RECEITA_INGREDIENTES", "CODRECEITA"):
cur.execute("SELECT FIRST 1 CODRECEITA FROM RECEITA_INGREDIENTES WHERE CODRECEITA = ?", (code_padded,))
if cur.fetchone():
ev["recipes"] = {code_padded}
if has("C000032", "CODPRODUTO", "TIPO"):
cur.execute("SELECT FIRST 1 CODPRODUTO FROM C000032 WHERE CODPRODUTO = ? AND TIPO = 'COMPOSICAO'", (code,))
if cur.fetchone():
# Counted, never a bare "ever happened" check: a single COMPOSICAO row among many VEND rows
# (e.g. a beer occasionally used as a recipe component) is not proof the product IS own
# production - central.app.field_contract.infer_role needs the ratio to judge that.
cur.execute("SELECT TRIM(TIPO), COUNT(*) FROM C000032 WHERE CODPRODUTO = ? GROUP BY 1", (code,))
counts = {(t or ""): n for t, n in cur.fetchall()}
composed_n = counts.get("COMPOSICAO", 0)
other_n = sum(n for t, n in counts.items() if t != "COMPOSICAO")
if composed_n or other_n:
ev["composed_ratio"][s(code)] = {"composed": composed_n, "other": other_n}
if composed_n:
ev["composed"] = {s(code)}
if has("C000088", "CODPRODUTO", "ICMS_RETIDO", "ICMS_VALORRETIDO", "SUB_VALOR"):
cur.execute("SELECT ICMS_RETIDO, ICMS_VALORRETIDO, SUB_VALOR FROM C000088 WHERE CODPRODUTO = ?", (code,))
@@ -141,13 +149,19 @@ def read_evidence(env=None, selected_company_code=None):
cols[t].add(f)
has = lambda t, *c: t in cols and all(x in cols[t] for x in c) # noqa: E731
s = lambda v: None if v is None else str(v).strip() # noqa: E731
ev = {"recipes": set(), "composed": set(), "purchases": defaultdict(list), "stock_entry": {}, "stock_in": {}, "company_default_csosn": None}
ev = {"recipes": set(), "composed": set(), "composed_ratio": {}, "purchases": defaultdict(list), "stock_entry": {}, "stock_in": {}, "company_default_csosn": None}
if has("RECEITA_INGREDIENTES", "CODRECEITA"):
cur.execute("SELECT CODRECEITA FROM RECEITA_INGREDIENTES WHERE CODRECEITA IS NOT NULL")
ev["recipes"] = {str(r[0]).zfill(6) for r in cur.fetchall()}
if has("C000032", "CODPRODUTO", "TIPO"):
cur.execute("SELECT DISTINCT CODPRODUTO FROM C000032 WHERE TIPO = 'COMPOSICAO'")
ev["composed"] = {s(r[0]) for r in cur.fetchall()}
# Counted per product, never a bare "ever happened" check - see read_product_evidence for why
# (a single COMPOSICAO row among many VEND rows is not proof of own production).
cur.execute("SELECT CODPRODUTO, TRIM(TIPO), COUNT(*) FROM C000032 GROUP BY 1, 2")
counts = defaultdict(lambda: {"composed": 0, "other": 0})
for cod, tipo, n in cur.fetchall():
counts[s(cod)]["composed" if tipo == "COMPOSICAO" else "other"] += n
ev["composed_ratio"] = dict(counts)
ev["composed"] = {code for code, c in counts.items() if c["composed"]}
if has("C000088", "CODPRODUTO", "ICMS_RETIDO", "ICMS_VALORRETIDO", "SUB_VALOR"):
cur.execute("SELECT CODPRODUTO, ICMS_RETIDO, ICMS_VALORRETIDO, SUB_VALOR FROM C000088")
for cod, ret, vr, sv in cur.fetchall():
agent/tests/test_enrichment.py
+1 -1
View File
@@ -136,7 +136,7 @@ def test_current_report_includes_fiscal_completion_and_classification_columns(tm
assert sent["CSOSN"] == "102" and sent["TIPO"] == "VENDA"
for f in E.FISCAL_COMPLETION_FIELDS + E.CLASSIFICATION_CONTEXT_FIELDS:
assert f in sent, f
assert hub.resolve_calls[0]["ev"] == {"recipes": [], "composed": [], "purchases": {}, "stock_entry": {}, "stock_in": {}} # targeted evidence rides along (empty here: sqlite has no RDB$RELATION_FIELDS)
assert hub.resolve_calls[0]["ev"] == {"recipes": [], "composed": [], "composed_ratio": {}, "purchases": {}, "stock_entry": {}, "stock_in": {}} # targeted evidence rides along (empty here: sqlite has no RDB$RELATION_FIELDS)
def test_existing_value_is_never_overwritten(env):
agent/tests/test_firebird_reader_product_evidence.py
+2 -2
View File
@@ -25,7 +25,7 @@ def test_no_firebird_system_table_degrades_to_no_evidence_never_raises():
gracefully to "no evidence", exactly like a missing table would, never raise."""
db = sqlite3.connect(":memory:")
ev = read_product_evidence(db, "000001")
assert ev == {"recipes": set(), "composed": set(), "purchases": {}, "stock_entry": {}, "stock_in": {}}
assert ev == {"recipes": set(), "composed": set(), "composed_ratio": {}, "purchases": {}, "stock_entry": {}, "stock_in": {}}
def test_targeted_lookup_is_scoped_to_the_one_product_code():
@@ -82,4 +82,4 @@ def test_targeted_lookup_is_scoped_to_the_one_product_code():
ev = read_product_evidence(conn, "000013")
assert ev["stock_entry"] == {"000013": True} and ev["stock_in"] == {"000013": True}
ev = read_product_evidence(conn, "999999")
assert ev == {"recipes": set(), "composed": set(), "purchases": {}, "stock_entry": {}, "stock_in": {}}
assert ev == {"recipes": set(), "composed": set(), "composed_ratio": {}, "purchases": {}, "stock_entry": {}, "stock_in": {}}