pipeline/openpagerank.py

"""Step 3: rank the candidate domains by authority using OpenPageRank.

API: GET https://openpagerank.com/api/v1.0/getPageRank
Auth: header  API-OPR: <key>
Params: repeated domains[]=<domain>  (up to 100 per call)
Response: response[] with page_rank_decimal / page_rank_integer / rank per domain.
Docs: https://www.domcop.com/openpagerank/documentation
"""
from __future__ import annotations

from typing import Dict, List

import requests

from . import config


def _chunks(items: List[str], size: int = 100):
    for i in range(0, len(items), size):
        yield items[i : i + size]


def fetch_pageranks(domains: List[str]) -> Dict[str, float]:
    """Return {domain: page_rank_decimal}. Missing/unknown domains map to 0.0.

    Raises RuntimeError if the API key is missing so the caller can surface it.
    """
    if not config.OPR_API_KEY:
        raise RuntimeError(
            "OPR_API_KEY is not set. The Space owner must add an OpenPageRank API key "
            "(free at domcop.com) as a Space secret."
        )
    scores: Dict[str, float] = {d: 0.0 for d in domains}
    headers = {"API-OPR": config.OPR_API_KEY}
    for chunk in _chunks(domains, 100):
        params = [("domains[]", d) for d in chunk]
        try:
            r = requests.get(
                config.OPR_ENDPOINT,
                params=params,
                headers=headers,
                timeout=config.HTTP_TIMEOUT,
            )
            r.raise_for_status()
            payload = r.json()
        except Exception as e:  # noqa: BLE001
            raise RuntimeError(f"OpenPageRank request failed: {e}") from e

        for item in payload.get("response", []) or []:
            dom = (item.get("domain") or "").lower()
            if dom not in scores:
                continue
            try:
                scores[dom] = float(item.get("page_rank_decimal") or 0.0)
            except (TypeError, ValueError):
                scores[dom] = 0.0
    return scores


def rank_results(results: List[dict], top_k: int = config.TOP_K) -> List[dict]:
    """Attach OpenPageRank scores to SearXNG results and return the top-K by authority.

    Falls back to the incoming (SearXNG) order if the API is unavailable.
    """
    domains = [r["domain"] for r in results]
    try:
        scores = fetch_pageranks(domains)
        for r in results:
            r["page_rank"] = scores.get(r["domain"], 0.0)
        ranked = sorted(results, key=lambda x: x.get("page_rank", 0.0), reverse=True)
    except RuntimeError:
        # Preserve SearXNG order; mark scores as unknown.
        for r in results:
            r.setdefault("page_rank", None)
        ranked = results
    return ranked[:top_k]