pipeline/searxng_client.py

"""Step 2: query the in-container SearXNG JSON API and aggregate results by domain."""
from __future__ import annotations

from typing import Dict, List
from urllib.parse import urlparse

import requests
import tldextract

from . import config


def registrable_domain(url: str) -> str:
    """Return the registrable domain (e.g. 'blog.example.co.uk' -> 'example.co.uk')."""
    ext = tldextract.extract(url)
    if ext.domain and ext.suffix:
        return f"{ext.domain}.{ext.suffix}".lower()
    return (urlparse(url).netloc or url).lower()


def _search_one(term: str) -> List[dict]:
    params = {"q": term, "format": "json", "safesearch": "1", "language": "en"}
    try:
        r = requests.get(
            f"{config.SEARXNG_URL}/search",
            params=params,
            timeout=config.HTTP_TIMEOUT,
            headers={"User-Agent": "BlogPostGenerator/1.0"},
        )
        r.raise_for_status()
        return r.json().get("results", []) or []
    except Exception:
        return []


def search(terms: List[str], top_n: int = config.TOP_N) -> List[dict]:
    """Run all queries, aggregate scores per domain, and return the top-N domains.

    Each returned item: {domain, url, title, snippet, score}. One representative URL
    (the highest-scoring) is kept per registrable domain so OpenPageRank ranks domains.
    """
    by_domain: Dict[str, dict] = {}
    for term in terms:
        for res in _search_one(term):
            url = res.get("url")
            if not url or not url.startswith("http"):
                continue
            dom = registrable_domain(url)
            if not dom:
                continue
            score = float(res.get("score") or 0.0)
            entry = by_domain.get(dom)
            if entry is None:
                by_domain[dom] = {
                    "domain": dom,
                    "url": url,
                    "title": res.get("title") or "",
                    "snippet": res.get("content") or "",
                    "score": score,
                    "_best": score,  # highest single-result score seen for this domain
                }
            else:
                entry["score"] += score
                # keep the single best-scoring page as the domain's representative URL
                if score > entry["_best"]:
                    entry["_best"] = score
                    entry["url"] = url
                    entry["title"] = res.get("title") or entry["title"]
                    entry["snippet"] = res.get("content") or entry["snippet"]

    ranked = sorted(by_domain.values(), key=lambda x: x["score"], reverse=True)
    for e in ranked:
        e.pop("_best", None)
    return ranked[:top_n]