blocked-page-recovery

1#!/usr/bin/env python32"""Recover a blocked / paywalled / WAF'd page from third-party copies.3 4Ladder (cheapest first):5  1. Wayback Machine "available" API  -> dated snapshot   (provenance: snapshot)6  2. archive.today domain rotation    -> dated snapshot   (provenance: snapshot)7  3. Jina Reader (JINA_API_KEY only)  -> live re-render   (provenance: live)8 9Every candidate body is validated before being declared a win: byte floors,10redirect-stub detection (meta-refresh/JS pointing back at the original host),11and interstitial-title rejection. Fake 200s are the norm in this space.12 13Stdlib only. Usage:14    python3 recover_page.py URL [--json] [--out FILE] [--timeout N]15 16Exit codes: 0 recovered, 1 nothing worked, 2 bad invocation.17"""18from __future__ import annotations19 20import argparse21import json22import os23import re24import sys25import time26import urllib.error27import urllib.parse28import urllib.request29 30USER_AGENT = (31    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "32    "(KHTML, like Gecko) Chrome/126.0 Safari/537.36"33)34 35ARCHIVE_TODAY_HOSTS = ["archive.ph", "archive.md", "archive.li", "archive.is"]36 37# Titles that mean "this is not the page you asked for".38INTERSTITIAL_TITLES = (39    "just a moment",40    "redirecting",41    "google search",42    "attention required",43    "access denied",44    "are you a robot",45    "one more step",46)47 48# Below these floors a body is a stub or an error page, not content.49MIN_BODY_BYTES = {"wayback": 3072, "archive_today": 3072, "jina": 512}50 51REDIRECT_STUB_RE = re.compile(52    r'http-equiv=["\']?refresh|window\.location|location\.replace', re.IGNORECASE53)54TITLE_RE = re.compile(r"<title[^>]*>(.*?)</title>", re.IGNORECASE | re.DOTALL)55 56 57def _fetch(58    url: str,59    timeout: int,60    headers: dict | None = None,61    retries_on_429: int = 2,62) -> tuple[int, bytes]:63    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, **(headers or {})})64    for attempt in range(retries_on_429 + 1):65        try:66            with urllib.request.urlopen(req, timeout=timeout) as resp:67                return resp.status, resp.read()68        except urllib.error.HTTPError as exc:69            if exc.code == 429 and attempt < retries_on_429:70                time.sleep(5 * (attempt + 1))71                continue72            return exc.code, exc.read() if exc.fp else b""73        except (urllib.error.URLError, OSError, ValueError):74            return 0, b""75    return 0, b""76 77 78def _fetch_follow(url: str, timeout: int) -> tuple[int, bytes, str]:79    """Like _fetch but also returns the final URL after redirects."""80    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})81    try:82        with urllib.request.urlopen(req, timeout=timeout) as resp:83            return resp.status, resp.read(), resp.geturl()84    except urllib.error.HTTPError as exc:85        return exc.code, exc.read() if exc.fp else b"", exc.geturl() or url86    except (urllib.error.URLError, OSError, ValueError):87        return 0, b"", url88 89 90def _page_title(body: bytes) -> str:91    m = TITLE_RE.search(body[:65536].decode("utf-8", "replace"))92    return re.sub(r"\s+", " ", m.group(1)).strip().lower() if m else ""93 94 95def validate(body: bytes, route: str, target_url: str) -> str | None:96    """Return a rejection reason, or None if the body looks like real content."""97    floor = MIN_BODY_BYTES.get(route, 3072)98    if len(body) < floor:99        return f"body_too_small:{len(body)}<{floor}"100    title = _page_title(body)101    for marker in INTERSTITIAL_TITLES:102        if marker in title:103            return f"interstitial_title:{marker!r}"104    # Redirect stub: small-ish page whose only job is bouncing back to the105    # original (blocked) host — the classic AMP-cache failure mode.106    if len(body) < 8192 and REDIRECT_STUB_RE.search(body.decode("utf-8", "replace")):107        target_host = urllib.parse.urlsplit(target_url).hostname or ""108        if target_host and target_host.encode() in body:109            return "redirect_stub_to_origin"110    return None111 112 113def try_wayback(url: str, timeout: int) -> dict | None:114    snap_url = None115    snap_ts = None116    discovery = "https://archive.org/wayback/available?url=" + urllib.parse.quote(url, safe="")117    status, raw = _fetch(discovery, timeout)118    if status == 200:119        try:120            closest = json.loads(raw).get("archived_snapshots", {}).get("closest", {})121        except (json.JSONDecodeError, AttributeError):122            closest = {}123        if closest.get("available") and closest.get("url"):124            snap_url = closest["url"].replace(125                "http://web.archive.org", "https://web.archive.org"126            )127            snap_ts = closest.get("timestamp")128    if snap_url is None:129        # Discovery API is rate-limited far more aggressively than snapshot130        # serving. Fall back to the redirect form: /web/2/<url> bounces to131        # the newest snapshot if one exists (404 page otherwise).132        snap_url = "https://web.archive.org/web/2/" + url133    status, body, final_url = _fetch_follow(snap_url, timeout)134    if status != 200 or validate(body, "wayback", url):135        return None136    if snap_ts is None:137        m = re.search(r"/web/(\d{14})", final_url)138        snap_ts = m.group(1) if m else None139    return {140        "route": "wayback",141        "provenance": "snapshot",142        "snapshot_timestamp": snap_ts,143        "source_url": final_url,144        "body": body,145    }146 147 148def try_archive_today(url: str, timeout: int) -> dict | None:149    for host in ARCHIVE_TODAY_HOSTS:150        fetch_url = f"https://{host}/newest/{url}"151        status, body = _fetch(fetch_url, timeout)152        if status != 200:153            continue154        if validate(body, "archive_today", url):155            continue  # 429 bodies and interstitials land here156        return {157            "route": f"archive_today:{host}",158            "provenance": "snapshot",159            "snapshot_timestamp": None,  # archive.today embeds the date in-page160            "source_url": fetch_url,161            "body": body,162        }163    return None164 165 166def try_jina(url: str, timeout: int) -> dict | None:167    key = os.environ.get("JINA_API_KEY")168    if not key:169        return None170    status, body = _fetch(171        "https://r.jina.ai/" + url, timeout, headers={"Authorization": f"Bearer {key}"}172    )173    if status != 200 or validate(body, "jina", url):174        return None175    return {176        "route": "jina_reader",177        "provenance": "live",178        "snapshot_timestamp": None,179        "source_url": "https://r.jina.ai/" + url,180        "body": body,181    }182 183 184ROUTES = (try_wayback, try_archive_today, try_jina)185 186 187def recover(url: str, timeout: int = 25) -> dict | None:188    for route_fn in ROUTES:189        result = route_fn(url, timeout)190        if result:191            return result192    return None193 194 195def main() -> int:196    ap = argparse.ArgumentParser(197        description="Recover a blocked / paywalled / WAF'd page from third-party copies."198    )199    ap.add_argument("url")200    ap.add_argument("--json", action="store_true", help="print metadata as JSON")201    ap.add_argument("--out", help="write recovered body to this file")202    ap.add_argument("--timeout", type=int, default=25)203    args = ap.parse_args()204 205    if not args.url.startswith(("http://", "https://")):206        print("error: URL must start with http:// or https://", file=sys.stderr)207        return 2208 209    result = recover(args.url, args.timeout)210    if not result:211        msg = {"recovered": False, "url": args.url,212               "hint": "No archive copy found. Try the API-first pivot or the browser tool."}213        print(json.dumps(msg, indent=2) if args.json else msg["hint"], file=sys.stderr)214        return 1215 216    body = result.pop("body")217    result.update({"recovered": True, "url": args.url, "body_bytes": len(body)})218    if args.out:219        with open(args.out, "wb") as fh:220            fh.write(body)221        result["saved_to"] = args.out222 223    if args.json:224        print(json.dumps(result, indent=2))225    else:226        for k, v in result.items():227            print(f"{k}: {v}")228        if not args.out:229            print("\n--- body (first 2000 chars) ---")230            print(body[:2000].decode("utf-8", "replace"))231    if result["provenance"] == "snapshot":232        print(233            "\nNOTE: this is an ARCHIVED SNAPSHOT, not the live page. "234            "Cite it with its timestamp.",235            file=sys.stderr,236        )237    return 0238 239 240if __name__ == "__main__":241    sys.exit(main())242