| |
| """Backfill gmeow ban records from Common Crawl snapshots (via proxy pool). |
| |
| For every collection that captured gkybot.gmeow.cc/latestban, fetch the WARC |
| record by byte range, decompress, parse with the same parser as the live |
| collector and append to the shared store with source=cc:<collection>. |
| |
| Usage: |
| python scripts/cc_backfill_gmeow.py --scan # index only |
| python scripts/cc_backfill_gmeow.py --fetch # download + parse |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import gzip |
| import json |
| import random |
| import subprocess |
| import sys |
| import time |
| import urllib.parse |
| from pathlib import Path |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parent)) |
| from collect_gmeow_bans import append_records, load_seen, parse_page |
|
|
| ROOT = Path("data/raw/gmeow") |
| COLLINFO = ROOT / "cc_collinfo.json" |
| INDEX_OUT = ROOT / "cc_index.jsonl" |
| PROXY_TOOL = Path("C:/Users/ll/.agents/skills/proxy-pool/scripts/proxy_pool.py") |
|
|
|
|
| def proxies(n: int = 6) -> list[str]: |
| try: |
| out = subprocess.run([sys.executable, str(PROXY_TOOL), "pick", |
| "--type", "socks5", "--n", str(n)], |
| capture_output=True, text=True, check=True).stdout |
| items = [ln.strip() for ln in out.splitlines() if ln.startswith("socks5h://")] |
| return items or [] |
| except Exception as exc: |
| print(f"[warn] proxy pick failed: {exc}") |
| return [] |
|
|
|
|
| def curl(url: str, proxy: str | None, timeout: int = 60, |
| out: Path | None = None, headers: list[str] | None = None) -> bytes: |
| cmd = ["curl", "-sS", "-m", str(timeout), "-L"] |
| if proxy: |
| cmd += ["-x", proxy] |
| for h in headers or []: |
| cmd += ["-H", h] |
| if out: |
| cmd += ["-o", str(out)] |
| cmd.append(url) |
| res = subprocess.run(cmd, capture_output=True) |
| if res.returncode != 0: |
| raise RuntimeError(f"curl exit {res.returncode}: {res.stderr.decode('utf-8','replace')[:200]}") |
| return res.stdout |
|
|
|
|
| def scan(collections: list[str]) -> None: |
| pool = proxies() |
| if not pool: |
| raise SystemExit("no proxies available") |
| seen_lines = set() |
| if INDEX_OUT.exists(): |
| seen_lines = {ln for ln in INDEX_OUT.read_text(encoding="utf-8").splitlines() if ln} |
| with INDEX_OUT.open("a", encoding="utf-8") as fh: |
| for i, cid in enumerate(collections): |
| q = urllib.parse.urlencode({ |
| "url": "gkybot.gmeow.cc", "matchType": "domain", |
| "output": "json", "limit": "200"}) |
| url = f"https://index.commoncrawl.org/{cid}-index?{q}" |
| try: |
| body = curl(url, pool[i % len(pool)], timeout=60) |
| text = body.decode("utf-8", "replace").strip() |
| except Exception as exc: |
| print(f"[scan] {cid} ERR {exc}") |
| time.sleep(3) |
| continue |
| if not text or text.startswith('{"message"'): |
| print(f"[scan] {cid} empty") |
| time.sleep(1.5) |
| continue |
| if text.startswith("<"): |
| print(f"[scan] {cid} non-json ({text[:40]!r})") |
| time.sleep(2) |
| continue |
| try: |
| rows = [json.loads(ln) for ln in text.splitlines() if ln.strip()] |
| except json.JSONDecodeError: |
| print(f"[scan] {cid} bad json ({text[:60]!r})") |
| time.sleep(2) |
| continue |
| new = 0 |
| for r in rows: |
| keys = {k: r.get(k) for k in |
| ("url", "timestamp", "filename", "offset", "length", "status", "mime")} |
| line = json.dumps(keys, ensure_ascii=False) |
| if line in seen_lines: |
| continue |
| seen_lines.add(line) |
| fh.write(line + "\n") |
| new += 1 |
| fh.flush() |
| print(f"[scan] {cid} captures={len(rows)} new={new}") |
| time.sleep(1.5) |
|
|
|
|
| def warc_body(raw: bytes) -> bytes | None: |
| """Extract the HTTP body from a raw WARC response record.""" |
| idx = raw.find(b"\r\n\r\n") |
| if idx < 0: |
| return None |
| rest = raw[idx + 4:] |
| |
| hidx = rest.find(b"\r\n\r\n") |
| if hidx < 0: |
| return None |
| headers = rest[:hidx].decode("latin-1", "replace").lower() |
| body = rest[hidx + 4:] |
| if "content-encoding: gzip" in headers or body[:2] == b"\x1f\x8b": |
| try: |
| body = gzip.decompress(body) |
| except Exception: |
| try: |
| body = gzip.GzipFile(fileobj=__import__("io").BytesIO(body)).read() |
| except Exception: |
| return None |
| return body |
|
|
|
|
| def fetch(store: Path) -> None: |
| if not INDEX_OUT.exists(): |
| raise SystemExit("run --scan first") |
| pool = proxies() |
| if not pool: |
| raise SystemExit("no proxies available") |
| rows = [json.loads(l) for l in INDEX_OUT.open(encoding="utf-8") if l.strip()] |
| latest = [r for r in rows if "/latestban" in r["url"] and r.get("status") == "200"] |
| latest.sort(key=lambda r: r["timestamp"], reverse=True) |
| print(f"[fetch] latestban captures={len(latest)}") |
| seen = load_seen(store) |
| for i, r in enumerate(latest): |
| start = int(r["offset"]) |
| end = start + int(r["length"]) - 1 |
| url = f"https://data.commoncrawl.org/{r['filename']}" |
| try: |
| raw = curl(url, pool[i % len(pool)], timeout=120, |
| headers=[f"Range: bytes={start}-{end}"]) |
| body = warc_body(raw) |
| if not body: |
| print(f"[fetch] {r['timestamp']} no body") |
| continue |
| html = body.decode("utf-8", "replace") |
| records = parse_page(html) |
| new = [x for x in records if x["id"] not in seen] |
| for x in new: |
| seen.add(x["id"]) |
| append_records(store, new, f"cc:{r['timestamp'][:8]}") |
| print(f"[fetch] {r['timestamp']} records={len(records)} new={len(new)}") |
| except Exception as exc: |
| print(f"[fetch] {r['timestamp']} ERR {exc}") |
| time.sleep(1.5) |
|
|
|
|
| def main() -> int: |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--scan", action="store_true") |
| ap.add_argument("--fetch", action="store_true") |
| ap.add_argument("--store", type=Path, default=Path("data/raw/gmeow/bans.jsonl")) |
| ap.add_argument("--since", default="CC-MAIN-2024") |
| ap.add_argument("--max-collections", type=int, default=0) |
| args = ap.parse_args() |
| if args.scan: |
| colls = [x["id"] for x in json.loads(COLLINFO.read_text(encoding="utf-8")) |
| if x["id"] >= args.since] |
| if args.max_collections: |
| colls = colls[-args.max_collections:] |
| print(f"scanning {len(colls)} collections") |
| scan(colls) |
| if args.fetch: |
| fetch(args.store) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
