Files
BSN-Chatsystem/reddit_scraper.py

270 lines
9.2 KiB
Python

#!/usr/bin/env python3
"""Scrape old.reddit.com/r/boardgames with regex-based extraction."""
import json
import re
import sys
import time
from datetime import datetime, timezone, date
from urllib.request import Request, urlopen
USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
BLOCKED_FLAIRS = {
"humor", "meme", "memes", "funny", "comc", "check out my collection",
"humour", "fluff", "just for fun", "shelfie", "collection",
"c.o.m.c",
}
BLOCKED_TITLE_TERMS = [
"meme", "shitpost", "shelfie saturday", "check out my collection",
"comc", "what did you play this week", "wdyp", "circlejerk",
]
STICKY_TITLES = [
"daily game recommendations thread",
"everything you wanted to know about board games",
]
def fetch_html(url):
req = Request(url, headers={"User-Agent": USER_AGENT})
with urlopen(req, timeout=30) as resp:
return resp.read().decode("utf-8", errors="replace")
def is_blocked(flair, title):
flair_lower = flair.lower().strip()
for blocked in BLOCKED_FLAIRS:
if blocked in flair_lower:
return True
title_lower = title.lower()
for term in BLOCKED_TITLE_TERMS:
if term in title_lower:
return True
return False
def extract_posts(html):
"""Extract posts using regex from old.reddit HTML."""
posts = []
things = re.split(r'(<div class="[^"]*?\bthing\b[^"]*?\bid-t3_[a-z0-9]+[^"]*"[^>]*>)', html)
current_id = None
current_attrs = ""
for part in things:
m = re.match(r'<div class="([^"]*?\bthing\b[^"]*?\bid-t3_([a-z0-9]+)[^"]*)"[^>]*>', part)
if m:
current_attrs = m.group(1)
current_id = m.group(2)
continue
if not current_id:
continue
post = {
"id": current_id,
"title": "",
"flair": "",
"score": 0,
"comments": 0,
"timestamp_utc": None,
"author": "",
"permalink": "",
"domain": "",
"sticky": "stickied" in current_attrs,
}
# Timestamp
ts_match = re.search(r'<time[^>]*datetime="([^"]+)"', part)
if ts_match:
try:
post["timestamp_utc"] = datetime.fromisoformat(ts_match.group(1))
except:
pass
# Score (prefer data-score, fallback to score div title)
score_match = re.search(r'data-score="(\d+)"', part)
if not score_match:
score_match = re.search(r'<div class="[^"]*score unvoted[^"]*"[^>]*title="(\d+)"', part)
if not score_match:
score_match = re.search(r'<div class="[^"]*\bscore\b[^"]*"[^>]*title="(\d+)"', part)
if score_match:
post["score"] = int(score_match.group(1))
# Comments
comments_match = re.search(r'data-comments-count="(\d+)"', part)
if not comments_match:
comments_match = re.search(r'>(\d+)\s*comments?</a>', part)
if comments_match:
post["comments"] = int(comments_match.group(1))
# Author from data-author
author_match = re.search(r'data-author="([^"]+)"', part)
if author_match and author_match.group(1) != "AutoModerator":
post["author"] = author_match.group(1)
else:
# Try from author link text
author_match2 = re.search(r'<a[^>]*class="[^"]*\bauthor\b[^"]*"[^>]*>([^<]+)</a>', part)
if author_match2:
post["author"] = author_match2.group(1)
# Domain from data-domain
domain_match = re.search(r'data-domain="([^"]+)"', part)
if domain_match:
post["domain"] = domain_match.group(1)
# Permalink
pm_match = re.search(r'data-permalink="([^"]+)"', part)
if pm_match:
post["permalink"] = "https://old.reddit.com" + pm_match.group(1)
else:
post["permalink"] = f"https://old.reddit.com/r/boardgames/comments/{current_id}/"
# Title from <a class="...title...">
title_match = re.search(r'<a class="[^"]*\btitle\b[^"]*"[^>]*>\s*(.*?)\s*</a>', part, re.DOTALL)
if title_match:
title_raw = title_match.group(1)
title_raw = title_raw.replace("&amp;", "&").replace("&lt;", "<").replace("&gt;", ">")
title_raw = title_raw.replace("&#32;", " ").replace("&quot;", '"').replace("&#39;", "'")
title_raw = re.sub(r'<[^>]+>', '', title_raw)
title_raw = " ".join(title_raw.split())
post["title"] = title_raw
# Flair
flair_match = re.search(r'<span class="[^"]*flairrichtext[^"]*"[^>]*>.*?<span[^>]*>([^<]+)</span></span>', part)
if not flair_match:
flair_match = re.search(r'<span class="[^"]*linkflairlabel[^"]*"[^>]*>([^<]+)</span>', part)
if flair_match:
post["flair"] = flair_match.group(1).strip()
# Skip sticky posts (pinned/megathreads)
if post["sticky"]:
title_lower = post["title"].lower()
for s in STICKY_TITLES:
if s in title_lower:
continue
posts.append(post)
current_id = None
return posts
def filter_posts(posts, target_dates):
filtered = []
for post in posts:
ts = post.get("timestamp_utc")
if ts is None:
continue
post_date = ts.date()
if post_date not in target_dates:
continue
if is_blocked(post.get("flair", ""), post.get("title", "")):
continue
# Skip sticky auto-mod posts
if post.get("sticky") and "daily" in post.get("title", "").lower():
continue
post["date_str"] = ts.strftime("%Y-%m-%d %H:%M UTC")
filtered.append(post)
return filtered
def main():
target_dates = {date(2026, 6, 6), date(2026, 6, 7)}
print("=== Scraping old.reddit.com/r/boardgames ===", file=sys.stderr)
urls = [
"https://old.reddit.com/r/boardgames/",
"https://old.reddit.com/r/boardgames/new/",
"https://old.reddit.com/r/boardgames/rising/",
]
all_posts = []
for url in urls:
print(f"Fetching: {url}", file=sys.stderr)
try:
html = fetch_html(url)
posts = extract_posts(html)
print(f" Extracted {len(posts)} posts", file=sys.stderr)
existing_ids = {p["id"] for p in all_posts}
for p in posts:
if p["id"] not in existing_ids:
all_posts.append(p)
existing_ids.add(p["id"])
except Exception as e:
print(f" Error: {e}", file=sys.stderr)
time.sleep(1)
print(f"\nTotal unique raw posts: {len(all_posts)}", file=sys.stderr)
# List all dates found
dates_found = set()
for p in all_posts:
if p.get("timestamp_utc"):
dates_found.add(p["timestamp_utc"].date())
print(f"Dates found: {sorted(dates_found)}", file=sys.stderr)
filtered = filter_posts(all_posts, target_dates)
print(f"After filtering: {len(filtered)} posts", file=sys.stderr)
# Sort by score desc
filtered.sort(key=lambda p: p.get("score", 0), reverse=True)
filtered = filtered[:20]
output = []
for i, post in enumerate(filtered, 1):
output.append({
"rank": i,
"title": post["title"],
"score": post["score"],
"comments": post["comments"],
"flair": post.get("flair", ""),
"author": post.get("author", ""),
"date": post.get("date_str", ""),
"url": post.get("permalink", ""),
"domain": post.get("domain", ""),
})
# Save JSON
with open("/home/hermes/workspace/boardgames_weekend_posts.json", "w") as f:
json.dump(output, f, indent=2, ensure_ascii=False)
print(json.dumps(output, indent=2, ensure_ascii=False))
# Summary
print("\n" + "="*80, file=sys.stderr)
print(f"TOP {len(output)} POSTS (Jun 6-7, 2026) from r/boardgames", file=sys.stderr)
print("="*80, file=sys.stderr)
news_posts = [p for p in output if p["flair"] == "News"]
reviews = [p for p in output if p["flair"] == "Review"]
questions = [p for p in output if "question" in p["flair"].lower() or "?" in p["title"]]
ks_posts = [p for p in output if "kickstarter" in p["title"].lower() or "crowdfunding" in p["title"].lower()]
if news_posts:
print("\n📰 NEWS:", file=sys.stderr)
for p in news_posts:
print(f"{p['title']} (🔺{p['score']}, 💬{p['comments']})", file=sys.stderr)
if reviews:
print("\n📝 REVIEWS:", file=sys.stderr)
for p in reviews:
print(f"{p['title']} (🔺{p['score']}, 💬{p['comments']})", file=sys.stderr)
print(f"\n📊 Total: {len(output)} posts", file=sys.stderr)
print(f"📅 Date range: Jun 6-7, 2026", file=sys.stderr)
print(f"📄 JSON saved to: /home/hermes/workspace/boardgames_weekend_posts.json", file=sys.stderr)
if __name__ == "__main__":
main()