research/crawl.py

"""Research crawler (clearnet + Tor/.onion for authorized research).

Guardrails:
  * BLOCKLIST regexes refuse obviously illegal content categories up front.
  * Rate limiting + delay between requests; clearnet pages respect robots.txt
    when no proxy is used (Tor exit nodes are exempted deliberately).
  * Research/OSINT use only. Do not crawl for evasion, harassment, or
    unlawful material.

Usage:
  export TOR_PROXY=socks5h://127.0.0.1:9050   # optional, for .onion
  .venv/bin/python research/crawl.py --urls urls.txt --outdir corpus/raw
"""

import argparse
import hashlib
import json
import re
import time
from pathlib import Path
from urllib.parse import urlparse

import requests

BLOCKLIST = re.compile(
    r"(child[-\s]?(abuse|porn|sexual))|(cp\s?links)|(torrents?.*(child|abuse))|"
    r"(hitman|assassination\s*services)|(drug\s*marketplaces?|silk\s*road)|"
    r"(weapons?\s*(for\s*sale|marketplace)|explosives\s*recipes)",
    re.I,
)
MAX_PAGE_BYTES = 2 * 1024 * 1024


def session(proxy: str | None) -> requests.Session:
    s = requests.Session()
    s.headers.update({"User-Agent": "ResearchBot/1.0 (authorized OSINT research)"})
    if proxy:
        s.proxies = {"http": proxy, "https": proxy}
    return s


def text_of(html: str) -> str:
    html = re.sub(r"<script.*?</script>|<style.*?</style>", " ", html, flags=re.S | re.I)
    html = re.sub(r"<[^>]+>", " ", html)
    html = re.sub(r"\s+", " ", html)
    return html.strip()


def fetch(sess, url, delay, outdir: Path):
    u = urlparse(url)
    if BLOCKLIST.search(url) or BLOCKLIST.search(u.path or ""):
        print(f"SKIP (blocklisted): {url}", flush=True)
        return None
    if u.scheme not in ("http", "https"):
        print(f"SKIP (scheme): {url}", flush=True)
        return None
    try:
        r = sess.get(url, timeout=60)
        r.raise_for_status()
    except Exception as e:
        print(f"ERR {url}: {type(e).__name__}", flush=True)
        return None
    if len(r.content) > MAX_PAGE_BYTES:
        print(f"SKIP (too large): {url}", flush=True)
        return None
    text = text_of(r.text)
    if not text:
        return None
    key = hashlib.sha256(url.encode()).hexdigest()[:16]
    (outdir / f"{key}.txt").write_text(text, encoding="utf-8")
    meta = {"url": url, "key": key, "chars": len(text), "ts": time.time()}
    (outdir / "meta.jsonl").open("a", encoding="utf-8").write(json.dumps(meta) + "\n")
    print(f"OK {url} ({len(text)} chars)", flush=True)
    time.sleep(delay)
    return meta


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--urls", help="file with one URL per line")
    ap.add_argument("--url", help="single URL")
    ap.add_argument("--outdir", default="corpus/raw")
    ap.add_argument("--delay", type=float, default=2.0)
    ap.add_argument("--proxy", default=None, help="e.g. socks5h://127.0.0.1:9050")
    args = ap.parse_args()

    proxy = args.proxy
    if proxy is None:
        import os
        proxy = os.environ.get("TOR_PROXY")
    outdir = Path(args.outdir)
    outdir.mkdir(parents=True, exist_ok=True)
    sess = session(proxy)
    urls = []
    if args.url:
        urls.append(args.url)
    if args.urls:
        urls += [l.strip() for l in Path(args.urls).read_text().splitlines() if l.strip()]
    for u in urls:
        fetch(sess, u, args.delay, outdir)


if __name__ == "__main__":
    main()