Blame for app/tars_web.py

"""openTARS 5.1: pesquisa web pelo SearXNG (decisão do Enzo em 11/10). Módulo puro, só biblioteca padrão.

Regras (plano 5.1, "Pesquisa web integrada"):
  * Ligada por padrão (26/10); sem SearXNG respondendo, o pedido cai na busca do navegador. Ligar mostra o que sai: o texto da pesquisa vai para o SearXNG, que o repassa aos buscadores.
  * Com ela desligada, nada é enviado (as ferramentas nem aparecem).
  * O que vem das páginas é DADO, nunca instrução, e entra cortado (contexto pequeno).
  * Só texto: sem scripts, sem baixar arquivos. read_page não abre endereço local nem de rede interna (senão uma página
    podia fazer a IA ler o Ollama ou outro serviço do próprio PC).

A configuração fica em ~/.config/opentars/config.json ("web": {"ligada", "endereco"}).
"""

import html.parser
import ipaddress
import json
import socket
import urllib.error
import urllib.parse
import urllib.request

import tars_i18n

ENDERECO_PADRAO = "http://127.0.0.1:8888"
LIMITE_RESULTADOS = 8
LIMITE_PAGINA_BYTES = 1_000_000
LIMITE_TEXTO_PAGINA = 8000
TIMEOUT = 10
AGENTE = "openTARS/5.1 (+https://project-opentars.github.io/OpenTARS/)"


def config():
    bruto = tars_i18n.ler_config().get("web")
    bruto = bruto if isinstance(bruto, dict) else {}
    endereco = str(bruto.get("endereco") or ENDERECO_PADRAO).strip().rstrip("/")
    return {"ligada": bool(bruto.get("ligada", True)), "endereco": endereco}


def salvar_config(ligada, endereco=None):
    atual = config()
    tars_i18n.salvar_config(web={"ligada": bool(ligada), "endereco": (endereco or atual["endereco"]).strip().rstrip("/")})


def buscar(consulta, endereco=None, abrir=None):
    """[{'titulo', 'url', 'trecho'}] do SearXNG (formato JSON). Levanta ValueError com um motivo curto se falhar."""
    consulta = " ".join(str(consulta or "").split())[:300]
    if not consulta:
        raise ValueError("empty query")
    url = f"{(endereco or config()['endereco'])}/search?" + urllib.parse.urlencode({"q": consulta, "format": "json"})
    pedido = urllib.request.Request(url, headers={"User-Agent": AGENTE, "Accept": "application/json"})
    try:
        with (abrir or urllib.request.urlopen)(pedido, timeout=TIMEOUT) as r:
            dados = json.loads(r.read(LIMITE_PAGINA_BYTES) or b"{}")
    except urllib.error.HTTPError as e:
        raise ValueError("json format disabled in SearXNG (settings.yml: search.formats)" if e.code == 403 else f"HTTP {e.code}")
    except (OSError, ValueError) as e:
        raise ValueError(f"SearXNG not reachable: {e}")
    saida = []
    for item in (dados.get("results") or [])[:LIMITE_RESULTADOS]:
        if isinstance(item, dict) and str(item.get("url", "")).startswith(("http://", "https://")):
            saida.append({"titulo": " ".join(str(item.get("title") or "").split())[:200], "url": item["url"],
                          "trecho": " ".join(str(item.get("content") or "").split())[:400]})
    return saida


def endereco_publico(url, resolver=None):
    """True só para http(s) cujo nome resolve para endereços públicos (nada de 127.0.0.1, 192.168.x, ::1, .local...)."""
    partes = urllib.parse.urlsplit(str(url or ""))
    if partes.scheme not in ("http", "https") or not partes.hostname:
        return False
    try:
        enderecos = {info[4][0] for info in (resolver or socket.getaddrinfo)(partes.hostname, partes.port or 443)}
    except (OSError, UnicodeError):
        return False
    for e in enderecos:
        ip = ipaddress.ip_address(e.split("%")[0])
        if not ip.is_global or ip.is_multicast:
            return False
    return bool(enderecos)


class _Texto(html.parser.HTMLParser):
    _PULAR = {"script", "style", "noscript", "svg", "template", "head"}
    _BLOCO = {"p", "div", "br", "li", "h1", "h2", "h3", "h4", "h5", "h6", "tr", "section", "article", "header", "footer"}

    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.partes, self._pulando = [], 0

    def handle_starttag(self, tag, attrs):
        if tag in self._PULAR:
            self._pulando += 1
        elif tag in self._BLOCO:
            self.partes.append("\n")

    def handle_endtag(self, tag):
        if tag in self._PULAR and self._pulando:
            self._pulando -= 1

    def handle_data(self, dados):
        if not self._pulando and dados.strip():
            self.partes.append(" ".join(dados.split()) + " ")


def texto_do_html(conteudo):
    p = _Texto()
    p.feed(conteudo)
    linhas = [" ".join(l.split()) for l in "".join(p.partes).splitlines()]
    return "\n".join(l for l in linhas if l)


class _SemRedirecionarParaDentro(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        if not endereco_publico(newurl):
            raise ValueError("redirect to a local address blocked")
        return super().redirect_request(req, fp, code, msg, headers, newurl)


def ler_pagina(url, abrir=None, resolver=None):
    """O texto de uma página pública (cortado). Levanta ValueError com o motivo."""
    if not endereco_publico(url, resolver):
        raise ValueError("only public http(s) addresses can be read")
    pedido = urllib.request.Request(url, headers={"User-Agent": AGENTE, "Accept": "text/html, text/plain"})
    abrir = abrir or urllib.request.build_opener(_SemRedirecionarParaDentro).open
    try:
        with abrir(pedido, timeout=TIMEOUT) as r:
            tipo = (r.headers.get("Content-Type") or "").lower()
            if not tipo.startswith(("text/html", "text/plain", "application/xhtml")):
                raise ValueError(f"not a text page ({tipo or 'unknown type'})")
            corpo = r.read(LIMITE_PAGINA_BYTES).decode(r.headers.get_content_charset() or "utf-8", errors="replace")
    except urllib.error.HTTPError as e:
        raise ValueError(f"HTTP {e.code}")
    except OSError as e:
        raise ValueError(str(e))
    texto = texto_do_html(corpo) if "html" in tipo else corpo
    return texto[:LIMITE_TEXTO_PAGINA] + ("\n(... cut)" if len(texto) > LIMITE_TEXTO_PAGINA else "")


CONFIG_SEARXNG = """# openTARS: SearXNG só nesta máquina, com o formato JSON ligado (o openTARS lê os resultados por ele)
use_default_settings: true
server:
  secret_key: "{segredo}"
  limiter: false
  image_proxy: false
search:
  formats:
    - html
    - json
"""