ca9b510922
- Reset master to upstream/main (16,697 commits) - Overlay 2,271 local-only files (skills, tools, workspace, configs, apps) - Restore IDENTITY.md and USER.md templates - Build verified, gateway running, Discord working Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
102 lines
2.7 KiB
Python
102 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Fetch recent posts from selected subreddits (no auth) and print a compact summary.
|
|
|
|
This is meant as a lightweight "researcher" helper for OpenClaw.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
from dataclasses import dataclass
|
|
|
|
|
|
SUBREDDITS_DEFAULT = ["openclaw", "clawd", "moltbot"]
|
|
|
|
|
|
@dataclass
|
|
class Post:
|
|
subreddit: str
|
|
title: str
|
|
url: str
|
|
permalink: str
|
|
created_utc: float
|
|
score: int
|
|
num_comments: int
|
|
is_self: bool
|
|
stickied: bool
|
|
|
|
|
|
def fetch_json(url: str) -> dict:
|
|
req = urllib.request.Request(
|
|
url,
|
|
headers={
|
|
"User-Agent": "openclaw-reddit-researcher/1.0 (no-auth; json feed)"
|
|
},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=20) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
|
|
|
|
def fetch_new(subreddit: str, limit: int = 10) -> list[Post]:
|
|
data = fetch_json(f"https://www.reddit.com/r/{subreddit}/new/.json?limit={limit}")
|
|
out: list[Post] = []
|
|
for child in data.get("data", {}).get("children", []):
|
|
d = child.get("data", {})
|
|
out.append(
|
|
Post(
|
|
subreddit=subreddit,
|
|
title=d.get("title", ""),
|
|
url=d.get("url", ""),
|
|
permalink="https://www.reddit.com" + d.get("permalink", ""),
|
|
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),
|
|
is_self=bool(d.get("is_self", False)),
|
|
stickied=bool(d.get("stickied", False)),
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
subs = SUBREDDITS_DEFAULT
|
|
if len(argv) > 1:
|
|
subs = [s.strip().lstrip("r/") for s in argv[1].split(",") if s.strip()]
|
|
|
|
now = time.time()
|
|
window_hours = 36
|
|
|
|
posts: list[Post] = []
|
|
for s in subs:
|
|
try:
|
|
posts.extend(fetch_new(s, limit=10))
|
|
except Exception as e:
|
|
print(f"[warn] failed r/{s}: {e}", file=sys.stderr)
|
|
|
|
# Filter out old + stickied
|
|
posts = [
|
|
p
|
|
for p in posts
|
|
if not p.stickied and (now - p.created_utc) <= window_hours * 3600
|
|
]
|
|
|
|
posts.sort(key=lambda p: p.created_utc, reverse=True)
|
|
|
|
print(f"Reddit researcher (last ~{window_hours}h)\n")
|
|
for p in posts[:30]:
|
|
age_h = (now - p.created_utc) / 3600
|
|
print(
|
|
f"- r/{p.subreddit}: {p.title}\n"
|
|
f" link: {p.permalink}\n"
|
|
f" age: {age_h:.1f}h | score: {p.score} | comments: {p.num_comments}\n"
|
|
)
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv))
|