#!/usr/bin/env python3
"""
radprice_sync.py - integracao de precos com o RadPrice (by Datatoms).

Le a view "ouro" do seu DW (padrao: radprice_ofertas) e envia para o RadPrice com a sua chave, OU responde
na hora as buscas do RadPrice (modo direto). Documentacao: https://radprice.datatoms.com/integracao/documentacao

Comandos:
  python radprice_sync.py check                 confere configuracao, DW, colunas, amostra e a chave
  python radprice_sync.py sync                  envio COMPLETO (o que nao vier e removido do RadPrice)
  python radprice_sync.py sync --incremental    envia so o que mudou desde o ultimo envio com sucesso
  python radprice_sync.py watch                 fica rodando: incremental a cada intervalo e completo 1x/dia
  python radprice_sync.py serve                 modo direto: responde as buscas assinadas pelo RadPrice

Configuracao: arquivo .env ao lado do script (ou variaveis de ambiente). Veja radprice.env.exemplo.
Requisitos: Python 3.10+, pip install -r requirements.txt (+ o driver do seu banco).
"""
from __future__ import annotations

import argparse
import base64
import datetime as dt
import json
import logging
import os
import re
import signal
import ssl
import sys
import threading
import time
import unicodedata
from decimal import Decimal, InvalidOperation
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any, Iterator
from urllib.parse import parse_qs, urlparse

import requests
import sqlalchemy as sa

VERSAO = "1.0.0"
LOTE = 1000
log = logging.getLogger("radprice")

# Coluna da view -> campo da API. Obrigatorias: id_externo, nome, preco, cidade, uf.
COLUNAS = {
    "id_externo": "externalId",
    "nome": "name",
    "marca": "brand",
    "gtin": "gtin",
    "sku": "sku",
    "categoria": "category",
    "unidade": "unit",
    "preco": "price",
    "preco_de": "regularPrice",
    "estoque": "stock",
    "link_produto": "url",
    "loja": "store",
    "endereco": "address",
    "cidade": "city",
    "uf": "uf",
    "codigo_ibge": "ibgeCode",
    "latitude": "lat",
    "longitude": "lon",
    "atualizado_em": "updatedAt",
}
OBRIGATORIAS = ["id_externo", "nome", "preco", "cidade", "uf"]
# Opcionais que nao vao para a API: "ativo" (0/false = remover do RadPrice) e "busca" (texto sem acento para o modo direto).
EXTRAS = ["ativo", "busca"]
ESTOQUE = {"em_estoque", "sob_encomenda", "sem_estoque"}
UFS = set("AC AL AP AM BA CE DF ES GO MA MT MS MG PA PB PR PE PI RJ RN RS RO RR SC SP SE TO".split())


# ----------------------------------------------------------------------------- configuracao
def carregar_env(caminho: Path) -> None:
    """Le KEY=VALOR de um .env simples (sem sobrescrever variaveis ja definidas no sistema)."""
    if not caminho.exists():
        return
    for linha in caminho.read_text(encoding="utf-8").splitlines():
        linha = linha.strip()
        if not linha or linha.startswith("#") or "=" not in linha:
            continue
        k, v = linha.split("=", 1)
        v = v.strip().strip('"').strip("'")
        os.environ.setdefault(k.strip(), v)


class Config:
    def __init__(self) -> None:
        self.url = os.environ.get("RADPRICE_URL", "https://radprice.datatoms.com").rstrip("/")
        self.chave = os.environ.get("RADPRICE_CHAVE", "").strip()
        self.dw_url = os.environ.get("DW_URL", "").strip()
        self.view = os.environ.get("DW_VIEW", "radprice_ofertas").strip()
        self.intervalo_min = int(os.environ.get("RADPRICE_INTERVALO_MIN", "60"))
        self.fuso_horas = int(os.environ.get("RADPRICE_FUSO_HORAS", "-3"))
        self.estado = Path(os.environ.get("RADPRICE_ESTADO", str(Path(__file__).with_name(".radprice_estado.json"))))
        self.serve_host = os.environ.get("RADPRICE_SERVE_HOST", "127.0.0.1")
        self.serve_porta = int(os.environ.get("RADPRICE_SERVE_PORTA", "8787"))
        self.serve_caminho = os.environ.get("RADPRICE_SERVE_CAMINHO", "/radprice/buscar")
        # Se um proxy muda o caminho (ex.: publico /api/radprice/buscar -> local /radprice/buscar), informe o publico.
        self.caminho_publico = os.environ.get("RADPRICE_CAMINHO_PUBLICO", "").strip()
        self.tls_cert = os.environ.get("RADPRICE_TLS_CERT", "").strip()
        self.tls_chave = os.environ.get("RADPRICE_TLS_CHAVE", "").strip()
        self.chave_publica_pem = os.environ.get("RADPRICE_CHAVE_PUBLICA_PEM", "").replace("\\n", "\n").strip()

    def validar(self, precisa_chave: bool = True) -> list[str]:
        erros = []
        if not self.dw_url:
            erros.append("DW_URL nao definido (ex.: postgresql+psycopg://leitor:senha@host:5432/dw)")
        if precisa_chave and not re.fullmatch(r"rpk_[A-Za-z0-9]{8}_[A-Za-z0-9_-]{43}", self.chave):
            erros.append("RADPRICE_CHAVE ausente ou em formato invalido (rpk_XXXXXXXX_...)")
        if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)?", self.view):
            erros.append("DW_VIEW invalido: use nome simples ou esquema.nome (letras, numeros e _)")
        if not self.url.startswith("https://") and not self.url.startswith("http://localhost"):
            erros.append("RADPRICE_URL precisa ser https://")
        return erros


# ----------------------------------------------------------------------------- leitura do DW
class DW:
    def __init__(self, cfg: Config) -> None:
        self.cfg = cfg
        self.engine = sa.create_engine(cfg.dw_url, pool_pre_ping=True, pool_size=4, max_overflow=2)
        self._colunas: list[str] | None = None

    def tabela(self) -> str:
        # Nome validado por regex em Config.validar; aspas evitam conflito com palavras reservadas.
        prep = self.engine.dialect.identifier_preparer
        return ".".join(prep.quote(p) for p in self.cfg.view.split("."))

    def colunas(self) -> list[str]:
        if self._colunas is None:
            with self.engine.connect() as c:
                res = c.execute(sa.text(f"SELECT * FROM {self.tabela()} WHERE 1 = 0"))
                self._colunas = [k.lower() for k in res.keys()]
        return self._colunas

    def usadas(self) -> list[str]:
        return [c for c in self.colunas() if c in COLUNAS or c in EXTRAS]

    def contar(self) -> int:
        with self.engine.connect() as c:
            return int(c.execute(sa.text(f"SELECT COUNT(*) FROM {self.tabela()}")).scalar() or 0)

    def ler(self, desde: dt.datetime | None = None, limite: int | None = None) -> Iterator[dict[str, Any]]:
        cols = self.usadas()
        sel = ", ".join(cols)
        sql = f"SELECT {sel} FROM {self.tabela()}"
        params: dict[str, Any] = {}
        if desde is not None:
            if "atualizado_em" not in cols:
                raise SystemExit("Envio incremental precisa da coluna atualizado_em na view.")
            sql += " WHERE atualizado_em > :desde"
            params["desde"] = desde.replace(tzinfo=None) if self.engine.dialect.name in ("mssql", "mysql", "sqlite") else desde
        with self.engine.connect().execution_options(stream_results=True) as c:
            res = c.execute(sa.text(sql), params)
            n = 0
            while True:
                linhas = res.fetchmany(LOTE)
                if not linhas:
                    break
                for row in linhas:
                    yield {k.lower(): v for k, v in row._mapping.items()}
                    n += 1
                    if limite and n >= limite:
                        return


# ----------------------------------------------------------------------------- conversao e validacao
def _txt(v: Any, maximo: int) -> str | None:
    if v is None:
        return None
    s = str(v).strip()
    return s[:maximo] if s else None


def _preco(v: Any) -> str | None:
    """Decimal/numero do banco ou texto ("38.90", "38,90", "1.234,56", "R$ 38,90") -> "38.9"."""
    if v is None or v == "":
        return None
    if isinstance(v, str):
        s = v.strip().replace("R$", "").replace(" ", "")
        if "," in s:
            s = s.replace(".", "").replace(",", ".")
    else:
        s = str(v)
    try:
        d = Decimal(s)
    except InvalidOperation:
        raise ValueError(f"preco invalido: {v!r}")
    if not d.is_finite() or d <= 0:
        raise ValueError("preco precisa ser maior que zero")
    texto = format(d.quantize(Decimal("0.0001")).normalize(), "f")
    if not re.fullmatch(r"\d{1,10}(\.\d{1,4})?", texto):
        raise ValueError(f"preco fora do formato: {v!r}")
    return texto


def _quando(v: Any, fuso: dt.timezone) -> str | None:
    if v is None or v == "":
        return None
    if isinstance(v, str):
        v = dt.datetime.fromisoformat(v.replace("Z", "+00:00"))
    if isinstance(v, dt.date) and not isinstance(v, dt.datetime):
        v = dt.datetime(v.year, v.month, v.day)
    if v.tzinfo is None:
        v = v.replace(tzinfo=fuso)
    return v.isoformat(timespec="seconds")


def _ativo(v: Any) -> bool:
    if v is None:
        return True
    return str(v).strip().lower() not in ("0", "false", "f", "nao", "não", "n", "inativo")


def converter(row: dict[str, Any], fuso: dt.timezone) -> dict[str, Any]:
    """Linha da view -> item da API. Levanta ValueError com o motivo se a linha nao puder ser enviada."""
    faltando = [c for c in OBRIGATORIAS if row.get(c) in (None, "")]
    if faltando:
        raise ValueError("coluna(s) vazia(s): " + ", ".join(faltando))
    item: dict[str, Any] = {
        "externalId": _txt(row["id_externo"], 80),
        "name": _txt(row["nome"], 200),
        "price": _preco(row["preco"]),
        "city": _txt(row["cidade"], 80),
        "uf": str(row["uf"]).strip().upper(),
    }
    if item["uf"] not in UFS:
        raise ValueError(f"uf invalida: {row['uf']!r}")
    for col, campo, maximo in (("marca", "brand", 80), ("sku", "sku", 60), ("categoria", "category", 80), ("unidade", "unit", 60),
                               ("loja", "store", 120), ("endereco", "address", 200), ("link_produto", "url", 500)):
        val = _txt(row.get(col), maximo)
        if val:
            item[campo] = val
    if item.get("url") and not item["url"].startswith("https://"):
        raise ValueError("link_produto precisa comecar com https://")
    if row.get("preco_de") not in (None, ""):
        item["regularPrice"] = _preco(row["preco_de"])
    if row.get("gtin") not in (None, ""):
        g = re.sub(r"\D", "", str(row["gtin"]))
        if not re.fullmatch(r"[0-9]{8}|[0-9]{12,14}", g):
            raise ValueError(f"gtin invalido: {row['gtin']!r}")
        item["gtin"] = g
    if row.get("estoque") not in (None, ""):
        e = str(row["estoque"]).strip().lower()
        if e not in ESTOQUE:
            raise ValueError(f"estoque invalido: {row['estoque']!r} (use em_estoque, sob_encomenda ou sem_estoque)")
        item["stock"] = e
    if row.get("codigo_ibge") not in (None, ""):
        item["ibgeCode"] = re.sub(r"\D", "", str(row["codigo_ibge"]))
    for col, campo in (("latitude", "lat"), ("longitude", "lon")):
        if row.get(col) not in (None, ""):
            item[campo] = float(row[col])
    quando = _quando(row.get("atualizado_em"), fuso)
    if quando:
        item["updatedAt"] = quando
    return item


# ----------------------------------------------------------------------------- cliente da API
class RadPrice:
    def __init__(self, cfg: Config) -> None:
        self.cfg = cfg
        self.s = requests.Session()
        self.s.headers.update({"Authorization": f"Bearer {cfg.chave}", "User-Agent": f"radprice-sync/{VERSAO}", "Content-Type": "application/json"})

    def _req(self, metodo: str, caminho: str, corpo: Any = None) -> dict[str, Any]:
        url = f"{self.cfg.url}/api/v1{caminho}"
        espera = 2.0
        for tentativa in range(1, 6):
            try:
                r = self.s.request(metodo, url, data=json.dumps(corpo) if corpo is not None else None, timeout=(10, 120))
            except requests.RequestException as e:
                if tentativa == 5:
                    raise RuntimeError(f"Sem conexao com o RadPrice: {e}") from e
                log.warning("Falha de rede (%s); nova tentativa em %.0fs", e.__class__.__name__, espera)
                time.sleep(espera)
                espera *= 2
                continue
            if r.status_code in (429, 502, 503, 504) and tentativa < 5:
                pausa = float(r.headers.get("retry-after") or espera)
                log.warning("RadPrice respondeu %s; nova tentativa em %.0fs", r.status_code, pausa)
                time.sleep(pausa)
                espera *= 2
                continue
            dados = r.json() if r.content and "json" in r.headers.get("content-type", "") else {}
            if r.status_code >= 400:
                err = dados.get("error", {}) if isinstance(dados, dict) else {}
                raise RuntimeError(f"RadPrice {r.status_code} {err.get('code', '')}: {err.get('message', r.text[:200])}")
            return dados
        raise RuntimeError("RadPrice indisponivel")

    def me(self) -> dict[str, Any]:
        return self._req("GET", "/integrations/me")

    def iniciar(self, tipo: str) -> str:
        return self._req("POST", "/integrations/sync", {"kind": tipo})["runId"]

    def lote(self, run_id: str, itens: list[dict[str, Any]], remover: list[str]) -> dict[str, Any]:
        return self._req("POST", f"/integrations/sync/{run_id}/items", {"items": itens, "remove": remover})

    def finalizar(self, run_id: str) -> dict[str, Any]:
        return self._req("POST", f"/integrations/sync/{run_id}/finish", {})


# ----------------------------------------------------------------------------- estado local
def ler_estado(cfg: Config) -> dict[str, Any]:
    try:
        return json.loads(cfg.estado.read_text(encoding="utf-8"))
    except (FileNotFoundError, ValueError):
        return {}


def gravar_estado(cfg: Config, estado: dict[str, Any]) -> None:
    tmp = cfg.estado.with_suffix(".tmp")
    tmp.write_text(json.dumps(estado, indent=2), encoding="utf-8")
    tmp.replace(cfg.estado)


# ----------------------------------------------------------------------------- comandos
def cmd_check(cfg: Config) -> int:
    ok = True

    def linha(estado: str, texto: str) -> None:
        print(f"[{estado}] {texto}")

    erros = cfg.validar()
    for e in erros:
        linha("erro", e)
    if erros:
        return 1
    linha("ok", f"configuracao lida (RadPrice: {cfg.url}, view: {cfg.view})")
    try:
        dw = DW(cfg)
        cols = dw.colunas()
        linha("ok", f"conexao com o DW ({dw.engine.dialect.name})")
    except Exception as e:  # noqa: BLE001 - mostrar qualquer falha do driver
        linha("erro", f"nao foi possivel ler a view {cfg.view}: {e}")
        return 1
    faltando = [c for c in OBRIGATORIAS if c not in cols]
    if faltando:
        linha("erro", "colunas obrigatorias ausentes na view: " + ", ".join(faltando))
        ok = False
    else:
        linha("ok", "colunas obrigatorias presentes")
    opcionais = [c for c in list(COLUNAS) + EXTRAS if c in cols and c not in OBRIGATORIAS]
    linha("ok", "colunas opcionais usadas: " + (", ".join(opcionais) or "nenhuma"))
    ignoradas = [c for c in cols if c not in COLUNAS and c not in EXTRAS]
    if ignoradas:
        linha("aviso", "colunas ignoradas (nao sao enviadas): " + ", ".join(ignoradas))
    if "atualizado_em" not in cols:
        linha("aviso", "sem atualizado_em: so envio completo (sem incremental)")
    if ok:
        total = dw.contar()
        linha("ok", f"{total} linha(s) na view")
        fuso = dt.timezone(dt.timedelta(hours=cfg.fuso_horas))
        ruins: list[str] = []
        n = 0
        for row in dw.ler(limite=500):
            n += 1
            try:
                converter(row, fuso)
            except (ValueError, TypeError) as e:
                ruins.append(f"{row.get('id_externo')}: {e}")
        if ruins:
            linha("aviso", f"{len(ruins)} de {n} linhas da amostra seriam recusadas. Primeiras:")
            for r in ruins[:5]:
                print("        " + r)
        else:
            linha("ok", f"amostra de {n} linha(s) valida")
    try:
        me = RadPrice(cfg).me()
        linha("ok", f"chave aceita: {me['partner']} - situacao {me['status']}, modo {me['mode']}, plano {me['plan']}, {me['items']} itens no RadPrice")
        if not me.get("visibleInSearch"):
            linha("aviso", "a integracao ainda NAO aparece na busca (situacao diferente de ativa). Fale com comercial@datatoms.com.")
        if me.get("mode") == "direct":
            linha("aviso", "modo direto: use 'serve' (nao 'sync'). Links aceitos: " + ", ".join(me.get("linkHosts", [])))
        else:
            linha("ok", "links de produto aceitos em: " + ", ".join(me.get("linkHosts", [])))
        ultimo = me.get("lastRun")
        if ultimo:
            linha("ok" if ultimo["status"] == "done" else "aviso", f"ultimo envio: {ultimo['status']} em {ultimo['started_at']} ({ultimo['upserted']} gravados, {ultimo['rejected']} recusados)")
    except RuntimeError as e:
        linha("erro", str(e))
        ok = False
    print("Tudo certo." if ok else "Corrija os itens [erro] acima.")
    return 0 if ok else 1


def cmd_sync(cfg: Config, incremental: bool) -> int:
    erros = cfg.validar()
    if erros:
        for e in erros:
            log.error(e)
        return 1
    estado = ler_estado(cfg)
    desde = None
    if incremental:
        if not estado.get("ultimo_sucesso"):
            log.info("Sem envio anterior registrado: fazendo envio completo.")
            incremental = False
        else:
            desde = dt.datetime.fromisoformat(estado["ultimo_sucesso"])
    tipo = "incremental" if incremental else "full"
    fuso = dt.timezone(dt.timedelta(hours=cfg.fuso_horas))
    inicio_leitura = dt.datetime.now(dt.timezone.utc).astimezone(fuso) - dt.timedelta(minutes=2)  # folga para relogios
    dw = DW(cfg)
    api = RadPrice(cfg)
    run_id = api.iniciar(tipo)
    log.info("Envio %s iniciado (%s)", tipo, run_id)
    itens: list[dict[str, Any]] = []
    remover: list[str] = []
    enviados = recusados_local = recusados_api = removidos = 0
    tem_ativo = "ativo" in dw.colunas()

    def descarregar() -> None:
        nonlocal enviados, recusados_api, removidos
        if not itens and not remover:
            return
        r = api.lote(run_id, itens, remover)
        enviados += r["accepted"]
        removidos += r["removed"]
        recusados_api += len(r["rejected"])
        for rej in r["rejected"][:5]:
            log.warning("Recusado pelo RadPrice: %s - %s", rej.get("externalId"), rej.get("error"))
        itens.clear()
        remover.clear()

    for row in dw.ler(desde=desde):
        if tem_ativo and not _ativo(row.get("ativo")):
            if row.get("id_externo") not in (None, ""):
                remover.append(str(row["id_externo"]).strip()[:80])
        else:
            try:
                itens.append(converter(row, fuso))
            except (ValueError, TypeError) as e:
                recusados_local += 1
                if recusados_local <= 5:
                    log.warning("Linha ignorada (%s): %s", row.get("id_externo"), e)
        if len(itens) >= LOTE or len(remover) >= LOTE:
            descarregar()
    descarregar()
    try:
        fim = api.finalizar(run_id)
    except RuntimeError as e:
        log.error("Envio nao concluido: %s", e)
        return 1
    estado["ultimo_sucesso"] = inicio_leitura.isoformat()
    if tipo == "full":
        estado["ultimo_completo"] = inicio_leitura.isoformat()
    gravar_estado(cfg, estado)
    log.info("Envio %s concluido: %s gravados, %s removidos, %s recusados aqui, %s recusados pelo RadPrice. Total no RadPrice: %s.",
             tipo, enviados, removidos + fim.get("removed", 0), recusados_local, recusados_api, fim.get("total"))
    return 0


def cmd_watch(cfg: Config) -> int:
    parar = threading.Event()
    for sig in (signal.SIGINT, signal.SIGTERM):
        signal.signal(sig, lambda *_: parar.set())
    log.info("Sincronizacao continua: a cada %s min (completo 1x por dia). Ctrl+C encerra.", cfg.intervalo_min)
    falhas = 0
    while not parar.is_set():
        estado = ler_estado(cfg)
        ultimo_completo = estado.get("ultimo_completo")
        completo = not ultimo_completo or dt.datetime.now(dt.timezone.utc) - dt.datetime.fromisoformat(ultimo_completo) > dt.timedelta(hours=24)
        try:
            codigo = cmd_sync(cfg, incremental=not completo)
            falhas = 0 if codigo == 0 else falhas + 1
        except Exception as e:  # noqa: BLE001 - o servico nunca para por um erro de ciclo
            falhas += 1
            log.error("Ciclo falhou: %s", e)
        espera = cfg.intervalo_min * 60 if falhas == 0 else min(cfg.intervalo_min * 60, 60 * 2 ** min(falhas, 5))
        parar.wait(espera)
    log.info("Encerrado.")
    return 0


# ----------------------------------------------------------------------------- modo direto
def _sem_acento(s: str) -> str:
    return "".join(c for c in unicodedata.normalize("NFD", s.lower()) if unicodedata.category(c) != "Mn")


STOP = {"de", "da", "do", "das", "dos", "com", "para", "por", "e", "em", "a", "o", "as", "os", "um", "uma", "kit"}


def tokens(q: str) -> list[str]:
    n = re.sub(r"(\d)([a-z])", r"\1 \2", _sem_acento(q))
    n = re.sub(r"([a-z])(\d)", r"\1 \2", n)
    out: list[str] = []
    for t in re.split(r"[^a-z0-9]+", n):
        if t and t not in STOP and t not in out:
            out.append(t)
    return out[:8]


class Verificador:
    """Confere a assinatura Ed25519 do RadPrice (chave publica baixada do RadPrice ou fixada no .env)."""

    def __init__(self, cfg: Config) -> None:
        from cryptography.hazmat.primitives.serialization import load_pem_public_key

        self._load = load_pem_public_key
        self.cfg = cfg
        self.chave = None
        self.quando = 0.0
        self.lock = threading.Lock()

    def _chave(self):
        with self.lock:
            if self.chave is None or time.time() - self.quando > 6 * 3600:
                pem = self.cfg.chave_publica_pem
                if not pem:
                    pem = requests.get(f"{self.cfg.url}/api/v1/integrations/public-key", timeout=10).json()["publicKeyPem"]
                self.chave = self._load(pem.encode())
                self.quando = time.time()
            return self.chave

    def valido(self, ts: str, assinatura: str, caminho: str) -> bool:
        from cryptography.exceptions import InvalidSignature

        if not ts.isdigit() or abs(time.time() - int(ts)) > 60:
            return False
        try:
            self._chave().verify(base64.b64decode(assinatura), f"{ts}\n{caminho}".encode())
            return True
        except (InvalidSignature, ValueError):
            return False


def cmd_serve(cfg: Config) -> int:
    erros = cfg.validar(precisa_chave=False)
    if erros:
        for e in erros:
            log.error(e)
        return 1
    dw = DW(cfg)
    cols = dw.colunas()
    faltando = [c for c in OBRIGATORIAS if c not in cols]
    if faltando:
        log.error("Colunas obrigatorias ausentes: %s", ", ".join(faltando))
        return 1
    usadas = dw.usadas()
    campo_busca = "busca" if "busca" in cols else "nome"
    verif = Verificador(cfg)
    fuso = dt.timezone(dt.timedelta(hours=cfg.fuso_horas))
    tem_ativo = "ativo" in cols

    def buscar(q: str, uf: str | None, limite: int) -> list[dict[str, Any]]:
        ts = sorted(tokens(q), key=len, reverse=True)[:4]
        if not ts:
            return []
        conds = " AND ".join(f"LOWER({campo_busca}) LIKE :t{i} ESCAPE '!'" for i in range(len(ts)))
        params: dict[str, Any] = {f"t{i}": "%" + re.sub(r"([!%_])", r"!\1", t) + "%" for i, t in enumerate(ts)}
        ordem = "CASE WHEN uf = :uf THEN 0 ELSE 1 END, preco" if uf else "preco"
        if uf:
            params["uf"] = uf
        sql = f"SELECT {', '.join(usadas)} FROM {dw.tabela()} WHERE {conds} ORDER BY {ordem}"
        itens: list[dict[str, Any]] = []
        with dw.engine.connect() as c:
            res = c.execute(sa.text(sql), params)
            for row in res.fetchmany(limite * 3):
                linha = {k.lower(): v for k, v in row._mapping.items()}
                if tem_ativo and not _ativo(linha.get("ativo")):
                    continue
                try:
                    itens.append(converter(linha, fuso))
                except (ValueError, TypeError):
                    continue
                if len(itens) >= limite:
                    break
        return itens

    class Handler(BaseHTTPRequestHandler):
        server_version = f"radprice-sync/{VERSAO}"

        def _json(self, status: int, corpo: dict[str, Any]) -> None:
            dados = json.dumps(corpo, ensure_ascii=False).encode("utf-8")
            self.send_response(status)
            self.send_header("Content-Type", "application/json; charset=utf-8")
            self.send_header("Content-Length", str(len(dados)))
            self.send_header("Cache-Control", "no-store")
            self.end_headers()
            self.wfile.write(dados)

        def do_GET(self) -> None:  # noqa: N802 - nome exigido pelo http.server
            u = urlparse(self.path)
            if u.path == "/saude":
                return self._json(200, {"ok": True})
            if u.path != cfg.serve_caminho:
                return self._json(404, {"erro": "nao encontrado"})
            caminho_assinado = (cfg.caminho_publico or u.path) + (f"?{u.query}" if u.query else "")
            if not verif.valido(self.headers.get("x-radprice-timestamp", ""), self.headers.get("x-radprice-signature", ""), caminho_assinado):
                log.warning("Consulta recusada: assinatura invalida ou expirada (%s)", self.client_address[0])
                return self._json(401, {"erro": "assinatura invalida"})
            qs = parse_qs(u.query)
            q = (qs.get("q") or [""])[0].strip()
            uf = (qs.get("uf") or [""])[0].strip().upper() or None
            try:
                limite = max(1, min(10, int((qs.get("limit") or ["5"])[0])))
            except ValueError:
                limite = 5
            if not 3 <= len(q) <= 80 or (uf and uf not in UFS):
                return self._json(400, {"erro": "consulta invalida"})
            inicio = time.perf_counter()
            try:
                itens = buscar(q, uf, limite)
            except Exception as e:  # noqa: BLE001
                log.error("Erro ao consultar a view: %s", e)
                return self._json(500, {"erro": "falha interna"})
            log.info("Busca %r uf=%s -> %s itens em %.0f ms", q, uf, len(itens), (time.perf_counter() - inicio) * 1000)
            return self._json(200, {"items": itens})

        def log_message(self, *_: Any) -> None:
            return

    httpd = ThreadingHTTPServer((cfg.serve_host, cfg.serve_porta), Handler)
    if cfg.tls_cert and cfg.tls_chave:
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        ctx.minimum_version = ssl.TLSVersion.TLSv1_2
        ctx.load_cert_chain(cfg.tls_cert, cfg.tls_chave)
        httpd.socket = ctx.wrap_socket(httpd.socket, server_side=True)
    esquema = "https" if cfg.tls_cert else "http"
    log.info("Modo direto ouvindo em %s://%s:%s%s (busca em %s)", esquema, cfg.serve_host, cfg.serve_porta, cfg.serve_caminho, campo_busca)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass
    return 0


def main() -> int:
    carregar_env(Path(__file__).with_name(".env"))
    p = argparse.ArgumentParser(description="Integracao de precos com o RadPrice")
    sub = p.add_subparsers(dest="comando", required=True)
    sub.add_parser("check", help="confere configuracao, DW, colunas, amostra e chave")
    s = sub.add_parser("sync", help="envia a view para o RadPrice")
    s.add_argument("--incremental", action="store_true", help="so o que mudou desde o ultimo envio")
    sub.add_parser("watch", help="fica rodando e envia no intervalo configurado")
    sub.add_parser("serve", help="modo direto: responde as buscas do RadPrice")
    p.add_argument("--env", help="arquivo .env alternativo")
    p.add_argument("-v", "--verbose", action="store_true")
    a = p.parse_args()
    if a.env:
        carregar_env(Path(a.env))
    logging.basicConfig(level=logging.DEBUG if a.verbose else os.environ.get("RADPRICE_LOG", "INFO"), format="%(asctime)s %(levelname)s %(message)s")
    cfg = Config()
    if a.comando == "check":
        return cmd_check(cfg)
    if a.comando == "sync":
        return cmd_sync(cfg, a.incremental)
    if a.comando == "watch":
        return cmd_watch(cfg)
    return cmd_serve(cfg)


if __name__ == "__main__":
    sys.exit(main())