From 24e8c5a6c62443bf76504b9682cc8e58f86d4246 Mon Sep 17 00:00:00 2001 From: tslil Date: Sun, 20 Sep 2026 10:18:31 +0100 Subject: [web_search.py] use kagi search api --- skills/web-search/web_search.py | 132 ++++++++++++++++------------------------ 1 file changed, 53 insertions(+), 79 deletions(-) (limited to 'skills/web-search/web_search.py') diff --git a/skills/web-search/web_search.py b/skills/web-search/web_search.py index 8124dc9..7889f96 100755 --- a/skills/web-search/web_search.py +++ b/skills/web-search/web_search.py @@ -1,20 +1,21 @@ #!/usr/bin/env python3 """ -Search the web via DuckDuckGo's html endpoint and print plain-text results. +Search the web via the Kagi Search API and print plain-text results. -Exits 2 on rate-limit/bot-block (retry later), 1 on other errors. +The API key is read from $KAGI_API_KEY. +Exits 2 on rate-limit (retry later), 1 on other errors. """ import argparse +import json +import os import sys import urllib.error -import urllib.parse import urllib.request from dataclasses import dataclass -from html.parser import HTMLParser -ENDPOINT = "https://html.duckduckgo.com/html/" -USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64; rv:155.0) Gecko/20100101 Firefox/155.0" +ENDPOINT = "https://kagi.com/api/v1/search" +API_KEY_ENV = "KAGI_API_KEY" @dataclass @@ -24,73 +25,41 @@ class SearchResult: snippet: str = "" -class ResultsParser(HTMLParser): - def __init__(self) -> None: - super().__init__() - self.results: list[SearchResult] = [] - # Title text accumulated while inside a result anchor. - self._title_parts: list[str] | None = None - self._href: str = "" - # Index of the result whose snippet we are currently inside. - self._snippet_for: int | None = None - self._snippet_parts: list[str] = [] - - def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: - if tag != "a": - return - classes = dict(attrs).get("class", "").split() - if "result__a" in classes: - self._title_parts = [] - self._href = dict(attrs).get("href", "") - elif "result__snippet" in classes: - self._snippet_for = len(self.results) - 1 - self._snippet_parts = [] - - def handle_data(self, data: str) -> None: - if self._title_parts is not None: - self._title_parts.append(data) - elif self._snippet_for is not None: - self._snippet_parts.append(data) - - def handle_endtag(self, tag: str) -> None: - if tag != "a": - return - if self._title_parts is not None: - title = " ".join("".join(self._title_parts).split()) - url = self._unwrap_redirect(self._href) - if title and url: - self.results.append(SearchResult(title, url)) - self._title_parts = None - elif self._snippet_for is not None: - if 0 <= self._snippet_for < len(self.results): - self.results[self._snippet_for].snippet = " ".join( - "".join(self._snippet_parts).split() - ) - self._snippet_for = None - - @staticmethod - def _unwrap_redirect(href: str) -> str: - parsed = urllib.parse.urlparse(href) - if parsed.netloc.endswith("duckduckgo.com") and parsed.path == "/l/": - target = urllib.parse.parse_qs(parsed.query).get("uddg", [None])[0] - return target if target else href - return href - - -def fetch(query: str) -> str: - data = urllib.parse.urlencode({"q": query}).encode() - request = urllib.request.Request( - ENDPOINT, data=data, headers={"User-Agent": USER_AGENT} - ) - with urllib.request.urlopen(request, timeout=15) as response: - return response.read().decode("utf-8", "replace") - - def fail(message: str, exit_code: int = 1) -> None: print(f"web_search: {message}", file=sys.stderr) sys.exit(exit_code) +def error_message(err: urllib.error.HTTPError) -> str: + try: + payload = json.loads(err.read().decode("utf-8", "replace")) + details = payload.get("errors") or payload.get("error") or [{}] + first = details[0] + bits = [b for b in (first.get("code"), first.get("message")) if b] + if bits: + return "; ".join(bits) + except (ValueError, AttributeError, IndexError): + pass + return err.reason or "unknown error" + + +def fetch(query: str, limit: int) -> dict: + key = os.environ.get(API_KEY_ENV, "").strip() + if not key: + fail(f"{API_KEY_ENV} is not set", exit_code=1) + body = json.dumps({"query": query, "limit": limit}).encode() + request = urllib.request.Request( + ENDPOINT, + data=body, + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {key}", + }, + ) + with urllib.request.urlopen(request, timeout=30) as response: + return json.loads(response.read().decode("utf-8", "replace")) + + def main() -> None: parser = argparse.ArgumentParser(description=__doc__) _ = parser.add_argument( @@ -101,24 +70,29 @@ def main() -> None: query = " ".join(args.query) try: - body = fetch(query) + payload = fetch(query, args.n) except urllib.error.HTTPError as err: - if err.code in (403, 429): - fail(f"rate-limited or blocked (HTTP {err.code}); try later", exit_code=2) - fail(f"HTTP {err.code}") + if err.code == 429: + fail("rate-limited (HTTP 429); try later", exit_code=2) + fail(f"HTTP {err.code}: {error_message(err)}") except OSError as err: fail(f"network error: {err}") - extractor = ResultsParser() - extractor.feed(body) - - if not extractor.results: - if "anomaly" in body or "challenge" in body: - fail("bot challenge served instead of results; try later", exit_code=2) + results = [ + SearchResult( + title=item.get("title", "").strip(), + url=item.get("url", "").strip(), + snippet=" ".join(item.get("snippet", "").split()), + ) + for item in payload.get("data", {}).get("search", []) + ] + results = [r for r in results if r.title and r.url] + + if not results: print(f"no results for: {query}") return - for number, result in enumerate(extractor.results[: args.n], 1): + for number, result in enumerate(results[: args.n], 1): print(f"{number}. {result.title}\n {result.url}") if result.snippet: print(f" {result.snippet}") -- cgit v1.3.1