reddit-reading

1#!/usr/bin/env python32"""Read Reddit without a browser: listings, search, threads with comments, user pages.3 4Two backends, chosen automatically:5 6* **OAuth API** (preferred when ``REDDIT_CLIENT_ID`` + ``REDDIT_CLIENT_SECRET`` are set):7  app-only ``client_credentials`` grant for a free "script" app registered at8  https://www.reddit.com/prefs/apps. No username, password or cookie is ever used and9  the script never acts as a user. ~100 requests/minute, full JSON including scores10  and nested comments.11* **Anonymous Atom feeds** (``.rss`` endpoints): the only unauthenticated path Reddit12  still serves to server IPs (``.json`` and old.reddit return 403 / an empty shell).13  Roughly ONE request per minute per IP; the script sleeps until the window resets14  when it hits a 429 and retries once.15 16    python3 reddit.py sub LocalLLaMA [--sort hot|new|top] [--limit N]17    python3 reddit.py search "hermes agent" [--sub LocalLLaMA] [--sort new] [--limit N]18    python3 reddit.py thread https://www.reddit.com/r/x/comments/abc123/... [--limit N]19    python3 reddit.py user spez [--limit N]20    python3 reddit.py doctor            # which backend is active, and why21 22Add ``--json`` to any read command for machine-readable output. Standard library only.23"""24 25from __future__ import annotations26 27import argparse28import base6429import html30import json31import os32import re33import sys34import time35import urllib.error36import urllib.parse37import urllib.request38import xml.etree.ElementTree as ET39 40USER_AGENT = "hermes-agent/1.0 (reddit-reading skill; +https://github.com/NousResearch/hermes-agent)"41TIMEOUT = 2542ATOM = {"a": "http://www.w3.org/2005/Atom"}43WWW = "https://www.reddit.com"44OAUTH = "https://oauth.reddit.com"45_TAG_RE = re.compile(r"<[^>]+>")46_WS_RE = re.compile(r"\s+")47_THREAD_RE = re.compile(r"reddit\.com/r/([^/]+)/comments/([a-z0-9]+)", re.I)48 49 50def strip_html(text: str | None) -> str:51    if not text:52        return ""53    # Reddit wraps entry bodies in a <table> with a "submitted by /u/x [link] [comments]" footer.54    text = _TAG_RE.sub(" ", html.unescape(text))55    text = re.sub(r"submitted by\s+/u/\S+|\[link\]|\[comments\]", " ", text)56    return _WS_RE.sub(" ", html.unescape(text)).strip()57 58 59# ── HTTP ─────────────────────────────────────────────────────────────────────60 61def _get(url: str, headers: dict | None = None, retry_on_429: bool = True) -> tuple[bytes, dict]:62    hdrs = {"User-Agent": USER_AGENT, "Accept": "*/*"}63    hdrs.update(headers or {})64    req = urllib.request.Request(url, headers=hdrs)65    try:66        with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:67            return resp.read(), dict(resp.headers)68    except urllib.error.HTTPError as exc:69        if exc.code == 429 and retry_on_429:70            wait = _reset_seconds(exc.headers)71            print(f"reddit: 429 rate-limited, sleeping {wait}s until the window resets", file=sys.stderr)72            time.sleep(wait)73            return _get(url, headers, retry_on_429=False)74        raise75 76 77def _reset_seconds(headers) -> int:78    for key in ("x-ratelimit-reset", "retry-after"):79        val = headers.get(key) if headers else None80        if val:81            try:82                return max(1, min(int(float(val)) + 1, 120))83            except ValueError:84                pass85    return 6186 87 88# ── OAuth backend ────────────────────────────────────────────────────────────89 90def oauth_credentials() -> tuple[str, str] | None:91    cid, secret = os.environ.get("REDDIT_CLIENT_ID"), os.environ.get("REDDIT_CLIENT_SECRET")92    return (cid, secret) if cid and secret else None93 94 95def oauth_token(cid: str, secret: str) -> str:96    body = urllib.parse.urlencode({"grant_type": "client_credentials"}).encode()97    auth = base64.b64encode(f"{cid}:{secret}".encode()).decode()98    req = urllib.request.Request(99        f"{WWW}/api/v1/access_token", data=body,100        headers={"Authorization": f"Basic {auth}", "User-Agent": USER_AGENT},101    )102    with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:103        return json.loads(resp.read())["access_token"]104 105 106def _api(path: str, token: str, **params):107    params.setdefault("raw_json", 1)108    url = f"{OAUTH}{path}?{urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})}"109    data, _ = _get(url, {"Authorization": f"Bearer {token}"})110    return json.loads(data)111 112 113def _post_from_api(child: dict) -> dict:114    d = child["data"]115    return {116        "title": d.get("title"),117        "author": d.get("author"),118        "subreddit": d.get("subreddit"),119        "score": d.get("score"),120        "num_comments": d.get("num_comments"),121        "created_utc": d.get("created_utc"),122        "url": f"{WWW}{d['permalink']}" if d.get("permalink") else d.get("url"),123        "external_url": None if d.get("is_self") else d.get("url"),124        "body": (d.get("selftext") or "")[:4000],125    }126 127 128def _flatten_comments(children: list, depth: int = 0, out: list | None = None) -> list:129    out = out if out is not None else []130    for c in children:131        if c.get("kind") != "t1":132            continue133        d = c["data"]134        out.append({135            "author": d.get("author"), "score": d.get("score"), "depth": depth,136            "created_utc": d.get("created_utc"), "body": (d.get("body") or "")[:4000],137            "url": f"{WWW}{d['permalink']}" if d.get("permalink") else None,138        })139        replies = d.get("replies")140        if isinstance(replies, dict):141            _flatten_comments(replies["data"]["children"], depth + 1, out)142    return out143 144 145def api_listing(token: str, path: str, limit: int, **params) -> list[dict]:146    data = _api(path, token, limit=limit, **params)147    return [_post_from_api(c) for c in data["data"]["children"] if c.get("kind") == "t3"]148 149 150def api_thread(token: str, sub: str, post_id: str, limit: int) -> dict:151    data = _api(f"/r/{sub}/comments/{post_id}", token, limit=limit, depth=10, sort="top")152    post = _post_from_api(data[0]["data"]["children"][0])153    post["comments"] = _flatten_comments(data[1]["data"]["children"])[:limit]154    return post155 156 157# ── Anonymous Atom backend ───────────────────────────────────────────────────158 159def _entries(url: str) -> list[dict]:160    data, _ = _get(url)161    root = ET.fromstring(data)162    out = []163    for e in root.findall("a:entry", ATOM):164        link = e.find("a:link", ATOM)165        out.append({166            "title": strip_html(e.findtext("a:title", default="", namespaces=ATOM)),167            "author": (e.findtext("a:author/a:name", default="", namespaces=ATOM) or "").replace("/u/", "") or None,168            "created": e.findtext("a:updated", default="", namespaces=ATOM) or None,169            "url": link.get("href") if link is not None else None,170            "body": strip_html(e.findtext("a:content", default="", namespaces=ATOM))[:4000],171        })172    return out173 174 175def atom_listing(path: str, limit: int, **params) -> list[dict]:176    params["limit"] = limit177    return _entries(f"{WWW}{path}.rss?{urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})}")178 179 180def atom_thread(sub: str, post_id: str, limit: int) -> dict:181    entries = _entries(f"{WWW}/r/{sub}/comments/{post_id}/.rss?limit={limit}")182    if not entries:183        raise SystemExit("thread feed returned no entries")184    post, comments = entries[0], entries[1:]185    post["comments"] = [{"author": c["author"], "created": c["created"], "body": c["body"], "url": c["url"]} for c in comments]186    post["note"] = ("anonymous feed: scores and nesting unavailable; register a free Reddit script app and set "187                    "REDDIT_CLIENT_ID/REDDIT_CLIENT_SECRET (no user login) for full data")188    return post189 190 191# ── Commands ─────────────────────────────────────────────────────────────────192 193def parse_thread_url(url: str) -> tuple[str, str]:194    m = _THREAD_RE.search(url)195    if not m:196        raise SystemExit(f"not a Reddit thread URL: {url}")197    return m.group(1), m.group(2)198 199 200def cmd_sub(a, token):201    path = f"/r/{a.name}/{a.sort}"202    if token:203        return api_listing(token, path, a.limit, t=a.time if a.sort == "top" else None)204    return atom_listing(path, a.limit, t=a.time if a.sort == "top" else None)205 206 207def cmd_search(a, token):208    path = f"/r/{a.sub}/search" if a.sub else "/search"209    params = {"q": a.query, "sort": a.sort, "restrict_sr": 1 if a.sub else None, "t": a.time}210    return api_listing(token, path, a.limit, **params) if token else atom_listing(path, a.limit, **params)211 212 213def cmd_thread(a, token):214    sub, post_id = parse_thread_url(a.url)215    return api_thread(token, sub, post_id, a.limit) if token else atom_thread(sub, post_id, a.limit)216 217 218def cmd_user(a, token):219    path = f"/user/{a.name}"220    if token:221        data = _api(f"{path}/overview", token, limit=a.limit)222        out = []223        for c in data["data"]["children"]:224            out.append(_post_from_api(c) if c["kind"] == "t3" else _flatten_comments([c])[0])225        return out226    return atom_listing(path, a.limit)227 228 229def cmd_doctor(a, token):230    report = {"oauth_credentials": bool(oauth_credentials()), "user_agent": USER_AGENT}231    if token:232        try:233            _api("/r/announcements/hot", token, limit=1)234            report["active_backend"] = "oauth"235        except (urllib.error.URLError, OSError, KeyError) as exc:236            report["active_backend"] = "oauth (broken)"237            report["oauth_error"] = str(exc)238    else:239        report["active_backend"] = "anonymous-atom"240    try:241        data, headers = _get(f"{WWW}/r/announcements/.rss?limit=1", retry_on_429=False)242        report["anonymous_feed"] = "ok" if b"<feed" in data[:200] else "unexpected body"243        report["anonymous_ratelimit"] = {k: v for k, v in headers.items() if k.lower().startswith("x-ratelimit")}244    except urllib.error.HTTPError as exc:245        report["anonymous_feed"] = f"HTTP {exc.code}"246    report["notes"] = [247        "anonymous .rss needs no account, login, cookie or key; ~1 request/minute per IP",248        "www.reddit.com .json, api.reddit.com and old.reddit are 403 / an empty shell for server IPs",249        "for more than a few calls per task register a free 'script' app at reddit.com/prefs/apps and set "250        "REDDIT_CLIENT_ID/REDDIT_CLIENT_SECRET in .env (app-only credentials; Hermes never logs in as the user)",251    ]252    return report253 254 255COMMANDS = {"sub": cmd_sub, "search": cmd_search, "thread": cmd_thread, "user": cmd_user, "doctor": cmd_doctor}256 257 258def render(cmd: str, result) -> str:259    if cmd == "doctor":260        return "\n".join(f"{k}: {v}" for k, v in result.items())261    if cmd == "thread":262        p = result263        lines = [f"# {p.get('title')}  — u/{p.get('author')}  score={p.get('score', '?')}  {p.get('url')}", p.get("body", "")[:1500], ""]264        for c in p["comments"]:265            indent = "  " * c.get("depth", 0)266            lines.append(f"{indent}- u/{c.get('author')} (score {c.get('score', '?')}): {c.get('body', '')[:600]}")267        if p.get("note"):268            lines.append(f"\n[{p['note']}]")269        return "\n".join(lines)270    lines = []271    for p in result:272        score = f" ↑{p['score']}" if p.get("score") is not None else ""273        nc = f" 💬{p['num_comments']}" if p.get("num_comments") is not None else ""274        lines.append(f"- {p.get('title') or p.get('body', '')[:80]}{score}{nc}  — u/{p.get('author')}\n  {p.get('url')}")275        if p.get("body") and p.get("title"):276            lines.append(f"  {p['body'][:300]}")277    return "\n".join(lines) or "(no results)"278 279 280def main(argv: list[str] | None = None) -> int:281    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)282    ap.add_argument("--json", action="store_true")283    sub = ap.add_subparsers(dest="cmd", required=True)284    s = sub.add_parser("sub"); s.add_argument("name"); s.add_argument("--sort", default="hot", choices=["hot", "new", "top", "rising"]); s.add_argument("--time", default="week", choices=["hour", "day", "week", "month", "year", "all"]); s.add_argument("--limit", type=int, default=15)285    q = sub.add_parser("search"); q.add_argument("query"); q.add_argument("--sub"); q.add_argument("--sort", default="relevance", choices=["relevance", "new", "top", "comments"]); q.add_argument("--time", default="all", choices=["hour", "day", "week", "month", "year", "all"]); q.add_argument("--limit", type=int, default=15)286    t = sub.add_parser("thread"); t.add_argument("url"); t.add_argument("--limit", type=int, default=40)287    u = sub.add_parser("user"); u.add_argument("name"); u.add_argument("--limit", type=int, default=15)288    sub.add_parser("doctor")289    args = ap.parse_args(argv)290 291    creds = oauth_credentials()292    token = None293    if creds:294        try:295            token = oauth_token(*creds)296        except (urllib.error.URLError, OSError, KeyError) as exc:297            print(f"reddit: OAuth token failed ({exc}); falling back to anonymous feeds", file=sys.stderr)298    try:299        result = COMMANDS[args.cmd](args, token)300    except urllib.error.HTTPError as exc:301        print(f"HTTP {exc.code} for {exc.url}", file=sys.stderr)302        return 2303    except (urllib.error.URLError, ET.ParseError, json.JSONDecodeError, KeyError) as exc:304        print(f"error: {exc}", file=sys.stderr)305        return 2306    print(json.dumps(result, indent=2, ensure_ascii=False) if args.json else render(args.cmd, result))307    return 0308 309 310if __name__ == "__main__":311    sys.exit(main())312