diff --git a/agent/scripts/run_agent_service.py b/agent/scripts/run_agent_service.py index 1eb2d98..24eeabd 100644 --- a/agent/scripts/run_agent_service.py +++ b/agent/scripts/run_agent_service.py @@ -36,6 +36,7 @@ from gti_fiscal_agent.pipeline import run_pipeline # noqa: E402 from gti_fiscal_agent.runner import REFERENCE_SCHEMA_V1, product_fingerprint, to_hub_product # noqa: E402 from central.app.operational_state import classify_product_state, summarize_operational_states # noqa: E402 from gti_fiscal_agent.state import AgentState # noqa: E402 +from gti_fiscal_agent import enrichment as _enrichment # noqa: E402 _MAX_BACKOFF_SECONDS = 60 SYNC_JOB_KINDS = ("INITIAL_SYNC", "SYNC_INCREMENTAL", "REPROCESS_ALL") # read-only ERP jobs that refresh coverage @@ -409,6 +410,17 @@ def resolve_installation_id(config, env, state, client): return state.installation_id(str(installation_id)) +def _start_enrichment(config, env, client, stopper): + """New-product enrichment worker (own thread, own SQLite file). It gates ITSELF on the Hub's per-installation features: DISABLED / not configured = zero Firebird activity.""" + if not hasattr(client, "resolve_product"): + return None + store = _enrichment.EnrichmentStore(Path(config.state_db).with_name("enrichment_state.db")) + worker = _enrichment.EnrichmentWorker(hub=client, store=store, read_conn=lambda: _firebird_reader._connect(env, True), write_conn=lambda: open_write_connection(env, config.database_path), + create_reprocess=lambda codes: client.create_job("REPROCESS_PRODUCTS", {"codigos": list(codes), "reason": "ENRICHMENT"})) + threading.Thread(target=worker.run_forever, args=(stopper,), name="product-enrichment", daemon=True).start() + return worker + + def run(config, env, *, max_iterations=None, stopper=None): """`stopper` defaults to a signal-based GracefulStop (SIGINT/SIGTERM), which only works from the main thread. A host that manages its own lifecycle @@ -420,6 +432,8 @@ def run(config, env, *, max_iterations=None, stopper=None): client = HubClient(config.hub_url, config.agent_token) installation_id = resolve_installation_id(config, env, state, client) stopper = stopper if stopper is not None else GracefulStop() + if max_iterations is None: + _start_enrichment(config, env, client, stopper) backoff = 1 was_down = False @@ -436,7 +450,7 @@ def run(config, env, *, max_iterations=None, stopper=None): "agent_version": config.agent_version, "last_job_id": last_job_id, "display_name": env.get("AGENT_DISPLAY_NAME", ""), "customer_code": env.get("CUSTOMER_CODE"), "hostname": env.get("AGENT_HOSTNAME") or os.environ.get("COMPUTERNAME"), "windows_version": env.get("WINDOWS_VERSION"), "company_code": config.company_id, "company_uf": config.company_uf, - "database_path": config.database_path, "capabilities": {"heartbeat": True, "jobs": True, "local_logs": True, "RTC_CCLASSTRIB_WRITE_V1": True}, + "database_path": config.database_path, "capabilities": {"heartbeat": True, "jobs": True, "local_logs": True, "RTC_CCLASSTRIB_WRITE_V1": True, "PRODUCT_ENRICHMENT_WRITE_V1": True}, "fiscal_engine_version": env.get("FISCAL_ENGINE_VERSION", config.agent_version), "dataset_versions": {}, } if hasattr(client, "heartbeat_metadata"): diff --git a/agent/src/gti_fiscal_agent/client.py b/agent/src/gti_fiscal_agent/client.py index 3e76cd8..2cd95c2 100644 --- a/agent/src/gti_fiscal_agent/client.py +++ b/agent/src/gti_fiscal_agent/client.py @@ -41,6 +41,13 @@ class HubClient: response.raise_for_status(); return response.json() def identity(self): return self.get_json("/v1/agents/me/identity") def authorizations(self): return self.get_json("/v1/agents/me/authorizations") + def resolve_product(self, payload): + """Asks the Hub to resolve a NEW product (the Agent never contacts external providers). Bounded: the Hub answers within its own time budget.""" + response = httpx.post(f"{self.url}/v1/products/resolve", json=payload, headers=self._headers(), timeout=20) + response.raise_for_status(); return response.json() + def report_enrichment(self, payload): + response = _resilient(lambda: httpx.post(f"{self.url}/v1/products/enrichment/applied", json=payload, headers=self._headers(), timeout=30)) + response.raise_for_status(); return response.json() def health(self): response = httpx.get(f"{self.url}/health", timeout=10) return response.status_code == 200 diff --git a/agent/src/gti_fiscal_agent/enrichment.py b/agent/src/gti_fiscal_agent/enrichment.py new file mode 100644 index 0000000..c135c30 --- /dev/null +++ b/agent/src/gti_fiscal_agent/enrichment.py @@ -0,0 +1,415 @@ +"""New-product enrichment on the Agent: detect NEW rows of C000025, ask the Hub, write ONLY safe empty fields. + +Safety model (defense in depth; nothing here is switchable locally - the Hub owns the per-installation feature flags): +- Feature gate: nothing touches Firebird unless the Hub says features_configured AND enrichment_mode != DISABLED. DRY_RUN never opens a write connection. +- Scope: only products that appear AFTER the baseline (taken when the function is first active). The historical catalog is never sent anywhere. +- No trigger, no table, no procedure in the SGV database: detection is one `SELECT COUNT(*)` per poll (+ a code diff only when the count moved, and every 60 s as a safety net). +- Typed capability PRODUCT_ENRICHMENT_WRITE_V1: ALLOWED_FIELDS is an explicit allowlist on C000025; the only statement is a fixed UPDATE ... WHERE CODIGO=? AND (field IS NULL OR TRIM(field)=''), + one short transaction per field, followed by a re-read. COD_ORIGEM_MERCADORIA and cClassTrib are NOT in the list (cClassTrib stays with the existing RTC gates). +- Right before EVERY field write: re-read the row, re-check the Hub authorization (fail closed), skip when a human already touched the field. +- Human overrides: after writing, the Agent watches the value; anything different from what it wrote (including a deleted value) marks the field HUMAN_OVERRIDDEN forever, is reported to the Hub + and is never written again by the enrichment (nor re-suggested by the Hub). +""" +from __future__ import annotations + +import json +import logging +import sqlite3 +import time +from datetime import datetime, timezone + +logger = logging.getLogger("gti_fiscal_agent.enrichment") + +CAPABILITY = "PRODUCT_ENRICHMENT_WRITE_V1" +ALLOWED_FIELDS = ("CLASSIFICACAO_FISCAL", "CEST", "IPI") # explicit allowlist; anything else raises before touching SQL +READ_FIELDS = ("CODIGO", "PRODUTO", "GTIN") + ALLOWED_FIELDS +POLL_SECONDS = 3.0 +SAFETY_DIFF_SECONDS = 60.0 +OVERRIDE_CHECK_SECONDS = 20.0 +WATCH_WINDOW_SECONDS = 120.0 # a fresh product is re-read for a while: the operator may add the EAN seconds after the first save +RETRY_BACKOFF = (30, 60, 120, 300, 600, 1800) + + +def _now() -> float: + return time.time() + + +def _text(value) -> str: + return "" if value is None else str(value).strip() + + +class EnrichmentStore: + """Own SQLite connection (the worker runs in its own thread).""" + + def __init__(self, path): + self.db = sqlite3.connect(str(path), timeout=10, check_same_thread=False) + self.db.row_factory = sqlite3.Row + self.db.executescript(""" + CREATE TABLE IF NOT EXISTS enrich_meta(k TEXT PRIMARY KEY, v TEXT); + CREATE TABLE IF NOT EXISTS enrich_known(code TEXT PRIMARY KEY); + CREATE TABLE IF NOT EXISTS enrich_products( + code TEXT PRIMARY KEY, state TEXT NOT NULL, detected_at REAL NOT NULL, next_attempt_at REAL, attempts INTEGER NOT NULL DEFAULT 0, + snapshot TEXT, written TEXT NOT NULL DEFAULT '{}', overridden TEXT NOT NULL DEFAULT '{}', last_state TEXT); + """) + self.db.commit() + + def meta(self, key, default=None): + row = self.db.execute("SELECT v FROM enrich_meta WHERE k=?", (key,)).fetchone() + return row["v"] if row else default + + def set_meta(self, key, value): + self.db.execute("INSERT INTO enrich_meta(k,v) VALUES(?,?) ON CONFLICT(k) DO UPDATE SET v=excluded.v", (key, str(value))) + self.db.commit() + + def known(self) -> set[str]: + return {r["code"] for r in self.db.execute("SELECT code FROM enrich_known")} + + def add_known(self, codes): + self.db.executemany("INSERT OR IGNORE INTO enrich_known(code) VALUES(?)", [(c,) for c in codes]) + self.db.commit() + + def enqueue(self, code, now=None): + now = _now() if now is None else now + self.db.execute("INSERT OR IGNORE INTO enrich_products(code,state,detected_at,next_attempt_at) VALUES(?,?,?,?)", (code, "PENDING", now, now)) + self.db.commit() + + def get(self, code): + row = self.db.execute("SELECT * FROM enrich_products WHERE code=?", (code,)).fetchone() + return dict(row) if row else None + + def due(self, now): + return [dict(r) for r in self.db.execute("SELECT * FROM enrich_products WHERE state IN ('PENDING','RETRY') AND next_attempt_at<=? ORDER BY detected_at LIMIT 20", (now,))] + + def watching(self, now): + return [dict(r) for r in self.db.execute("SELECT * FROM enrich_products WHERE state IN ('DONE','DRY_RUN_DONE') AND detected_at>=?", (now - WATCH_WINDOW_SECONDS,))] + + def tracked_written(self): + return [dict(r) for r in self.db.execute("SELECT * FROM enrich_products WHERE written != '{}'")] + + def update(self, code, **values): + for key in ("written", "overridden"): + if key in values and not isinstance(values[key], str): + values[key] = json.dumps(values[key]) + sets = ",".join(f"{k}=?" for k in values) + self.db.execute(f"UPDATE enrich_products SET {sets} WHERE code=?", (*values.values(), code)) + self.db.commit() + + def is_overridden(self, code, field) -> bool: + row = self.get(code) + return bool(row and json.loads(row["overridden"] or "{}").get(field)) + + +def is_human_overridden(store_path, code, field) -> bool: + """Used by every automatic writer of the Agent (the Fiscal Corrector included) before touching a field the enrichment tracks.""" + try: + return EnrichmentStore(store_path).is_overridden(str(code), field) + except Exception: # noqa: BLE001 - no state means nothing is protected yet + return False + + +# ---- Firebird access (injected: `read_conn()` returns a READ-ONLY connection, `write_conn()` a write-capable one, only ever opened by ENABLED mode) ---------------------------------------- +def _digits(value) -> str: + return "".join(ch for ch in _text(value) if ch.isdigit()) + + +def read_codes(conn) -> set[str]: + cur = conn.cursor() + cur.execute("SELECT CODIGO FROM C000025") + codes = {_text(r[0]) for r in cur.fetchall()} + conn.commit() + return codes + + +def read_count(conn) -> int: + cur = conn.cursor() + cur.execute("SELECT COUNT(*) FROM C000025") + n = int(cur.fetchone()[0]) + conn.commit() + return n + + +def read_rows(conn, codes) -> dict[str, dict]: + out = {} + codes = list(codes) + for start in range(0, len(codes), 100): + chunk = codes[start:start + 100] + cur = conn.cursor() + cur.execute(f"SELECT {','.join(READ_FIELDS)} FROM C000025 WHERE CODIGO IN ({','.join('?' for _ in chunk)})", tuple(chunk)) + for row in cur.fetchall(): + rec = dict(zip(READ_FIELDS, row)) + out[_text(rec["CODIGO"])] = rec + conn.commit() + return out + + +def write_field(conn, code: str, field: str, value) -> dict: + """PRODUCT_ENRICHMENT_WRITE_V1: one field, one short transaction, only if the field is still empty. Returns {outcome, reason?}.""" + if field not in ALLOWED_FIELDS: + raise ValueError(f"field {field!r} is outside the PRODUCT_ENRICHMENT_WRITE_V1 allowlist") + if field == "IPI": + value = float(value) + else: + value = _digits(value) + if not value or (field == "CLASSIFICACAO_FISCAL" and len(value) != 8) or (field == "CEST" and len(value) != 7): + return {"outcome": "ABORTED", "reason": "VALUE_SHAPE_INVALID"} + cur = conn.cursor() + try: + cur.execute(f"SELECT {field} FROM C000025 WHERE CODIGO = ?", (code,)) + rows = cur.fetchall() + if len(rows) != 1: + conn.rollback() + return {"outcome": "ABORTED", "reason": "PRODUCT_ROW_NOT_UNIQUE" if rows else "PRODUCT_NOT_FOUND"} + if _text(rows[0][0]): + conn.rollback() + return {"outcome": "ABORTED", "reason": "VALUE_ALREADY_SET_NEVER_OVERWRITTEN"} + cur.execute(f"UPDATE C000025 SET {field} = ? WHERE CODIGO = ? AND ({field} IS NULL OR TRIM({field}) = '')", (value, code)) + if cur.rowcount != 1: + conn.rollback() + return {"outcome": "ABORTED", "reason": "UPDATE_ROWCOUNT_NOT_ONE"} + conn.commit() + cur = conn.cursor() + cur.execute(f"SELECT {field} FROM C000025 WHERE CODIGO = ?", (code,)) + after = cur.fetchone()[0] + conn.commit() + same = (float(after) == value) if field == "IPI" and after is not None else _text(after) == value + if not same: + return {"outcome": "FAILED", "reason": "POST_WRITE_VERIFICATION_FAILED"} + return {"outcome": "APPLIED", "value": value, "rowcount": 1} + except Exception as error: # noqa: BLE001 + try: + conn.rollback() + except Exception: # noqa: BLE001 + pass + return {"outcome": "FAILED", "reason": type(error).__name__} + + +def _same_value(field, current, written) -> bool: + if field == "IPI": + try: + return current is not None and _text(current) != "" and abs(float(current) - float(written)) < 1e-9 + except (TypeError, ValueError): + return False + return _text(current) == _text(written) and _text(current) != "" + + +class EnrichmentWorker: + def __init__(self, *, hub, store: EnrichmentStore, read_conn, write_conn, create_reprocess=None, clock=_now, sleep=time.sleep): + self.hub, self.store, self.read_conn, self.write_conn = hub, store, read_conn, write_conn + self.create_reprocess, self.clock, self.sleep = create_reprocess, clock, sleep + self._features = None + self._features_at = 0.0 + self._last_count = None + self._last_diff = 0.0 + self._last_override = 0.0 + self._auth_cache = (None, 0.0) + self.metrics = {"detected": 0, "resolved": 0, "fields_written": 0, "detect_latency_ms": []} + + # ---- feature gate ------------------------------------------------------------------------------------------------------------------------------ + def features(self, max_age=10.0) -> dict: + if self._features is None or self.clock() - self._features_at > max_age: + try: + auth = self.hub.authorizations() + self._features = (auth or {}).get("features") or {} + except Exception: # noqa: BLE001 - Hub offline: keep the last known gate, never invent one + self._features = self._features or {} + self._features_at = self.clock() + return self._features + + def mode(self) -> str: + f = self.features() + if not f.get("features_configured"): + return "WAITING_FEATURE_CONFIGURATION" + return f.get("enrichment_mode") or "DISABLED" + + def _fresh_enabled(self) -> bool: + """Fail-closed authorization read right before a write (cached ~3 s).""" + auth, at = self._auth_cache + if auth is None or self.clock() - at > 3.0: + try: + auth = self.hub.authorizations().get("features") or {} + except Exception: # noqa: BLE001 + return False + self._auth_cache = (auth, self.clock()) + return bool(auth.get("features_configured")) and auth.get("enrichment_mode") == "ENABLED" + + # ---- detection --------------------------------------------------------------------------------------------------------------------------------- + def detect(self) -> list[str]: + conn = self.read_conn() + try: + if self.store.meta("baseline_done") != "1": + codes = read_codes(conn) + self.store.add_known(codes) # historical catalog: known, never sent + self.store.set_meta("baseline_done", "1") + self.store.set_meta("baseline_at", datetime.now(timezone.utc).isoformat()) + self.store.set_meta("baseline_count", len(codes)) + self._last_count, self._last_diff = len(codes), self.clock() + return [] + started = self.clock() + count = read_count(conn) + moved = self._last_count is None or count != self._last_count + if not moved and self.clock() - self._last_diff < SAFETY_DIFF_SECONDS: + return [] + codes = read_codes(conn) + self._last_count, self._last_diff = len(codes), self.clock() + new = sorted(codes - self.store.known()) + self.store.add_known(new) + for code in new: + self.store.enqueue(code, self.clock()) + if new: + self.metrics["detected"] += len(new) + self.metrics["detect_latency_ms"].append(int((self.clock() - started) * 1000)) + return new + finally: + conn.close() + + # ---- one product --------------------------------------------------------------------------------------------------------------------------------- + def process(self, item: dict): + code, mode = item["code"], self.mode() + conn = self.read_conn() + try: + rows = read_rows(conn, [code]) + finally: + conn.close() + row = rows.get(code) + if row is None: + self.store.update(code, state="GONE") + return + snapshot = f"{_text(row['GTIN'])}|{_text(row['PRODUTO'])}" + if mode not in ("DRY_RUN", "ENABLED"): + self.store.update(code, state="SKIPPED_FEATURE", snapshot=snapshot) + return + attempts = int(item["attempts"] or 0) + 1 + current = {f: row[f] for f in ALLOWED_FIELDS} + try: + result = self.hub.resolve_product({"product_code": code, "gtin": _text(row["GTIN"]), "description": _text(row["PRODUTO"]), "current": {k: (None if v is None else str(v)) for k, v in current.items()}}) + except Exception as error: # noqa: BLE001 - Hub offline / slow: the product stays pending, the SGV is unaffected + delay = RETRY_BACKOFF[min(attempts - 1, len(RETRY_BACKOFF) - 1)] + self.store.update(code, state="RETRY", attempts=attempts, next_attempt_at=self.clock() + delay, last_state=f"HUB_ERROR:{type(error).__name__}") + return + self.metrics["resolved"] += 1 + state = result.get("state") + if state == "FEATURE_DISABLED": + self.store.update(code, state="SKIPPED_FEATURE", snapshot=snapshot, last_state=state) + return + if state in ("PENDING_PROVIDER_QUOTA", "PENDING_PROVIDERS_OFFLINE"): + retry_at = result.get("retry_at") + due = self.clock() + RETRY_BACKOFF[min(attempts - 1, len(RETRY_BACKOFF) - 1)] + if retry_at: + try: + due = max(due, datetime.fromisoformat(retry_at).timestamp()) + except ValueError: + pass + self.store.update(code, state="RETRY", attempts=attempts, next_attempt_at=due, last_state=state, snapshot=snapshot) + return + fields = result.get("fields") or {} + reports = [] + written = json.loads(item.get("written") or "{}") + overridden = json.loads(item.get("overridden") or "{}") + for field in ALLOWED_FIELDS: + decision = fields.get(field) or {} + status, value = decision.get("status"), decision.get("value") + base = {"field": field, "rule": decision.get("rule"), "kb_status": decision.get("kb_status"), "evidence": decision.get("evidence") or [], "old": None, "new": value} + if status == "HUMAN_OVERRIDDEN" or overridden.get(field): + reports.append({**base, "outcome": "HUMAN_OVERRIDDEN", "reason": "HUMAN_OVERRIDDEN", "new": None}) + continue + if status != "AUTO_FILL_SAFE" or value in (None, ""): + continue + if mode == "DRY_RUN": + reports.append({**base, "outcome": "DRY_RUN", "reason": "DRY_RUN_NO_WRITE"}) + continue + outcome = self._write_safely(code, field, value) + reports.append({**base, "outcome": outcome["outcome"], "reason": outcome.get("reason")}) + if outcome["outcome"] == "APPLIED": + written[field] = value + self.metrics["fields_written"] += 1 + self.store.update(code, state="DRY_RUN_DONE" if mode == "DRY_RUN" else "DONE", attempts=attempts, snapshot=snapshot, written=written, last_state=state, next_attempt_at=None) + if reports: + try: + self.hub.report_enrichment({"product_code": code, "fields": reports, "dataset_versions": result.get("dataset_versions") or {}}) + except Exception: # noqa: BLE001 - audit delivery is retried on the next cycle via last_state + self.store.update(code, last_state="AUDIT_PENDING") + if written and self.create_reprocess is not None and mode == "ENABLED": + try: + self.create_reprocess([code]) # cClassTrib is decided by the existing RTC gates, never here + except Exception: # noqa: BLE001 + logger.warning("reprocess request for %s failed", code) + + def _write_safely(self, code, field, value) -> dict: + if not self._fresh_enabled(): + return {"outcome": "ABORTED", "reason": "AUTHORIZATION_CHANGED"} + conn = self.read_conn() + try: + fresh = read_rows(conn, [code]).get(code) + finally: + conn.close() + if fresh is None: + return {"outcome": "ABORTED", "reason": "PRODUCT_NOT_FOUND"} + if _text(fresh[field]): + return {"outcome": "ABORTED", "reason": "VALUE_ALREADY_SET_NEVER_OVERWRITTEN"} + wconn = self.write_conn() + try: + return write_field(wconn, code, field, value) + finally: + wconn.close() + + # ---- human override watch -------------------------------------------------------------------------------------------------------------------------- + def check_overrides(self): + tracked = self.store.tracked_written() + if not tracked: + return + conn = self.read_conn() + try: + rows = read_rows(conn, [t["code"] for t in tracked]) + finally: + conn.close() + for item in tracked: + row = rows.get(item["code"]) + written, overridden = json.loads(item["written"]), json.loads(item["overridden"] or "{}") + new = [] + for field, value in written.items(): + if overridden.get(field) or row is None: + continue + if not _same_value(field, row[field], value): # different value OR blanked/deleted: a person decided, forever + overridden[field] = True + new.append({"field": field, "outcome": "HUMAN_OVERRIDDEN", "reason": "VALUE_CHANGED_AFTER_WRITE", "old": value, "new": None}) + if new: + self.store.update(item["code"], overridden=overridden) + try: + self.hub.report_enrichment({"product_code": item["code"], "fields": new}) + except Exception: # noqa: BLE001 - local protection is already in place; the Hub copy follows on the next cycle + logger.warning("override report for %s failed", item["code"]) + + # ---- loop ---------------------------------------------------------------------------------------------------------------------------------------- + def step(self): + """One iteration; returns the list of processed product codes.""" + mode = self.mode() + if mode not in ("DRY_RUN", "ENABLED"): + return [] # WAITING_FEATURE_CONFIGURATION / DISABLED: zero Firebird activity + self.detect() + done = [] + for item in self.store.due(self.clock()): + self.process(item) + done.append(item["code"]) + for item in self.store.watching(self.clock()): # operator completes the record (EAN typed a moment later): re-resolve once per change + rows = None + conn = self.read_conn() + try: + rows = read_rows(conn, [item["code"]]) + finally: + conn.close() + row = (rows or {}).get(item["code"]) + if row and f"{_text(row['GTIN'])}|{_text(row['PRODUTO'])}" != item.get("snapshot"): + self.process(item) + done.append(item["code"]) + if self.clock() - self._last_override >= OVERRIDE_CHECK_SECONDS: + self._last_override = self.clock() + self.check_overrides() + return done + + def run_forever(self, stopper): + while not stopper.stop: + try: + self.step() + except Exception: # noqa: BLE001 - the enrichment thread must never die or affect the SGV + logger.exception("enrichment step failed") + self.sleep(POLL_SECONDS) diff --git a/agent/tests/test_enrichment.py b/agent/tests/test_enrichment.py new file mode 100644 index 0000000..cfdc8cf --- /dev/null +++ b/agent/tests/test_enrichment.py @@ -0,0 +1,201 @@ +"""Agent enrichment: baseline/new-product detection, feature gating, typed write, second run = 0 writes, human override (incl. deleted value).""" +import sqlite3 +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "agent" / "src")) +from gti_fiscal_agent import enrichment as E # noqa: E402 + + +class Conn: + """sqlite stand-in for the Firebird connection; close() is a no-op so the worker can 'open' it repeatedly.""" + + def __init__(self, db): + self.db = db + + def cursor(self): + return self.db.cursor() + + def commit(self): + self.db.commit() + + def rollback(self): + self.db.rollback() + + def close(self): + pass + + +class Hub: + def __init__(self, features, decisions=None, state="EXACT_GTIN_VERIFIED"): + self.features = features + self.decisions = decisions or {} + self.state = state + self.resolve_calls, self.reports, self.overridden = [], [], {} + + def authorizations(self): + return {"features": dict(self.features)} + + def resolve_product(self, payload): + self.resolve_calls.append(payload) + fields = {k: dict(v) for k, v in self.decisions.items()} + for f in list(fields): + if f in self.overridden.get(payload["product_code"], set()): + fields[f] = {"status": "HUMAN_OVERRIDDEN", "value": None} + elif payload["current"].get(f): + fields[f] = {"status": "ALREADY_SET", "value": None} + return {"state": self.state, "fields": fields, "dataset_versions": {"NCM": {"sha256": "abc"}}} + + def report_enrichment(self, payload): + self.reports.append(payload) + for item in payload["fields"]: + if item["outcome"] == "HUMAN_OVERRIDDEN": + self.overridden.setdefault(payload["product_code"], set()).add(item["field"]) + return {"ok": True} + + +ENABLED = {"features_configured": True, "corrector_enabled": True, "enrichment_mode": "ENABLED"} +DECISIONS = {"CLASSIFICACAO_FISCAL": {"status": "AUTO_FILL_SAFE", "value": "22021000", "rule": "NCM_KB_VERIFIED", "kb_status": "VERIFIED_KB"}, + "CEST": {"status": "AUTO_FILL_SAFE", "value": "0300700", "rule": "CEST_OFFICIAL_UNIQUE_MAPPING"}, + "IPI": {"status": "AUTO_FILL_SAFE", "value": "0.0", "rule": "IPI_TIPI_NUMERIC"}} + + +@pytest.fixture() +def env(tmp_path): + db = sqlite3.connect(":memory:", check_same_thread=False) + db.execute("CREATE TABLE C000025(CODIGO TEXT PRIMARY KEY, PRODUTO TEXT, GTIN TEXT, CLASSIFICACAO_FISCAL TEXT, CEST TEXT, IPI REAL, OTHER TEXT)") + db.executemany("INSERT INTO C000025(CODIGO,PRODUTO,GTIN,OTHER) VALUES(?,?,?,?)", [(f"H{i}", f"HIST {i}", "", "keep") for i in range(50)]) + db.commit() + hub = Hub(dict(ENABLED), DECISIONS) + store = E.EnrichmentStore(tmp_path / "e.db") + now = [1000.0] + reproc = [] + worker = E.EnrichmentWorker(hub=hub, store=store, read_conn=lambda: Conn(db), write_conn=lambda: Conn(db), create_reprocess=reproc.append, clock=lambda: now[0], sleep=lambda s: None) + return db, hub, store, worker, now, reproc + + +def add(db, code, name="REFRIGERANTE COLA LATA 350ML", gtin="7891000100103"): + db.execute("INSERT INTO C000025(CODIGO,PRODUTO,GTIN,OTHER) VALUES(?,?,?,?)", (code, name, gtin, "keep")) + db.commit() + + +def row(db, code): + return db.execute("SELECT CLASSIFICACAO_FISCAL,CEST,IPI,OTHER FROM C000025 WHERE CODIGO=?", (code,)).fetchone() + + +def test_baseline_never_sends_historical_products(env): + db, hub, store, worker, now, _ = env + assert worker.step() == [] # baseline + now[0] += 5 + assert worker.step() == [] and hub.resolve_calls == [] + assert len(store.known()) == 50 + + +def test_new_product_is_enriched_only_in_empty_fields_then_second_run_zero_writes(env): + db, hub, store, worker, now, reproc = env + worker.step() + add(db, "N1") + now[0] += 5 + assert worker.step() == ["N1"] + assert row(db, "N1") == ("22021000", "0300700", 0.0, "keep") + assert reproc == [["N1"]] + assert row(db, "H1") == (None, None, None, "keep") # historical untouched + assert [r["outcome"] for r in hub.reports[0]["fields"]] == ["APPLIED"] * 3 + now[0] += 5 + before = db.total_changes + worker.step() + assert db.total_changes == before and len(hub.resolve_calls) == 1 # nothing re-processed, 0 writes + + +def test_existing_value_is_never_overwritten(env): + db, hub, store, worker, now, _ = env + worker.step() + add(db, "N2") + db.execute("UPDATE C000025 SET CLASSIFICACAO_FISCAL='19059090' WHERE CODIGO='N2'") + db.commit() + now[0] += 5 + worker.step() + assert row(db, "N2")[0] == "19059090" and row(db, "N2")[1] == "0300700" + + +def test_dry_run_and_disabled_and_waiting_do_not_write_or_touch_firebird(env): + db, hub, store, worker, now, _ = env + worker.step() + add(db, "N3") + for features in ({"features_configured": False}, {"features_configured": True, "enrichment_mode": "DISABLED"}): + hub.features = features + worker._features_at = -999 + assert worker.step() == [] and hub.resolve_calls == [] + hub.features = {"features_configured": True, "enrichment_mode": "DRY_RUN"} + worker._features_at = -999 + now[0] += 5 + worker.step() + assert row(db, "N3") == (None, None, None, "keep") and hub.reports[0]["fields"][0]["outcome"] == "DRY_RUN" + + +def test_authorization_revoked_right_before_write_fails_closed(env): + db, hub, store, worker, now, _ = env + worker.step() + add(db, "N4") + calls = {"n": 0} + + def flip(): + calls["n"] += 1 + return {"features": dict(ENABLED if calls["n"] == 1 else {"features_configured": True, "enrichment_mode": "DISABLED"})} + hub.authorizations = flip + now[0] += 5 + worker._features_at = -999 + worker.step() + assert row(db, "N4") == (None, None, None, "keep") + assert all(f["outcome"] == "ABORTED" and f["reason"] == "AUTHORIZATION_CHANGED" for f in hub.reports[0]["fields"]) + + +def test_human_override_including_deleted_value_is_final(env): + db, hub, store, worker, now, _ = env + worker.step() + add(db, "N5") + now[0] += 5 + worker.step() + db.execute("UPDATE C000025 SET CEST='', CLASSIFICACAO_FISCAL='19059090' WHERE CODIGO='N5'") # a person deletes the CEST and changes the NCM + db.commit() + now[0] += 30 + worker.step() + assert store.is_overridden("N5", "CEST") and store.is_overridden("N5", "CLASSIFICACAO_FISCAL") and not store.is_overridden("N5", "IPI") + # the description changes a moment later: the worker re-resolves, but overridden fields must NOT be written again + db.execute("UPDATE C000025 SET PRODUTO='REFRIGERANTE COLA LATA 350ML ZERO' WHERE CODIGO='N5'") + db.commit() + now[0] += 30 + worker.step() + assert row(db, "N5")[:2] == ("19059090", "") + + +def test_write_field_allowlist_and_shape(): + db = sqlite3.connect(":memory:") + db.execute("CREATE TABLE C000025(CODIGO TEXT, COD_ORIGEM_MERCADORIA TEXT, CLASSIFICACAO_FISCAL TEXT)") + with pytest.raises(ValueError): + E.write_field(Conn(db), "1", "COD_ORIGEM_MERCADORIA", "1") + with pytest.raises(ValueError): + E.write_field(Conn(db), "1", "CCLASSTRIB", "000001") + db.execute("INSERT INTO C000025(CODIGO) VALUES('1')") + db.commit() + assert E.write_field(Conn(db), "1", "CLASSIFICACAO_FISCAL", "123")["reason"] == "VALUE_SHAPE_INVALID" + + +def test_hub_offline_keeps_product_pending_and_retries(env): + db, hub, store, worker, now, _ = env + worker.step() + add(db, "N6") + + def boom(payload): + raise ConnectionError("down") + good = hub.resolve_product + hub.resolve_product = boom + now[0] += 5 + worker.step() + assert store.get("N6")["state"] == "RETRY" and row(db, "N6")[0] is None + hub.resolve_product = good + now[0] += 40 + assert worker.step() == ["N6"] and row(db, "N6")[0] == "22021000" diff --git a/central/app/agent_distribution.py b/central/app/agent_distribution.py index b7a6bfd..cca45f9 100644 --- a/central/app/agent_distribution.py +++ b/central/app/agent_distribution.py @@ -175,7 +175,9 @@ def router(session_factory): session.add(agent) session.flush() session.add(Installation(agent_id=agent.id, tenant_id=enr.tenant_id, installation_id=enr.enrollment_uuid, - display_name=enr.display_name, hostname=hostname or None)) + display_name=enr.display_name, hostname=hostname or None, + features_configured=bool(enr.features_selected), corrector_enabled=bool(enr.features_selected and enr.feature_corrector), + enrichment_mode=(enr.feature_enrichment_mode if enr.features_selected else "DISABLED"))) enr.status = "ENROLLED" enr.enrollment_used_at = datetime.now(timezone.utc) diff --git a/central/app/api.py b/central/app/api.py index 0188876..aac8344 100644 --- a/central/app/api.py +++ b/central/app/api.py @@ -19,6 +19,7 @@ from .events import record_event from .installation_identity import compute_agent_health, compute_version_drift, reference_agent_version, sanitize_registration from .control_plane import list_jobs from .human_auth import require_human +from .enrichment.api import router as enrichment_router logger = logging.getLogger("gti_fiscal_hub") @@ -180,7 +181,8 @@ def router(session_factory, bootstrap_token): if installation is not None and installation.installation_id != installation_key: raise HTTPException(409, "installation_id conflicts with registered Agent") if installation is None: - installation = Installation(agent_id=agent.id, tenant_id=agent.tenant_id, installation_id=installation_key, display_name=supplied.get("display_name") or agent.name) + installation = Installation(agent_id=agent.id, tenant_id=agent.tenant_id, installation_id=installation_key, display_name=supplied.get("display_name") or agent.name, + features_configured=True, corrector_enabled=True) # legacy/bootstrap self-registration keeps the pre-existing behaviour session.add(installation); session.flush() fields = {"display_name": supplied.get("display_name"), "customer_code": supplied.get("customer_code"), "notes": supplied.get("notes"), "hostname": supplied.get("hostname"), "windows_version": supplied.get("windows_version"), "machine_fingerprint": supplied.get("machine_fingerprint"), "company_code": supplied.get("company_code"), "company_name": supplied.get("company_name"), "company_cnpj_masked": supplied.get("company_cnpj_masked"), "company_uf": supplied.get("company_uf"), "company_crt": supplied.get("company_crt"), "firebird_version": supplied.get("firebird_version"), "firebird_ods": supplied.get("firebird_ods"), "database_identifier": supplied.get("database_identifier"), "capabilities": supplied.get("capabilities"), "fiscal_engine_version": supplied.get("fiscal_engine_version"), "dataset_versions": supplied.get("dataset_versions"), "last_status": supplied.get("status") or payload.get("status"), "last_seen_at": now} for key, value in fields.items(): @@ -464,4 +466,5 @@ def router(session_factory, bootstrap_token): return _job_view(session.get(Job, job_id)) finally: session.close() + routes.include_router(enrichment_router(principal)) return routes diff --git a/central/app/config.py b/central/app/config.py index 29b0501..04111cc 100644 --- a/central/app/config.py +++ b/central/app/config.py @@ -77,6 +77,11 @@ class Settings: # Z.ai general endpoint (NOT the /api/coding/paas/v4 coding-plan endpoint). ai_api_key: str = os.getenv("AI_API_KEY", "") ai_base_url: str = os.getenv("AI_BASE_URL", "https://api.z.ai/api/paas/v4") + # ---- Product enrichment (docs/product_enrichment.md): free providers only, tokens live ONLY in the Hub environment ---- + cosmos_token: str = os.getenv("COSMOS_TOKEN", "") # optional; without it Cosmos is NOT_CONFIGURED (not an error) + allow_paid_providers: bool = os.getenv("ALLOW_PAID_PROVIDERS", "0") == "1" + official_refresh_enabled: bool = os.getenv("OFFICIAL_REFRESH_ENABLED", "1") == "1" + enrichment_time_budget_seconds: float = float(os.getenv("ENRICHMENT_TIME_BUDGET_SECONDS", "4.5")) settings = Settings() diff --git a/central/app/datasets/updater.py b/central/app/datasets/updater.py new file mode 100644 index 0000000..ec09783 --- /dev/null +++ b/central/app/datasets/updater.py @@ -0,0 +1,85 @@ +"""Automatic refresh of the official NCM dataset (Siscomex JSON): staging -> validate -> atomic promotion, keeping the last valid dataset on ANY failure. + +Every attempt leaves a SourceRefreshRun (url, time, sha256, version label, record count, status, error) for the panel. Nothing here touches Agents or providers. +""" +from __future__ import annotations + +import logging +import threading +import time +from datetime import datetime, timedelta, timezone + +import httpx +from sqlalchemy import select + +from ..config import settings +from ..models import Dataset, SourceRefreshRun +from .ncm import parse_ncm_payload +from .service import DatasetService + +logger = logging.getLogger("gti_fiscal_hub") +MIN_RATIO = 0.9 # a payload with < 90% of the active record count is treated as a broken download, never promoted +INTERVAL = timedelta(hours=24) +RETRY_AFTER_FAILURE = timedelta(hours=6) + + +def _fetch(url: str) -> bytes: + response = httpx.get(url, timeout=max(settings.http_timeout_seconds, 60), follow_redirects=True, headers={"User-Agent": "GTIFiscalHub/1.0 (official dataset refresh)"}) + response.raise_for_status() + return response.content + + +def refresh_ncm(session, fetch=_fetch, url: str | None = None) -> SourceRefreshRun: + url = url or settings.ncm_dataset_url + run = SourceRefreshRun(source="NCM", source_url=url, status="RUNNING") + session.add(run) + session.commit() + try: + content = fetch(url) # staging: bytes in memory, nothing written to the active dataset yet + parsed = parse_ncm_payload(content) # schema validation + run.sha256, run.version_label, run.record_count = parsed.sha256, (parsed.version or "")[:64], len(parsed.entries) + active = session.scalar(select(Dataset).where(Dataset.kind == "NCM", Dataset.status == "ACTIVE")) + if active is not None and active.sha256 == parsed.sha256: + run.status = "UNCHANGED" + else: + if active is not None and len(parsed.entries) < MIN_RATIO * (active.record_count or 0): + raise ValueError(f"implausible record count {len(parsed.entries)} vs active {active.record_count}") + DatasetService(session).import_payload(parsed, url, datetime.now(timezone.utc)) # atomic: one transaction, old ACTIVE superseded only at the end + run.status = "UPDATED" + except Exception as error: # keep the last valid dataset + session.rollback() + run = session.get(SourceRefreshRun, run.id) + run.status, run.error = "FAILED", f"{type(error).__name__}: {error}"[:300] + logger.warning("NCM refresh failed: %s", run.error) + run.finished_at = datetime.now(timezone.utc) + session.commit() + return run + + +def last_runs(session, source: str = "NCM", limit: int = 5): + return session.scalars(select(SourceRefreshRun).where(SourceRefreshRun.source == source).order_by(SourceRefreshRun.id.desc()).limit(limit)).all() + + +def _due(session) -> bool: + last = session.scalar(select(SourceRefreshRun).where(SourceRefreshRun.source == "NCM", SourceRefreshRun.status != "RUNNING").order_by(SourceRefreshRun.id.desc()).limit(1)) + if last is None or last.finished_at is None: + return True + finished = last.finished_at if last.finished_at.tzinfo else last.finished_at.replace(tzinfo=timezone.utc) + return datetime.now(timezone.utc) - finished >= (RETRY_AFTER_FAILURE if last.status == "FAILED" else INTERVAL) + + +def start_scheduler(session_factory, first_delay: float = 90.0): + """Daemon thread: checks hourly whether a refresh is due (survives restarts because due-ness is derived from SourceRefreshRun).""" + def loop(): + time.sleep(first_delay) + while True: + try: + with session_factory() as session: + if _due(session): + refresh_ncm(session) + except Exception: # never let the scheduler die + logger.exception("official refresh loop error") + time.sleep(3600) + thread = threading.Thread(target=loop, name="official-refresh", daemon=True) + thread.start() + return thread diff --git a/central/app/enrichment/api.py b/central/app/enrichment/api.py new file mode 100644 index 0000000..3c23e65 --- /dev/null +++ b/central/app/enrichment/api.py @@ -0,0 +1,140 @@ +"""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 ..models import Installation, ProductEnrichment, ProductFieldAudit +from .providers import build_providers +from .quota import ProviderGateway +from .resolver import FIELDS, resolve + +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 FIELDS}, + client_label=f"installation:{installation.id}", budget_s=settings.enrichment_time_budget_seconds) + 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 + row.gtin, row.description, row.latency_ms = result.get("gtin"), description, latency + row.attempts = (row.attempts or 0) + 1 + 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 + 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/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 diff --git a/central/app/enrichment/resolver.py b/central/app/enrichment/resolver.py index c5bd6ce..d29153e 100644 --- a/central/app/enrichment/resolver.py +++ b/central/app/enrichment/resolver.py @@ -7,7 +7,8 @@ Field decisions: AUTO_FILL_SAFE | SUGGESTION_ONLY | CONFLICT | INSUFFICIENT_EVID """ from __future__ import annotations -from datetime import date, datetime +import time +from datetime import date from sqlalchemy import select @@ -79,7 +80,7 @@ def _fiscal_decisions(session, ncm_eval: dict, gtin_kb_verified: bool, current: if not _empty(current, "CEST"): out["CEST"] = _field(ALREADY, reason="campo já preenchido") elif decision["cest_status"] == "CEST_CONFIRMED" and len(cest) == 1 and ncm["status"] == AUTO: - out["CEST"] = _field(AUTO, cest[0], "CEST_OFFICIAL_UNIQUE_MAPPING", ncm["kb_status"], [f"NCM {ncm_for_engine}"]) + out["CEST"] = _field(AUTO, normalize_code(cest[0]), "CEST_OFFICIAL_UNIQUE_MAPPING", ncm["kb_status"], [f"NCM {ncm_for_engine}"]) elif decision["cest_status"] == "NO_CEST_MAPPING_FOUND": out["CEST"] = _field(NA, reason="NCM sem mapeamento de CEST no dataset oficial") else: @@ -96,7 +97,8 @@ def _fiscal_decisions(session, ncm_eval: dict, gtin_kb_verified: bool, current: return out -def resolve(session, gateway: ProviderGateway, *, gtin_raw, description, current: dict | None = None, client_label: str | None = None, allow_external: bool = True) -> dict: +def resolve(session, gateway: ProviderGateway, *, gtin_raw, description, current: dict | None = None, client_label: str | None = None, allow_external: bool = True, budget_s: float | None = None) -> dict: + started = time.monotonic() """Returns {state, gtin_status, gtin, fields:{...}, providers:[...], retry_at, dataset_versions}. Never raises for provider trouble.""" current = {k: v for k, v in (current or {}).items()} g = normalize_gtin(gtin_raw) @@ -117,6 +119,10 @@ def resolve(session, gateway: ProviderGateway, *, gtin_raw, description, current found, pending = [], [] if allow_external: for provider in gateway.providers: + if budget_s is not None and time.monotonic() - started > budget_s: + result["providers"].append({"provider": provider.name, "state": "SKIPPED", "reason": "TIME_BUDGET", "from_cache": False}) + pending.append(type("P", (), {"reason": "TIME_BUDGET", "retry_at": None})()) + continue out = gateway.lookup(provider, query) result["providers"].append({"provider": out.provider, "state": out.state, "reason": out.reason, "from_cache": out.from_cache}) if out.state in ("FOUND", "CACHE_HIT") and out.product: diff --git a/central/app/jobs.py b/central/app/jobs.py index 5ceed44..035068b 100644 --- a/central/app/jobs.py +++ b/central/app/jobs.py @@ -61,8 +61,8 @@ def claim_next_job(session, agent): processing is enabled centrally and no other job of this Agent is RUNNING (one fiscal job at a time per Firebird).""" installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id)) - if installation is not None and not installation.processing_enabled: - return None + if installation is not None and (not installation.processing_enabled or not installation.features_configured or not installation.corrector_enabled): + return None # WAITING_FEATURE_CONFIGURATION / corrector switched off centrally: nothing is handed to the Agent recover_stale_running(session, agent) if session.scalar(select(Job.id).where(Job.agent_id == agent.id, Job.status == "RUNNING").limit(1)) is not None: return None @@ -127,7 +127,7 @@ def schedule_automatic_sync(session, agent, installation, snapshot, now=None): - otherwise, if processing is enabled and the last sync attempt is older than the installation's interval, and nothing is queued/running -> ONE SYNC_INCREMENTAL. Returns the created job or None.""" - if installation is None or not installation.processing_enabled: + if installation is None or not installation.processing_enabled or not installation.features_configured or not installation.corrector_enabled: return None now = now or datetime.now(timezone.utc) sync_jobs = session.scalars(select(Job).where(Job.agent_id == agent.id, Job.kind.in_(SYNC_KINDS)).order_by(Job.id.desc())).all() diff --git a/central/app/main.py b/central/app/main.py index 5111089..90fe893 100644 --- a/central/app/main.py +++ b/central/app/main.py @@ -1,6 +1,7 @@ """FastAPI application factory.""" import logging +import sys from collections.abc import Callable from fastapi import FastAPI, HTTPException @@ -40,6 +41,12 @@ def create_app(database_probe: Callable[[], bool] = database_is_available, sessi app.include_router(control_api_router(session_factory)) app.include_router(agent_distribution_router(session_factory)) app.include_router(web_router(session_factory)) + if settings.official_refresh_enabled and session_factory is SessionLocal and "pytest" not in sys.modules: + @app.on_event("startup") + def _start_official_refresh(): + from .datasets.updater import start_scheduler + start_scheduler(session_factory) + app.mount("/static", StaticFiles(directory="central/static"), name="static") return app diff --git a/central/app/panel_presentation.py b/central/app/panel_presentation.py index 4e3c80f..7d1ca2e 100644 --- a/central/app/panel_presentation.py +++ b/central/app/panel_presentation.py @@ -206,6 +206,8 @@ def _rtc_products_view(resolution, enabled, cclasstrib_applicability=None): "catalog": (resolution.get("catalog_compatibility") or {}).get("compatibility")} +ENRICHMENT_MODES = ("DISABLED", "DRY_RUN", "ENABLED") +ENRICHMENT_MODE_LABELS = {"DISABLED": "Desativado", "DRY_RUN": "Simulação (não grava)", "ENABLED": "Automático"} RTC_MODE_LABELS = {"DISABLED": "Desativada", "DRY_RUN": "Dry-run", "APPROVAL_REQUIRED": "Aprovação manual", "ENABLED": "Automática"} RTC_MODE_EXPLANATION = "Somente cClassTrib com regra fiscal determinística e catálogo validado pode ser preenchido automaticamente." diff --git a/central/app/rtc_audit.py b/central/app/rtc_audit.py index ff30a0e..ee0fe15 100644 --- a/central/app/rtc_audit.py +++ b/central/app/rtc_audit.py @@ -20,7 +20,8 @@ def _clip(value, size): def job_authorizations(session, installation) -> dict: """What the authenticated job carries to the Agent. Decided centrally per installation; the Agent has no local switch.""" mode = installation.rtc_write_mode if installation.rtc_write_mode in WRITE_MODES else "DISABLED" - out = {"rtc_write_mode": mode, "rtc_approved_codes": [], "rtc_rollbacks": []} + out = {"rtc_write_mode": mode, "rtc_approved_codes": [], "rtc_rollbacks": [], + "features": {"corrector_enabled": bool(installation.corrector_enabled), "enrichment_mode": installation.enrichment_mode, "features_configured": bool(installation.features_configured)}} if mode == "APPROVAL_REQUIRED": out["rtc_approved_codes"] = list(session.scalars(select(RtcCandidate.product_code).where(RtcCandidate.installation_id == installation.id, RtcCandidate.status == "APPROVED").limit(MAX_APPROVED_PER_JOB))) for audit in session.scalars(select(RtcWriteAudit).where(RtcWriteAudit.installation_id == installation.id, RtcWriteAudit.action == "APPLY", RtcWriteAudit.outcome == "APPLIED", diff --git a/central/app/web.py b/central/app/web.py index 0ed6828..59fa081 100644 --- a/central/app/web.py +++ b/central/app/web.py @@ -15,11 +15,11 @@ from .config import settings from .blocker_resolution import BlockerResolutionUnavailable, _company_key, resolve_blocker_with_context from .context_keys import ALLOWED_KEYS, ENUMS from .installation_identity import reference_agent_version -from .panel_presentation import RTC_MODE_EXPLANATION, RTC_MODE_LABELS, abort_reason_label, audit_outcome_label, candidate_status_label, catalog_state_label, job_kind_label, job_status_label, rtc_rule_label, rtc_status_view +from .panel_presentation import ENRICHMENT_MODES, ENRICHMENT_MODE_LABELS, RTC_MODE_EXPLANATION, RTC_MODE_LABELS, abort_reason_label, audit_outcome_label, candidate_status_label, catalog_state_label, job_kind_label, job_status_label, rtc_rule_label, rtc_status_view from .control_plane import blocker_detail, blocker_filter_summary, dashboard_summary, installation_view, jobs_status_summary, list_blockers, list_installations, list_jobs, list_panel_events, rtc_blocker_summary from .human_auth import create_session, current_session, require_csrf, verify_password from .panel_presentation import format_ui_datetime -from .models import AdminUser, Agent, AgentEnrollment, AgentRelease, AgentSnapshot, BlockedProduct, FiscalContextEntry, Installation, Job, RtcCandidate, RtcWriteAudit, Tenant +from .models import AdminUser, Agent, AgentEnrollment, AgentRelease, AgentSnapshot, BlockedProduct, FiscalContextEntry, Installation, Job, ProductEnrichment, ProductFieldAudit, RtcCandidate, RtcWriteAudit, Tenant from .jobs import create_job from .audit import record_audit @@ -256,14 +256,18 @@ def router(session_factory): return render(request, "agent_new_installation.html", {"tenants": tenants, "clients_json": _clients_json(tenants), "releases": releases, "auth": auth_data, "error": " ".join(errors), "search": ""}, 400) enrollment_uuid = str(uuid.uuid4()) enr = AgentEnrollment(enrollment_uuid=enrollment_uuid, tenant_id=tenant.id, display_name=display_name, expected_hostname=expected_hostname, release_id=release.id, status="PENDING", created_by=admin.id) + if values.get("features_selected") == "1": # chosen explicitly at creation; otherwise the installation waits for an admin (WAITING_FEATURE_CONFIGURATION) + mode = (values.get("feature_enrichment_mode") or "DISABLED").strip().upper() + enr.features_selected, enr.feature_corrector = True, values.get("feature_corrector") == "1" + enr.feature_enrichment_mode = mode if mode in ENRICHMENT_MODES else "DISABLED" session.add(enr) - record_audit(session, admin_user_id=admin.id, tenant_id=tenant.id, action="INSTALLATION_CREATED", target_type="agent_enrollment", target_id=enrollment_uuid, context={"display_name": display_name, "release_version": release.version}) + record_audit(session, admin_user_id=admin.id, tenant_id=tenant.id, action="INSTALLATION_CREATED", target_type="agent_enrollment", target_id=enrollment_uuid, context={"display_name": display_name, "release_version": release.version, "features_selected": bool(enr.features_selected), "corrector": bool(enr.feature_corrector), "enrichment_mode": enr.feature_enrichment_mode}) session.commit() return RedirectResponse(f"/agents/pending/{enrollment_uuid}", status_code=303) finally: session.close() @routes.get("/agents/{agent_id}", response_class=HTMLResponse) - def agent_detail(request: Request, agent_id: str, tab: str = "overview", event_page: int = 1, severity: str | None = None, event_type: str | None = None, job_id: int | None = None, product_code: str | None = None, since: str | None = None, until: str | None = None): + def agent_detail(request: Request, agent_id: str, tab: str = "overview", event_page: int = 1, severity: str | None = None, event_type: str | None = None, job_id: int | None = None, product_code: str | None = None, since: str | None = None, until: str | None = None, enrich_code: str | None = None): # agent_id is parsed here (not typed int in the path) so a non-numeric # slug like /agents/abc gets a controlled HTML 404 instead of a raw # 422 — and can never shadow the static routes above. @@ -285,7 +289,49 @@ def router(session_factory): for event in events: event["occurred_display"] = format_ui_datetime(datetime.fromisoformat(event["occurred_at"]) if event["occurred_at"] else None) agent_blockers = list_blockers(session, tenant_id=installation.tenant_id, agent_id=agent_id, page=1, page_size=25, scope="all") - return render(request, "agent_detail.html", {"detail": detail, "tab": tab, "blockers": agent_blockers, "blocker_breakdown": blocker_filter_summary(session, tenant_id=installation.tenant_id, agent_id=agent_id, status="OPEN", scope="all")["blocker_breakdown"], "jobs": list_jobs(session, agent_id=agent_id, page=1, page_size=25, scope="all"), "events": events, "event_page": max(1, event_page), "has_more": has_more, "severity": severity or "", "event_type": event_type or "", "job_id": job_id or "", "product_code": product_code or "", "since": since or "", "until": until or "", "event_query_base": query_base(tab=tab, severity=severity, event_type=event_type, job_id=job_id, product_code=product_code, since=since, until=until), "auth": auth_data, "csrf_token": request.cookies.get("gti_csrf", ""), "ops": _ops_view(session, installation, agent)}) + return render(request, "agent_detail.html", {"detail": detail, "tab": tab, "blockers": agent_blockers, "blocker_breakdown": blocker_filter_summary(session, tenant_id=installation.tenant_id, agent_id=agent_id, status="OPEN", scope="all")["blocker_breakdown"], "jobs": list_jobs(session, agent_id=agent_id, page=1, page_size=25, scope="all"), "events": events, "event_page": max(1, event_page), "has_more": has_more, "severity": severity or "", "event_type": event_type or "", "job_id": job_id or "", "product_code": product_code or "", "since": since or "", "until": until or "", "event_query_base": query_base(tab=tab, severity=severity, event_type=event_type, job_id=job_id, product_code=product_code, since=since, until=until), "auth": auth_data, "csrf_token": request.cookies.get("gti_csrf", ""), "ops": _ops_view(session, installation, agent), "features": _features_view(session, installation, enrich_code), "enrich_code": enrich_code or "", "is_admin": _is_privileged(auth_data[1])}) + finally: session.close() + + def _features_view(session, installation, enrich_code=None): + """FUNCIONALIDADES + small ENRIQUECIMENTO summary of the Agent detail (no separate dashboard).""" + counts = dict(session.execute(select(ProductEnrichment.status, func.count()).where(ProductEnrichment.installation_id == installation.id).group_by(ProductEnrichment.status)).all()) + applied = session.scalar(select(func.count()).select_from(ProductFieldAudit).where(ProductFieldAudit.installation_id == installation.id, ProductFieldAudit.outcome == "APPLIED")) or 0 + lookup = None + if enrich_code: + row = session.scalar(select(ProductEnrichment).where(ProductEnrichment.installation_id == installation.id, ProductEnrichment.product_code == enrich_code.strip()[:64])) + audits = session.scalars(select(ProductFieldAudit).where(ProductFieldAudit.installation_id == installation.id, ProductFieldAudit.product_code == enrich_code.strip()[:64]).order_by(ProductFieldAudit.id.desc()).limit(20)).all() if row else [] + lookup = {"found": row is not None, "row": row, "audits": audits} + mode = installation.enrichment_mode + waiting = not installation.features_configured + return {"configured": bool(installation.features_configured), "waiting": waiting, "corrector_enabled": bool(installation.corrector_enabled), "enrichment_mode": mode, + "mode_label": ENRICHMENT_MODE_LABELS.get(mode, mode), "counts": counts, "total": sum(counts.values()), "applied": applied, "lookup": lookup, + "state_label": "Aguardando configuração de funcionalidades" if waiting else ENRICHMENT_MODE_LABELS.get(mode, mode)} + + @routes.post("/agents/{agent_id}/features") + async def set_features(request: Request, agent_id: int): + """Per-installation functions (ADMIN only, CSRF, audited). Central decision: the Agent has no local switch.""" + session, auth_data, redirect = page_auth(request) + if redirect: return redirect + try: + values = await form_values(request); web_session, admin = auth_data + require_csrf(web_session, values.get("csrf_token")) + agent, installation = _rtc_target(session, agent_id) + if not agent or not installation: return HTMLResponse("Agent not found", status_code=404) + if not _is_privileged(admin): + return _deny(session, admin, installation.tenant_id, "features:set", installation.installation_id) + mode = (values.get("enrichment_mode") or "").strip().upper() + if mode not in ENRICHMENT_MODES: + return HTMLResponse("modo inválido", status_code=422) + if mode == "ENABLED" and (values.get("confirm") or "").strip().upper() != "PREENCHER": + return HTMLResponse("Digite PREENCHER para confirmar o enriquecimento automático", status_code=422) + before = {"corrector": bool(installation.corrector_enabled), "enrichment_mode": installation.enrichment_mode, "configured": bool(installation.features_configured)} + installation.corrector_enabled = values.get("corrector_enabled") == "1" + installation.enrichment_mode = mode + installation.features_configured = True + record_audit(session, admin_user_id=admin.id, tenant_id=installation.tenant_id, action="INSTALLATION_FEATURES", target_type="installation", target_id=installation.installation_id, + context={"from": before, "to": {"corrector": installation.corrector_enabled, "enrichment_mode": mode}, "operator": admin.display_name or admin.username, "role": admin.external_role or admin.auth_source}, severity="WARNING") + session.commit() + return RedirectResponse(f"/agents/{agent_id}", status_code=303) finally: session.close() def _ops_view(session, installation, agent): diff --git a/central/templates/agent_detail.html b/central/templates/agent_detail.html index 7217443..5dc2377 100644 --- a/central/templates/agent_detail.html +++ b/central/templates/agent_detail.html @@ -10,6 +10,14 @@ {% if ops.enabled %}
{% else %}
{% endif %}
Processamento
{{ "Ativo" if ops.enabled else "Pausado" }} · sincronização automática a cada {{ ops.interval_minutes }} min
Última sincronização
{{ ops.last_sync or "ainda não sincronizado" }}{% if ops.last_changed is not none %} · {{ ops.last_changed }} produto(s) alterado(s){% endif %}
Sincronização agendada
{{ ops.next_sync or ("aguardando o primeiro sync" if ops.enabled else "pausado") }}
{% if ops.rtc %}
Reforma Tributária
{{ ops.rtc.label }}
{% endif %}{% if ops.rtc and ops.rtc.enabled %}{% if ops.rtc.products and not ops.rtc.products.applicable %}
cClassTrib
{{ ops.rtc.products.label }}
{% elif ops.rtc.products %}
Produtos RTC
{{ ops.rtc.products.headline }}
Total {{ ops.rtc.products.total }} · com cClassTrib válido {{ ops.rtc.products.valid }} · prontos para preenchimento {{ ops.rtc.products.ready }} · ambíguos/bloqueados {{ ops.rtc.products.blocked }}
{% endif %}
Preenchimento automático
{{ ops.rtc.apply.state }}{% if ops.rtc.apply.applied_last_job %} · {{ ops.rtc.apply.applied_last_job }} preenchido(s) no último job{% endif %} · Controle e candidatos cClassTrib
{% endif %}
Último job
{% if ops.last_job %}{{ ops.last_job.kind_label }} · {{ ops.last_job.status_label }}{% else %}nenhum{% endif %}
+

Funcionalidades

Decididas no Hub para esta instalação — o Agent não tem chave local.

+{% if features.waiting %}

Aguardando configuração de funcionalidades: esta instalação ainda não processa nada.

{% endif %} +
Corretor fiscal
{{ "Ativo" if features.corrector_enabled else "Desativado" }}
Enriquecimento de produtos novos
{{ features.state_label }}
+{% if is_admin %}
{% else %}

Somente administradores alteram as funcionalidades.

{% endif %}
+

Enriquecimento

Produtos novos cadastrados no SGV; o catálogo histórico nunca é enviado a fontes externas.

+
Produtos acompanhados
{{ features.total }}
Campos preenchidos
{{ features.applied }}
{% for status, n in features.counts.items() %}
{{ status }}
{{ n }}
{% endfor %}
+
+{% if features.lookup %}{% if features.lookup.found %}
Produto
{{ features.lookup.row.product_code }} · {{ features.lookup.row.description }}
GTIN
{{ features.lookup.row.gtin or "sem GTIN válido" }}
Estado
{{ features.lookup.row.status }}{% if features.lookup.row.decisions.state %} · {{ features.lookup.row.decisions.state }}{% endif %}
{% for a in features.lookup.audits %}
{{ a.field }}: {{ a.old_value or "vazio" }} → {{ a.new_value or "—" }}{{ a.outcome }} · regra {{ a.rule or "—" }} · base {{ a.kb_status or "—" }}{% if a.evidence %} · {{ a.evidence|join(", ") }}{% endif %}
{% endfor %}{% else %}

Produto sem registro de enriquecimento.

{% endif %}{% endif %}

Contexto da empresa

Cliente
{{ detail.client or "Não informado" }}
Nome de exibição
{{ detail.display_name or "Não informado" }}
Hostname
{{ detail.hostname or "Não informado" }}
Empresa
{{ detail.company_name or detail.company_code or "Não informado" }}
UF
{{ detail.company_uf or "Não informado" }}
CRT
{{ detail.company_crt or "Não informado" }}
Versão do agent
{{ detail.agent_version_display }}
Último heartbeat
{{ detail.last_seen_display }} · {{ detail.last_seen_age }}
Firebird
{{ detail.firebird_version or "Não informado" }} · ODS {{ detail.firebird_ods or "Não informado" }}
IDs técnicos
Agent {{ detail.technical_ids.agent_id }} · instalação {{ detail.technical_ids.installation_db_id }}

Eventos recentes

Ver eventos
{% for event in events[:8] %}
{{ event.occurred_display }}{{ event.event_type }}{{ event.message }}{% if event.job_id %} · Job {{ event.job_id }}{% endif %}{% if event.product_code %} · Produto {{ event.product_code }}{% endif %}Severidade: {{ event.severity }}
{% else %}

Nenhum evento.

{% endfor %}
diff --git a/central/templates/agent_new_installation.html b/central/templates/agent_new_installation.html index a63ec24..54f629d 100644 --- a/central/templates/agent_new_installation.html +++ b/central/templates/agent_new_installation.html @@ -32,6 +32,19 @@ {% for release in releases %}{% endfor %} +
+ Funcionalidades desta instalação + + + + Nada é processado até as funcionalidades estarem definidas. Pode ser alterado depois, no detalhe do Agent. +