GTI_GERENCIADOR_FISCAL/fiscal/central/app/enrichment/backfill.py

"""Gradual, quota-safe verification of locally CORROBORATED GTINs by an NCM-bearing free provider (OSCBR). Never spends the operational reserve.

Rules: only CORROBORATED single-NCM GTINs (never VERIFIED_KB, never CONFLICT, never an invalid GTIN); each GTIN is checked once per `RECHECK_DAYS` (next_checked_at kept in the KB evidence, plus the
provider cache); a provider answer is EVIDENCE only - the promotion/conflict decision is kb.evaluate's (local strong evidence + independent provider NCM + vigente + zero conflict).
"""
from __future__ import annotations

import time
from datetime import datetime, timedelta, timezone

from sqlalchemy import select

from ..models import KbKnowledge, KbProduct
from . import kb
from .gtin import normalize_gtin

RECHECK_DAYS = 30


def _now():
    return datetime.now(timezone.utc)


def select_sample(session, is_official, limit: int, provider: str = "oscbr") -> list[dict]:
    """Priority: multi-client, more documents, more years; current-vigente NCM; zero conflict; not checked recently."""
    ids = list(session.scalars(select(KbKnowledge.kb_product_id).where(KbKnowledge.knowledge_key == "ncm", KbKnowledge.status == kb.CORROBORATED).distinct()))
    by_product: dict[int, list] = {}
    for start in range(0, len(ids), 500):                                        # every NCM row of those products: a second (even weaker) value already disqualifies the GTIN
        for r in session.scalars(select(KbKnowledge).where(KbKnowledge.knowledge_key == "ncm", KbKnowledge.kb_product_id.in_(ids[start:start + 500]))):
            by_product.setdefault(r.kb_product_id, []).append(r)
    now = _now().isoformat()
    picks = []
    for pid, items in by_product.items():
        valid = [i for i in items if is_official(i.value)]
        if len(valid) != 1 or valid[0].status != kb.CORROBORATED:
            continue                                                         # conflicts / no vigente NCM / already verified are not asked
        row = valid[0]
        info = (row.evidence or {}).get("info") or {}
        if info.get(f"{provider}_next_checked_at", "") > now:
            continue
        prod = session.get(KbProduct, pid)
        if not prod or normalize_gtin(prod.gtin[1:] if prod.gtin and prod.gtin.startswith("0") else prod.gtin)["status"] != "VALID_GTIN":
            continue
        summary = kb._lineage_summary([row])
        if any(f.startswith("PROVIDER:" + provider) for f in ((row.evidence or {}).get("families") or {})):
            continue
        picks.append({"gtin_key": prod.gtin, "ncm": row.value, "clients": len(summary["clients"]), "docs": summary["docs"], "years": len(summary["years"]), "product_id": pid})
    picks.sort(key=lambda p: (-p["clients"], -p["docs"], -p["years"], p["gtin_key"]))
    return picks[:limit]


def apply_outcome(session, gtin_key: str, provider: str, outcome, is_official) -> dict:
    """Turns one gateway Outcome into KB evidence. Returns the per-GTIN measurement record."""
    product = kb.find_by_gtin(session, gtin_key)
    rec = {"gtin": gtin_key[-13:] if gtin_key.startswith("0") else gtin_key, "state": outcome.state, "reason": outcome.reason, "found": False, "ncm": None, "cest": None, "description": None, "brand": None,
           "local_ncm": None, "official_valid": None, "kb_status": None, "external_conflict": False}
    local_rows = list(session.scalars(select(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == "ncm"))) if product else []
    rec["local_ncm"] = local_rows[0].value if local_rows else None
    stamp = (_now() + timedelta(days=RECHECK_DAYS)).isoformat()
    if outcome.state in ("FOUND", "CACHE_HIT") and outcome.product:
        p = outcome.product
        rec.update({"found": True, "ncm": p.ncm, "cest": p.cest, "description": p.title, "brand": p.brand})
        if p.ncm:
            rec["official_valid"] = bool(is_official(p.ncm))
            kb.add_evidence(session, product, "ncm", p.ncm, f"provider:{provider}"[:32], family=f"PROVIDER:{provider}", info={"raw_hash": p.raw_hash})
        if p.cest:
            kb.add_evidence(session, product, "cest", p.cest, f"provider:{provider}"[:32], family=f"PROVIDER:{provider}", info={"raw_hash": p.raw_hash})
        if p.brand and not product.brand:
            product.brand = p.brand
        if p.title:
            kb.add_alias(session, product, p.title.upper()[:200])
    ev = kb.evaluate(session, product, "ncm", is_official)
    rec["kb_status"] = ev["status"]
    rec["external_conflict"] = bool(ev["status"] == kb.CONFLICT and rec["ncm"])
    if outcome.state in ("FOUND", "CACHE_HIT", "NOT_FOUND"):                    # remember the check so it is not repeated (skipped/errors are retried later)
        for r in session.scalars(select(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == "ncm")):
            r.evidence = {**(r.evidence or {}), "info": {**((r.evidence or {}).get("info") or {}), f"{provider}_next_checked_at": stamp}}
    return rec


def provider_query(key: str) -> str:
    """Canonical form sent to providers: GTIN-8 stays 8 digits, UPC-12/EAN-13 as 13 digits (same rule as the resolver), GTIN-14 with an indicator digit stays 14."""
    if len(key) == 14 and key.startswith("000000"):
        return key[-8:]
    if len(key) == 14 and key.startswith("0"):
        return key[1:]
    return key


def run(session, gateway, provider, *, limit: int, reserve_percent: int, is_official, dry_run: bool = False, wait_rate_window: bool = True) -> dict:
    """One backfill pass. dry_run only lists the sample (no request)."""
    sample = select_sample(session, is_official, limit, provider.name)
    report = {"provider": provider.name, "sample_size": len(sample), "requests": 0, "records": [], "stopped": None, "dry_run": dry_run}
    if dry_run:
        report["records"] = [{k: v for k, v in s.items() if k != "product_id"} for s in sample]
        return report
    for item in sample:
        query = provider_query(item["gtin_key"])
        outcome = gateway.lookup(provider, query, "BACKFILL", reserve_percent)
        while outcome.state == "SKIPPED" and outcome.reason == "RATE_WINDOW" and wait_rate_window:      # per-minute governor: wait for the window, never exceed it
            time.sleep(max(1, min(65, int((outcome.retry_at - _now()).total_seconds()) + 1)) if outcome.retry_at else 12)
            outcome = gateway.lookup(provider, query, "BACKFILL", reserve_percent)
        if outcome.state == "SKIPPED":
            report["stopped"] = outcome.reason                                 # quota / reserve / per-minute window: stop, never insist
            break
        if not outcome.from_cache:
            report["requests"] += 1
        report["records"].append(apply_outcome(session, item["gtin_key"], provider.name, outcome, is_official))
        session.commit()
    return report