Spaces:
Sleeping
Sleeping
File size: 2,690 Bytes
31fa536 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """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]
|