""" Provider layer: every search backend behind one interface. Keyless providers work with zero configuration. Premium providers (Tavily/Exa/Brave/Serper) activate automatically when their API key is present in the environment. All of them are queried in parallel and their rankings are fused downstream by Reciprocal Rank Fusion. """ from __future__ import annotations import asyncio import base64 import json import logging import re from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional, Tuple from urllib.parse import parse_qs, quote_plus, unquote, urlparse from selectolax.parser import HTMLParser from .config import BINARY_EXTENSIONS, BLOCKED_PATTERNS, settings from .core import TTLCache, fetch_json, fetch_text, serp_cache from .schemas import RawResult, canonical_url log = logging.getLogger("deepsearch.providers") # -------------------------------------------------------------------------- # helpers # -------------------------------------------------------------------------- def _usable(url: str) -> bool: if not url: return False # magnet: links are a first-class result type for the file tier. if url.startswith("magnet:"): return "btih:" in url.lower() if not url.startswith(("http://", "https://")): return False low = url.lower() if any(p in low for p in BLOCKED_PATTERNS): return False path = urlparse(low).path if path.endswith(BINARY_EXTENSIONS): return False return True def _unwrap_bing(url: str) -> str: """Bing wraps results in /ck/a redirects with a base64url payload.""" if "bing.com/ck/a" not in url: return url try: u = parse_qs(urlparse(url).query).get("u", [""])[0] if u.startswith("a1"): b = u[2:] b += "=" * (-len(b) % 4) return base64.urlsafe_b64decode(b).decode("utf-8", "ignore") except Exception: pass return url def _unwrap_ddg(url: str) -> str: """DuckDuckGo HTML sometimes emits /l/?uddg=.""" if "duckduckgo.com/l/" in url or url.startswith("//duckduckgo.com/l/"): try: q = parse_qs(urlparse(url if url.startswith("http") else "https:" + url).query) if "uddg" in q: return unquote(q["uddg"][0]) except Exception: pass if url.startswith("//"): return "https:" + url return url def _clean(text: str, limit: int = 500) -> str: return " ".join((text or "").split())[:limit] # -------------------------------------------------------------------------- # base class # -------------------------------------------------------------------------- class Provider(ABC): name: str = "base" kind: str = "web" # web | news | academic | code | social | reference keyless: bool = True weight: float = 1.0 # trust multiplier applied during fusion def available(self) -> bool: return self.keyless @abstractmethod async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: ... # Adaptive latency budget. A provider that is IP-blocked will happily # burn the full timeout returning nothing on every single sub-query - # that alone added ~8s to every search. After two empty/slow rounds its # budget collapses so it can never gate the pipeline again. _empty_streak: Dict[str, int] = {} def _budget(self) -> float: streak = Provider._empty_streak.get(self.name, 0) if streak >= 4: return min(1.5, settings.provider_timeout) if streak >= 2: return min(3.0, settings.provider_timeout) return settings.provider_timeout async def safe_search(self, query: str, limit: int, on_event: Optional[Any] = None, **kw: Any) -> List[RawResult]: cache_key = TTLCache.key(self.name, query, limit, sorted(kw.items())) cached = serp_cache.get(cache_key) if cached is not None: if on_event: try: await on_event(self.name, len(cached), True) except Exception: pass return cached try: res = await asyncio.wait_for( self.search(query, limit, **kw), timeout=self._budget() ) except (asyncio.TimeoutError, Exception) as e: # noqa: BLE001 log.debug("%s failed: %s", self.name, e) Provider._empty_streak[self.name] = \ Provider._empty_streak.get(self.name, 0) + 1 return [] out: List[RawResult] = [] for i, r in enumerate(res or []): if not _usable(r.url): continue r.provider = self.name r.rank = i + 1 out.append(r) if on_event: try: await on_event(self.name, len(out), False) except Exception: pass if out: Provider._empty_streak[self.name] = 0 else: Provider._empty_streak[self.name] = \ Provider._empty_streak.get(self.name, 0) + 1 serp_cache.set(cache_key, out, settings.serp_cache_ttl) return out # ========================================================================== # KEYLESS WEB PROVIDERS # ========================================================================== class DuckDuckGoHTML(Provider): name = "duckduckgo" kind = "web" weight = 1.15 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = {"q": query, "kl": kw.get("region", "wt-wt")} tf = {"day": "d", "week": "w", "month": "m", "year": "y"}.get(kw.get("freshness") or "") if tf: data["df"] = tf # NOTE: do NOT add Referer/Content-Type here. DuckDuckGo answers those # requests with a 202 "anomaly" bot-check page containing zero results. # The default header set is what gets a real SERP. html = await fetch_text( "https://html.duckduckgo.com/html/", method="POST", data=data, provider=self.name, timeout=settings.provider_timeout, ) if not html: return [] tree = HTMLParser(html) out: List[RawResult] = [] for node in tree.css("div.result"): if len(out) >= limit: break cls = node.attributes.get("class", "") or "" if "result--ad" in cls: # sponsored continue a = node.css_first("a.result__a") if not a: continue href = _unwrap_ddg(a.attributes.get("href", "") or "") if "duckduckgo.com/y.js" in href: # ad redirect continue sn = node.css_first(".result__snippet") out.append(RawResult(url=href, title=_clean(a.text(), 300), snippet=_clean(sn.text() if sn else ""))) return out class DuckDuckGoLite(Provider): name = "ddg_lite" kind = "web" weight = 1.0 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: html = await fetch_text( "https://lite.duckduckgo.com/lite/", method="POST", data={"q": query}, provider=self.name, timeout=settings.provider_timeout, ) if not html: return [] tree = HTMLParser(html) out: List[RawResult] = [] rows = tree.css("tr") pending: Optional[RawResult] = None for tr in rows: a = tr.css_first("a.result-link") if a: if pending: out.append(pending) href = _unwrap_ddg(a.attributes.get("href", "") or "") pending = RawResult(url=href, title=_clean(a.text(), 300)) continue td = tr.css_first("td.result-snippet") if td and pending: pending.snippet = _clean(td.text()) out.append(pending) pending = None if len(out) >= limit: break if pending and len(out) < limit: out.append(pending) return out[:limit] class BingScrape(Provider): name = "bing" kind = "web" weight = 1.20 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: params: Dict[str, Any] = {"q": query, "count": min(30, max(10, limit * 2)), "setlang": "en", "form": "QBLH"} fresh = {"day": "ex1%3a%22ez1%22", "week": "ex1%3a%22ez2%22", "month": "ex1%3a%22ez3%22"}.get(kw.get("freshness") or "") if fresh: params["filters"] = fresh html = await fetch_text( "https://www.bing.com/search", params=params, provider=self.name, timeout=settings.provider_timeout, headers={"Referer": "https://www.bing.com/"}, ) if not html: return [] tree = HTMLParser(html) out: List[RawResult] = [] for li in tree.css("li.b_algo"): if len(out) >= limit: break h = li.css_first("h2 a") if not h: continue url = _unwrap_bing(h.attributes.get("href", "") or "") body = li.css_first(".b_caption p") or li.css_first("p") out.append(RawResult(url=url, title=_clean(h.text(), 300), snippet=_clean(body.text() if body else ""))) return out class YahooScrape(Provider): """Yahoo Search - independent SERP, tolerant of datacenter IPs.""" name = "yahoo" kind = "web" weight = 1.10 _RU = re.compile(r"/RU=([^/]+)/R[KS]=") @classmethod def _unwrap(cls, url: str) -> str: m = cls._RU.search(url or "") return unquote(m.group(1)) if m else url @staticmethod def _fix_title(title: str, url: str) -> str: """ Yahoo glues a breadcrumb onto titles, e.g. 'OpenAIhttps://openai.com › index › introducing-gpt-5Introducing GPT-5 | OpenAI' Strip the breadcrumb and the trailing URL slug fused to the real title. """ t = " ".join((title or "").split()) if "›" in t: t = t.split("›")[-1].strip() host = urlparse(url).netloc.replace("www.", "") if host and host in t: t = t.split(host)[-1].lstrip(" ›/|-·") # Remove a leading URL slug ('introducing-gpt-5Introducing GPT-5'). seg = unquote(urlparse(url).path.rstrip("/").split("/")[-1]) if seg: seg_clean = re.sub(r"\.(html?|php|aspx?)$", "", seg) if t.lower().startswith(seg_clean.lower()): t = t[len(seg_clean):].lstrip(" -–—:|·") else: # slug may appear with separators normalised away compact = re.sub(r"[^a-z0-9]", "", seg_clean.lower()) head = re.sub(r"[^a-z0-9]", "", t[: len(seg_clean) + 6].lower()) if compact and head.startswith(compact): cut = 0 seen = 0 for i, ch in enumerate(t): if re.match(r"[a-z0-9]", ch, re.I): seen += 1 if seen >= len(compact): cut = i + 1 break t = t[cut:].lstrip(" -–—:|·") return t[:300] or host async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: params: Dict[str, Any] = {"p": query, "n": min(25, max(10, limit * 2))} age = {"day": "1d", "week": "1w", "month": "1m"}.get(kw.get("freshness") or "") if age: params["btf"] = age html = await fetch_text( "https://search.yahoo.com/search", params=params, provider=self.name, timeout=settings.provider_timeout, ) if not html: return [] tree = HTMLParser(html) out: List[RawResult] = [] seen: set = set() for node in tree.css("div.algo, div.dd.algo"): if len(out) >= limit: break a = node.css_first("h3 a") or node.css_first("a") if not a: continue url = self._unwrap(a.attributes.get("href", "") or "") if not url or url in seen: continue seen.add(url) sn = node.css_first("div.compText") or node.css_first("p") out.append(RawResult( url=url, title=self._fix_title(a.text(), url), snippet=_clean(sn.text() if sn else ""), )) return out class MojeekScrape(Provider): name = "mojeek" kind = "web" weight = 0.9 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: html = await fetch_text( "https://www.mojeek.com/search", params={"q": query, "t": min(30, limit * 2)}, provider=self.name, timeout=settings.provider_timeout, headers={"Referer": "https://www.mojeek.com/"}, ) if not html: return [] tree = HTMLParser(html) out: List[RawResult] = [] for li in tree.css("ul.results-standard li, li.result"): if len(out) >= limit: break a = li.css_first("a.title") or li.css_first("h2 a") if not a: continue p = li.css_first("p.s") or li.css_first("p") out.append(RawResult(url=a.attributes.get("href", "") or "", title=_clean(a.text(), 300), snippet=_clean(p.text() if p else ""))) return out class SearxNG(Provider): """ Meta-search across public SearxNG instances - each one already fuses Google/Bing/Qwant/etc, so it adds real recall beyond our own scrapers. Most public instances disable the JSON API (403/HTML), so we try JSON first and transparently fall back to parsing the HTML SERP. """ name = "searxng" kind = "web" weight = 1.05 # Remember which instance last worked so we stop probing dead ones on # every single sub-query (that behaviour got us 429'd everywhere). _preferred: Optional[str] = None def _instance_order(self) -> List[str]: order = list(settings.searx_instances) if self._preferred and self._preferred in order: order.remove(self._preferred) order.insert(0, self._preferred) return order[:3] # never probe more than 3 per query async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: cat = kw.get("searx_category", "general") tr = {"day": "day", "week": "week", "month": "month", "year": "year"}.get(kw.get("freshness") or "") for base in self._instance_order(): base = base.rstrip("/") params: Dict[str, Any] = { "q": query, "categories": cat, "language": kw.get("language", "en"), "safesearch": 0, } if tr: params["time_range"] = tr # 1) JSON API (fast path, rarely enabled publicly) data = await fetch_json( f"{base}/search", params={**params, "format": "json"}, provider=f"{self.name}:{base}", timeout=settings.provider_timeout, headers={"Accept": "application/json"}, ) if isinstance(data, dict) and data.get("results"): out = [ RawResult(url=r.get("url", ""), title=_clean(r.get("title", ""), 300), snippet=_clean(r.get("content", "")), published=r.get("publishedDate"), score_hint=float(r.get("score") or 0.0)) for r in data["results"][:limit] ] if out: SearxNG._preferred = base return out # 2) HTML fallback html = await fetch_text( f"{base}/search", params=params, provider=f"{self.name}:{base}", timeout=settings.provider_timeout, ) if not html: continue tree = HTMLParser(html) out = [] for art in tree.css("article.result, div.result"): if len(out) >= limit: break a = art.css_first("h3 a") or art.css_first("a.url_wrapper") or art.css_first("a") if not a: continue href = a.attributes.get("href", "") or "" if not href.startswith("http"): continue p = art.css_first("p.content") or art.css_first("p") out.append(RawResult(url=href, title=_clean(a.text(), 300), snippet=_clean(p.text() if p else ""))) if out: SearxNG._preferred = base return out return [] # ========================================================================== # KEYLESS SPECIALIST PROVIDERS # ========================================================================== def _parse_rss(xml: str, limit: int, strip_html: bool = True) -> List[RawResult]: """Minimal RSS/Atom parser - no external dependency, very fast.""" out: List[RawResult] = [] items = re.findall(r"]*>(.*?)", xml, re.S | re.I) if not items: items = re.findall(r"]*>(.*?)", xml, re.S | re.I) for it in items[:limit]: def tag(name: str) -> str: m = re.search(rf"<{name}[^>]*>(.*?)", it, re.S | re.I) if not m: return "" v = m.group(1) v = re.sub(r"", r"\1", v, flags=re.S) if strip_html: v = re.sub(r"<[^>]+>", " ", v) return _clean(v, 600) link = tag("link") if not link: m = re.search(r']*href="([^"]+)"', it, re.I) link = m.group(1) if m else "" if not link: continue out.append(RawResult( url=link, title=tag("title"), snippet=tag("description") or tag("summary") or tag("content"), published=tag("pubDate") or tag("updated") or tag("published"), )) return out class GoogleNewsRSS(Provider): """ Google News RSS - no key, no bot-wall, and it indexes essentially the whole news web. Our single most reliable source for current events. Item links are opaque `news.google.com/rss/articles/` redirects, so we resolve them to the real publisher URLs (in parallel) before they enter the pipeline - otherwise every citation would read "news.google.com" and the reader could never extract article text. """ name = "google_news" kind = "news" weight = 1.25 @staticmethod async def _resolve(url: str) -> str: m = re.search(r"/articles/([^?/]+)", url) if not m: return url aid = m.group(1) page = await fetch_text(f"https://news.google.com/rss/articles/{aid}", timeout=6.0) if not page: return url sig = re.search(r'data-n-a-sg="([^"]+)"', page) ts = re.search(r'data-n-a-ts="([^"]+)"', page) if not (sig and ts): return url inner = json.dumps([ "garturlreq", [["X", "X", ["X", "X"], None, None, 1, 1, "US:en", None, 1, None, None, None, None, None, 0, 1], "X", "EN", None, None, None, None, None, None, 0], aid, int(ts.group(1)), sig.group(1), ]) freq = json.dumps([[["Fbv4je", inner, None, "1"]]]) resp = await fetch_text( "https://news.google.com/_/DotsSplashUi/data/batchexecute", method="POST", data={"f.req": freq}, headers={"Content-Type": "application/x-www-form-urlencoded;charset=UTF-8"}, timeout=6.0, ) if not resp: return url found = re.findall(r'https?://(?!news\.google)[^\\"\s]+', resp) return found[0] if found else url async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: when = {"day": "1d", "week": "7d", "month": "1m", "year": "1y"}.get(kw.get("freshness") or "") q = f"{query} when:{when}" if when else query xml = await fetch_text( "https://news.google.com/rss/search", params={"q": q, "hl": "en-US", "gl": "US", "ceid": "US:en"}, provider=self.name, timeout=settings.provider_timeout, ) if not xml: return [] results = _parse_rss(xml, limit) for r in results: # titles arrive as "Headline - Publisher" if " - " in r.title: head, _, pub = r.title.rpartition(" - ") if head: r.title = head r.extra["publisher"] = pub real = await asyncio.gather( *[self._resolve(r.url) for r in results], return_exceptions=True ) out: List[RawResult] = [] for r, u in zip(results, real): if isinstance(u, str) and u and "news.google.com" not in u: r.url = canonical_url(u) out.append(r) elif "news.google.com" not in r.url: out.append(r) return out or results class BingNewsRSS(Provider): name = "bing_news" kind = "news" weight = 1.10 @staticmethod def _unwrap_apiclick(url: str) -> str: """Bing News RSS links are /news/apiclick.aspx?...&url=.""" if "apiclick.aspx" not in url: return url try: q = parse_qs(urlparse(url.replace("&", "&")).query) for key in ("url", "u"): if q.get(key): return unquote(q[key][0]) except Exception: pass return url async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: xml = await fetch_text( "https://www.bing.com/news/search", params={"q": query, "format": "rss", "count": min(30, limit * 2)}, provider=self.name, timeout=settings.provider_timeout, ) if not xml: return [] out = _parse_rss(xml, limit) for r in out: real = self._unwrap_apiclick(r.url) if real != r.url: r.url = canonical_url(real) # drop anything still pointing back at bing return [r for r in out if "bing.com" not in r.url] or out class Wikipedia(Provider): name = "wikipedia" kind = "reference" weight = 1.1 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = await fetch_json( "https://en.wikipedia.org/w/api.php", params={"action": "query", "list": "search", "srsearch": query, "format": "json", "srlimit": min(limit, 15), "srprop": "snippet|timestamp"}, provider=self.name, timeout=settings.provider_timeout, ) if not isinstance(data, dict): return [] out = [] for item in data.get("query", {}).get("search", []): title = item.get("title", "") out.append(RawResult( url=f"https://en.wikipedia.org/wiki/{quote_plus(title.replace(' ', '_'))}", title=title, snippet=_clean(re.sub(r"<[^>]+>", "", item.get("snippet", ""))), published=item.get("timestamp"), )) return out class HackerNews(Provider): name = "hackernews" kind = "social" weight = 0.85 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = await fetch_json( "https://hn.algolia.com/api/v1/search", params={"query": query, "hitsPerPage": min(limit, 20), "tags": "(story,comment)"}, provider=self.name, timeout=settings.provider_timeout, ) if not isinstance(data, dict): return [] out = [] for h in data.get("hits", []): url = h.get("url") or f"https://news.ycombinator.com/item?id={h.get('objectID')}" title = h.get("title") or h.get("story_title") or "" text = h.get("comment_text") or h.get("story_text") or "" if not title and not text: continue out.append(RawResult( url=url, title=_clean(title or f"HN discussion ({h.get('points', 0)} pts)", 300), snippet=_clean(re.sub(r"<[^>]+>", "", text)), published=h.get("created_at"), score_hint=float(h.get("points") or 0) / 100.0, )) return out class StackExchange(Provider): name = "stackexchange" kind = "code" weight = 1.0 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = await fetch_json( "https://api.stackexchange.com/2.3/search/advanced", params={"order": "desc", "sort": "relevance", "q": query, "site": "stackoverflow", "pagesize": min(limit, 20), "filter": "!nNPvSNdWme"}, provider=self.name, timeout=settings.provider_timeout, ) if not isinstance(data, dict): return [] out = [] for item in data.get("items", []): out.append(RawResult( url=item.get("link", ""), title=_clean(item.get("title", ""), 300), snippet=_clean(f"Score {item.get('score', 0)} · " f"{item.get('answer_count', 0)} answers · " f"{'ANSWERED' if item.get('is_answered') else 'open'}"), score_hint=min(1.0, float(item.get("score") or 0) / 50.0), extra={"tags": item.get("tags", [])}, )) return out class GitHubSearch(Provider): name = "github" kind = "code" weight = 0.95 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = await fetch_json( "https://api.github.com/search/repositories", params={"q": query, "sort": "stars", "order": "desc", "per_page": min(limit, 15)}, provider=self.name, timeout=settings.provider_timeout, headers={"Accept": "application/vnd.github+json"}, ) if not isinstance(data, dict): return [] out = [] for r in data.get("items", []): stars = r.get("stargazers_count", 0) out.append(RawResult( url=r.get("html_url", ""), title=_clean(f"{r.get('full_name', '')} ({stars:,}★)", 300), snippet=_clean(f"{r.get('description') or ''} " f"[{r.get('language') or 'n/a'}]"), published=r.get("pushed_at"), score_hint=min(1.0, stars / 50000.0), )) return out class ArxivSearch(Provider): name = "arxiv" kind = "academic" weight = 1.05 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: # arXiv aggressively rate-limits shared/cloud IPs (429). Try both # hosts, then give up quietly - OpenAlex/Crossref cover the same ground. xml = None for host in ("https://export.arxiv.org", "http://export.arxiv.org"): xml = await fetch_text( f"{host}/api/query", params={"search_query": f"all:{query}", "start": 0, "max_results": min(limit, 15), "sortBy": "relevance"}, provider=self.name, timeout=settings.provider_timeout, ) if xml and "" in xml: break if not xml: return [] out = [] for entry in re.findall(r"(.*?)", xml, re.S)[:limit]: def tag(t: str) -> str: m = re.search(rf"<{t}[^>]*>(.*?)", entry, re.S) return _clean(re.sub(r"<[^>]+>", "", m.group(1))) if m else "" link = re.search(r'(.*?)', entry, re.S) if not link: continue out.append(RawResult(url=link.group(1).strip(), title=tag("title"), snippet=tag("summary")[:500], published=tag("published"))) return out class OpenAlex(Provider): name = "openalex" kind = "academic" weight = 1.0 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = await fetch_json( "https://api.openalex.org/works", params={"search": query, "per-page": min(limit, 15), "mailto": "deepsearch@example.com"}, provider=self.name, timeout=settings.provider_timeout, ) if not isinstance(data, dict): return [] out = [] for w in data.get("results", []): url = (w.get("primary_location") or {}).get("landing_page_url") \ or w.get("doi") or w.get("id") or "" cited = w.get("cited_by_count", 0) inv = w.get("abstract_inverted_index") or {} abstract = "" if inv: try: pos: Dict[int, str] = {} for word, idxs in inv.items(): for i in idxs: pos[i] = word abstract = " ".join(pos[k] for k in sorted(pos))[:400] except Exception: abstract = "" out.append(RawResult( url=url, title=_clean(w.get("title") or w.get("display_name") or "", 300), snippet=_clean(abstract or f"Cited by {cited}"), published=str(w.get("publication_year") or ""), score_hint=min(1.0, cited / 2000.0), )) return out class CrossrefSearch(Provider): name = "crossref" kind = "academic" weight = 0.9 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = await fetch_json( "https://api.crossref.org/works", params={"query": query, "rows": min(limit, 12), "sort": "relevance"}, provider=self.name, timeout=settings.provider_timeout, ) if not isinstance(data, dict): return [] out = [] for it in data.get("message", {}).get("items", []): title = (it.get("title") or [""])[0] if not title: continue out.append(RawResult( url=it.get("URL", ""), title=_clean(title, 300), snippet=_clean(f"{(it.get('container-title') or [''])[0]} · " f"{it.get('type', '')} · cited {it.get('is-referenced-by-count', 0)}"), published=str((it.get("issued", {}).get("date-parts") or [[""]])[0][0] or ""), )) return out class PubMedSearch(Provider): name = "pubmed" kind = "academic" weight = 1.05 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: ids_data = await fetch_json( "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi", params={"db": "pubmed", "term": query, "retmode": "json", "retmax": min(limit, 12), "sort": "relevance"}, provider=self.name, timeout=settings.provider_timeout, ) if not isinstance(ids_data, dict): return [] ids = ids_data.get("esearchresult", {}).get("idlist", []) if not ids: return [] summ = await fetch_json( "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esummary.fcgi", params={"db": "pubmed", "id": ",".join(ids), "retmode": "json"}, provider=self.name, timeout=settings.provider_timeout, ) out = [] result = (summ or {}).get("result", {}) if isinstance(summ, dict) else {} for pid in ids: rec = result.get(pid) if not isinstance(rec, dict): continue out.append(RawResult( url=f"https://pubmed.ncbi.nlm.nih.gov/{pid}/", title=_clean(rec.get("title", ""), 300), snippet=_clean(f"{rec.get('source', '')} · {rec.get('pubdate', '')} · " f"{', '.join(a.get('name', '') for a in (rec.get('authors') or [])[:3])}"), published=rec.get("pubdate"), )) return out class RedditSearch(Provider): """Reddit via public JSON; falls back to the old.reddit mirror.""" name = "reddit" kind = "social" weight = 0.75 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: # JSON endpoint first (richer metadata)... for host in ("https://www.reddit.com", "https://old.reddit.com"): data = await fetch_json( f"{host}/search.json", params={"q": query, "limit": min(limit, 20), "sort": "relevance", "t": kw.get("freshness") or "all", "raw_json": 1}, provider=self.name, timeout=settings.provider_timeout, headers={"Accept": "application/json"}, ) if not isinstance(data, dict): continue children = data.get("data", {}).get("children", []) out = [] for c in children: d = c.get("data", {}) out.append(RawResult( url=f"https://www.reddit.com{d.get('permalink', '')}", title=_clean(d.get("title", ""), 300), snippet=_clean(d.get("selftext", "") or f"r/{d.get('subreddit','')} · " f"{d.get('score',0)} upvotes · {d.get('num_comments',0)} comments"), score_hint=min(1.0, float(d.get("score") or 0) / 5000.0), )) if out: return out # ...then the RSS feed, which survives IPs that the JSON API blocks. xml = await fetch_text( "https://www.reddit.com/search.rss", params={"q": query, "limit": min(limit, 25), "sort": "relevance"}, provider=self.name, timeout=settings.provider_timeout, ) return _parse_rss(xml, limit) if xml else [] # ========================================================================== # ADULT / NSFW PROVIDERS # Mainstream engines hard-filter these queries at the index level (a probe # for "best adult video sites" returned bestbuy.com and a dictionary entry), # so explicit intent is routed to dedicated sources that return real data. # These only ever activate when the query actually calls for them. # ========================================================================== class RedTubeAPI(Provider): name = "redtube" kind = "adult" weight = 1.15 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = await fetch_json( "https://api.redtube.com/", params={"data": "redtube.Videos.searchVideos", "output": "json", "search": query, "thumbsize": "medium", "ordering": kw.get("adult_order", "mostviewed")}, provider=self.name, timeout=settings.provider_timeout, ) if not isinstance(data, dict): return [] out: List[RawResult] = [] for item in (data.get("videos") or [])[:limit]: v = item.get("video", item) or {} url = v.get("url", "") if not url: continue views = int(v.get("views") or 0) out.append(RawResult( url=url, title=_clean(v.get("title", ""), 300), snippet=_clean(f"{v.get('duration','')} · {views:,} views · " f"rating {v.get('rating','n/a')} · " f"{', '.join(t.get('tag_name','') for t in (v.get('tags') or [])[:6])}"), published=v.get("publish_date"), score_hint=min(1.0, views / 3_000_000), extra={"thumb": v.get("default_thumb", ""), "media": "video"}, )) return out class PornhubAPI(Provider): """Official public webmasters search endpoint - JSON, no key.""" name = "pornhub" kind = "adult" weight = 1.25 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: params: Dict[str, Any] = {"search": query} if kw.get("freshness") in ("day", "week", "month"): params["period"] = {"day": "daily", "week": "weekly", "month": "monthly"}[kw["freshness"]] data = await fetch_json( "https://www.pornhub.com/webmasters/search", params=params, provider=self.name, timeout=settings.provider_timeout, ) if not isinstance(data, dict): return [] out: List[RawResult] = [] for v in (data.get("videos") or [])[:limit]: url = v.get("url", "") if not url: continue views = int(v.get("views") or 0) tags = ", ".join(t.get("tag_name", "") for t in (v.get("tags") or [])[:6]) out.append(RawResult( url=url, title=_clean(v.get("title", ""), 300), snippet=_clean(f"{v.get('duration','')} · {views:,} views · " f"rating {v.get('rating','n/a')}% · {tags}"), published=v.get("publish_date"), score_hint=min(1.0, views / 5_000_000), extra={"thumb": v.get("default_thumb", ""), "media": "video"}, )) return out class EpornerAPI(Provider): name = "eporner" kind = "adult" weight = 1.10 # Eporner does literal AND matching, so long natural-language queries # return zero. Reduce to the content-bearing keywords. _NOISE = {"best", "top", "site", "sites", "video", "videos", "free", "watch", "online", "list", "good", "the", "and", "for", "with", "2024", "2025", "2026", "hd", "full", "new"} def _reduce(self, query: str) -> str: words = [w for w in re.findall(r"[a-z0-9]+", query.lower()) if w not in self._NOISE and len(w) > 2] return " ".join(words[:3]) or query[:40] async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = await fetch_json( "https://www.eporner.com/api/v2/video/search/", params={"query": self._reduce(query), "per_page": min(limit, 30), "page": 1, "format": "json", "thumbsize": "medium", "order": kw.get("adult_order", "top-weekly")}, provider=self.name, timeout=settings.provider_timeout, ) if not isinstance(data, dict): return [] out: List[RawResult] = [] for v in (data.get("videos") or [])[:limit]: url = v.get("url", "") if not url: continue views = int(v.get("views") or 0) out.append(RawResult( url=url, title=_clean(v.get("title", ""), 300), snippet=_clean(f"{v.get('length_min','')} · {views:,} views · " f"rate {v.get('rate','')} · {v.get('keywords','')}"), published=v.get("added"), score_hint=min(1.0, views / 3_000_000), extra={"thumb": (v.get("default_thumb") or {}).get("src", ""), "media": "video"}, )) return out class XvideosScrape(Provider): name = "xvideos" kind = "adult" weight = 1.05 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: html = await fetch_text("https://www.xvideos.com/", params={"k": query}, provider=self.name, timeout=settings.provider_timeout) if not html: return [] tree = HTMLParser(html) out: List[RawResult] = [] for node in tree.css("div.thumb-block"): if len(out) >= limit: break a = node.css_first("p.title a") if not a: continue href = a.attributes.get("href", "") if not href: continue meta = node.css_first("p.metadata") out.append(RawResult( url=f"https://www.xvideos.com{href}" if href.startswith("/") else href, title=_clean(a.attributes.get("title") or a.text(), 300), snippet=_clean(meta.text() if meta else ""), extra={"media": "video"}, )) return out class XhamsterScrape(Provider): name = "xhamster" kind = "adult" weight = 1.0 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: slug = quote_plus(query.strip().replace(" ", "-")) html = await fetch_text(f"https://xhamster.com/search/{slug}", provider=self.name, timeout=settings.provider_timeout) if not html: return [] tree = HTMLParser(html) out: List[RawResult] = [] seen: set = set() for a in tree.css("a[href*='/videos/']"): if len(out) >= limit: break href = a.attributes.get("href", "") title = _clean(a.attributes.get("title") or a.text(), 300) if not href or href in seen or len(title) < 8: continue seen.add(href) out.append(RawResult(url=href, title=title, extra={"media": "video"})) return out # ========================================================================== # EXTENDED ADULT TIER # ========================================================================== class _AdultTube(Provider): """Shared scraper base for adult tube sites with a common thumb layout.""" kind = "adult" weight = 1.0 base = "" path = "/search/{q}" item_sel = "div.thumb-block" link_sel = "p.title a" meta_sel = "p.metadata" def _url(self, query: str) -> str: return self.base + self.path.format(q=quote_plus(query)) async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: html = await fetch_text(self._url(query), provider=self.name, timeout=settings.provider_timeout) if not html: return [] tree = HTMLParser(html) out: List[RawResult] = [] seen: set = set() nodes = tree.css(self.item_sel) if not nodes: # Layout changed: fall back to harvesting the link pattern. nodes = tree.css(self.link_sel) for node in nodes: if len(out) >= limit: break a = node.css_first(self.link_sel) or ( node if node.tag == "a" else None) if not a: continue href = a.attributes.get("href", "") or "" if not href: continue if href.startswith("/"): href = self.base + href if href in seen: continue seen.add(href) meta = node.css_first(self.meta_sel) title = _clean(a.attributes.get("title") or a.text(), 300) if len(title) < 4: # Many tubes put the caption in a sibling block. for alt in ("div.thumb-under a", "p.title a", "a.title", "div.video-title", "span.title"): t2 = node.css_first(alt) if t2: title = _clean(t2.attributes.get("title") or t2.text(), 300) if len(title) >= 4: break if len(title) < 4: title = _clean(re.sub(r"[-_/]+", " ", href.rsplit("/", 1)[-1])[:120], 300) if len(title) < 4: continue out.append(RawResult(url=href, title=title, snippet=_clean(meta.text() if meta else ""), extra={"media": "video"})) return out class XnxxScrape(_AdultTube): name = "xnxx" base = "https://www.xnxx.com" path = "/search/{q}" item_sel = "div.thumb-block" link_sel = "a[href*='/video']" meta_sel = "p.metadata" weight = 1.05 class PornhatScrape(_AdultTube): name = "pornhat" base = "https://pornhat.com" path = "/search/{q}/" item_sel = "div.thumb" link_sel = "a[href*='/video/']" meta_sel = "span.duration" class SexComScrape(_AdultTube): name = "sexcom" base = "https://www.sex.com" path = "/search/pics?query={q}" item_sel = "article, div.masonry_box" link_sel = "a[href*='/pin/'], a[href*='/picture/']" meta_sel = "span.tag" weight = 0.85 class HQPornerScrape(_AdultTube): name = "hqporner" base = "https://hqporner.com" path = "/?q={q}" item_sel = "article" link_sel = "a[href*='/hdporn/']" meta_sel = "span.duration" class TnaflixScrape(_AdultTube): name = "tnaflix" base = "https://www.tnaflix.com" path = "/search.php?what={q}" item_sel = "div.thumb, div.video-item" link_sel = "a[href*='/video']" meta_sel = "span.duration" weight = 0.85 class BeegAPI(Provider): name = "beeg" kind = "adult" weight = 0.95 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = await fetch_json( "https://store.externulls.com/facts/search", params={"q": query, "limit": min(limit, 30)}, provider=self.name, timeout=settings.provider_timeout) if not isinstance(data, list): return [] out: List[RawResult] = [] for v in data[:limit]: fc = (v.get("fc_facts") or [{}])[0] if isinstance(v, dict) else {} vid = v.get("id") or fc.get("id") if not vid: continue out.append(RawResult( url=f"https://beeg.com/-{vid}", title=_clean((v.get("sf_resources") or {}).get("title") or fc.get("title") or f"beeg {vid}", 300), extra={"media": "video"})) return out # ---- booru / anime / manga ----------------------------------------------- class _Booru(Provider): kind = "adult" weight = 0.9 endpoint = "" is_list = True # Boorus AND-match tags, so a natural-language query returns nothing. # Use the single most distinctive token as the tag. _NOISE = {"best", "top", "free", "site", "sites", "video", "videos", "watch", "online", "the", "and", "for", "with", "new", "hd", "full", "porn", "xxx", "2024", "2025", "2026", "how", "find"} def _tag(self, query: str) -> str: words = [w for w in re.findall(r"[a-z0-9]+", query.lower()) if w not in self._NOISE and len(w) > 2] return words[0] if words else "rating:explicit" def _params(self, query: str, limit: int) -> Dict[str, Any]: return {"limit": min(limit, 30), "tags": self._tag(query)} def _rows(self, data: Any) -> List[Dict[str, Any]]: if isinstance(data, list): return data if isinstance(data, dict): for k in ("posts", "post", "results"): if isinstance(data.get(k), list): return data[k] return [] async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = await fetch_json(self.endpoint, params=self._params(query, limit), provider=self.name, timeout=settings.provider_timeout) out: List[RawResult] = [] for p in self._rows(data)[:limit]: if not isinstance(p, dict): continue f = p.get("file") url = (p.get("file_url") or (f.get("url") if isinstance(f, dict) else None) or p.get("source") or "") pid = p.get("id") if not url and pid: url = f"{self.endpoint.split('/index.php')[0]}/index.php?page=post&s=view&id={pid}" if not url: continue tags = p.get("tags") if isinstance(tags, dict): tags = " ".join(sum((v for v in tags.values() if isinstance(v, list)), [])[:12]) out.append(RawResult( url=url, title=_clean(f"{self.name} #{pid}", 120), snippet=_clean(str(tags or "")[:400]), score_hint=min(1.0, float( (p.get("score") or {}).get("total", 0) if isinstance(p.get("score"), dict) else (p.get("score") or 0)) / 200.0), extra={"media": "image"})) return out class E621(_Booru): name = "e621" endpoint = "https://e621.net/posts.json" class Yandere(_Booru): name = "yandere" endpoint = "https://yande.re/post.json" class Konachan(_Booru): name = "konachan" endpoint = "https://konachan.com/post.json" class Safebooru(_Booru): name = "safebooru" endpoint = "https://safebooru.org/index.php" def _params(self, query: str, limit: int) -> Dict[str, Any]: return {"page": "dapi", "s": "post", "q": "index", "json": 1, "limit": min(limit, 30), "tags": self._tag(query)} class Rule34US(Provider): name = "rule34us" kind = "adult" weight = 0.85 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: html = await fetch_text("https://rule34.us/index.php", params={"r": "posts/index", "q": query.replace(" ", "_")}, provider=self.name, timeout=settings.provider_timeout) if not html: return [] tree = HTMLParser(html) out: List[RawResult] = [] for a in tree.css("a[href*='posts/view'], div.thumbail-container a"): if len(out) >= limit: break href = a.attributes.get("href", "") if not href: continue if href.startswith("/") or href.startswith("?"): href = "https://rule34.us/" + href.lstrip("/") img = a.css_first("img") out.append(RawResult( url=href, title=_clean((img.attributes.get("alt") if img else "") or f"rule34 {query}", 200), extra={"media": "image"})) return out class NHentaiAPI(Provider): name = "nhentai" kind = "adult" weight = 0.9 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: for url in ("https://nhentai.net/api/v2/galleries/search", "https://nhentai.net/api/galleries/search"): data = await fetch_json(url, params={"query": query, "page": 1}, provider=self.name, timeout=settings.provider_timeout) if not isinstance(data, dict): continue out: List[RawResult] = [] for g in (data.get("result") or [])[:limit]: gid = g.get("id") if not gid: continue title = (g.get("title") or {}) out.append(RawResult( url=f"https://nhentai.net/g/{gid}/", title=_clean(title.get("english") or title.get("pretty") or str(gid), 300), snippet=_clean(", ".join( t.get("name", "") for t in (g.get("tags") or [])[:10])), extra={"media": "manga"})) if out: return out return [] # ========================================================================== # DEEP / ONION SURFACE INDEXES # Clearnet gateways that index .onion services - reachable without Tor. # ========================================================================== _ONION_RE = re.compile(r"[a-z2-7]{16,56}\.onion") class OnionLandSearch(Provider): name = "onionland" kind = "deep" weight = 1.10 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: html = await fetch_text("https://onionlandsearchengine.net/search", params={"q": query}, provider=self.name, timeout=settings.provider_timeout) if not html: return [] tree = HTMLParser(html) out: List[RawResult] = [] seen: set = set() for blk in tree.css("div.result-block"): if len(out) >= limit: break a = blk.css_first("div.title a") or blk.css_first("a") if not a: continue href = a.attributes.get("href", "") if not href or ".onion" not in href or href in seen: continue seen.add(href) desc = blk.css_first("div.description") or blk.css_first("div.link") out.append(RawResult( url=href, title=_clean(a.text(), 300), snippet=_clean(desc.text() if desc else ""), extra={"network": "tor"})) return out class OnionSearchEngine(Provider): name = "onionsearch" kind = "deep" weight = 1.05 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: html = await fetch_text("https://onionsearchengine.com/search.php", params={"search": query}, provider=self.name, timeout=settings.provider_timeout) if not html: return [] tree = HTMLParser(html) out: List[RawResult] = [] seen: set = set() for a in tree.css("a[href*='onion']"): if len(out) >= limit: break href = a.attributes.get("href", "") or "" title = _clean(a.text(), 300) if not _ONION_RE.search(href) or href in seen: continue if "report_page" in href or len(title) < 6: continue seen.add(href) out.append(RawResult(url=href, title=title, extra={"network": "tor"})) return out class TorchClearnet(Provider): """Onion-topic index reachable over clearnet.""" name = "torch" kind = "deep" weight = 0.9 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: html = await fetch_text("https://torchsearch.wordpress.com/", params={"s": query}, provider=self.name, timeout=settings.provider_timeout) if not html: return [] tree = HTMLParser(html) out: List[RawResult] = [] for art in tree.css("article, div.post"): if len(out) >= limit: break a = art.css_first("h1 a") or art.css_first("h2 a") or art.css_first("a") if not a: continue href = a.attributes.get("href", "") if not href: continue p = art.css_first("p") out.append(RawResult(url=href, title=_clean(a.text(), 300), snippet=_clean(p.text() if p else ""), extra={"network": "tor-index"})) return out # ========================================================================== # INDEPENDENT / UNFILTERED SURFACE ENGINES # Own crawlers - not reselling a filtered mainstream index. # ========================================================================== class MarginaliaSearch(Provider): name = "marginalia" kind = "web" weight = 1.0 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: for base in ("https://old-search.marginalia.nu/search", "https://search.marginalia.nu/search"): html = await fetch_text(base, params={"query": query}, provider=self.name, timeout=settings.provider_timeout) if not html: continue tree = HTMLParser(html) out: List[RawResult] = [] for card in tree.css("section.card, div.card"): if len(out) >= limit: break a = card.css_first("h2 a") or card.css_first("a.title") if not a: continue href = a.attributes.get("href", "") if not href or not href.startswith("http"): continue p = card.css_first("p") out.append(RawResult(url=href, title=_clean(a.text(), 300), snippet=_clean(p.text() if p else ""))) if out: return out return [] class WibySearch(Provider): name = "wiby" kind = "web" weight = 0.8 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = await fetch_json("https://wiby.me/json/", params={"q": query}, provider=self.name, timeout=settings.provider_timeout) if not isinstance(data, list): return [] return [RawResult(url=r.get("URL", ""), title=_clean(r.get("Title", ""), 300), snippet=_clean(r.get("Snippet") or r.get("Description") or "")) for r in data[:limit] if r.get("URL")] # ========================================================================== # ARCHIVES / FILES / UNMODERATED FORUMS # ========================================================================== class ArchiveOrgSearch(Provider): name = "archive" kind = "archive" weight = 1.0 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = await fetch_json( "https://archive.org/advancedsearch.php", params={"q": query, "rows": min(limit, 25), "output": "json", "fl[]": ["identifier", "title", "description", "year", "downloads", "mediatype"]}, provider=self.name, timeout=settings.provider_timeout) if not isinstance(data, dict): return [] out: List[RawResult] = [] for d in ((data.get("response") or {}).get("docs") or [])[:limit]: ident = d.get("identifier") if not ident: continue desc = d.get("description") if isinstance(desc, list): desc = " ".join(str(x) for x in desc) out.append(RawResult( url=f"https://archive.org/details/{ident}", title=_clean(str(d.get("title") or ident), 300), snippet=_clean(f"{d.get('mediatype','')} · " f"{d.get('downloads',0)} downloads · {desc or ''}"), published=str(d.get("year") or ""), score_hint=min(1.0, float(d.get("downloads") or 0) / 100000))) return out class OpenLibrarySearch(Provider): name = "openlibrary" kind = "archive" weight = 0.9 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = await fetch_json("https://openlibrary.org/search.json", params={"q": query, "limit": min(limit, 20)}, provider=self.name, timeout=settings.provider_timeout) if not isinstance(data, dict): return [] out: List[RawResult] = [] for d in (data.get("docs") or [])[:limit]: key = d.get("key") if not key: continue out.append(RawResult( url=f"https://openlibrary.org{key}", title=_clean(d.get("title", ""), 300), snippet=_clean(f"{', '.join((d.get('author_name') or [])[:3])} · " f"{d.get('first_publish_year','')}"), published=str(d.get("first_publish_year") or ""))) return out class ChanCatalog(Provider): """Unmoderated imageboard discussion - 4chan JSON API (SFW+NSFW boards).""" name = "chan" kind = "social" weight = 0.7 BOARDS_SFW = ["g", "sci", "biz", "his", "lit", "diy", "k", "o", "n"] BOARDS_NSFW = ["b", "pol", "x", "r9k", "d", "aco", "gif", "hc", "s", "e"] async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: boards = list(self.BOARDS_SFW) if kw.get("explicit"): boards = self.BOARDS_NSFW + boards terms = [w for w in re.findall(r"[a-z0-9]{3,}", query.lower())][:6] if not terms: return [] out: List[RawResult] = [] for board in boards[:5]: data = await fetch_json(f"https://a.4cdn.org/{board}/catalog.json", provider=self.name, timeout=6.0) if not isinstance(data, list): continue for page in data: for th in (page.get("threads") or []): blob = f"{th.get('sub','')} {th.get('com','')}".lower() hits = sum(1 for t in terms if t in blob) if hits < max(1, len(terms) // 3): continue text = re.sub(r"<[^>]+>", " ", th.get("com", "")) out.append(RawResult( url=f"https://boards.4chan.org/{board}/thread/{th.get('no')}", title=_clean(th.get("sub") or text[:90] or f"/{board}/ thread", 300), snippet=_clean(text, 400), score_hint=min(1.0, hits / max(1, len(terms))), extra={"board": board})) if len(out) >= limit: return out return out class TorrentIndex(Provider): """Public torrent metadata (apibay) - unfiltered file index.""" name = "torrents" kind = "files" weight = 0.85 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: data = await fetch_json("https://apibay.org/q.php", params={"q": query}, provider=self.name, timeout=settings.provider_timeout) if not isinstance(data, list): return [] out: List[RawResult] = [] for d in data[:limit]: if not isinstance(d, dict): continue name = d.get("name", "") if not name or name == "No results returned": continue seeders = int(d.get("seeders") or 0) size_gb = int(d.get("size") or 0) / 1e9 out.append(RawResult( url=f"https://thepiratebay.org/description.php?id={d.get('id')}", title=_clean(name, 300), snippet=_clean(f"{seeders} seeders · {size_gb:.2f} GB · " f"{d.get('num_files','?')} files"), score_hint=min(1.0, seeders / 2000.0), extra={"magnet": f"magnet:?xt=urn:btih:{d.get('info_hash','')}"})) return out class NyaaRSS(Provider): name = "nyaa" kind = "files" weight = 0.75 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: xml = await fetch_text("https://nyaa.si/", params={"page": "rss", "q": query, "f": 0, "c": "0_0"}, provider=self.name, timeout=settings.provider_timeout) return _parse_rss(xml, limit) if xml else [] # ========================================================================== # FILE / DOWNLOAD TIER # Verified live: every provider below returns links that actually resolve. # The old ThePirateBay description URLs timed out, so magnets are now # extracted directly and exposed in `extra` for one-click use. # ========================================================================== _MAGNET_RE = re.compile(r"magnet:\?xt=urn:btih:[a-zA-Z0-9]{32,40}[^\"'\s<>]*") def _clean_magnet(m: str) -> str: """WordPress sites HTML-encode ampersands inside hrefs.""" return (m.replace("&", "&").replace("&", "&") .replace("&", "&").strip()) _BTIH_RE = re.compile(r"btih:([a-fA-F0-9]{40}|[a-zA-Z2-7]{32})") def _size_to_gb(text: str) -> float: m = re.search(r"([\d.]+)\s*(TB|GB|MB|KB)", text or "", re.I) if not m: return 0.0 v, unit = float(m.group(1)), m.group(2).upper() return v * {"TB": 1024, "GB": 1, "MB": 1 / 1024, "KB": 1 / 1048576}[unit] class KnabenIndex(Provider): """ Knaben - meta-index across dozens of torrent trackers. The single most reliable file source we tested: 50 rows per query with magnet, size, date, seeders and leechers all present in the markup. """ name = "knaben" kind = "files" weight = 1.30 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: html = await fetch_text( f"https://knaben.org/search/{quote_plus(query)}/0/1/seeders", provider=self.name, timeout=settings.provider_timeout) if not html: return [] tree = HTMLParser(html) out: List[RawResult] = [] for tr in tree.css("tbody tr"): if len(out) >= limit: break tds = tr.css("td") if len(tds) < 6: continue a = tr.css_first("a[title]") or tr.css_first("a") if not a: continue title = _clean(a.attributes.get("title") or a.text(), 300) if len(title) < 3: continue mag = tr.css_first("a[href^='magnet:']") magnet = mag.attributes.get("href", "") if mag else "" try: seeders = int(re.sub(r"\D", "", tds[4].text(strip=True)) or 0) except ValueError: seeders = 0 size = tds[2].text(strip=True) cat = tds[0].text(strip=True) date = tds[3].text(strip=True) link = magnet or (a.attributes.get("href") or "") if not link: continue ih = _BTIH_RE.search(magnet) out.append(RawResult( url=link, title=title, snippet=_clean(f"{cat} · {size} · {seeders} seeders · {date}"), published=date, score_hint=min(1.0, seeders / 500.0), extra={"magnet": magnet, "size": size, "size_gb": _size_to_gb(size), "seeders": seeders, "category": cat, "infohash": ih.group(1) if ih else "", "kind": "torrent"}, )) return out class BitSearchIndex(Provider): """BitSearch / SolidTorrents - DHT-backed index, magnets inline.""" name = "bitsearch" kind = "files" weight = 1.15 async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: html = await fetch_text("https://bitsearch.to/search", params={"q": query, "sort": "seeders"}, provider=self.name, timeout=settings.provider_timeout) if not html: return [] tree = HTMLParser(html) out: List[RawResult] = [] seen: set = set() for mag in tree.css("a[href^='magnet:']"): if len(out) >= limit: break magnet = mag.attributes.get("href", "") ih = _BTIH_RE.search(magnet) key = ih.group(1) if ih else magnet[:60] if key in seen: continue seen.add(key) # walk up to the card that holds the title + stats node = mag title, stats = "", "" for _ in range(5): node = node.parent if node is None: break for sel in ("h5 a", "h5", "a[href*='/torrent/']", "h3 a", "h4 a"): h = node.css_first(sel) cand = _clean(h.text(), 300) if h else "" if cand and cand.lower() not in ("torrent", "magnet", "download"): title = cand stats = _clean(node.text(separator=" ", strip=True), 220) break if title: break if not title: continue sm = re.search(r"([\d.]+\s*(?:TB|GB|MB|KB))", stats, re.I) seed = re.search(r"(\d+)\s*(?:seed|Seeders)", stats, re.I) size = sm.group(1) if sm else "" seeders = int(seed.group(1)) if seed else 0 out.append(RawResult( url=magnet, title=title, snippet=_clean(f"{size} · {seeders} seeders" if size else stats[:160]), score_hint=min(1.0, seeders / 500.0), extra={"magnet": magnet, "size": size, "size_gb": _size_to_gb(size), "seeders": seeders, "infohash": ih.group(1) if ih else "", "kind": "torrent"}, )) return out class _RepackSite(Provider): """ Shared base for game-repack / direct-download sites. These publish an article per release; the magnet or mirror list lives on the detail page, so the top hits are opened and their links harvested. """ kind = "files" weight = 1.05 base = "" path = "/?s={q}" item_sel = "h1.entry-title a" fetch_detail = 3 # how many detail pages to open per query async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: html = await fetch_text(self.base + self.path.format(q=quote_plus(query)), provider=self.name, timeout=settings.provider_timeout) if not html: return [] tree = HTMLParser(html) picks: List[Tuple[str, str]] = [] for a in tree.css(self.item_sel): href = a.attributes.get("href", "") title = _clean(a.text(), 300) if href and title and len(title) > 3: picks.append((href, title)) if len(picks) >= limit: break if not picks: return [] async def detail(href: str, title: str) -> RawResult: page = await fetch_text(href, timeout=settings.scrape_timeout) magnets = [_clean_magnet(m) for m in _MAGNET_RE.findall(page or "")] size = "" if page: m = re.search(r"(?:Repack Size|Size)\s*:?\s*~?\s*" r"([\d.]+\s*(?:GB|MB))", page, re.I) size = m.group(1) if m else "" ih = _BTIH_RE.search(magnets[0]) if magnets else None return RawResult( url=magnets[0] if magnets else href, title=title, snippet=_clean(f"{self.name} · {size} · " f"{len(magnets)} download link(s)" if magnets else f"{self.name} · {size} · page"), extra={"magnet": magnets[0] if magnets else "", "all_magnets": magnets[:6], "page": href, "size": size, "size_gb": _size_to_gb(size), "infohash": ih.group(1) if ih else "", "kind": "repack"}, ) head = picks[: self.fetch_detail] rows = await asyncio.gather(*[detail(h, t) for h, t in head], return_exceptions=True) out = [r for r in rows if isinstance(r, RawResult) and (r.extra.get("magnet") or r.url.startswith("http"))] for href, title in picks[self.fetch_detail:]: out.append(RawResult(url=href, title=title, snippet=f"{self.name} · release page", extra={"page": href, "kind": "repack"})) return out[:limit] class FitGirlRepacks(_RepackSite): name = "fitgirl" base = "https://fitgirl-repacks.site" weight = 1.20 class DodiRepacks(_RepackSite): name = "dodi" base = "https://dodi-repacks.site" item_sel = "h1.entry-title a, h2 a" weight = 1.10 class GLoadDirect(_RepackSite): name = "gload" base = "https://gload.to" item_sel = "article h2 a, article h1 a" weight = 1.0 class X1337Index(Provider): """ 1337x - large release index (games, crack/repack scene, media). The primary domains sit behind Cloudflare (403), so we fail over across known-good mirrors. Search rows carry seeders/size; the magnet lives on the detail page, so the top hits are opened to harvest it. """ name = "1337x" kind = "files" weight = 1.25 MIRRORS = ["https://1337xx.to", "https://1377x.to", "https://1337x.ws", "https://1337x.to"] _preferred: Optional[str] = None fetch_detail = 4 def _order(self) -> List[str]: m = list(self.MIRRORS) if self._preferred and self._preferred in m: m.remove(self._preferred) m.insert(0, self._preferred) return m[:3] async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: rows = [] base_used = "" for base in self._order(): html = await fetch_text( f"{base}/search/{quote_plus(query)}/1/", provider=f"{self.name}:{base}", timeout=settings.provider_timeout) if not html: continue tree = HTMLParser(html) rows = tree.css("table.table-list tbody tr") if rows: base_used = base X1337Index._preferred = base break if not rows: return [] picks: List[Tuple[str, str, int, str]] = [] for tr in rows: if len(picks) >= limit: break tds = tr.css("td") if len(tds) < 3: continue links = [a for a in tds[0].css("a") if "/torrent/" in (a.attributes.get("href") or "")] if not links: continue href = links[0].attributes.get("href", "") title = _clean(links[0].text(), 300) if not href or len(title) < 3: continue try: seeders = int(re.sub(r"\D", "", tds[1].text(strip=True)) or 0) except ValueError: seeders = 0 size = tds[4].text(strip=True) if len(tds) > 4 else "" picks.append((base_used + href, title, seeders, size)) async def detail(url: str, title: str, seeders: int, size: str) -> RawResult: page = await fetch_text(url, timeout=settings.scrape_timeout) mags = [_clean_magnet(m) for m in _MAGNET_RE.findall(page or "")] ih = _BTIH_RE.search(mags[0]) if mags else None return RawResult( url=mags[0] if mags else url, title=title, snippet=_clean(f"1337x · {size} · {seeders} seeders"), score_hint=min(1.0, seeders / 500.0), extra={"magnet": mags[0] if mags else "", "page": url, "size": size, "size_gb": _size_to_gb(size), "seeders": seeders, "infohash": ih.group(1) if ih else "", "kind": "torrent"}, ) head = picks[: self.fetch_detail] got = await asyncio.gather( *[detail(u, t, s, z) for u, t, s, z in head], return_exceptions=True) out = [r for r in got if isinstance(r, RawResult)] for u, t, s, z in picks[self.fetch_detail:]: out.append(RawResult( url=u, title=t, snippet=_clean(f"1337x · {z} · {s} seeders"), score_hint=min(1.0, s / 500.0), extra={"page": u, "size": z, "size_gb": _size_to_gb(z), "seeders": s, "kind": "torrent"})) return out[:limit] # ========================================================================== # PREMIUM (auto-enabled when keys exist) # ========================================================================== class Tavily(Provider): name = "tavily" kind = "web" keyless = False weight = 1.35 def available(self) -> bool: return bool(settings.tavily_key) async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: body = { "api_key": settings.tavily_key, "query": query, "max_results": min(limit, 20), "search_depth": kw.get("tavily_depth", "advanced"), "include_answer": False, "include_raw_content": False, } if kw.get("include_domains"): body["include_domains"] = kw["include_domains"] if kw.get("exclude_domains"): body["exclude_domains"] = kw["exclude_domains"] data = await fetch_json("https://api.tavily.com/search", method="POST", json_body=body, provider=self.name, timeout=settings.provider_timeout) if not isinstance(data, dict): return [] return [RawResult(url=r.get("url", ""), title=_clean(r.get("title", ""), 300), snippet=_clean(r.get("content", "")), score_hint=float(r.get("score") or 0)) for r in data.get("results", [])] class Exa(Provider): name = "exa" kind = "web" keyless = False weight = 1.35 def available(self) -> bool: return bool(settings.exa_key) async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: body: Dict[str, Any] = { "query": query, "numResults": min(limit, 25), "type": kw.get("exa_type", "auto"), "contents": {"text": {"maxCharacters": 1200}}, } if kw.get("include_domains"): body["includeDomains"] = kw["include_domains"] if kw.get("exclude_domains"): body["excludeDomains"] = kw["exclude_domains"] data = await fetch_json("https://api.exa.ai/search", method="POST", json_body=body, headers={"x-api-key": settings.exa_key}, provider=self.name, timeout=settings.provider_timeout) if not isinstance(data, dict): return [] return [RawResult(url=r.get("url", ""), title=_clean(r.get("title") or "", 300), snippet=_clean(r.get("text") or ""), published=r.get("publishedDate"), score_hint=float(r.get("score") or 0)) for r in data.get("results", [])] class BraveAPI(Provider): name = "brave" kind = "web" keyless = False weight = 1.25 def available(self) -> bool: return bool(settings.brave_key) async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: params = {"q": query, "count": min(limit, 20)} fr = {"day": "pd", "week": "pw", "month": "pm", "year": "py"}.get(kw.get("freshness") or "") if fr: params["freshness"] = fr data = await fetch_json("https://api.search.brave.com/res/v1/web/search", params=params, headers={"X-Subscription-Token": settings.brave_key, "Accept": "application/json"}, provider=self.name, timeout=settings.provider_timeout) if not isinstance(data, dict): return [] return [RawResult(url=r.get("url", ""), title=_clean(r.get("title", ""), 300), snippet=_clean(r.get("description", "")), published=r.get("age")) for r in data.get("web", {}).get("results", [])] class Serper(Provider): """Google results via serper.dev.""" name = "serper" kind = "web" keyless = False weight = 1.40 def available(self) -> bool: return bool(settings.serper_key) async def search(self, query: str, limit: int, **kw: Any) -> List[RawResult]: body: Dict[str, Any] = {"q": query, "num": min(limit, 20)} tbs = {"day": "qdr:d", "week": "qdr:w", "month": "qdr:m", "year": "qdr:y"}.get(kw.get("freshness") or "") if tbs: body["tbs"] = tbs data = await fetch_json("https://google.serper.dev/search", method="POST", json_body=body, headers={"X-API-KEY": settings.serper_key, "Content-Type": "application/json"}, provider=self.name, timeout=settings.provider_timeout) if not isinstance(data, dict): return [] out = [] for r in data.get("organic", []): out.append(RawResult(url=r.get("link", ""), title=_clean(r.get("title", ""), 300), snippet=_clean(r.get("snippet", "")), published=r.get("date"))) return out # -------------------------------------------------------------------------- # registry # -------------------------------------------------------------------------- ALL_PROVIDERS: List[Provider] = [ DuckDuckGoHTML(), DuckDuckGoLite(), BingScrape(), YahooScrape(), MojeekScrape(), SearxNG(), GoogleNewsRSS(), BingNewsRSS(), Wikipedia(), HackerNews(), StackExchange(), GitHubSearch(), ArxivSearch(), OpenAlex(), CrossrefSearch(), PubMedSearch(), RedditSearch(), RedTubeAPI(), PornhubAPI(), EpornerAPI(), XvideosScrape(), XhamsterScrape(), XnxxScrape(), PornhatScrape(), SexComScrape(), HQPornerScrape(), TnaflixScrape(), BeegAPI(), E621(), Yandere(), Konachan(), Safebooru(), Rule34US(), NHentaiAPI(), OnionLandSearch(), OnionSearchEngine(), TorchClearnet(), MarginaliaSearch(), WibySearch(), ArchiveOrgSearch(), OpenLibrarySearch(), ChanCatalog(), TorrentIndex(), NyaaRSS(), KnabenIndex(), BitSearchIndex(), FitGirlRepacks(), DodiRepacks(), GLoadDirect(), X1337Index(), Tavily(), Exa(), BraveAPI(), Serper(), ] PROVIDERS_BY_NAME: Dict[str, Provider] = {p.name: p for p in ALL_PROVIDERS} def active_providers( names: Optional[List[str]] = None, kinds: Optional[List[str]] = None, ) -> List[Provider]: pool = [p for p in ALL_PROVIDERS if p.available()] if names: want = {n.lower() for n in names} pool = [p for p in pool if p.name in want] if kinds: kset = set(kinds) pool = [p for p in pool if p.kind in kset] return pool def provider_report() -> Dict[str, Any]: return { "active": [p.name for p in ALL_PROVIDERS if p.available()], "inactive_needs_key": [p.name for p in ALL_PROVIDERS if not p.available()], "by_kind": { k: [p.name for p in ALL_PROVIDERS if p.kind == k and p.available()] for k in ("web", "reference", "academic", "code", "social", "news", "adult", "deep", "archive", "files") }, "premium_enabled": settings.has_premium(), }