research/web.py

#!/usr/bin/env python3
"""Mechanical web-fetch helper for the /research skill.

Talks to the local SearXNG (search) and Firecrawl (scrape) services. Stdlib
only — no third-party runtime dependencies. The /research skill (Claude) does
all the thinking; this file only fetches and parses.

Endpoints resolve from env (WEBTOOLS_HOST / SEARXNG_PORT / FIRECRAWL_PORT) with
defaults for the `webtools` box. Any failure exits nonzero with a stderr message
so the calling skill can react.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path


class WebToolsError(Exception):
    """Raised on any unrecoverable search/scrape failure."""


def endpoints() -> tuple[str, str]:
    """Return (searxng_base, firecrawl_base) from env, with webtools defaults."""
    host = os.environ.get("WEBTOOLS_HOST", "10.10.20.37")
    searx_port = os.environ.get("SEARXNG_PORT", "8080")
    fire_port = os.environ.get("FIRECRAWL_PORT", "3002")
    return (f"http://{host}:{searx_port}", f"http://{host}:{fire_port}")


def _request(url, *, data=None, headers=None, timeout=30, retries=2, backoff=1.5):
    """GET (data=None) or POST JSON; parse JSON; retry transient network errors.

    Network failures (URLError/timeout) are retried with backoff. A non-JSON
    response is NOT retried — it is almost always a deterministic
    misconfiguration (e.g. SearXNG json format disabled), so we fail fast with
    a clear message.
    """
    headers = dict(headers or {})
    body = None
    if data is not None:
        body = json.dumps(data).encode("utf-8")
        headers.setdefault("Content-Type", "application/json")

    last = None
    for attempt in range(retries + 1):
        try:
            req = urllib.request.Request(
                url, data=body, headers=headers,
                method="POST" if body is not None else "GET",
            )
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                raw = resp.read()
            break
        except (urllib.error.URLError, TimeoutError) as exc:
            last = exc
            if attempt < retries:
                time.sleep(backoff ** attempt)
    else:
        raise WebToolsError(f"request failed after {retries + 1} attempt(s): {url}: {last}")

    try:
        return json.loads(raw.decode("utf-8"))
    except ValueError as exc:
        raise WebToolsError(
            f"non-JSON response from {url} (is SearXNG json format enabled?): {exc}"
        )


def search(query, max_results=12, timeout=30):
    """Query SearXNG and return up to max_results normalized hits."""
    searx_base, _ = endpoints()
    qs = urllib.parse.urlencode({"q": query, "format": "json"})
    payload = _request(f"{searx_base}/search?{qs}", timeout=timeout)
    hits = []
    for r in payload.get("results", []):
        url = r.get("url")
        if not url:
            continue
        hits.append({
            "title": r.get("title", ""),
            "url": url,
            "snippet": r.get("content", ""),
            "engine": r.get("engine", ""),
        })
    return hits[:max_results]


def scrape(url, timeout=60):
    """Scrape a URL to markdown via Firecrawl. Raises WebToolsError on failure."""
    _, fire_base = endpoints()
    payload = _request(
        f"{fire_base}/v1/scrape",
        data={"url": url, "formats": ["markdown"]},
        timeout=timeout,
    )
    if not payload.get("success", True):
        raise WebToolsError(f"firecrawl failed for {url}: {payload.get('error')}")
    markdown = (payload.get("data") or {}).get("markdown", "")
    if not markdown:
        raise WebToolsError(f"firecrawl returned no markdown for {url}")
    return markdown


def main(argv=None):
    parser = argparse.ArgumentParser(
        prog="web.py",
        description="Search (SearXNG) and scrape (Firecrawl) on the webtools box.",
    )
    sub = parser.add_subparsers(dest="cmd", required=True)

    sp = sub.add_parser("search", help="search SearXNG")
    sp.add_argument("query")
    sp.add_argument("--max", type=int, default=12, help="max results (default 12)")
    sp.add_argument("--json", action="store_true", help="emit JSON")

    cp = sub.add_parser("scrape", help="scrape a URL to markdown")
    cp.add_argument("url")
    cp.add_argument("--out", help="write markdown to this path instead of stdout")

    args = parser.parse_args(argv)

    try:
        if args.cmd == "search":
            hits = search(args.query, max_results=args.max)
            if args.json:
                print(json.dumps(hits, indent=2))
            else:
                for h in hits:
                    print(f"{h['title']}\n  {h['url']}\n  {h['snippet'][:160]}\n")
        elif args.cmd == "scrape":
            markdown = scrape(args.url)
            if args.out:
                Path(args.out).write_text(markdown, encoding="utf-8")
                print(f"wrote {args.out}")
            else:
                print(markdown)
    except WebToolsError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())