web_search.py

"""Small fixed-source web-search backend for the local OpenClaude proxy."""

from __future__ import annotations

import html
import os
import re
import unicodedata
import xml.etree.ElementTree as ET
from datetime import timezone
from email.utils import parsedate_to_datetime
from html.parser import HTMLParser
from typing import Any, Callable
from urllib.parse import parse_qs, quote, urlparse

import httpx


MAX_RESULTS = 10
TARGET_PROVIDER_COUNT = 2
SEARCH_TIMEOUT = float(os.getenv("LOCAL_WEB_SEARCH_TIMEOUT", "20"))
USER_AGENT = (
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/124.0 Safari/537.36"
)
RECENT_NEWS_TERMS = {
    "agora",
    "atual",
    "atualizada",
    "atualizado",
    "hoje",
    "latest",
    "news",
    "noticia",
    "noticias",
    "recente",
    "recentes",
    "ultima",
    "ultimas",
    "ultimo",
    "ultimos",
}
QUERY_STOP_WORDS = RECENT_NEWS_TERMS | {
    "a",
    "as",
    "da",
    "das",
    "de",
    "do",
    "dos",
    "e",
    "em",
    "na",
    "nas",
    "no",
    "nos",
    "o",
    "os",
    "para",
    "sobre",
}


class SearchUnavailable(RuntimeError):
    pass


def _clean_text(value: str) -> str:
    cleaned = re.sub(r"\s+", " ", html.unescape(value)).strip()
    return re.sub(r"\s+([,.;:!?])", r"\1", cleaned)


def _fold_text(value: str) -> str:
    normalized = unicodedata.normalize("NFKD", str(value))
    return "".join(
        character for character in normalized if not unicodedata.combining(character)
    ).casefold()


def _query_words(value: str) -> list[str]:
    words = re.findall(r"[a-z0-9]+", _fold_text(value))
    return list(dict.fromkeys(word for word in words if word not in QUERY_STOP_WORDS))


def _is_recent_news_query(query: str) -> bool:
    return bool(set(re.findall(r"[a-z0-9]+", _fold_text(query))) & RECENT_NEWS_TERMS)


def _targets_rio_de_janeiro(query: str) -> bool:
    folded = _fold_text(query)
    return "rio de janeiro" in folded or bool(re.search(r"\brj\b", folded))


def _hostname(url: str) -> str:
    return (urlparse(url).hostname or "").lower()


def _result_url(raw_url: str) -> str | None:
    value = html.unescape(raw_url).strip()
    if value.startswith("//"):
        value = "https:" + value
    parsed = urlparse(value)
    if parsed.hostname in {"duckduckgo.com", "www.duckduckgo.com"}:
        target = parse_qs(parsed.query).get("uddg", [])
        if target:
            value = target[0]
            parsed = urlparse(value)
    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
        return None
    return value


class DuckDuckGoLiteParser(HTMLParser):
    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.results: list[dict[str, str]] = []
        self._anchor_depth = 0
        self._anchor_href = ""
        self._anchor_text: list[str] = []
        self._active_result_index: int | None = None
        self._snippet_depth = 0
        self._snippet_text: list[str] = []
        self._snippet_result_index: int | None = None

    @staticmethod
    def _classes(attributes: list[tuple[str, str | None]]) -> set[str]:
        value = next((value for key, value in attributes if key == "class"), "")
        return set((value or "").split())

    def handle_starttag(
        self, tag: str, attributes: list[tuple[str, str | None]]
    ) -> None:
        if tag == "a" and "result-link" in self._classes(attributes):
            self._anchor_depth = 1
            self._anchor_href = next(
                (value or "" for key, value in attributes if key == "href"), ""
            )
            self._anchor_text = []
            self._active_result_index = None
            return
        if self._anchor_depth:
            self._anchor_depth += 1

        if tag == "td" and "result-snippet" in self._classes(attributes):
            self._snippet_depth = 1
            self._snippet_text = []
            self._snippet_result_index = self._active_result_index
            return
        if self._snippet_depth:
            self._snippet_depth += 1

    def handle_endtag(self, tag: str) -> None:
        if self._anchor_depth:
            self._anchor_depth -= 1
            if self._anchor_depth == 0 and tag == "a":
                url = _result_url(self._anchor_href)
                title = _clean_text("".join(self._anchor_text))
                if url and title:
                    self.results.append(
                        {
                            "title": title,
                            "url": url,
                            "description": "",
                            "source": _hostname(url),
                        }
                    )
                    self._active_result_index = len(self.results) - 1

        if self._snippet_depth:
            self._snippet_depth -= 1
            if self._snippet_depth == 0 and tag == "td":
                if self._snippet_result_index is not None:
                    self.results[self._snippet_result_index]["description"] = (
                        _clean_text("".join(self._snippet_text))
                    )
                self._snippet_result_index = None

    def handle_data(self, data: str) -> None:
        if self._anchor_depth:
            self._anchor_text.append(data)
        if self._snippet_depth:
            self._snippet_text.append(data)


def parse_duckduckgo_lite(payload: str) -> list[dict[str, str]]:
    parser = DuckDuckGoLiteParser()
    parser.feed(payload)
    return _deduplicate(parser.results)


def parse_bing_rss(payload: str) -> list[dict[str, str]]:
    root = ET.fromstring(payload)
    results: list[dict[str, str]] = []
    for item in root.findall("./channel/item"):
        url = _result_url(item.findtext("link", ""))
        title = _clean_text(item.findtext("title", ""))
        if not url or not title:
            continue
        description = _clean_text(
            re.sub(r"<[^>]+>", " ", item.findtext("description", ""))
        )
        results.append(
            {
                "title": title,
                "url": url,
                "description": description,
                "source": _hostname(url),
            }
        )
    return _deduplicate(results)


def _format_publication_date(raw_value: str) -> str:
    value = _clean_text(raw_value)
    if not value:
        return ""
    try:
        parsed = parsedate_to_datetime(value)
    except (TypeError, ValueError, OverflowError):
        return ""
    if parsed.tzinfo is not None:
        parsed = parsed.astimezone(timezone.utc)
        return parsed.strftime("%d/%m/%Y %H:%M UTC")
    return parsed.strftime("%d/%m/%Y %H:%M")


def _news_description(
    raw_description: str,
    title: str,
    publisher: str,
    publication_date: str,
) -> str:
    snippet = _clean_text(re.sub(r"<[^>]+>", " ", raw_description))
    for repeated in (title, publisher):
        if repeated:
            snippet = re.sub(re.escape(repeated), " ", snippet, flags=re.IGNORECASE)
    snippet = _clean_text(snippet)

    metadata: list[str] = []
    if publication_date:
        metadata.append(f"Publicado em {publication_date}")
    if publisher:
        metadata.append(f"Fonte: {publisher}")
    prefix = " — ".join(metadata)
    if prefix and snippet:
        return f"{prefix}. {snippet}"
    if prefix:
        return prefix + "."
    return snippet


def parse_google_news_rss(payload: str) -> list[dict[str, str]]:
    root = ET.fromstring(payload)
    results: list[dict[str, str]] = []
    for item in root.findall("./channel/item"):
        url = _result_url(item.findtext("link", ""))
        title = _clean_text(item.findtext("title", ""))
        if not url or not title:
            continue

        source_node = item.find("source")
        publisher = (
            _clean_text(source_node.text or "") if source_node is not None else ""
        )
        source_url = (
            source_node.attrib.get("url", "") if source_node is not None else ""
        )
        source = publisher or _hostname(source_url) or _hostname(url)
        publication_date = _format_publication_date(item.findtext("pubDate", ""))
        description = _news_description(
            item.findtext("description", ""),
            title,
            publisher,
            publication_date,
        )
        results.append(
            {
                "title": title,
                "url": url,
                "description": description,
                "source": source,
            }
        )
    return _deduplicate(results)


def _deduplicate(
    results: list[dict[str, str]], limit: int | None = MAX_RESULTS
) -> list[dict[str, str]]:
    unique: list[dict[str, str]] = []
    seen_urls: set[str] = set()
    seen_titles: set[str] = set()
    for result in results:
        url = result.get("url", "")
        title = _fold_text(result.get("title", "")).strip()
        if not url or url in seen_urls or (title and title in seen_titles):
            continue
        seen_urls.add(url)
        if title:
            seen_titles.add(title)
        unique.append(result)
        if limit is not None and len(unique) >= limit:
            break
    return unique


def _contains_word(text: str, word: str) -> bool:
    return bool(re.search(rf"(?<![a-z0-9]){re.escape(word)}(?![a-z0-9])", text))


def _result_score(result: dict[str, str], query: str) -> int:
    title = _fold_text(result.get("title", ""))
    description = _fold_text(result.get("description", ""))
    source = _fold_text(result.get("source", ""))
    url = _fold_text(result.get("url", ""))
    score = 0

    for word in _query_words(query):
        if _contains_word(title, word):
            score += 8
        if _contains_word(description, word):
            score += 3
        if _contains_word(source, word) or _contains_word(url, word):
            score += 1

    if description.startswith("publicado em "):
        score += 2

    if _targets_rio_de_janeiro(query):
        combined = f"{title} {description} {source} {url}"
        if "rio de janeiro" in title:
            score += 28
        elif "rio de janeiro" in combined:
            score += 16
        if _contains_word(title, "rj"):
            score += 18
        elif _contains_word(combined, "rj"):
            score += 10
        if re.search(r"(?:^|[/.?&=_-])rj(?:$|[/.?&=_-])", url):
            score += 14

        if "rio grande do sul" in combined:
            score -= 40
        if "porto alegre" in combined:
            score -= 28
        if _contains_word(combined, "rs"):
            score -= 20
        if any(
            clue in combined
            for clue in (
                "agorars.com",
                "gauchazh",
                "jornal o sul",
                "poa24horas",
                "/rs/rio-grande-do-sul",
            )
        ):
            score -= 28

    return score


def _rank_results(
    results: list[dict[str, str]], query: str
) -> list[dict[str, str]]:
    unique = _deduplicate(results, limit=None)
    indexed = list(enumerate(unique))
    indexed.sort(key=lambda pair: (-_result_score(pair[1], query), pair[0]))
    return [result for _, result in indexed[:MAX_RESULTS]]


def _duckduckgo_lite(client: httpx.Client, query: str) -> list[dict[str, str]]:
    response = client.get(
        "https://lite.duckduckgo.com/lite/",
        params={"q": query, "kl": "br-pt"},
    )
    response.raise_for_status()
    return parse_duckduckgo_lite(response.text)


def _bing_rss(client: httpx.Client, query: str) -> list[dict[str, str]]:
    response = client.get(
        "https://www.bing.com/search",
        params={"q": query, "format": "rss", "setlang": "pt-BR"},
    )
    response.raise_for_status()
    return parse_bing_rss(response.text)


def _google_news_rss(
    client: httpx.Client, query: str
) -> list[dict[str, str]]:
    response = client.get(
        "https://news.google.com/rss/search",
        params={
            "q": query,
            "hl": "pt-BR",
            "gl": "BR",
            "ceid": "BR:pt-419",
        },
    )
    response.raise_for_status()
    return parse_google_news_rss(response.text)


def _wikipedia(client: httpx.Client, query: str) -> list[dict[str, str]]:
    response = client.get(
        "https://pt.wikipedia.org/w/api.php",
        params={
            "action": "query",
            "list": "search",
            "srsearch": query,
            "format": "json",
            "utf8": "1",
        },
    )
    response.raise_for_status()
    rows = response.json().get("query", {}).get("search", [])
    results: list[dict[str, str]] = []
    for row in rows:
        if not isinstance(row, dict) or not row.get("title"):
            continue
        title = str(row["title"])
        url = "https://pt.wikipedia.org/wiki/" + quote(
            title.replace(" ", "_"), safe="()_-"
        )
        results.append(
            {
                "title": title,
                "url": url,
                "description": _clean_text(
                    re.sub(r"<[^>]+>", " ", str(row.get("snippet", "")))
                ),
                "source": "pt.wikipedia.org",
            }
        )
    return _deduplicate(results)


def search_web(query: str) -> dict[str, Any]:
    normalized = _clean_text(query)
    if not normalized:
        raise ValueError("A consulta de busca não pode estar vazia.")
    if len(normalized) > 500:
        raise ValueError("A consulta de busca excede 500 caracteres.")

    providers: list[
        tuple[str, Callable[[httpx.Client, str], list[dict[str, str]]]]
    ]
    if _is_recent_news_query(normalized):
        providers = [
            ("google-news", _google_news_rss),
            ("duckduckgo-lite", _duckduckgo_lite),
            ("bing-rss", _bing_rss),
        ]
    else:
        providers = [
            ("duckduckgo-lite", _duckduckgo_lite),
            ("bing-rss", _bing_rss),
            ("wikipedia-pt", _wikipedia),
        ]

    errors: list[str] = []
    successful_providers: list[str] = []
    aggregated_results: list[dict[str, str]] = []
    with httpx.Client(
        timeout=SEARCH_TIMEOUT,
        follow_redirects=True,
        headers={
            "User-Agent": USER_AGENT,
            "Accept-Language": "pt-BR,pt;q=0.9,en;q=0.7",
        },
    ) as client:
        for provider_name, provider in providers:
            try:
                results = provider(client, normalized)
            except (httpx.HTTPError, ET.ParseError, ValueError, TypeError) as error:
                errors.append(f"{provider_name}: {error}")
                continue
            if results:
                successful_providers.append(provider_name)
                aggregated_results.extend(results)
                if len(successful_providers) >= TARGET_PROVIDER_COUNT:
                    break
            else:
                errors.append(f"{provider_name}: nenhum resultado")

    if aggregated_results:
        return {
            "query": normalized,
            "provider": "+".join(successful_providers),
            "results": _rank_results(aggregated_results, normalized),
        }

    detail = "; ".join(errors) if errors else "nenhuma fonte disponível"
    raise SearchUnavailable(f"A busca web local falhou: {detail}")