1 files changed, 128 insertions, 0 deletions
diff --git a/skills/web-search/web_search.py b/skills/web-search/web_search.py
new file mode 100755
index 0000000..8124dc9
--- /dev/null
+++ b/skills/web-search/web_search.py
@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+"""
+Search the web via DuckDuckGo's html endpoint and print plain-text results.
+
+Exits 2 on rate-limit/bot-block (retry later), 1 on other errors.
+"""
+
+import argparse
+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"
+
+
+@dataclass
+class SearchResult:
+    title: str
+    url: str
+    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 main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    _ = parser.add_argument(
+        "-n", type=int, default=5, metavar="N", help="max results (default 5)"
+    )
+    _ = parser.add_argument("query", nargs="+", help="search terms")
+    args = parser.parse_args()
+    query = " ".join(args.query)
+
+    try:
+        body = fetch(query)
+    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}")
+    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)
+        print(f"no results for: {query}")
+        return
+
+    for number, result in enumerate(extractor.results[: args.n], 1):
+        print(f"{number}. {result.title}\n   {result.url}")
+        if result.snippet:
+            print(f"   {result.snippet}")
+
+
+if __name__ == "__main__":
+    main()