diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6cc5c4a..ba0c4cc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -35,6 +35,14 @@ jobs: python -c "from switchback import scrape, scrape_detailed, search; print('import OK')" python -m switchback --help || true + # Offline tests (fake tiers, no network). Includes the backward-compatibility + # tests that pin the result fields and CLI/HTTP JSON keys; fastapi + httpx + # let the HTTP shape test run too. + - name: Offline tests + run: | + pip install pytest fastapi httpx + pytest -q tests + - name: Build distribution run: | pip install build diff --git a/CHANGELOG.md b/CHANGELOG.md index 9d17d5b..cba1bb6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,36 @@ versioning while pre-1.0. ## [Unreleased] +### Added +- **Agent-ready extras, all opt-in** (`scrape(..., state=, appendix=, query=)`, + the same flags on `scrape_detailed`, `--state` / `--appendix` / `--query` on the + CLI, and `state` / `appendix` / `query` in the HTTP `/scrape` body). Without them + results, CLI JSON and HTTP JSON are unchanged: `ScrapeResult` and `ScrapeOutcome` + keep exactly their fields. With them, results are the new subclasses + `ScrapeResultWithExtras` / `ScrapeOutcomeWithExtras`. + - `state` — the page's embedded JSON (Next.js `__NEXT_DATA__`, Nuxt, + `window.__X__` globals, JSON-LD). + - `appendix` — nav/header/footer/aside as separate markdown regions. + - `query` → `highlights` — the passages that answer the question, each with a + text-fragment citation link. Graded by local BM25 by default, or by TypeSafe's + Jev model with `SCRAPER_HIGHLIGHTS_GRADER=jev` + `TYPESAFE_API_KEY` (falls back + to BM25 on any Jev error). +- **MCP server** — `switchback-mcp` (stdio) exposes `scrape` and `search` as tools. + Install with `switchback[mcp]`. +- **YouTube transcripts** in tier_1 (title, channel, timestamped transcript linked + to the moment in the video). Opt in with `SCRAPER_YOUTUBE=1` plus + `switchback[youtube]`; otherwise, or for videos with no captions, YouTube URLs + go through the cascade as before. + YouTube blocks most cloud IPs; a blocked request is retried once through + `SCRAPER_EGRESS_PROXY` when one is configured. Transcript highlights cite the + timestamp in the video. +- **Office and ebook files** (.docx, .pptx, .xlsx, .xls, .epub) converted to + markdown in tier_2 via markitdown. Opt in with `SCRAPER_DOCUMENTS=1` plus + `switchback[documents]`; otherwise these URLs are handled exactly as before. + +The new extras are deliberately not added to `[all]`, so upgrading an existing +`[all]` install pulls no new dependencies. + ## [0.5.0] - 2026-06-30 ### Changed diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index cff69e2..63bd83d 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -25,8 +25,40 @@ policy (`switchback/policy/`). Start with the cascade runner in - **Don't commit secrets or run artifacts.** `.env`, `state/`, and `*.csv` are gitignored; keep it that way. +## Backward compatibility comes first +Upgrading must never change what an existing call returns. Breaking a user's app +is worse than any feature is good. + +What went wrong once: the agent extras added `state` / `appendix` / `highlights` +as defaulted fields on `ScrapeResult` and `ScrapeOutcome`. That looked additive, +but `asdict(r)` (how people serialise results) gained three `null` keys and +`astuple(r)` grew from 4 values to 7. The CLI and HTTP JSON were fine; the Python +objects were not. It was caught before merge and fixed with opt-in subclasses. + +Rules for any new feature: +- **Never add, remove, rename or reorder fields on a public result type** + (`ScrapeResult`, `ScrapeOutcome`, `TierAttempt`, `SearchResult`). Put new data + on a subclass that is returned only when the caller opts in (see + `ScrapeResultWithExtras` in `switchback/orchestrator.py`). +- **New parameters are keyword-only and default to the old behaviour.** Existing + parameters keep their names, order and defaults. +- **CLI and HTTP JSON gain a key only when the request asked for it** (see + `result_dict` in `switchback/api.py`). +- **A new dependency gets its own extra and stays out of `[all]`**, and a missing + dependency means the old behaviour. Installing a package is not opt-in: users + may already have it for something else. Anything that changes what an existing + call returns stays off until a `SCRAPER_*` variable turns it on (see + `SCRAPER_YOUTUBE`, `SCRAPER_DOCUMENTS`). +- **The pinned tests are the contract.** The "backward compatibility" tests in + `tests/test_extras.py` pin the result fields and the CLI/HTTP JSON keys, and CI + runs them. If one fails, the change is breaking: redesign the change, don't + edit the pin. +- **Before opening the PR, compare against `main` live:** scrape a few pages + with the default call on both; the output should be byte-identical. + ## Tests -`tests/test_suite.py` exercises the cascade across the anti-bot difficulty +`pytest tests` runs the offline tests (fake tiers, no network, a few seconds); +CI runs them on every PR. `tests/test_suite.py` exercises the cascade across the anti-bot difficulty spectrum (needs network + browser tiers): `python tests/test_suite.py --quick` for a fast tier-0/1 pass. diff --git a/README.md b/README.md index f050f7c..ad18278 100644 --- a/README.md +++ b/README.md @@ -66,8 +66,8 @@ That's the whole loop. Add tiers as you need them (see [Install](#install)). | Tier | Strategy | Cost | |---|---|---| -| tier_1 | Direct APIs / mirrors (arxiv, wikipedia, EuropePMC; extend: job boards) | free, cleanest | -| tier_2 | Plain HTTP + TLS impersonation (`curl_cffi`), incl. PDFs | cheap | +| tier_1 | Direct APIs / mirrors (arxiv, wikipedia, EuropePMC, YouTube transcripts; extend: job boards) | free, cleanest | +| tier_2 | Plain HTTP + TLS impersonation (`curl_cffi`), incl. PDFs and office/ebook files | cheap | | tier_3 | Cloudflare / anti-bot solver (`cloudscraper`, install `.[cloudflare]`) | cheap-ish (~5s/host) | | tier_4 | Stealth headless browser (`patchright`, Chromium) | heavy | | tier_5 | Camoufox (Firefox stealth) — **on by default** (opt out: `SCRAPER_DISABLE_CAMOUFOX`) | heavy + slow (~40s on hard CF) | @@ -94,9 +94,18 @@ pip install "switchback[browser]" && patchright install chromium # + tier_4 st pip install "switchback[camoufox]" && camoufox fetch # + tier_5 Firefox stealth pip install "switchback[firecrawl]" # + tier_7 paid API (needs FIRECRAWL_API_KEY) pip install "switchback[tracing]" # + OpenTelemetry -> any OTLP backend -pip install "switchback[all]" # everything +pip install "switchback[all]" # all of the above +pip install "switchback[mcp]" # + MCP server for agents (switchback-mcp) +pip install "switchback[youtube]" # + YouTube transcripts in tier_1 (and set SCRAPER_YOUTUBE=1) +pip install "switchback[documents]" # + .docx/.pptx/.xlsx/.xls/.epub → markdown in tier_2 (and set SCRAPER_DOCUMENTS=1) ``` +`mcp`, `youtube` and `documents` are not part of `[all]`, so upgrading an existing +`[all]` install doesn't pull new dependencies (`documents` brings markitdown and +onnxruntime). Add them by name. YouTube transcripts and office files also need +their switch (`SCRAPER_YOUTUBE=1`, `SCRAPER_DOCUMENTS=1`), because they change what +those URLs return; without it they behave exactly as before, installed or not. + For tier_3's **full** v3 JS-VM + Turnstile + stealth, install the Enhanced Edition 3.x fork (PyPI's `cloudscraper` is the older v1/v2 — PyPI forbids pinning a git-URL dep inside a published package, so install it alongside): @@ -172,6 +181,61 @@ Non-Python callers: see [clients/node_bridge.md](clients/node_bridge.md). Python callers that want HTTP-with-CLI-fallback can drop in [clients/python_client.py](clients/python_client.py). +**MCP server** (for agents: Claude Code, Cursor, any MCP client) +```bash +pip install "switchback[mcp]" +claude mcp add switchback -- switchback-mcp # stdio; tools: scrape, search +``` + +## Agent-ready extras + +Off by default: without them every result is exactly what it was before, the same +`ScrapeResult` / `ScrapeOutcome` with the same fields, and the same CLI and HTTP +JSON. Ask for any combination per call; results then come back as +`ScrapeResultWithExtras` / `ScrapeOutcomeWithExtras` (subclasses, so existing +`isinstance` checks still hold) with the extras you asked for filled in and the +rest `None`. The CLI and HTTP JSON add only the keys you asked for. + +```python +from switchback import scrape +[r] = scrape("https://example.com/product", state=True, appendix=True, + query="what does it cost?") +r.state # {"__NEXT_DATA__": {...}, "json_ld": [...], "__INITIAL_STATE__": {...}} +r.appendix # [{"region": "nav", "markdown": "..."}, {"region": "footer", ...}] +r.highlights # [{"text": "...", "score": 7.1, "grader": "bm25", + # "link": "https://example.com/product#:~:text=Price%20from,per%20month"}] +``` + +```bash +switchback --state --appendix --query "what does it cost?" https://example.com/product +curl -s localhost:8799/scrape -d '{"urls":["https://example.com"],"query":"pricing","state":true}' +``` + +| extra | what you get | +| --- | --- | +| `state` | The JSON the page embeds for its own frontend: Next.js `__NEXT_DATA__`, Nuxt, `window.__X__` globals (incl. `JSON.parse("…")`), schema.org JSON-LD. Often cleaner than the text (prices, ids, dates). Only valid JSON is returned. `{}` when the winning tier had no HTML (API mirrors, PDFs, docs, Firecrawl markdown). | +| `appendix` | The menus, header, footer and sidebars that are stripped from the main content, as separate markdown, one entry per region. | +| `query` → `highlights` | The passages that answer the question, best first, each with a citation `link`: a [text fragment](https://developer.mozilla.org/en-US/docs/Web/URI/Reference/Fragment/Text_fragments) URL that opens the page scrolled to, and highlighting, that passage (for a YouTube transcript, the video at that moment). If the text isn't visible on the page (e.g. it came from the `` or image alt text) the browser simply opens the page at the top. | + +`state` and `appendix` need the fetched HTML, so they skip the content cache +(`SCRAPER_CONTENT_TTL_S`) and fetch; `highlights` work on cached content too. + +**Grading highlights.** The default grader is local BM25 keyword scoring: free, +offline, no data leaves the box. For better relevance, opt into +[TypeSafe](https://typesafe.ai)'s Jev model, which answers "does this passage +answer the question?" for every passage and returns a calibrated probability +(passages under `SCRAPER_HIGHLIGHTS_MIN_SCORE`, default 0.5, are dropped, so an +off-topic page honestly returns `[]`): + +```bash +export SCRAPER_HIGHLIGHTS_GRADER=jev +export TYPESAFE_API_KEY=... # sends the page passages + query to api.typesafe.ai +``` + +Jev is billed per input token (about $0.0004 for a 10k-token page). If the call +fails, that request falls back to BM25 (`"grader": "bm25"`); the scrape never fails +because of highlights. + ## Cost-scoped residential egress The dominant reason hard hosts wall you is the **datacenter IP**, not the @@ -258,6 +322,13 @@ export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 - `SCRAPER_TIER_RETRY_ON` — failure classes eligible for a same-tier retry (default `timeout,rate_limited,connection`; widen to include `botwall,http_block` behind a rotating residential proxy so each retry gets a fresh IP). Retries are bounded by `SCRAPER_DEADLINE_S`; enabling them on the paid Firecrawl tier bills per attempt - `SCRAPER_LOGIN_HOOK` — `pkg.module:func` returning `{cookie: value}` for a host (see [Logged-in sessions](#logged-in-sessions)) - `SCRAPER_EXTRACTION_FILE` — per-domain extraction prefs JSON (default `config/extraction.json`) +- `SCRAPER_HIGHLIGHTS_GRADER` — `bm25` (default) or `jev` (needs `TYPESAFE_API_KEY`; see [Agent-ready extras](#agent-ready-extras)) +- `SCRAPER_HIGHLIGHTS_TOP_K` — highlights returned per URL (5) +- `SCRAPER_HIGHLIGHTS_MIN_SCORE` — Jev probability below which a passage is dropped (0.5) +- `SCRAPER_JEV_MODEL` / `SCRAPER_JEV_TIMEOUT_S` — Jev model alias (`jev-latest`) and per-request timeout (20s) +- `SCRAPER_YOUTUBE` — opt-in: `1` returns YouTube transcripts from tier_1 (needs `switchback[youtube]`) +- `SCRAPER_DOCUMENTS` — opt-in: `1` converts .docx/.pptx/.xlsx/.xls/.epub to markdown in tier_2 (needs `switchback[documents]`) +- `SCRAPER_YOUTUBE_LANGS` — transcript languages to try, in order (`en`; e.g. `en,hi`). YouTube blocks most cloud IPs; when it does, the transcript is retried once through `SCRAPER_EGRESS_PROXY` if set - `SCRAPER_TRACE_SESSION` — opt-in: capture a Playwright trace (screenshots + DOM + network) per browser-tier attempt, written to `state/traces/` - `BENCH_FIRECRAWL_USD` / `BENCH_FIRECRAWL_HARD_MULT` — cost model for the savings report </details> @@ -309,6 +380,20 @@ strip site-specific noise, declare prefs per host in `config/extraction.json` (see [config/extraction.example.json](config/extraction.example.json)); every tier's normalize step picks them up automatically. +## Roadmap + +Not built yet: + +- **Layout-aware markdown** — keep pricing grids and card layouts built from + `<div>`s (not `<table>`s) as tables, using the browser tiers' rendered layout. +- **Scanned pages and images** — OCR / vision-model text for PDFs with no text + layer and for images. +- **Audio and caption-less video** — speech-to-text transcription. +- **Hosted service** — managed browsers at high concurrency, stored results and + shareable result links. switchback stays a library you run yourself. +- Smaller gaps: Next.js App Router flight data (`self.__next_f`) in `state`, and + YouTube chapters in transcripts. + ## Contributing Issues and PRs welcome — see [CONTRIBUTING.md](CONTRIBUTING.md). Start with the diff --git a/pyproject.toml b/pyproject.toml index fe1db45..439022d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,6 +53,9 @@ browser = ["patchright"] # tier3: stealth headless Chromium camoufox = ["camoufox[geoip]"] # tier3b: Firefox stealth (geoip: match locale/tz to proxy IP) firecrawl = ["firecrawl-py"] # tier4: paid last-resort API server = ["fastapi", "uvicorn"] # HTTP service (switchback.server) + /metrics + /traces +mcp = ["mcp"] # MCP server for agents (switchback-mcp) +youtube = ["youtube-transcript-api>=1.0"] # tier_1: YouTube transcripts +documents = ["markitdown[docx,pptx,xlsx,xls]"] # tier_2: office/ebook files → markdown tracing = [ # OTel → any OTLP backend "opentelemetry-sdk", "opentelemetry-exporter-otlp-proto-grpc", @@ -62,11 +65,15 @@ all = [ "fastapi", "uvicorn", "opentelemetry-sdk", "opentelemetry-exporter-otlp-proto-grpc", ] +# Not in `all` on purpose: added in 0.6, kept out so existing `[all]` installs +# don't suddenly pull markitdown's onnxruntime (via magika) or the MCP SDK. +# Install them by name: "switchback[mcp,youtube,documents]". dev = ["pytest"] [project.scripts] switchback = "switchback.api:_main" switchback-server = "switchback.server:main" +switchback-mcp = "switchback.mcp_server:main" [project.urls] Homepage = "https://github.com/akash-kr/switchback" diff --git a/switchback/__init__.py b/switchback/__init__.py index 76a1a18..a91ccc8 100644 --- a/switchback/__init__.py +++ b/switchback/__init__.py @@ -5,8 +5,10 @@ results = scrape(["https://example.com/article"]) hits = search("web scraping") # query → URLs (SearXNG) """ -from .api import scrape, scrape_detailed, ScrapeOutcome, ScrapeResult, TierAttempt +from .api import (scrape, scrape_detailed, ScrapeOutcome, ScrapeResult, TierAttempt, + ScrapeOutcomeWithExtras, ScrapeResultWithExtras) from .search import search, SearchResult __all__ = ["scrape", "scrape_detailed", "ScrapeOutcome", "ScrapeResult", - "TierAttempt", "search", "SearchResult"] + "TierAttempt", "search", "SearchResult", + "ScrapeOutcomeWithExtras", "ScrapeResultWithExtras"] diff --git a/switchback/api.py b/switchback/api.py index dfa3694..d1fa7e1 100644 --- a/switchback/api.py +++ b/switchback/api.py @@ -11,29 +11,62 @@ import sys from .normalize import output_key -from .orchestrator import ScrapeOutcome, ScrapeResult, TierAttempt, run, run_detailed +from .orchestrator import (Extras, ScrapeOutcome, ScrapeOutcomeWithExtras, ScrapeResult, + ScrapeResultWithExtras, TierAttempt, run, run_detailed) from .search import search # re-export: query → URLs (SearXNG) -def scrape(urls: str | list[str], fmt: str | None = None) -> list[ScrapeResult]: +def _extras(state: bool, appendix: bool, query: str | None) -> Extras | None: + """None when nothing extra was asked for, so the default path is untouched.""" + if state or appendix or (query and query.strip()): + return Extras(state=bool(state), appendix=bool(appendix), + query=query.strip() if query and query.strip() else None) + return None + + +def scrape(urls: str | list[str], fmt: str | None = None, *, state: bool = False, + appendix: bool = False, query: str | None = None) -> list[ScrapeResult]: """Scrape one or many URLs through the cascade. Returns successes only. fmt selects the output format (markdown | markdown_trimmed | html | html_selectors); None uses the SCRAPER_OUTPUT_FORMAT default (markdown). For failures with classified reasons + the per-tier cascade, use - scrape_detailed().""" + scrape_detailed(). + + Opt-in extras. Without any, results are plain ScrapeResults, unchanged. With + one, results are ScrapeResultWithExtras (a ScrapeResult subclass) whose + matching field is filled; the others stay None: + state=True embedded page JSON (__NEXT_DATA__, JSON-LD, window.__X__ …) + appendix=True the nav/header/footer/aside stripped from the main content + query="…" highlights: the passages that answer it, with citation links""" if isinstance(urls, str): urls = [urls] - return run(urls, fmt) + return run(urls, fmt, _extras(state, appendix, query)) -def scrape_detailed(urls: str | list[str], fmt: str | None = None) -> list[ScrapeOutcome]: +def scrape_detailed(urls: str | list[str], fmt: str | None = None, *, + state: bool = False, appendix: bool = False, + query: str | None = None) -> list[ScrapeOutcome]: """Like scrape() but returns a ScrapeOutcome per URL — successes *and* failures, each with final_outcome, error_class, status_code, and the - per-tier attempts that were made. fmt as in scrape().""" + per-tier attempts that were made. fmt and the extras as in scrape().""" if isinstance(urls, str): urls = [urls] - return run_detailed(urls, fmt) + return run_detailed(urls, fmt, _extras(state, appendix, query)) + + +def result_dict(r: ScrapeResult | ScrapeOutcome, state: bool = False, appendix: bool = False, + query: str | None = None) -> dict: + """The JSON shape the CLI, HTTP service and MCP server return for a result. + Extra keys appear only when that extra was requested.""" + d = {"url": r.url, "source_method": r.source_method, output_key(r.format): r.markdown} + if state: + d["state"] = r.state + if appendix: + d["appendix"] = r.appendix + if query and query.strip(): + d["highlights"] = r.highlights + return d def _main() -> int: @@ -53,11 +86,14 @@ def _main() -> int: _k = _k.strip() if _k and _k not in _os.environ: _os.environ[_k] = _v.strip() - usage = ("usage: switchback [--format FMT] <url> [<url> ...]\n" + usage = ("usage: switchback [--format FMT] [--state] [--appendix] [--query Q] <url> [<url> ...]\n" " switchback --search <query ...>\n" " switchback --doctor\n" " (or: python -m switchback <url> ...)\n" - " FMT: markdown (default) | markdown_trimmed | html | html_selectors") + " FMT: markdown (default) | markdown_trimmed | html | html_selectors\n" + " --state add the page's embedded JSON (Next.js, JSON-LD, ...)\n" + " --appendix add the nav/header/footer/aside as separate markdown\n" + " --query Q add highlights: passages that answer Q, with citation links") # --help/-h is an explicit request: usage to stdout, exit 0 (don't treat it # as a URL to scrape). Check before any work so it stays fast and side-effect-free. if any(a in ("--help", "-h") for a in sys.argv[1:]): @@ -78,8 +114,11 @@ def _main() -> int: [{"title": h.title, "url": h.url, "snippet": h.snippet} for h in hits], indent=2)) return 0 if hits else 1 - # Optional --format / --format=FMT flag; everything else is a URL. + # Optional --format / --format=FMT, --state, --appendix, --query / --query=Q + # flags; everything else is a URL. fmt: str | None = None + state = appendix = False + query: str | None = None rest: list[str] = [] argv = sys.argv[1:] i = 0 @@ -89,16 +128,21 @@ def _main() -> int: fmt = argv[i + 1]; i += 2; continue if a.startswith("--format="): fmt = a.split("=", 1)[1]; i += 1; continue + if a == "--state": + state = True; i += 1; continue + if a == "--appendix": + appendix = True; i += 1; continue + if a == "--query" and i + 1 < len(argv): + query = argv[i + 1]; i += 2; continue + if a.startswith("--query="): + query = a.split("=", 1)[1]; i += 1; continue rest.append(a); i += 1 if not rest: print(usage, file=sys.stderr) return 2 - results = scrape(rest, fmt=fmt) - print(json.dumps( - [{"url": r.url, "source_method": r.source_method, - output_key(r.format): r.markdown} - for r in results], - indent=2)) + results = scrape(rest, fmt=fmt, state=state, appendix=appendix, query=query) + print(json.dumps([result_dict(r, state, appendix, query) for r in results], + indent=2)) return 0 if results else 1 diff --git a/switchback/highlights.py b/switchback/highlights.py new file mode 100644 index 0000000..2eae128 --- /dev/null +++ b/switchback/highlights.py @@ -0,0 +1,311 @@ +"""Query highlights — the passages of a scraped page that answer a question. + + highlights(markdown, query, url) -> [{"text", "score", "grader", "link"}, ...] + +The page markdown is cut into passages (a heading travels with the paragraphs +under it) and graded against the query: + + bm25 default. Local keyword scoring, no network, no cost. Returns the top + passages that share terms with the query (score = BM25, unbounded). + jev opt-in: SCRAPER_HIGHLIGHTS_GRADER=jev + TYPESAFE_API_KEY. TypeSafe's Jev + model answers "does this passage answer the query?" per passage and + returns a calibrated probability (score = 0..1). Passages below + SCRAPER_HIGHLIGHTS_MIN_SCORE (0.5) are dropped, so an off-topic page + honestly returns []. Sends the passages + query to api.typesafe.ai. + Any Jev failure falls back to bm25 for that call; the scrape never fails. + +Every highlight carries a `link`: the page URL with a text fragment +(`#:~:text=start,end`) that scrolls a browser to, and highlights, that passage — +or, for a YouTube transcript, the video at that passage's timestamp. +""" +from __future__ import annotations + +import logging +import math +import os +import re +from collections import Counter +from urllib.parse import quote + +logger = logging.getLogger(__name__) + +TOP_K = int(os.getenv("SCRAPER_HIGHLIGHTS_TOP_K", "5")) +MIN_SCORE = float(os.getenv("SCRAPER_HIGHLIGHTS_MIN_SCORE", "0.5")) # jev only + +# Passage sizing, in words: grow a passage until TARGET, split anything past MAX. +_TARGET_WORDS = 80 +_MAX_WORDS = 220 + +# ── Passages ──────────────────────────────────────────────────────────────── + + +def _words(s: str) -> int: + return len(s.split()) + + +def _pack(units: list[str], sep: str, head: list[str] = ()) -> list[str]: + """Greedily join units into pieces of ≤_MAX_WORDS words, each prefixed by head.""" + out, cur = [], list(head) + for u in units: + if len(cur) > len(head) and _words(sep.join(cur + [u])) > _MAX_WORDS: + out.append(sep.join(cur)) + cur = list(head) + cur.append(u) + if len(cur) > len(head): + out.append(sep.join(cur)) + return out + + +def _split_long(block: str) -> list[str]: + """Split an oversized block into ≤_MAX_WORDS pieces: tables and lists by line + (a table's header rows repeat on every piece so each stays readable), prose by + sentence, and anything still too long by words.""" + lines = block.splitlines() + if len(lines) > 1: + # Header = anything above the table's |---| separator (title + header row), + # when that separator is near the top. + sep = next((i for i, l in enumerate(lines[:4]) + if i and re.fullmatch(r"\s*\|?[\s:|-]*-[\s:|-]*\|?\s*", l) + and lines[i - 1].lstrip().startswith("|")), None) + head = lines[:sep + 1] if sep is not None else [] + pieces = _pack(lines[len(head):], "\n", head) + else: + pieces = _pack(re.split(r"(?<=[.!?])\s+", block), " ") + out = [] + for p in pieces: + out.extend(_pack(p.split(" "), " ") if _words(p) > _MAX_WORDS * 1.5 else [p]) + return out + + +def passages(markdown: str) -> list[str]: + """Cut markdown into passages of roughly _TARGET_WORDS words. A heading starts + a new passage and stays attached to the text under it.""" + out: list[str] = [] + buf: list[str] = [] + + def flush(): + text = "\n\n".join(buf).strip() + if text and not re.fullmatch(r"(#+\s*[^\n]*)", text): # skip heading-only + out.append(text) + buf.clear() + + for block in re.split(r"\n\s*\n", markdown or ""): + block = block.strip() + if not block: + continue + if block.startswith("#"): + flush() + for piece in (_split_long(block) if _words(block) > _MAX_WORDS else [block]): + # Close the passage before it would overflow, unless it's just a heading. + if (buf and _words("\n\n".join(buf + [piece])) > _MAX_WORDS + and not (len(buf) == 1 and buf[0].startswith("#"))): + flush() + buf.append(piece) + if _words("\n\n".join(buf)) >= _TARGET_WORDS: + flush() + flush() + return out + + +# ── BM25 (local, free) ────────────────────────────────────────────────────── + +_STOP = frozenset("""a an and are as at be but by for from has have how i if in into +is it its of on or that the their there these this to was were what when where +which who why will with you your do does did can""".split()) + + +def _stem(t: str) -> str: + """Light suffix stripping so "countries"/"country", "listed"/"list" match.""" + for suf, rep in (("ies", "y"), ("sses", "ss"), ("ing", ""), ("ed", ""), ("es", ""), ("s", "")): + if t.endswith(suf) and len(t) - len(suf) >= 3: + return t[: len(t) - len(suf)] + rep + return t + + +def _tokens(s: str) -> list[str]: + return [_stem(t) for t in re.findall(r"[a-z0-9]+", s.lower()) if t not in _STOP] + + +def bm25_scores(chunks: list[str], query: str, k1: float = 1.5, b: float = 0.75) -> list[float]: + """BM25 score of each chunk for the query (0 when no query term appears).""" + docs = [_tokens(c) for c in chunks] + q = set(_tokens(query)) + if not docs or not q: + return [0.0] * len(chunks) + n = len(docs) + avgdl = sum(len(d) for d in docs) / n or 1.0 + df = Counter(t for d in docs for t in set(d) if t in q) + idf = {t: math.log(1 + (n - df[t] + 0.5) / (df[t] + 0.5)) for t in q} + scores = [] + for d in docs: + tf = Counter(d) + s = 0.0 + for t in q: + if tf[t]: + s += idf[t] * tf[t] * (k1 + 1) / (tf[t] + k1 * (1 - b + b * len(d) / avgdl)) + scores.append(s) + return scores + + +# ── Jev (opt-in, TypeSafe API) ────────────────────────────────────────────── + +_JEV_URL = os.getenv("TYPESAFE_API_URL", "https://api.typesafe.ai/v1/systemone") +_JEV_MODEL = os.getenv("SCRAPER_JEV_MODEL", "jev-latest") +_JEV_TIMEOUT_S = float(os.getenv("SCRAPER_JEV_TIMEOUT_S", "20")) +# Request sizing. Jev allows 64k tokens per request and 32k for the state plus +# the longest question; ~4 chars/token keeps a batch well inside both. +_JEV_BATCH_CHARS = 80_000 +_JEV_BATCH_N = 60 +_JEV_MAX_PASSAGES = 200 # larger pages are pre-ranked by bm25 down to this + +_QUESTION_CRITERIA = { + "true": "The passage states information that answers the question, or a specific part of it.", + "false": "The passage is off-topic, or only mentions the topic without answering the question.", +} + + +def _jev_batches(chunks: list[str]) -> list[list[int]]: + batches, cur, size = [], [], 0 + for i, c in enumerate(chunks): + if cur and (size + len(c) > _JEV_BATCH_CHARS or len(cur) >= _JEV_BATCH_N): + batches.append(cur) + cur, size = [], 0 + cur.append(i) + size += len(c) + if cur: + batches.append(cur) + return batches + + +def jev_scores(chunks: list[str], query: str, api_key: str) -> list[float]: + """Jev's probability that each chunk answers the query. One Noul question per + passage, batched so each request stays inside Jev's token budget. Raises on + any API failure (the caller falls back to bm25).""" + import requests + scores = [0.0] * len(chunks) + for batch in _jev_batches(chunks): + state = {"query": query, "passages": [chunks[i] for i in batch]} + questions = { + f"p{j}": { + "type": "noul", + "instructions": (f"Does `passages[{j}]` contain information that answers " + f"the question in `query`?"), + "criteria": _QUESTION_CRITERIA, + } + for j in range(len(batch)) + } + r = requests.post(_JEV_URL, timeout=_JEV_TIMEOUT_S, + headers={"Authorization": f"Bearer {api_key}"}, + json={"model": _JEV_MODEL, "state": state, "questions": questions}) + r.raise_for_status() + answers = r.json()["answers"] + for j, i in enumerate(batch): + scores[i] = float(answers[f"p{j}"]["noul"]) + return scores + + +def grader() -> str: + """The configured grader: "jev" when selected and a TypeSafe key is set, else "bm25".""" + want = os.getenv("SCRAPER_HIGHLIGHTS_GRADER", "bm25").strip().lower() + if want == "jev" and os.getenv("TYPESAFE_API_KEY"): + return "jev" + if want == "jev": + logger.warning("SCRAPER_HIGHLIGHTS_GRADER=jev but TYPESAFE_API_KEY is unset; using bm25") + return "bm25" + + +# ── Citations ─────────────────────────────────────────────────────────────── + +# A link target: the URL (which may hold balanced parens) plus an optional "title". +_TARGET = r"\((?:[^()\s]|\([^()]*\))*(?:\s+(?:\"[^\"]*\"|'[^']*'))?\)" +_MD_IMG = re.compile(r"!\[[^\]]*\]" + _TARGET) +_MD_LINK = re.compile(r"\[((?:\[[^\]]*\]|[^\[\]])*)\]" + _TARGET) # incl. [[3]](url) +_MD_AUTOLINK = re.compile(r"<(https?://[^>\s]+)>") +_MD_MARKS = re.compile(r"(\*\*|__|~~|`|^\s*#+\s*|^\s*[-*>]\s+|^\s*\d+\.\s+|\|)", re.M) + + +def plain_text(md: str) -> str: + """Markdown → the visible text a browser would match (links/images/marks off).""" + s = _MD_IMG.sub(" ", md) + s = _MD_LINK.sub(r"\1", s) + s = _MD_AUTOLINK.sub(r"\1", s) + s = _MD_MARKS.sub(" ", s) + s = re.sub(r"(?<!\w)(?<!\\)[*_](?=\S)|(?<=\S)(?<!\\)[*_](?!\w)", "", s) # emphasis marks + s = re.sub(r"\\([\\`*_{}\[\]()#+\-.!|<>~])", r"\1", s) # markdownify's escapes + return re.sub(r"\s+", " ", s).strip() + + +def _frag(s: str) -> str: + # Text-fragment values must also escape '-', ',' and '&' (directive syntax). + return quote(s, safe="").replace("-", "%2D") + + +# A transcript passage cites its moment in the video (tier_1's timestamp links), +# since the transcript text isn't on the video page for a text fragment to find. +_TIMESTAMP_LINK = re.compile(r"\[\d+(?::\d\d){1,2}\]\((https://www\.youtube\.com/watch\?v=[^)\s]+&t=\d+s)\)") + + +def cite_link(url: str, passage: str, query: str = "") -> str: + """The citation for a passage: for a YouTube transcript, the timestamp of its + paragraph that best matches the query; else a text-fragment link into the page.""" + if re.search(r"(?:youtube\.com|youtu\.be)/", url): + q = set(_tokens(query)) + best = None # (overlap, link); first paragraph wins ties + for para in passage.split("\n\n"): + m = _TIMESTAMP_LINK.search(para) + if m: + overlap = len(q & set(_tokens(para))) + if best is None or overlap > best[0]: + best = (overlap, m.group(1)) + if best: + return best[1] + return fragment_link(url, passage) + + +def fragment_link(url: str, passage: str, edge_words: int = 5) -> str: + """`url` + a text fragment pointing at `passage`. Uses its longest prose line + (headings and table rows are skipped when there is prose, since their visible + text rarely matches the markdown cell-for-cell): the whole line when short, + else its first and last `edge_words` words as a start,end range.""" + raw = [l for l in passage.splitlines() if l.strip()] + prose = [l for l in raw if not l.lstrip().startswith(("#", "|"))] + line = max((plain_text(l) for l in (prose or raw)), key=len, default="") + words = line.split() + if not words: + return url + if len(words) <= edge_words * 2: + directive = _frag(" ".join(words)) + else: + directive = f"{_frag(' '.join(words[:edge_words]))},{_frag(' '.join(words[-edge_words:]))}" + base, sep, frag = url.partition("#") + return f"{base}#{frag}:~:text={directive}" if sep and frag else f"{base}#:~:text={directive}" + + +# ── Entry point ───────────────────────────────────────────────────────────── + + +def highlights(markdown: str, query: str, url: str) -> list[dict]: + """The passages of `markdown` that best answer `query`, best first, each with + its score, the grader that produced it, and a citation link into `url`.""" + chunks = passages(markdown) + if not chunks or not (query or "").strip(): + return [] + bm25 = bm25_scores(chunks, query) + used = grader() + if used == "jev": + # Grade every passage when the page is small enough, so Jev can find what + # keyword matching misses; on huge pages pre-rank by bm25 first. + pool = sorted(range(len(chunks)), key=lambda i: bm25[i], reverse=True)[:_JEV_MAX_PASSAGES] + pool.sort() + try: + s = jev_scores([chunks[i] for i in pool], query, os.environ["TYPESAFE_API_KEY"]) + ranked = [(s[k], i) for k, i in enumerate(pool) if s[k] >= MIN_SCORE] + except Exception as e: + logger.warning(f"jev grading failed, using bm25 for this call: {e}") + used = "bm25" + if used == "bm25": + ranked = [(sc, i) for i, sc in enumerate(bm25) if sc > 0] + ranked.sort(key=lambda t: t[0], reverse=True) + return [{"text": chunks[i], "score": round(sc, 4), "grader": used, + "link": cite_link(url, chunks[i], query)} + for sc, i in ranked[:TOP_K]] diff --git a/switchback/mcp_server.py b/switchback/mcp_server.py new file mode 100644 index 0000000..5129887 --- /dev/null +++ b/switchback/mcp_server.py @@ -0,0 +1,81 @@ +"""MCP server — the scrape cascade and search as tools for AI agents. + + pip install "switchback[mcp]" + switchback-mcp # stdio transport + claude mcp add switchback -- switchback-mcp # e.g. register with Claude Code + +Tools: + scrape(url, format, query, state, appendix) one URL through the cascade + search(query) query → URLs (local SearXNG) + +Scrapes run one at a time on a worker thread: the browser tiers use Playwright's +sync API, which refuses to run on a thread with an asyncio loop, and serialising +keeps the per-host policy DB writes from racing (same as the CLI). +""" +from __future__ import annotations + +import asyncio + +try: # mcp 2.x renamed FastMCP to MCPServer + from mcp.server.mcpserver import MCPServer +except ImportError: # mcp 1.x + from mcp.server.fastmcp import FastMCP as MCPServer + +from .api import result_dict, scrape_detailed +from .search import search as _search + +server = MCPServer( + "switchback", + instructions=("Fetch web pages as clean markdown. Use `scrape` for a known URL " + "(pass `query` to get only the passages that answer a question, " + "with citation links), and `search` to find URLs first."), +) +_lock = asyncio.Lock() + + +@server.tool() +async def scrape(url: str, format: str = "markdown", query: str | None = None, + state: bool = False, appendix: bool = False) -> dict: + """Fetch one URL as clean content, cheapest method first (HTTP, then stealth + browsers, then a paid API if configured). + + Args: + url: the page or PDF to fetch (office files and YouTube videos too, when + the server runs with SCRAPER_DOCUMENTS=1 / SCRAPER_YOUTUBE=1). + format: markdown (default) | markdown_trimmed | html | html_selectors. + query: optional question; adds `highlights`, the passages that answer it, + each with a score and a link that jumps to that text on the page. + state: add `state`, the JSON the page embeds for its own frontend + (Next.js data, JSON-LD, window globals) — often cleaner than the text. + appendix: add `appendix`, the page's menus/header/footer/sidebars as + separate markdown (they are stripped from the main content). + + Returns {"ok": true, "url", "source_method", "markdown"|"html", ...extras} + or {"ok": false, "url", "error", "status_code"} when every method failed.""" + async with _lock: + outcome = (await asyncio.to_thread( + scrape_detailed, [url], format, state=state, appendix=appendix, query=query))[0] + if not outcome.ok: + return {"ok": False, "url": url, + "error": outcome.error_class or outcome.final_outcome, + "status_code": outcome.status_code} + return {"ok": True, **result_dict(outcome, state, appendix, query)} + + +@server.tool() +async def search(query: str) -> list[dict]: + """Web search through the local SearXNG instance (SEARXNG_URL). Returns + [{"title", "url", "snippet"}]; feed a URL into `scrape` to read it.""" + hits = await asyncio.to_thread(_search, query) + return [{"title": h.title, "url": h.url, "snippet": h.snippet} for h in hits] + + +def main() -> None: + import logging + # stdout is the MCP channel: logs must go to stderr (logging's default). + logging.basicConfig(level=logging.WARNING, format="%(levelname)s %(message)s") + server.run() + + +if __name__ == "__main__": + main() diff --git a/switchback/normalize.py b/switchback/normalize.py index 240cbb9..8d20f79 100644 --- a/switchback/normalize.py +++ b/switchback/normalize.py @@ -1,4 +1,4 @@ -"""Shared content normalization — HTML→Markdown and PDF→text. +"""Shared content normalization — HTML→Markdown, PDF→text, office docs→Markdown. Ported from musings-by-hermes/scripts/muse_helpers.py (the most mature version): strips boilerplate, promotes lazy-loaded images, resolves relative URLs. @@ -64,6 +64,32 @@ def active_format() -> str: return getattr(_scope, "fmt", None) or OUTPUT_FORMAT +# Opt-in raw-HTML capture. The extras that need the page itself (state JSON, the +# appendix) set a box here around a tier's fetch; render() drops the HTML it was +# given into the box, so no tier has to change what it returns. The last render +# wins, which is the one that produced the tier's content (a browser tier's +# interstitial check renders first, then the real page). Thread-local like the +# format scope; tiers that render on a worker thread re-apply it with capture_into. +_capture = threading.local() + + +@contextmanager +def capture_into(box: dict | None): + """Record the HTML render() sees into `box` for the enclosed work (per-thread). + None disables capture. Always restored on exit.""" + prev = getattr(_capture, "box", None) + _capture.box = box + try: + yield box + finally: + _capture.box = prev + + +def current_capture() -> dict | None: + """The active capture box on this thread, for a tier to hand to its worker.""" + return getattr(_capture, "box", None) + + def output_key(fmt: str) -> str: """The JSON/result key for a format's content family: html-family → "html", markdown-family → "markdown". Lets the default path stay {"...","markdown"}.""" @@ -149,6 +175,9 @@ def render(html: str, base_url: str | None = None, fmt: str | None = None) -> st - html the raw HTML, untouched - html_selectors cleaned HTML (per-domain prefs applied), not converted """ + box = current_capture() + if box is not None: + box["html"], box["base_url"] = html, base_url fmt = _validate(fmt) if fmt is not None else active_format() if fmt == "html": return html or "" @@ -172,6 +201,40 @@ def html_to_markdown(html: str, base_url: str | None = None) -> str: return render(html, base_url) +# Page chrome that _clean_html strips from the main content. The appendix keeps it +# instead, one entry per outermost region, so an agent can ask for the menus, +# footer or "related" links on demand without them padding the main markdown. +_APPENDIX_TAGS = ["nav", "header", "footer", "aside"] + + +def appendix(html: str, base_url: str | None = None) -> list[dict]: + """The secondary regions of a page as [{"region": tag, "markdown": md}], in + page order. Nested regions stay inside their outermost parent. Empty regions + are left out; on any failure returns [].""" + try: + from bs4 import BeautifulSoup + from markdownify import markdownify + from urllib.parse import urljoin + + soup = BeautifulSoup(html or "", "html.parser") + for tag in soup(["script", "style", "noscript"]): + tag.decompose() + regions = [] + for tag in soup.find_all(_APPENDIX_TAGS): + if tag.find_parent(_APPENDIX_TAGS): + continue + if base_url: + for a in tag.find_all("a", href=True): + a["href"] = urljoin(base_url, a["href"]) + md = (markdownify(str(tag), heading_style="ATX", bullets="-") or "").strip() + if md: + regions.append({"region": tag.name, "markdown": re.sub(r"\n{3,}", "\n\n", md)}) + return regions + except Exception as e: + logger.warning(f"appendix extraction failed: {e}") + return [] + + def pdf_bytes_to_text(data: bytes) -> str: """Extract text from PDF bytes. In-memory only — nothing written to disk.""" from pypdf import PdfReader @@ -181,3 +244,37 @@ def pdf_bytes_to_text(data: bytes) -> str: return "\n\n".join((p.extract_text() or "") for p in reader.pages).strip() finally: buf.close() + + +# Office / ebook files converted by markitdown (optional: `pip install +# "switchback[documents]"`). Matched by Content-Type, else by the URL's extension. +# PDFs are not listed: they keep their own pypdf path above. +DOC_TYPES = { + "application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx", + "application/vnd.openxmlformats-officedocument.presentationml.presentation": ".pptx", + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx", + "application/vnd.ms-excel": ".xls", + "application/epub+zip": ".epub", +} + + +def doc_extension(content_type: str, url: str) -> str | None: + """The office/ebook extension this response is, or None when it isn't one.""" + ctype = (content_type or "").split(";")[0].strip().lower() + ext = DOC_TYPES.get(ctype) + if ext or "html" in ctype: # a page served as HTML is HTML, whatever its path + return ext + path = (url or "").lower().split("?")[0].split("#")[0] + return next((e for e in DOC_TYPES.values() if path.endswith(e)), None) + + +def doc_bytes_to_markdown(data: bytes, extension: str) -> str: + """Convert an office/ebook file to markdown with markitdown. In-memory only. + Raises ImportError when markitdown isn't installed.""" + from markitdown import MarkItDown, StreamInfo + buf = io.BytesIO(data) + try: + res = MarkItDown().convert_stream(buf, stream_info=StreamInfo(extension=extension)) + return (res.text_content or "").strip() + finally: + buf.close() diff --git a/switchback/orchestrator.py b/switchback/orchestrator.py index 7ce9e94..a190fbf 100644 --- a/switchback/orchestrator.py +++ b/switchback/orchestrator.py @@ -14,14 +14,15 @@ import os import random import time -from dataclasses import dataclass, field +from contextlib import nullcontext +from dataclasses import dataclass, field, fields from . import content_cache, egress, session_cache -from .normalize import active_format, output_format_scope +from .normalize import active_format, appendix, capture_into, output_format_scope from .policy import botwall from .policy.gates import (BotWall, RateLimited, ShortContent, Unavailable, classify_error, host_of) -from .tiers import TIERS, INDEX +from .tiers import TIERS, INDEX, tier_1 from .tracing import Attr, flush, span logger = logging.getLogger(__name__) @@ -117,6 +118,16 @@ class ScrapeResult: format: str = "markdown" # markdown | markdown_trimmed | html | html_selectors +@dataclass +class ScrapeResultWithExtras(ScrapeResult): + """A ScrapeResult plus the opt-in extras. Returned only when an extra was + requested, so a plain scrape() keeps exactly the ScrapeResult shape (fields, + asdict(), astuple()). Each extra is None unless it was asked for.""" + state: dict | None = None # embedded page JSON (switchback.state) + appendix: list[dict] | None = None # nav/header/footer/aside as markdown + highlights: list[dict] | None = None # passages answering the query, with links + + @dataclass class TierAttempt: """One tier's attempt on a URL — what it was and why it ended.""" @@ -144,6 +155,63 @@ class ScrapeOutcome: attempts: list[TierAttempt] = field(default_factory=list) +@dataclass +class ScrapeOutcomeWithExtras(ScrapeOutcome): + """A ScrapeOutcome plus the opt-in extras, as on ScrapeResultWithExtras. + Every outcome of a run with extras has this type (failures too, extras None).""" + state: dict | None = None + appendix: list[dict] | None = None + highlights: list[dict] | None = None + + +def _with_extras(o: ScrapeOutcome) -> ScrapeOutcomeWithExtras: + if isinstance(o, ScrapeOutcomeWithExtras): + return o + return ScrapeOutcomeWithExtras(**{f.name: getattr(o, f.name) for f in fields(ScrapeOutcome)}) + + +@dataclass(frozen=True) +class Extras: + """Opt-in additions to a successful result. All off by default; a run without + Extras takes exactly the pre-extras code path. + + state embedded page JSON (needs the fetched HTML) + appendix the page chrome stripped from the main content (needs the HTML) + query rank the content's passages against this question (highlights)""" + state: bool = False + appendix: bool = False + query: str | None = None + + @property + def needs_html(self) -> bool: + return self.state or self.appendix + + +def _add_page_extras(res: ScrapeOutcomeWithExtras, cap: dict, extras: Extras) -> None: + """Fill state/appendix from the HTML the winning tier rendered. Tiers that + never had HTML (API mirrors, PDFs, office docs, Firecrawl markdown) give + empty values. Never raises: an extra must not turn a success into a failure.""" + html, base_url = cap.get("html"), cap.get("base_url") + if extras.state: + from .state import extract_state + res.state = extract_state(html) if html else {} + if extras.appendix: + res.appendix = appendix(html, base_url) if html else [] + + +def _add_highlights(res: ScrapeOutcomeWithExtras, query: str) -> None: + try: + from .highlights import highlights + from .normalize import render + # html formats carry HTML in `markdown`: grade (and cite) its markdown form. + text = (render(res.markdown, res.url, fmt="markdown") + if res.format.startswith("html") else res.markdown) + res.highlights = highlights(text, query, res.url) + except Exception as e: # never fail a scrape over its highlights + logger.warning(f"highlights failed for {res.url}: {e}") + res.highlights = [] + + def _dominant_failure(attempts: list[TierAttempt]) -> tuple[str, int | None]: """The failure that best explains 'why this URL failed': the highest-priority real failing attempt (ties resolve to the later, more capable tier).""" @@ -223,7 +291,7 @@ def _skipped(url, root, outcome, reason) -> ScrapeOutcome: return ScrapeOutcome(url, False, final_outcome=outcome, error_class=outcome) -def _run_one(url: str, db: dict) -> ScrapeOutcome: +def _run_one(url: str, db: dict, extras: Extras | None = None) -> ScrapeOutcome: host = host_of(url) t0 = time.monotonic() deadline = t0 + _DEADLINE_S @@ -235,7 +303,10 @@ def _run_one(url: str, db: dict) -> ScrapeOutcome: if botwall.is_url_skipped(url, db): return _skipped(url, root, "url_excluded", db.get("urls", {}).get(url, {}).get("reason", "")) - hit = content_cache.get(url, active_format()) + # The cache keeps content only, so a run that needs the page HTML (state / + # appendix) skips the read and fetches; it still refreshes the cache below. + hit = (None if extras and extras.needs_html + else content_cache.get(url, active_format())) if hit: md, method = hit root.set(Attr.OUTCOME, "cache_hit") @@ -248,7 +319,7 @@ def _run_one(url: str, db: dict) -> ScrapeOutcome: # tiers route through SCRAPER_EGRESS_PROXY (when set); easy hosts stay # direct and never spend residential bandwidth. with egress.egress_scope(botwall.needs_egress(host, db)): - res = _run_cascade(url, host, db, root, t0, deadline) + res = _run_cascade(url, host, db, root, t0, deadline, extras) # Dead logged-in session? For an authed host with a login hook wired, # refresh the cookies once and re-run on a fresh budget — the # refreshed cookies overlay every tier (and persist for later runs). @@ -258,7 +329,7 @@ def _run_one(url: str, db: dict) -> ScrapeOutcome: and session_cache.refresh_login(url)): logger.info(f"re-running after login refresh: {url}") rt = time.monotonic() - res = _run_cascade(url, host, db, root, rt, rt + _DEADLINE_S) + res = _run_cascade(url, host, db, root, rt, rt + _DEADLINE_S, extras) # Record which egress this URL's cascade ran on, while still in scope — # "egress" means its bytes were metered residential-proxy bandwidth. res.egress = egress.scope_label() @@ -277,12 +348,17 @@ def _enabled_paid_ahead(i: int) -> bool: return False -def _run_cascade(url, host, db, root, t0, deadline) -> ScrapeOutcome: +def _run_cascade(url, host, db, root, t0, deadline, extras=None) -> ScrapeOutcome: attempts: list[TierAttempt] = [] + want_html = bool(extras and extras.needs_html) transient = 0 # count of rate_limited/timeout misses so far (drives backoff) start = _start_index(url, db) for i, tier in enumerate(TIERS): - if i < start: + # A host's learned start skips tier_1, except for an opted-in YouTube URL: + # youtube.com may have been won by a browser tier before transcripts were + # on. If tier_1 has no transcript, the next tiers are still skipped up to start. + if i < start and not (tier.NAME == "tier_1" and tier_1.youtube_enabled() + and tier_1.YOUTUBE_RE.search(url)): continue # Env/feature gate (e.g. paid Firecrawl off, residential not wired). @@ -352,8 +428,10 @@ def _run_cascade(url, host, db, root, t0, deadline) -> ScrapeOutcome: botwall.record(db, url, tier.NAME, "firecrawl_used") ts = time.monotonic() outcome = exc = status = challenge = unavailable_exc = None + cap: dict | None = {} if want_html else None try: - md = tier.fetch(url) + with capture_into(cap) if want_html else nullcontext(): + md = tier.fetch(url) except BotWall as e: outcome, exc, challenge = "botwall", e, getattr(e, "vendor", None) except ShortContent as e: @@ -419,9 +497,15 @@ def _run_cascade(url, host, db, root, t0, deadline) -> ScrapeOutcome: attempts.append(TierAttempt(tier.NAME, "ok", latency_ms=dt)) logger.info( f"{tier.NAME} OK {url} md_len={len(md)} {dt}ms (total {total}ms)") - return ScrapeOutcome(url, True, markdown=md, source_method=tier.NAME, - final_outcome="ok", latency_ms=total, - format=active_format(), attempts=attempts) + if not want_html: + return ScrapeOutcome(url, True, markdown=md, source_method=tier.NAME, + final_outcome="ok", latency_ms=total, + format=active_format(), attempts=attempts) + res = ScrapeOutcomeWithExtras( + url, True, markdown=md, source_method=tier.NAME, final_outcome="ok", + latency_ms=total, format=active_format(), attempts=attempts) + _add_page_extras(res, cap, extras) + return res total = int((time.monotonic() - t0) * 1000) ec, sc = _dominant_failure(attempts) @@ -436,24 +520,36 @@ def _run_cascade(url, host, db, root, t0, deadline) -> ScrapeOutcome: latency_ms=total, attempts=attempts) -def run_detailed(urls: list[str], fmt: str | None = None) -> list[ScrapeOutcome]: +def run_detailed(urls: list[str], fmt: str | None = None, + extras: Extras | None = None) -> list[ScrapeOutcome]: """Scrape each URL; return a full ScrapeOutcome (success or failure with the per-tier cascade and a classified reason) for every URL. - fmt overrides SCRAPER_OUTPUT_FORMAT for this call (None = use the default).""" + fmt overrides SCRAPER_OUTPUT_FORMAT for this call (None = use the default). + extras opts into state / appendix / highlights on successful results.""" db = botwall.load_db() out = [] try: with output_format_scope(fmt): for url in urls: - out.append(_run_one(url, db)) + res = _run_one(url, db, extras) + if extras: + res = _with_extras(res) + if res.ok and extras.query: + _add_highlights(res, extras.query) + out.append(res) finally: botwall.save_db(db) flush() return out -def run(urls: list[str], fmt: str | None = None) -> list[ScrapeResult]: +def run(urls: list[str], fmt: str | None = None, + extras: Extras | None = None) -> list[ScrapeResult]: """Successes only (backward-compatible). Use run_detailed() for failures.""" - return [ScrapeResult(o.url, o.markdown, o.source_method, o.format) - for o in run_detailed(urls, fmt) if o.ok] + if not extras: + return [ScrapeResult(o.url, o.markdown, o.source_method, o.format) + for o in run_detailed(urls, fmt) if o.ok] + return [ScrapeResultWithExtras(o.url, o.markdown, o.source_method, o.format, + o.state, o.appendix, o.highlights) + for o in run_detailed(urls, fmt, extras) if o.ok] diff --git a/switchback/server.py b/switchback/server.py index c8e7f80..3e0cb51 100644 --- a/switchback/server.py +++ b/switchback/server.py @@ -24,8 +24,7 @@ from pydantic import BaseModel from . import session_trace -from .api import scrape -from .normalize import output_key +from .api import result_dict, scrape from .reporting import build_report, domain_report from .search import search from .tracing import setup_logs @@ -36,6 +35,9 @@ class ScrapeRequest(BaseModel): urls: list[str] format: str | None = None # markdown (default) | markdown_trimmed | html | html_selectors + state: bool = False # add the page's embedded JSON + appendix: bool = False # add nav/header/footer/aside as separate markdown + query: str | None = None # add highlights answering this question @app.get("/healthz") @@ -47,10 +49,11 @@ def healthz() -> dict: def scrape_endpoint(req: ScrapeRequest) -> list[dict]: """Run URLs through the cascade. Returns successes only (failed URLs omitted). Optional "format" selects the output shape; the content key is "markdown" for - markdown formats and "html" for html formats.""" - return [{"url": r.url, "source_method": r.source_method, - output_key(r.format): r.markdown} - for r in scrape(req.urls, fmt=req.format)] + markdown formats and "html" for html formats. Optional "state", "appendix" and + "query" add those keys to each result; without them the shape is unchanged.""" + results = scrape(req.urls, fmt=req.format, state=req.state, + appendix=req.appendix, query=req.query) + return [result_dict(r, req.state, req.appendix, req.query) for r in results] @app.get("/search") diff --git a/switchback/state.py b/switchback/state.py new file mode 100644 index 0000000..88e6c16 --- /dev/null +++ b/switchback/state.py @@ -0,0 +1,115 @@ +"""State JSON — the structured data a page ships for its own frontend. + +Many sites render from a JSON blob embedded in the HTML: Next.js `__NEXT_DATA__`, +Nuxt, Apollo / Redux initial state, schema.org JSON-LD. It is often cleaner than +the rendered text (prices, ids, specs, dates) and it is already in the HTML we +fetched, so pulling it out costs nothing extra. + + extract_state(html) -> {"json_ld": [...], "__NEXT_DATA__": {...}, ...} + +Only valid JSON is returned; a blob that is a JavaScript literal rather than JSON +(e.g. contains `undefined`) is skipped, never guessed at. {} when nothing is found. + +Not covered yet: the Next.js App Router flight payload (`self.__next_f.push`), +which is a streamed RSC format rather than JSON. +""" +from __future__ import annotations + +import json +import logging +import re + +logger = logging.getLogger(__name__) + +# `window.__INITIAL_STATE__ = {...}`, `self.__APOLLO_STATE__=`, `window["__X__"] =`, +# `window.__remixContext =` — a global assigned a JSON object/array literal, or +# JSON.parse("...") of an escaped string. +_GLOBAL_RE = re.compile( + r"""(?:window|self|globalThis)\s*(?:\.\s*(__[A-Za-z0-9_]+)|\[\s*["'](__[A-Za-z0-9_]+)["']\s*\])\s*=\s*""") +_JSON_PARSE_RE = re.compile(r"JSON\.parse\(\s*") + +_SCRIPT_JS_TYPES = ("", "text/javascript", "application/javascript", "module") + +# Bundler bookkeeping, not page data (e.g. BBC's __LOADABLE_REQUIRED_CHUNKS__ is a +# list of webpack chunk ids). +_NOISE_RE = re.compile(r"LOADABLE|CHUNK|WEBPACK|BUILD_MANIFEST|SSG_MANIFEST", re.I) + + +def _loads(text: str): + try: + return json.loads(text) + except (ValueError, TypeError): + return None + + +def _globals_in(js: str) -> dict: + """JSON values assigned to `__name` globals in one script body.""" + found: dict = {} + dec = json.JSONDecoder() + for m in _GLOBAL_RE.finditer(js): + name = m.group(1) or m.group(2) + if _NOISE_RE.search(name): + continue + rest = js[m.end():] + try: + pm = _JSON_PARSE_RE.match(rest) + if pm: + # JSON.parse("…escaped json…") / JSON.parse('…'): decode the string + # literal first. Only double-quoted literals are valid JSON strings. + lit, _ = dec.raw_decode(rest[pm.end():]) + value = json.loads(lit) if isinstance(lit, str) else None + elif rest[:1] in ("{", "["): + value, _ = dec.raw_decode(rest) + else: + continue + except ValueError: + continue + if value is not None and name not in found: + found[name] = value + return found + + +def extract_state(html: str) -> dict: + """Embedded JSON state from a page's <script> tags. + + - JSON-LD blocks -> "json_ld": [block, ...] (page order) + - <script type="application/json" id="X"> (e.g. __NEXT_DATA__, __NUXT_DATA__) + -> "X" (or its first data-* value; else application_json_N) + - `window.__X = {...}` / `JSON.parse("...")` globals -> "__X" + + On any failure returns what was collected so far; never raises.""" + out: dict = {} + try: + from bs4 import BeautifulSoup + soup = BeautifulSoup(html or "", "html.parser") + json_ld: list = [] + n_unnamed = 0 + for s in soup.find_all("script"): + typ = (s.get("type") or "").split(";")[0].strip().lower() + text = s.string if s.string is not None else s.get_text() + if not text or not text.strip(): + continue + if typ == "application/ld+json": + value = _loads(text) + if value is not None: + json_ld.append(value) + elif typ == "application/json": + value = _loads(text) + if value is None: + continue + key = s.get("id") or next( + (v for k, v in s.attrs.items() if k.startswith("data-") and isinstance(v, str) and v), + None) + if not key: + n_unnamed += 1 + key = f"application_json_{n_unnamed}" + if not _NOISE_RE.search(key): + out.setdefault(key, value) + elif typ in _SCRIPT_JS_TYPES and "__" in text: + for name, value in _globals_in(text).items(): + out.setdefault(name, value) + if json_ld: + out["json_ld"] = json_ld + except Exception as e: + logger.warning(f"state extraction failed: {e}") + return out diff --git a/switchback/tiers/tier_1.py b/switchback/tiers/tier_1.py index d60308d..0f6ee65 100644 --- a/switchback/tiers/tier_1.py +++ b/switchback/tiers/tier_1.py @@ -4,20 +4,28 @@ matches (caller falls through to Tier 1). Bypasses the botwall skip-list because these are open, stable endpoints — not the scraped page. +YouTube videos return their transcript when SCRAPER_YOUTUBE=1 and +youtube-transcript-api is installed (`switchback[youtube]`); otherwise, or when a +video has no captions, the URL falls through to the HTTP tiers exactly as before. + Extend here with career-ops-style structured providers (greenhouse/lever/ashby). Web *search* (query → URLs) is a different shape and lives in switchback/search.py (local SearXNG), not in this fetch cascade. """ from __future__ import annotations +import logging import os import re from urllib.parse import unquote from xml.etree import ElementTree as ET +from ..egress import has_egress_proxy, requests_proxies from ..normalize import html_to_markdown, UA from ..policy.gates import check +logger = logging.getLogger(__name__) + NAME = "tier_1" PAID = False @@ -27,6 +35,14 @@ ARXIV_RE = re.compile(r"arxiv\.org/(?:abs|pdf)/(\d{4}\.\d{4,5})(?:v\d+)?(?:\.pdf)?", re.I) WIKI_RE = re.compile(r"en\.wikipedia\.org/wiki/([^?#]+)", re.I) PMC_RE = re.compile(r"pmc\.ncbi\.nlm\.nih\.gov/articles/(PMC\d+)", re.I) +YOUTUBE_RE = re.compile( + r"(?:(?:www\.|m\.)?youtube\.com/(?:watch\?(?:[^#]*&)?v=|shorts/|embed/|live/)" + r"|youtu\.be/)([A-Za-z0-9_-]{11})", re.I) + +# Transcript languages to try, in order (comma-separated codes). +_YT_LANGS = [c.strip() for c in os.getenv("SCRAPER_YOUTUBE_LANGS", "en").split(",") if c.strip()] +# Group transcript lines into one timestamped paragraph per this many seconds. +_YT_PARAGRAPH_S = 30 def fetch(url: str) -> str | None: @@ -38,6 +54,9 @@ def fetch(url: str) -> str | None: return _wikipedia(m.group(1), url) if PMC_RE.search(url): return _europepmc(url) + m = YOUTUBE_RE.search(url) + if m and youtube_enabled(): + return _youtube(m.group(1), url) return None # no mirror — fall through @@ -79,3 +98,96 @@ def _europepmc(url: str) -> str: if len(r.text) < 1000: raise RuntimeError(f"europepmc empty: {len(r.text)}") return check(url, html_to_markdown(r.text, base_url=url)) + + +def youtube_enabled() -> bool: + """Opt-in (SCRAPER_YOUTUBE=1): transcripts change what YouTube URLs return.""" + return os.getenv("SCRAPER_YOUTUBE") in ("1", "true", "True") + + +def _youtube(video_id: str, url: str) -> str | None: + """Title + timestamped transcript for a YouTube video. Returns None (not + applicable — the cascade carries on as it did before) when the library is + missing or the video has no usable transcript, rather than recording a miss + against youtube.com.""" + try: + import youtube_transcript_api # noqa: F401 + except ImportError: + return None + proxies = requests_proxies() + try: + snippets = _yt_snippets(video_id, proxies["https"] if proxies else None) + except Exception as e: # no captions, wrong language, blocked IP, network + # YouTube blocks most cloud/datacenter IPs. Like the rest of the cascade, + # retry through the residential egress proxy only for this blocked case. + blocked = type(e).__name__ in ("RequestBlocked", "IpBlocked") + if blocked and not proxies and has_egress_proxy(): + logger.info(f"{NAME} youtube blocked our IP; retrying {video_id} via egress proxy") + try: + snippets = _yt_snippets(video_id, os.environ["SCRAPER_EGRESS_PROXY"]) + except Exception as e2: + logger.info(f"{NAME} youtube transcript unavailable for {video_id}: " + f"{type(e2).__name__}") + return None + else: + logger.info(f"{NAME} youtube transcript unavailable for {video_id}: " + f"{type(e).__name__}" + (" (set SCRAPER_EGRESS_PROXY)" if blocked else "")) + return None + snippets = [(t, text) for t, text in snippets if text] + if not snippets: + return None + + title, channel = _youtube_meta(url) + paras: list[str] = [] + start, buf = snippets[0][0], [] + for t, text in snippets: + if buf and t - start >= _YT_PARAGRAPH_S: + paras.append(_yt_para(video_id, start, buf)) + start, buf = t, [] + buf.append(text) + paras.append(_yt_para(video_id, start, buf)) + head = f"# {title}\n\n" if title else "" + if channel: + head += f"**Channel:** {channel}\n\n" + return f"{head}**Video:** https://www.youtube.com/watch?v={video_id}\n\n## Transcript\n\n" + "\n\n".join(paras) + + +def _yt_snippets(video_id: str, proxy_url: str | None) -> list[tuple[float, str]]: + from youtube_transcript_api import YouTubeTranscriptApi + from youtube_transcript_api.proxies import GenericProxyConfig + proxy = GenericProxyConfig(http_url=proxy_url, https_url=proxy_url) if proxy_url else None + api = YouTubeTranscriptApi(proxy_config=proxy, http_client=_TimeoutSession()) + return [(float(s.start), (s.text or "").strip()) + for s in api.fetch(video_id, languages=_YT_LANGS)] + + +def _TimeoutSession(): + """A requests.Session whose calls default to the tier timeout — the transcript + library passes none, and a hung socket here would stall the whole cascade.""" + import requests + + class _S(requests.Session): + def request(self, *a, **kw): + kw.setdefault("timeout", _TIMEOUT_S) + return super().request(*a, **kw) + return _S() + + +def _yt_para(video_id: str, start: float, texts: list[str]) -> str: + s = int(start) + stamp = f"{s // 3600}:{s % 3600 // 60:02d}:{s % 60:02d}" if s >= 3600 else f"{s // 60}:{s % 60:02d}" + body = " ".join(texts).replace("\n", " ") + return f"[{stamp}](https://www.youtube.com/watch?v={video_id}&t={s}s) {body}" + + +def _youtube_meta(url: str) -> tuple[str, str]: + """(title, channel) from YouTube's public oEmbed endpoint; blanks on failure.""" + import requests + try: + r = requests.get("https://www.youtube.com/oembed", timeout=_TIMEOUT_S, + params={"url": url, "format": "json"}, proxies=requests_proxies()) + r.raise_for_status() + d = r.json() + return (d.get("title") or "").strip(), (d.get("author_name") or "").strip() + except Exception: + return "", "" diff --git a/switchback/tiers/tier_2.py b/switchback/tiers/tier_2.py index 73eb4b8..3631345 100644 --- a/switchback/tiers/tier_2.py +++ b/switchback/tiers/tier_2.py @@ -1,7 +1,9 @@ """Tier 1 — plain HTTP with TLS fingerprint impersonation. curl_cffi impersonates a real Chrome TLS handshake, which clears many naive bot -walls without a browser. Handles PDFs inline. Fast and cheap. +walls without a browser. Handles PDFs inline, and office/ebook files when +SCRAPER_DOCUMENTS=1 and markitdown is installed (`switchback[documents]`). +Fast and cheap. The bare "chrome" alias resolves to an old default; we pin recent targets and rotate them deterministically per host, so our traffic isn't one shared JA3 yet @@ -15,14 +17,18 @@ from __future__ import annotations import hashlib +import logging import os from urllib.parse import urlsplit from .. import session_cache from ..egress import requests_proxies, add_wire_bytes -from ..normalize import html_to_markdown, pdf_bytes_to_text +from ..normalize import (doc_bytes_to_markdown, doc_extension, html_to_markdown, + pdf_bytes_to_text) from ..policy.gates import BotWall, check, is_cf_challenge +logger = logging.getLogger(__name__) + NAME = "tier_2" PAID = False @@ -66,4 +72,19 @@ def fetch(url: str) -> str: finally: r.close() return check(url, text) + # Opt-in: converting changes what these URLs return (see CONTRIBUTING.md). + doc_ext = (doc_extension(ctype, r.url) + if os.getenv("SCRAPER_DOCUMENTS") in ("1", "true", "True") else None) + if doc_ext: + # Office/ebook file. Without markitdown (or if it can't read the file) we + # fall through to the HTML path below, exactly as before this was added. + try: + text = doc_bytes_to_markdown(r.content, doc_ext) + except ImportError: + text = None + except Exception as e: + logger.info(f"{NAME} {doc_ext} conversion failed for {url}: {e}") + text = None + if text is not None: + return check(url, text) return check(url, html_to_markdown(r.text, base_url=r.url)) diff --git a/switchback/tiers/tier_3.py b/switchback/tiers/tier_3.py index 6648c85..8e11f4f 100644 --- a/switchback/tiers/tier_3.py +++ b/switchback/tiers/tier_3.py @@ -21,7 +21,7 @@ from .. import egress, session_cache from ..egress import requests_proxies -from ..normalize import html_to_markdown +from ..normalize import capture_into, current_capture, html_to_markdown from ..policy.gates import Unavailable, check logger = logging.getLogger(__name__) @@ -145,12 +145,13 @@ def fetch(url: str) -> str: # Run the (blocking, occasionally runaway) solve under a hard wall-clock cap. # A daemon worker means an abandoned solve can't block process exit; it dies on # its own socket timeout shortly after. Thread-locals don't inherit, so the - # egress scope is re-applied inside the worker. + # egress scope (and any HTML capture) is re-applied inside the worker. scoped = egress.in_egress_scope() + capture = current_capture() box: dict = {} def work(): - with egress.egress_scope(scoped): + with egress.egress_scope(scoped), capture_into(capture): try: box["md"], box["bytes"] = _fetch(url) except BaseException as e: # noqa: BLE001 — propagated to caller below diff --git a/switchback/tiers/tier_7.py b/switchback/tiers/tier_7.py index 62b3a39..42d0fa6 100644 --- a/switchback/tiers/tier_7.py +++ b/switchback/tiers/tier_7.py @@ -9,7 +9,7 @@ import os import threading -from ..normalize import active_format, render +from ..normalize import active_format, capture_into, current_capture, render from ..policy.gates import check NAME = "tier_7" @@ -48,10 +48,12 @@ def fetch(url: str) -> str: # active_format() is thread-local, so read it here (main thread) and pass it in. box: dict = {} fmt = active_format() + capture = current_capture() # thread-local too; re-applied in the worker def work(): try: - box["md"] = _scrape(url, fmt) + with capture_into(capture): + box["md"] = _scrape(url, fmt) except BaseException as e: # noqa: BLE001 — re-raised to the caller below box["err"] = e diff --git a/tests/test_extras.py b/tests/test_extras.py new file mode 100644 index 0000000..9d00974 --- /dev/null +++ b/tests/test_extras.py @@ -0,0 +1,569 @@ +"""Unit tests for the opt-in extras: state JSON, appendix, query highlights (bm25 + +Jev), citation links, office docs, YouTube transcripts, and the MCP server — plus +the guarantee that a call without extras behaves exactly as before. + +Pure + offline: fake tiers, botwall/content_cache stubbed, network mocked. +Run with: pytest tests/test_extras.py +""" +from __future__ import annotations + +import io +import json +import os +import sys +import types +from dataclasses import asdict, astuple, fields + +import pytest + +import switchback.orchestrator as orch +from switchback import api, highlights as H, normalize as N, state as S +from switchback.tiers import tier_1, tier_2 + +URL = "https://example.test/post" + +PAGE = """<html><head><title>T + + + + + + + + +
Site banner
+
+

Solar power costs

+

Solar panel prices fell by 30 percent in 2025 as factories in China expanded output +and polysilicon became cheaper. Installers passed most of the saving to households.

+

Wind

+

Offshore wind projects were delayed by higher interest rates and supply chain trouble, +and several developers cancelled contracts with state governments last year.

+

""" + ("Filler sentence about unrelated matters of local gardening. " * 40) + """

+
+ + +""" + + +# ── harness ───────────────────────────────────────────────────────────────── + +def _tier(name, fetch): + return types.SimpleNamespace(NAME=name, PAID=False, fetch=fetch) + + +def _html_tier(url): + return N.html_to_markdown(PAGE, base_url=url) + + +@pytest.fixture(autouse=True) +def _isolate(monkeypatch): + # The CLI entry point loads the repo's .env into os.environ; undo that after. + saved_env = dict(os.environ) + monkeypatch.setattr(orch.botwall, "load_db", lambda: {}) + monkeypatch.setattr(orch.botwall, "save_db", lambda db: None) + monkeypatch.setattr(orch.botwall, "is_skipped", lambda h, db: False) + monkeypatch.setattr(orch.botwall, "is_url_skipped", lambda u, db: False) + monkeypatch.setattr(orch.botwall, "needs_egress", lambda h, db: False) + monkeypatch.setattr(orch.botwall, "winning_tier", lambda h, db: None) + monkeypatch.setattr(orch.botwall, "log_final", lambda *a, **k: None) + monkeypatch.setattr(orch.botwall, "record", lambda *a, **k: None) + monkeypatch.setattr(orch.content_cache, "get", lambda u, fmt: None) + monkeypatch.setattr(orch.content_cache, "put", lambda *a, **k: None) + monkeypatch.setattr(N, "OUTPUT_FORMAT", "markdown") + monkeypatch.delenv("SCRAPER_HIGHLIGHTS_GRADER", raising=False) + monkeypatch.setattr(orch, "TIERS", [_tier("fake", _html_tier)]) + monkeypatch.setattr(orch, "INDEX", {"fake": 0}) + yield + os.environ.clear() + os.environ.update(saved_env) + + +# ── backward compatibility ────────────────────────────────────────────────── + +# The result shapes as shipped before the extras existed. A default scrape must +# keep exactly these: users asdict()/astuple() results into their own payloads. +RESULT_FIELDS = ["url", "markdown", "source_method", "format"] +OUTCOME_FIELDS = ["url", "ok", "markdown", "source_method", "final_outcome", "error_class", + "status_code", "latency_ms", "egress", "wire_bytes", "format", "attempts"] + + +def test_result_classes_keep_their_fields(): + assert [f.name for f in fields(orch.ScrapeResult)] == RESULT_FIELDS + assert [f.name for f in fields(orch.ScrapeOutcome)] == OUTCOME_FIELDS + + +def test_default_scrape_is_unchanged(): + """No extras: same markdown as a plain render, same types and payload shape.""" + [r] = api.scrape(URL) + assert r.markdown == N.html_to_markdown(PAGE, base_url=URL) + assert type(r) is orch.ScrapeResult and list(asdict(r)) == RESULT_FIELDS + [o] = api.scrape_detailed(URL) + assert o.ok and type(o) is orch.ScrapeOutcome and list(asdict(o)) == OUTCOME_FIELDS + + +def _short_tier(url): + raise orch.ShortContent("too short") + + +def test_default_failure_is_unchanged(monkeypatch): + monkeypatch.setattr(orch, "TIERS", [_tier("fake", _short_tier)]) + [o] = api.scrape_detailed(URL) + assert not o.ok and type(o) is orch.ScrapeOutcome + assert api.scrape(URL) == [] + + +def test_extras_return_subclasses(monkeypatch): + """Opt-in results are subclasses: isinstance checks and old fields still hold.""" + [r] = api.scrape(URL, query="solar") + assert type(r) is orch.ScrapeResultWithExtras and isinstance(r, orch.ScrapeResult) + assert list(asdict(r)) == RESULT_FIELDS + ["state", "appendix", "highlights"] + assert r.state is None and r.appendix is None and isinstance(r.highlights, list) + monkeypatch.setattr(orch, "TIERS", [_tier("fake", _short_tier)]) + [o] = api.scrape_detailed(URL, state=True) + assert not o.ok and type(o) is orch.ScrapeOutcomeWithExtras and o.state is None + + +def test_default_path_does_not_capture(monkeypatch): + """Without state/appendix the tier runs with no capture box active.""" + seen = [] + monkeypatch.setattr(orch, "TIERS", [_tier("fake", lambda u: seen.append(N.current_capture()) + or _html_tier(u))]) + api.scrape(URL, query="solar") + assert seen == [None] + + +def test_capture_does_not_change_markdown(): + plain = N.render(PAGE, URL) + box: dict = {} + with N.capture_into(box): + captured = N.render(PAGE, URL) + assert captured == plain and box["html"] == PAGE and box["base_url"] == URL + + +def test_positional_scrape_result_still_works(): + r = orch.ScrapeResult("u", "md", "tier_2") + assert astuple(r) == ("u", "md", "tier_2", "markdown") + + +def test_cli_json_shape_unchanged(monkeypatch, capsys): + monkeypatch.setattr(sys, "argv", ["switchback", URL]) + monkeypatch.setattr("switchback.tracing.setup_logs", lambda: None) + assert api._main() == 0 + out = json.loads(capsys.readouterr().out) + assert list(out[0]) == ["url", "source_method", "markdown"] + + +def test_cli_extras_flags(monkeypatch, capsys): + monkeypatch.setattr(sys, "argv", ["switchback", "--state", "--appendix", + "--query", "solar panel prices", URL]) + monkeypatch.setattr("switchback.tracing.setup_logs", lambda: None) + assert api._main() == 0 + [d] = json.loads(capsys.readouterr().out) + assert list(d) == ["url", "source_method", "markdown", "state", "appendix", "highlights"] + assert d["state"]["__NEXT_DATA__"]["props"]["pageProps"]["price"] == 42 + assert "Solar panel prices fell" in d["highlights"][0]["text"] + + +def test_server_shape(monkeypatch): + pytest.importorskip("fastapi") + pytest.importorskip("httpx") + from fastapi.testclient import TestClient + from switchback.server import app + c = TestClient(app) + [d] = c.post("/scrape", json={"urls": [URL]}).json() + assert list(d) == ["url", "source_method", "markdown"] + [d] = c.post("/scrape", json={"urls": [URL], "state": True, "query": "wind"}).json() + assert list(d) == ["url", "source_method", "markdown", "state", "highlights"] + + +# ── state JSON ────────────────────────────────────────────────────────────── + +def test_extract_state(): + st = S.extract_state(PAGE) + assert st["json_ld"] == [{"@type": "NewsArticle", "headline": "Solar costs fall"}] + assert st["__NEXT_DATA__"]["props"]["pageProps"]["price"] == 42 + assert st["__INITIAL_STATE__"] == {"user": None, "items": [1, 2]} + assert st["__APOLLO_STATE__"] == {"k": "v"} + assert "__BROKEN__" not in st # JS literal, not JSON: skipped, not guessed + assert "__LOADABLE_REQUIRED_CHUNKS__" not in st # bundler noise + + +def test_extract_state_empty_and_garbage(): + assert S.extract_state("") == {} + assert S.extract_state("") == {} + + +def test_state_via_scrape(): + [r] = api.scrape(URL, state=True) + assert r.state["json_ld"][0]["headline"] == "Solar costs fall" + assert r.appendix is None + + +def test_state_empty_when_tier_has_no_html(monkeypatch): + """API/PDF tiers never render HTML: requested extras come back empty, not None.""" + monkeypatch.setattr(orch, "TIERS", [_tier("fake", lambda u: "TEXT " * 600)]) + [r] = api.scrape(URL, state=True, appendix=True) + assert r.state == {} and r.appendix == [] + + +def test_state_skips_content_cache(monkeypatch): + monkeypatch.setattr(orch.content_cache, "get", lambda u, fmt: ("cached md", "tier_2")) + [r] = api.scrape(URL) + assert r.markdown == "cached md" # default still uses the cache + [r] = api.scrape(URL, state=True) + assert r.source_method == "fake" and r.state # needs HTML → fetched + + +def test_state_from_worker_thread_tier(monkeypatch): + """tier_3/tier_7 render on a worker thread; capture must follow them there.""" + import threading + + def threaded(url): + cap, box = N.current_capture(), {} + + def work(): + with N.capture_into(cap): + box["md"] = _html_tier(url) + t = threading.Thread(target=work) + t.start(); t.join() + return box["md"] + monkeypatch.setattr(orch, "TIERS", [_tier("fake", threaded)]) + [r] = api.scrape(URL, state=True) + assert r.state["__NEXT_DATA__"] + + +# ── appendix ──────────────────────────────────────────────────────────────── + +def test_appendix_regions(): + regions = N.appendix(PAGE, URL) + assert [r["region"] for r in regions] == ["nav", "header", "aside", "footer"] + assert "](https://example.test/world)" in regions[0]["markdown"] # links resolved + assert "Example News" in regions[-1]["markdown"] + + +def test_appendix_via_scrape_leaves_markdown_alone(): + [plain] = api.scrape(URL) + [r] = api.scrape(URL, appendix=True) + assert r.markdown == plain.markdown and len(r.appendix) == 4 + + +# ── highlights ────────────────────────────────────────────────────────────── + +def test_passages_keep_headings_with_text(): + md = N.html_to_markdown(PAGE, URL) + ps = H.passages(md) + assert any(p.startswith("## Wind") and "Offshore wind" in p for p in ps) + assert all(len(p.split()) <= H._MAX_WORDS + 10 for p in ps) # + a heading + + +def test_long_table_is_split_with_header_repeated(): + rows = "\n".join(f"| Canada | Widget {i} | {i} |" for i in range(200)) + md = "## Sheet1\n| Country | Product | Units |\n| --- | --- | --- |\n" + rows + ps = H.passages(md) + assert len(ps) > 3 + assert all("| Country | Product | Units |" in p for p in ps) + assert all(len(p.split()) <= H._MAX_WORDS * 1.5 for p in ps) + + +def test_bm25_stems_terms(): + md = "| Country | Units |\n| --- | --- |\n| Canada | 3 |\n\nSomething else entirely here." + hs = H.highlights(md, "which countries are listed", URL) + assert hs and "Canada" in hs[0]["text"] + + +def test_bm25_ranks_relevant_passage_first(): + md = N.html_to_markdown(PAGE, URL) + hs = H.highlights(md, "why were offshore wind projects delayed", URL) + assert "Offshore wind" in hs[0]["text"] and hs[0]["grader"] == "bm25" + assert hs[0]["link"].startswith(URL + "#:~:text=") + + +def test_no_match_returns_empty(): + assert H.highlights("Nothing relevant here at all.", "quantum chromodynamics", URL) == [] + assert H.highlights("", "x", URL) == [] + + +def test_plain_text_unescapes_markdown(): + assert H.plain_text(r"Use snake\_case and 2\*3, *not* __bold__") == "Use snake_case and 2*3, not bold" + + +def test_plain_text_links(): + assert H.plain_text("tuned.[[3]](https://w.org/x#cite_note-3) More") == "tuned.[3] More" + assert H.plain_text('the [compute](https://w.org/C_(x) "Compute (machine)") of') == "the compute of" + assert H.plain_text("see ") == "see https://youtu.be/abc" + + +def test_fragment_link_prefers_prose_over_tables(): + passage = "| Name | Last commit message and a very long cell |\n| --- | --- |\nShort prose line here." + assert H.fragment_link("https://a.test/", passage).endswith("text=Short%20prose%20line%20here.") + + +def test_fragment_link(): + link = H.fragment_link("https://a.test/p", "## Head\n\nThe **quick** [brown](https://x) fox") + assert link == "https://a.test/p#:~:text=The%20quick%20brown%20fox" + long = "one-two three four five six seven eight nine ten eleven twelve, end" + link = H.fragment_link("https://a.test/p#sec", long) + assert link.startswith("https://a.test/p#sec:~:text=one%2Dtwo%20three") + assert link.endswith(",nine%20ten%20eleven%20twelve%2C%20end") + + +class _Resp: + def __init__(self, payload): + self._p = payload + + def raise_for_status(self): + pass + + def json(self): + return self._p + + +def test_jev_grader(monkeypatch): + monkeypatch.setenv("SCRAPER_HIGHLIGHTS_GRADER", "jev") + monkeypatch.setenv("TYPESAFE_API_KEY", "test-key") + calls = [] + + def fake_post(url, json=None, headers=None, timeout=None): + calls.append((url, json, headers)) + passages = json["state"]["passages"] + return _Resp({"answers": {f"p{j}": {"type": "noul", + "noul": 0.9 if "Offshore" in p else 0.1} + for j, p in enumerate(passages)}}) + monkeypatch.setattr("requests.post", fake_post) + md = N.html_to_markdown(PAGE, URL) + hs = H.highlights(md, "what delayed wind farms?", URL) + assert [h["grader"] for h in hs] == ["jev"] # below 0.5 dropped + assert "Offshore wind" in hs[0]["text"] and hs[0]["score"] == 0.9 + _, body, headers = calls[0] + assert headers["Authorization"] == "Bearer test-key" + assert body["questions"]["p0"]["type"] == "noul" + assert "`passages[0]`" in body["questions"]["p0"]["instructions"] + + +def test_jev_failure_falls_back_to_bm25(monkeypatch): + monkeypatch.setenv("SCRAPER_HIGHLIGHTS_GRADER", "jev") + monkeypatch.setenv("TYPESAFE_API_KEY", "test-key") + + def down(*a, **k): + raise ConnectionError("api down") + monkeypatch.setattr("requests.post", down) + hs = H.highlights(N.html_to_markdown(PAGE, URL), "offshore wind delayed", URL) + assert hs and hs[0]["grader"] == "bm25" + + +def test_jev_without_key_uses_bm25(monkeypatch): + monkeypatch.setenv("SCRAPER_HIGHLIGHTS_GRADER", "jev") + monkeypatch.delenv("TYPESAFE_API_KEY", raising=False) + assert H.grader() == "bm25" + + +def test_jev_batches_respect_limits(): + chunks = ["x" * 30_000] * 5 + ["y"] * 100 + batches = H._jev_batches(chunks) + assert sorted(i for b in batches for i in b) == list(range(105)) + assert all(len(b) <= H._JEV_BATCH_N for b in batches) + assert all(sum(len(chunks[i]) for i in b) <= H._JEV_BATCH_CHARS for b in batches if len(b) > 1) + + +def test_highlights_error_never_fails_scrape(monkeypatch): + monkeypatch.setattr(H, "highlights", lambda *a, **k: 1 / 0) + [r] = api.scrape(URL, query="solar") + assert r.highlights == [] and r.markdown + + +# ── office documents (tier_2) ─────────────────────────────────────────────── + +def test_doc_extension(): + xlsx = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + assert N.doc_extension(xlsx, "https://a.test/x") == ".xlsx" + assert N.doc_extension("application/octet-stream", "https://a.test/r.docx?dl=1") == ".docx" + assert N.doc_extension("text/html; charset=utf-8", "https://a.test/r.docx") is None + assert N.doc_extension("application/pdf", "https://a.test/r.pdf") is None + + +def _docx_bytes(): + docx = pytest.importorskip("docx") + d = docx.Document() + d.add_heading("Quarterly report", 1) + for _ in range(60): + d.add_paragraph("Revenue grew strongly across every region this quarter.") + buf = io.BytesIO() + d.save(buf) + return buf.getvalue() + + +class _CffiResp: + def __init__(self, content, ctype, url): + self.status_code, self.content, self.url = 200, content, url + self.headers = {"Content-Type": ctype} + self.text = content.decode("latin-1") + + def close(self): + pass + + +def _patch_cffi(monkeypatch, resp): + fake = types.SimpleNamespace(requests=types.SimpleNamespace(get=lambda *a, **k: resp)) + monkeypatch.setitem(sys.modules, "curl_cffi", fake) + monkeypatch.setitem(sys.modules, "curl_cffi.requests", fake.requests) + monkeypatch.setattr(tier_2.session_cache, "cookie_header", lambda *a, **k: None) + + +def test_tier2_docs_off_by_default(monkeypatch): + """Without SCRAPER_DOCUMENTS the old path runs, even with markitdown installed.""" + ctype = "application/vnd.openxmlformats-officedocument.wordprocessingml.document" + _patch_cffi(monkeypatch, _CffiResp(b"

" + b"legacy body text " * 200 + b"

", + ctype, "https://a.test/r.docx")) + monkeypatch.setattr(tier_2, "doc_bytes_to_markdown", lambda *a: pytest.fail("converted")) + assert "legacy body text" in tier_2.fetch("https://a.test/r.docx") + + +def test_tier2_converts_docx(monkeypatch): + pytest.importorskip("markitdown") + monkeypatch.setenv("SCRAPER_DOCUMENTS", "1") + ctype = "application/vnd.openxmlformats-officedocument.wordprocessingml.document" + _patch_cffi(monkeypatch, _CffiResp(_docx_bytes(), ctype, "https://a.test/r.docx")) + md = tier_2.fetch("https://a.test/r.docx") + assert md.startswith("# Quarterly report") and "Revenue grew" in md + + +def test_tier2_docx_without_markitdown_takes_old_path(monkeypatch): + """No markitdown → the pre-existing HTML path runs, as before this feature.""" + monkeypatch.setenv("SCRAPER_DOCUMENTS", "1") + ctype = "application/vnd.openxmlformats-officedocument.wordprocessingml.document" + _patch_cffi(monkeypatch, _CffiResp(b"

" + b"legacy body text " * 200 + b"

", + ctype, "https://a.test/r.docx")) + def no_markitdown(*a, **k): + raise ImportError("no markitdown") + monkeypatch.setattr(tier_2, "doc_bytes_to_markdown", no_markitdown) + assert "legacy body text" in tier_2.fetch("https://a.test/r.docx") + + +# ── YouTube (tier_1) ──────────────────────────────────────────────────────── + +def test_youtube_url_matching(): + m = tier_1.YOUTUBE_RE.search + assert m("https://www.youtube.com/watch?v=dQw4w9WgXcQ").group(1) == "dQw4w9WgXcQ" + assert m("https://youtu.be/dQw4w9WgXcQ?t=3").group(1) == "dQw4w9WgXcQ" + assert m("https://www.youtube.com/shorts/dQw4w9WgXcQ").group(1) == "dQw4w9WgXcQ" + assert m("https://www.youtube.com/@channel") is None + + +def test_youtube_off_by_default(monkeypatch): + """Without SCRAPER_YOUTUBE, YouTube URLs skip tier_1, even with the library.""" + _fake_yt(monkeypatch, [(0.0, "hello")]) + monkeypatch.delenv("SCRAPER_YOUTUBE") + assert tier_1.fetch("https://youtu.be/dQw4w9WgXcQ") is None + + +def test_youtube_without_library_falls_through(monkeypatch): + monkeypatch.setenv("SCRAPER_YOUTUBE", "1") + monkeypatch.setitem(sys.modules, "youtube_transcript_api", None) # ImportError + assert tier_1.fetch("https://www.youtube.com/watch?v=dQw4w9WgXcQ") is None + + +def _fake_yt(monkeypatch, snippets=None, error=None): + class Api: + def __init__(self, proxy_config=None, http_client=None): + pass + + def fetch(self, vid, languages=("en",)): + if error: + raise error + return [types.SimpleNamespace(start=s, text=t) for s, t in snippets] + mod = types.ModuleType("youtube_transcript_api") + mod.YouTubeTranscriptApi = Api + prox = types.ModuleType("youtube_transcript_api.proxies") + prox.GenericProxyConfig = lambda **k: k + monkeypatch.setitem(sys.modules, "youtube_transcript_api", mod) + monkeypatch.setitem(sys.modules, "youtube_transcript_api.proxies", prox) + monkeypatch.setattr(tier_1, "_youtube_meta", lambda url: ("A Talk", "A Channel")) + monkeypatch.setenv("SCRAPER_YOUTUBE", "1") + + +def test_youtube_transcript(monkeypatch): + _fake_yt(monkeypatch, [(0.0, "hello"), (12.5, "world"), (31.0, "second para"), (3700, "late")]) + md = tier_1.fetch("https://youtu.be/dQw4w9WgXcQ") + assert md.startswith("# A Talk\n\n**Channel:** A Channel") + assert "[0:00](https://www.youtube.com/watch?v=dQw4w9WgXcQ&t=0s) hello world" in md + assert "[0:31](https://www.youtube.com/watch?v=dQw4w9WgXcQ&t=31s) second para" in md + assert "[1:01:40]" in md + + +def test_youtube_blocked_ip_retries_via_egress_proxy(monkeypatch): + class RequestBlocked(Exception): + pass + seen = [] + + def fake(vid, proxy_url): + seen.append(proxy_url) + if proxy_url is None: + raise RequestBlocked("cloud IP") + return [(0.0, "via proxy")] + monkeypatch.setitem(sys.modules, "youtube_transcript_api", types.ModuleType("y")) + monkeypatch.setattr(tier_1, "_yt_snippets", fake) + monkeypatch.setattr(tier_1, "_youtube_meta", lambda url: ("", "")) + monkeypatch.setenv("SCRAPER_EGRESS_PROXY", "http://user:pass@proxy.test:80") + monkeypatch.setenv("SCRAPER_YOUTUBE", "1") + md = tier_1.fetch("https://youtu.be/dQw4w9WgXcQ") + assert seen == [None, "http://user:pass@proxy.test:80"] and "via proxy" in md + monkeypatch.delenv("SCRAPER_EGRESS_PROXY") + seen.clear() + assert tier_1.fetch("https://youtu.be/dQw4w9WgXcQ") is None and seen == [None] + + +def test_youtube_tried_first_despite_learned_start(monkeypatch): + """youtube.com learned tier_5: an opted-in video still tries tier_1 first, and + without a transcript resumes at tier_5 (tier_2 stays skipped). Off: unchanged.""" + calls = [] + def t(name, md): + return _tier(name, lambda u: calls.append(name) or md) + monkeypatch.setattr(orch, "TIERS", [t("tier_1", None), t("tier_2", "x" * 3000), + t("tier_5", "y" * 3000)]) + monkeypatch.setattr(orch, "INDEX", {"tier_1": 0, "tier_2": 1, "tier_5": 2}) + monkeypatch.setattr(orch.botwall, "winning_tier", lambda h, db: "tier_5") + yt = "https://www.youtube.com/watch?v=dQw4w9WgXcQ" + api.scrape_detailed(yt) + assert calls == ["tier_5"] + monkeypatch.setenv("SCRAPER_YOUTUBE", "1") + calls.clear() + [o] = api.scrape_detailed(yt) + assert calls == ["tier_1", "tier_5"] and o.source_method == "tier_5" + calls.clear() + api.scrape_detailed("https://example.test/page") + assert calls == ["tier_5"] + + +def test_youtube_highlight_cites_timestamp(): + md = ("# T\n\n## Transcript\n\n[0:00](https://www.youtube.com/watch?v=abcdefghijk&t=0s) intro\n\n" + "[12:05](https://www.youtube.com/watch?v=abcdefghijk&t=725s) attention is all you need") + [h] = H.highlights(md, "attention", "https://youtu.be/abcdefghijk") + assert h["link"] == "https://www.youtube.com/watch?v=abcdefghijk&t=725s" + + +def test_youtube_no_captions_falls_through(monkeypatch): + _fake_yt(monkeypatch, error=RuntimeError("TranscriptsDisabled")) + assert tier_1.fetch("https://www.youtube.com/watch?v=dQw4w9WgXcQ") is None + + +# ── MCP server ────────────────────────────────────────────────────────────── + +def test_mcp_tools(monkeypatch): + pytest.importorskip("mcp") + import asyncio + from switchback import mcp_server as M + tools = {t.name for t in asyncio.run(M.server.list_tools())} + assert tools == {"scrape", "search"} + ok = asyncio.run(M.scrape(URL, query="solar prices")) + assert ok["ok"] and ok["markdown"] and ok["highlights"] + monkeypatch.setattr(orch, "TIERS", [_tier("fake", lambda u: (_ for _ in ()).throw( + orch.BotWall("cf", vendor="cloudflare")))]) + bad = asyncio.run(M.scrape(URL)) + assert bad == {"ok": False, "url": URL, "error": "botwall", "status_code": None} + + +def test_highlights_on_html_format_use_markdown_text(): + [r] = api.scrape(URL, fmt="html", query="offshore wind delayed") + assert r.markdown.startswith("") # content unchanged + assert "Offshore wind" in r.highlights[0]["text"] and "

" not in r.highlights[0]["text"]