GTI_GERENCIADOR_FISCAL/fiscal/scripts/enrichment_local_hub_e2e.py

"""E2E of the enrichment across REAL HTTP: the real Agent worker (real Firebird COPY) <-> the real Hub container (dev PostgreSQL, real datasets, real providers).

    docker compose up -d --build fiscal-hub   # Hub at http://localhost:8000
    python scripts/enrichment_local_hub_e2e.py

Registers a throw-away dev Agent (AGENT_DEV_TOKEN), flips its per-installation features straight in the dev database (the panel route is covered by tests), seeds the KB of the dev Hub with a VERIFIED NCM for
one test GTIN (two independent provider families), then walks: DISABLED -> DRY_RUN -> ENABLED -> DISABLED, an unknown GTIN (real providers, internet) and an invalid GTIN (no provider call).
Writes go ONLY to scratch_rtc_e2e/TESTESOFICIAIS_RTC_COPY.FDB (guarded by rtc_copy_write_e2e.guard). The dev database receives test rows only.
"""
import json
import os
import subprocess
import sys
import threading
import time
import uuid
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT)); sys.path.insert(0, str(ROOT / "scripts")); sys.path.insert(0, str(ROOT / "agent" / "src"))
from dotenv import load_dotenv  # noqa: E402

load_dotenv(ROOT / ".env")
import enrichment_copy_e2e as ce  # noqa: E402  (also patches base.connect with the brief retry)
from gti_fiscal_agent import enrichment as E  # noqa: E402
from gti_fiscal_agent.client import HubClient  # noqa: E402

base = ce.base
HUB = "http://127.0.0.1:8000"          # NOT "localhost": on Windows it resolves ::1 first and adds ~500 ms per request
TOKEN = os.environ["AGENT_DEV_TOKEN"]
CONTAINER = "projeto-postgres-1"


def sql(query):
    done = subprocess.run(["docker", "exec", CONTAINER, "psql", "-U", "gti", "-d", "gti_fiscal", "-At", "-c", query], capture_output=True, text=True, timeout=60)
    if done.returncode != 0:
        raise RuntimeError(done.stderr[-300:])
    return done.stdout.strip()


def seed_kb(gtin14, description, ncm):
    code = f'''
from central.app.db import SessionLocal
from central.app.enrichment import kb
with SessionLocal() as s:
    p = kb.upsert_product(s, "{gtin14}", "{description}")
    kb.add_evidence(s, p, "ncm", "{ncm}", "provider:cosmos", family="PROVIDER:cosmos")
    kb.add_evidence(s, p, "ncm", "{ncm}", "provider:other", family="PROVIDER:other")
    s.commit()
print("seeded")
'''
    done = subprocess.run(["docker", "exec", "-i", "projeto-fiscal-hub-1", "python", "-"], input=code, capture_output=True, text=True, timeout=120, cwd=str(ROOT))
    assert "seeded" in done.stdout, done.stderr[-400:]


def set_features(installation_id, mode, corrector=True):
    sql(f"UPDATE installations SET enrichment_mode='{mode}', corrector_enabled={'true' if corrector else 'false'}, features_configured=true WHERE installation_id='{installation_id}'")


def main():
    checks = {}
    base.restore_copy()
    client = HubClient(HUB, TOKEN)
    try:
        installation_id = client.identity().get("installation_id") or str(uuid.uuid4())        # the dev Agent may already own an installation
    except Exception:  # noqa: BLE001
        installation_id = str(uuid.uuid4())
    client.heartbeat_metadata({"installation_id": installation_id, "status": "IDLE", "agent_version": "1.3.0", "display_name": "E2E-ENRICH", "capabilities": {"PRODUCT_ENRICHMENT_WRITE_V1": True}})
    set_features(installation_id, "DISABLED")                                   # start every run from the default
    checks["REGISTERED_DEFAULT_DISABLED"] = client.authorizations()["features"]["enrichment_mode"] == "DISABLED"
    gtin_ok = ce.gtin("789500000001")
    seed_kb("0" + gtin_ok, "REFRIGERANTE COLA LATA 350ML", "22021000")
    for suffix in ("", "-wal", "-shm"):
        try:
            (ROOT / "scratch_enrich" / f"enrich_hub_e2e.db{suffix}").unlink()
        except FileNotFoundError:
            pass
    store = E.EnrichmentStore(ROOT / "scratch_enrich" / "enrich_hub_e2e.db")
    worker = E.EnrichmentWorker(hub=client, store=store, read_conn=lambda: base.connect(False), write_conn=lambda: base.connect(True), create_reprocess=lambda codes: client.create_job("REPROCESS_PRODUCTS", {"codigos": list(codes), "reason": "ENRICHMENT"}))

    class Stop:
        stop = False
    stopper = Stop()
    threading.Thread(target=worker.run_forever, args=(stopper,), daemon=True).start()

    def refresh():
        worker._features_at = -999
        worker._auth_cache = (None, 0.0)

    time.sleep(4)
    checks["DISABLED_NO_ACTIVITY"] = store.meta("baseline_done") is None
    set_features(installation_id, "DRY_RUN"); refresh(); time.sleep(6)
    checks["BASELINE_ON_ACTIVATION"] = store.meta("baseline_done") == "1"
    ce.insert_product("HD001", "REFRIGERANTE COLA LATA 350ML", gtin_ok); time.sleep(8)
    checks["DRY_RUN_ZERO_WRITES"] = ce.row("HD001") == (None, None, None)
    audits = int(sql(f"SELECT count(*) FROM product_field_audits a JOIN installations i ON i.id=a.installation_id WHERE i.installation_id='{installation_id}' AND a.outcome='DRY_RUN' AND a.product_code='HD001'"))
    checks["DRY_RUN_AUDITED_IN_HUB"] = audits >= 1
    set_features(installation_id, "ENABLED"); refresh()
    ce.insert_product("HE001", "REFRIGERANTE COLA LATA 350ML", gtin_ok)
    t0 = time.monotonic()
    while time.monotonic() - t0 < 25 and ce.row("HE001")[0] is None:
        time.sleep(0.2)
    latency = round(time.monotonic() - t0, 2)
    r = ce.row("HE001")
    checks["ENABLED_NCM_WRITTEN"] = r[0] == "22021000"
    time.sleep(3)                                                              # the Agent reports the audit right after the write
    applied = sql(f"SELECT field||':'||coalesce(new_value,'') FROM product_field_audits a JOIN installations i ON i.id=a.installation_id WHERE i.installation_id='{installation_id}' AND a.outcome='APPLIED' AND a.product_code='HE001' ORDER BY a.id")
    checks["WRITTEN_FIELDS_MATCH_HUB_AUDIT"] = ("CLASSIFICACAO_FISCAL:22021000" in applied) and all(v in applied for v in ([f"CEST:{r[1]}"] if r[1] else []) + ([f"IPI:{r[2]}"] if r[2] not in (None,) else []))
    time.sleep(6)
    hd = ce.row("HD001")
    checks["EARLIER_DRY_RUN_PRODUCT_STILL_UNTOUCHED"] = hd == (None, None, None)
    # unknown GTIN goes to the REAL providers (internet); nothing may be written, a human blocker must appear
    unknown = ce.gtin("789600000009")
    ce.insert_product("HE002", "PRODUTO INEXISTENTE ESPECIAL XYZ 123", unknown); time.sleep(14)
    checks["UNKNOWN_GTIN_NO_WRITE"] = ce.row("HE002") == (None, None, None)
    blockers = sql("SELECT reason_code FROM blocked_products WHERE field='ENRICHMENT' AND product_code='HE002'")
    checks["UNKNOWN_GTIN_BLOCKER_PT"] = blockers in ("NOT_FOUND", "SINGLE_WEAK_SOURCE", "PENDING_PROVIDER_QUOTA", "PENDING_PROVIDERS_OFFLINE") or blockers.startswith("NOT_FOUND")
    bad = unknown[:-1] + str((int(unknown[-1]) + 1) % 10)
    before_req = sql("SELECT coalesce(sum(requests),0) FROM provider_usage_daily")
    ce.insert_product("HE003", "PRODUTO GTIN INVALIDO", bad); time.sleep(8)
    checks["INVALID_GTIN_NO_PROVIDER_CALL"] = sql("SELECT coalesce(sum(requests),0) FROM provider_usage_daily") == before_req and ce.row("HE003") == (None, None, None)
    # feature switched off centrally: the next new product is not touched
    set_features(installation_id, "DISABLED"); refresh(); time.sleep(4)
    ce.insert_product("HE004", "REFRIGERANTE COLA LATA 350ML", gtin_ok); time.sleep(8)
    checks["DISABLED_AGAIN_ZERO_WRITES"] = ce.row("HE004") == (None, None, None)
    # corrector off + enrichment on: independent switches
    set_features(installation_id, "ENABLED", corrector=False); refresh(); time.sleep(3)
    checks["CORRECTOR_OFF_NO_JOBS"] = client.next_job() is None
    stopper.stop = True
    result = {"copy_only": True, "hub": HUB, "detect_to_written_seconds_enabled": latency, "checks": checks, "ALL_PASS": all(checks.values())}
    (ROOT / "docs" / "runs" / "enrichment_local_hub_e2e.json").write_text(json.dumps(result, indent=1), encoding="utf-8")
    print(json.dumps(result, indent=1))
    sys.exit(0 if result["ALL_PASS"] else 1)


if __name__ == "__main__":
    main()