gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
194 lines
13 KiB
Python
194 lines
13 KiB
Python
"""Agent-facing product enrichment routes (authenticated as the Agent; the Agent never talks to external providers)."""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy import select
|
|
|
|
from ..blocked_products import upsert_blocker
|
|
from ..config import settings
|
|
from ..field_contract import WRITE_FIELDS
|
|
from ..models import Installation, ProductEnrichment, ProductFieldAudit
|
|
from . import status as enrichment_status, textnorm
|
|
from .placeholders import is_placeholder
|
|
from .providers import build_providers
|
|
from .quota import ProviderGateway
|
|
from .resolver import FIELDS, resolve
|
|
|
|
CLASSIFICATION_CONTEXT_FIELDS = ("CODGRUPO", "TIPO", "PROD_COZINHA", "PRODUTO_SERVICO") # read-only evidence for field_contract.infer_role; never written
|
|
CURRENT_FIELDS = tuple(dict.fromkeys(FIELDS + tuple(WRITE_FIELDS) + CLASSIFICATION_CONTEXT_FIELDS)) # everything resolve() may read from the Agent's `current`: NCM/CEST/IPI + every fiscal-completion field + role evidence
|
|
|
|
|
|
def sanitize_product_evidence(raw) -> dict | None:
|
|
"""Bounds the Agent's targeted, single-product evidence (firebird_reader.read_product_evidence) before it reaches field_contract: shape-only
|
|
validation (never trusts an authenticated Agent to send something huge or malformed for a single-product payload)."""
|
|
if not isinstance(raw, dict):
|
|
return None
|
|
out = {"recipes": set(), "composed": set(), "composed_ratio": {}, "purchases": {}, "stock_entry": {}, "stock_in": {}}
|
|
for key in ("recipes", "composed"):
|
|
if isinstance(raw.get(key), list):
|
|
out[key] = {str(v) for v in raw[key][:8]}
|
|
if isinstance(raw.get("composed_ratio"), dict):
|
|
for code, counts in list(raw["composed_ratio"].items())[:8]:
|
|
if isinstance(counts, dict) and isinstance(counts.get("composed"), int) and isinstance(counts.get("other"), int):
|
|
out["composed_ratio"][str(code)] = {"composed": max(0, counts["composed"]), "other": max(0, counts["other"])}
|
|
if isinstance(raw.get("purchases"), dict):
|
|
for code, docs in list(raw["purchases"].items())[:8]:
|
|
if isinstance(docs, list):
|
|
out["purchases"][str(code)] = [d for d in docs[:32] if isinstance(d, dict)]
|
|
for key in ("stock_entry", "stock_in"):
|
|
if isinstance(raw.get(key), dict):
|
|
out[key] = {str(k): bool(v) for k, v in list(raw[key].items())[:8]}
|
|
return out
|
|
|
|
|
|
PROVIDERS_OVERRIDE = None # tests inject fake providers here (never the network)
|
|
PENDING_STATES = {"PENDING_PROVIDER_QUOTA", "PENDING_PROVIDERS_OFFLINE", "PENDING_TIME_BUDGET"}
|
|
REASONS = { # human PT-BR reasons shown through the existing blockers (category ENRICHMENT); never internal jargon
|
|
"INVALID_GTIN": "Código de barras (GTIN) inválido: o produto não foi consultado em fontes externas.",
|
|
"AMBIGUOUS": "Descrição genérica demais para identificar o produto com segurança; informe o código de barras.",
|
|
"NOT_FOUND": "Produto não encontrado nas fontes consultadas; preencha os campos fiscais manualmente.",
|
|
"CONFLICT": "As fontes divergem sobre a classificação deste produto; nenhuma alteração foi feita.",
|
|
"SINGLE_WEAK_SOURCE": "Apenas uma fonte identificou o produto; a classificação foi apenas sugerida.",
|
|
"PENDING_PROVIDER_QUOTA": "Limite diário das fontes de produtos atingido; a consulta será repetida automaticamente.",
|
|
"PENDING_PROVIDERS_OFFLINE": "Fontes de produtos indisponíveis no momento; a consulta será repetida automaticamente.",
|
|
}
|
|
|
|
|
|
def _aware(dt):
|
|
return dt.replace(tzinfo=timezone.utc) if dt is not None and dt.tzinfo is None else dt
|
|
|
|
|
|
def features_of(installation) -> dict:
|
|
return {"corrector_enabled": bool(installation.corrector_enabled), "enrichment_mode": installation.enrichment_mode, "features_configured": bool(installation.features_configured)}
|
|
|
|
|
|
def router(principal, providers_factory=None):
|
|
routes = APIRouter(prefix="/products")
|
|
holder = {}
|
|
|
|
def providers():
|
|
if PROVIDERS_OVERRIDE is not None:
|
|
return PROVIDERS_OVERRIDE
|
|
if providers_factory is not None:
|
|
return providers_factory()
|
|
if "p" not in holder:
|
|
holder["p"] = build_providers(settings)
|
|
return holder["p"]
|
|
|
|
@routes.post("/resolve")
|
|
def resolve_product(payload: dict, identity=Depends(principal)):
|
|
session, agent = identity
|
|
try:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
|
|
if installation is None:
|
|
raise HTTPException(409, "installation not registered")
|
|
features = features_of(installation)
|
|
if not features["features_configured"] or installation.enrichment_mode == "DISABLED":
|
|
return {"state": "FEATURE_DISABLED", "features": features, "fields": {}}
|
|
code = str(payload.get("product_code") or "")[:64]
|
|
if not code:
|
|
raise HTTPException(422, "product_code required")
|
|
description, current = str(payload.get("description") or "")[:255], payload.get("current") if isinstance(payload.get("current"), dict) else {}
|
|
started = time.monotonic()
|
|
result = resolve(session, ProviderGateway(session, providers()), gtin_raw=payload.get("gtin"), description=description, current={k: current.get(k) for k in CURRENT_FIELDS},
|
|
client_label=f"installation:{installation.id}", budget_s=settings.enrichment_time_budget_seconds,
|
|
product_code=code, installation=installation, tenant_id=agent.tenant_id, product_evidence=sanitize_product_evidence(payload.get("ev")))
|
|
latency = int((time.monotonic() - started) * 1000)
|
|
row = session.scalar(select(ProductEnrichment).where(ProductEnrichment.installation_id == installation.id, ProductEnrichment.product_code == code))
|
|
if row is None:
|
|
row = ProductEnrichment(tenant_id=agent.tenant_id, installation_id=installation.id, product_code=code)
|
|
session.add(row)
|
|
for field in (row.human_overridden or {}): # a human decision (even deleting the value) is final for this product
|
|
if field in result["fields"]:
|
|
result["fields"][field] = {"status": "HUMAN_OVERRIDDEN", "value": None, "rule": None, "kb_status": None, "evidence": [], "reason": "valor alterado por uma pessoa; nunca mais preenchido automaticamente"}
|
|
pending = result["state"] in PENDING_STATES
|
|
for name, decision in result["fields"].items(): # explain in the audit trail that a technical placeholder was treated as "not classified yet"
|
|
if is_placeholder(name, current.get(name)):
|
|
decision["technical_placeholder"] = str(current.get(name))
|
|
row.gtin, row.description, row.latency_ms = result.get("gtin"), description, latency
|
|
row.attempts = (row.attempts or 0) + 1
|
|
row.ui_status, row.reason_code = enrichment_status.classify(result, has_gtin=bool(result.get("gtin")), specific_name=textnorm.is_specific(description))
|
|
row.last_attempt_at, row.reevaluate = datetime.now(timezone.utc), False
|
|
row.status = "PENDING" if pending else "RESOLVED"
|
|
row.decisions = {"state": result["state"], "fields": {k: {kk: vv for kk, vv in v.items() if kk != "evidence"} for k, v in result["fields"].items()}}
|
|
row.next_attempt_at = _aware(result["retry_at"]) if pending else None
|
|
if pending and row.next_attempt_at is None:
|
|
row.next_attempt_at = datetime.now(timezone.utc).replace(microsecond=0)
|
|
session.commit()
|
|
reason = REASONS.get(result["state"])
|
|
if reason:
|
|
upsert_blocker(session, agent.tenant_id, {"company": installation.hostname or "", "product_code": code, "gtin": result.get("gtin"), "description": description, "blocker_type": "ENRICHMENT_" + result["state"],
|
|
"field": "ENRICHMENT", "reason_code": result["state"], "evidence_summary": {"message": reason}, "input_fingerprint": f"{result.get('gtin')}|{description}"}, installation_id=installation.id)
|
|
result.update({"features": features, "latency_ms": latency, "mode": installation.enrichment_mode})
|
|
result["retry_at"] = result["retry_at"].isoformat() if result["retry_at"] else None
|
|
return result
|
|
finally:
|
|
session.close()
|
|
|
|
@routes.post("/enrichment/applied")
|
|
def applied(payload: dict, identity=Depends(principal)):
|
|
"""The Agent reports what it wrote / refused. The Hub re-checks that the feature is still ENABLED and audits every field."""
|
|
session, agent = identity
|
|
try:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
|
|
if installation is None:
|
|
raise HTTPException(409, "installation not registered")
|
|
code = str(payload.get("product_code") or "")[:64]
|
|
row = session.scalar(select(ProductEnrichment).where(ProductEnrichment.installation_id == installation.id, ProductEnrichment.product_code == code))
|
|
if row is None:
|
|
raise HTTPException(404, "unknown product enrichment")
|
|
versions = payload.get("dataset_versions") if isinstance(payload.get("dataset_versions"), dict) else {}
|
|
applied_fields, overridden = dict(row.fields_applied or {}), dict(row.human_overridden or {})
|
|
for item in (payload.get("fields") or [])[:32]:
|
|
if not isinstance(item, dict) or item.get("field") not in FIELDS:
|
|
continue
|
|
outcome = str(item.get("outcome") or "APPLIED")[:24]
|
|
session.add(ProductFieldAudit(tenant_id=agent.tenant_id, installation_id=installation.id, agent_id=agent.id, product_code=code, gtin=row.gtin, description_snapshot=row.description, field=item["field"],
|
|
old_value=(str(item["old"])[:64] if item.get("old") is not None else None), new_value=(str(item["new"])[:64] if item.get("new") is not None else None), rule=str(item.get("rule") or "")[:64] or None,
|
|
kb_status=str(item.get("kb_status") or "")[:16] or None, evidence=item.get("evidence") if isinstance(item.get("evidence"), list) else [], dataset_versions=versions,
|
|
authorization_origin=f"enrichment_mode:{installation.enrichment_mode}", job_id=payload.get("job_id") if isinstance(payload.get("job_id"), int) else None, outcome=outcome,
|
|
reason=str(item.get("reason") or "")[:96] or None))
|
|
if outcome == "APPLIED":
|
|
applied_fields[item["field"]] = item.get("new")
|
|
elif outcome == "HUMAN_OVERRIDDEN":
|
|
overridden[item["field"]] = True
|
|
row.fields_applied, row.human_overridden = applied_fields, overridden
|
|
ui, why = enrichment_status.after_write([i for i in (payload.get("fields") or []) if isinstance(i, dict)], installation.enrichment_mode)
|
|
if ui:
|
|
row.ui_status, row.reason_code = ui, why
|
|
row.reevaluate = False
|
|
row.status = "APPLIED" if applied_fields and not overridden else ("PARTIAL" if applied_fields else row.status if row.status != "PENDING" else "RESOLVED")
|
|
row.updated_at = datetime.now(timezone.utc)
|
|
session.commit()
|
|
return {"ok": True, "status": row.status, "mode": installation.enrichment_mode}
|
|
finally:
|
|
session.close()
|
|
|
|
@routes.get("/enrichment/pending")
|
|
def pending_reevaluation(identity=Depends(principal)):
|
|
"""Tracked products flagged for a new evaluation (panel button, REPROCESS/SYNC, KB decision). The Agent polls this cheaply and re-runs resolve for each code."""
|
|
session, agent = identity
|
|
try:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
|
|
if installation is None or not installation.features_configured or installation.enrichment_mode == "DISABLED":
|
|
return {"codes": []}
|
|
codes = list(session.scalars(select(ProductEnrichment.product_code).where(ProductEnrichment.installation_id == installation.id, ProductEnrichment.reevaluate.is_(True)).order_by(ProductEnrichment.id).limit(25)))
|
|
return {"codes": codes}
|
|
finally:
|
|
session.close()
|
|
|
|
@routes.get("/enrichment/authorization")
|
|
def authorization(identity=Depends(principal)):
|
|
"""Fresh central authorization read by the Agent right before EVERY field write."""
|
|
session, agent = identity
|
|
try:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
|
|
return features_of(installation) if installation is not None else {"corrector_enabled": False, "enrichment_mode": "DISABLED", "features_configured": False}
|
|
finally:
|
|
session.close()
|
|
|
|
return routes
|