openclaw/tools/reddit_deep_dive.py

#!/usr/bin/env python3
"""Deep-dive Reddit subreddits without auth.

- Fetches top posts for a given time window (day/week/month/year/all)
- Paginates up to a max number of posts
- Emits a markdown report + a JSONL cache

Usage:
  reddit_deep_dive.py openclaw,clawd,moltbot --t month --limit 200

Notes:
- Uses public reddit JSON endpoints. Be nice: sleeps between requests.
"""

from __future__ import annotations

import argparse
import json
import time
import urllib.parse
import urllib.request
from dataclasses import dataclass
from pathlib import Path


@dataclass
class Post:
    subreddit: str
    id: str
    title: str
    permalink: str
    url: str
    created_utc: float
    score: int
    num_comments: int
    author: str
    selftext: str
    is_self: bool
    stickied: bool


def fetch_json(url: str) -> dict:
    req = urllib.request.Request(
        url,
        headers={
            "User-Agent": "openclaw-reddit-deep-dive/1.0 (no-auth; json feed)"
        },
    )
    with urllib.request.urlopen(req, timeout=25) as resp:
        return json.loads(resp.read().decode("utf-8"))


def fetch_top(subreddit: str, t: str, limit: int, after: str | None) -> tuple[list[Post], str | None]:
    q = {
        "t": t,
        "limit": str(min(limit, 100)),
    }
    if after:
        q["after"] = after
    url = f"https://www.reddit.com/r/{subreddit}/top/.json?{urllib.parse.urlencode(q)}"
    data = fetch_json(url)
    children = data.get("data", {}).get("children", [])
    out: list[Post] = []
    for child in children:
        d = child.get("data", {})
        out.append(
            Post(
                subreddit=subreddit,
                id=d.get("id", ""),
                title=d.get("title", ""),
                permalink="https://www.reddit.com" + d.get("permalink", ""),
                url=d.get("url", ""),
                created_utc=float(d.get("created_utc", 0.0) or 0.0),
                score=int(d.get("score", 0) or 0),
                num_comments=int(d.get("num_comments", 0) or 0),
                author=d.get("author", ""),
                selftext=(d.get("selftext", "") or "")[:4000],
                is_self=bool(d.get("is_self", False)),
                stickied=bool(d.get("stickied", False)),
            )
        )
    return out, data.get("data", {}).get("after")


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("subreddits", help="comma-separated, e.g. openclaw,clawd,moltbot")
    ap.add_argument("--t", default="month", choices=["day", "week", "month", "year", "all"], help="time window")
    ap.add_argument("--limit", type=int, default=200, help="max posts per subreddit")
    ap.add_argument("--sleep", type=float, default=0.8, help="sleep between requests (seconds)")
    ap.add_argument("--out", default="/home/alex/clawd/notes/reddit-deep-dive-month.md")
    args = ap.parse_args()

    subs = [s.strip().lstrip("r/") for s in args.subreddits.split(",") if s.strip()]
    out_path = Path(args.out)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    cache_path = out_path.with_suffix(".jsonl")

    all_posts: list[Post] = []

    for sub in subs:
        got: list[Post] = []
        after: str | None = None
        while len(got) < args.limit:
            batch, after = fetch_top(sub, args.t, args.limit - len(got), after)
            got.extend(batch)
            if not after or not batch:
                break
            time.sleep(args.sleep)
        # drop stickies + dup ids
        seen: set[str] = set()
        cleaned: list[Post] = []
        for p in got:
            if p.stickied:
                continue
            if p.id in seen:
                continue
            seen.add(p.id)
            cleaned.append(p)
        all_posts.extend(cleaned)

    # Write cache
    with cache_path.open("w", encoding="utf-8") as f:
        for p in all_posts:
            f.write(json.dumps(p.__dict__, ensure_ascii=False) + "\n")

    # Markdown report
    all_posts.sort(key=lambda p: (p.subreddit, -p.score, -p.num_comments))
    by_sub: dict[str, list[Post]] = {}
    for p in all_posts:
        by_sub.setdefault(p.subreddit, []).append(p)

    md: list[str] = []
    md.append(f"# Reddit deep dive (top/{args.t})\n")
    md.append(f"Subreddits: {', '.join('r/'+s for s in subs)}\n")
    md.append(f"Generated: {time.strftime('%Y-%m-%d %H:%M %Z')}\n")

    for sub in subs:
        posts = by_sub.get(sub, [])
        md.append(f"\n## r/{sub} (n={len(posts)})\n")
        for p in posts[:50]:
            md.append(
                f"- **{p.score}** (💬 {p.num_comments}) — {p.title}\n"
                f"  - link: {p.permalink}\n"
            )

    out_path.write_text("\n".join(md), encoding="utf-8")

    print(str(out_path))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())