#!/usr/bin/env python3 """ Aria Knowledge Pipeline (AKP) — Reddit-Connector Stage 1 (KAR-81, Plan B Cookie-Mode) Reddit API-Approval wurde 12.05.2026 abgelehnt. Plan B: Browser-Cookie-Auth gegen reddit.com/r//new.json — funktioniert solange Cookies gültig sind. Liest Subreddit-Liste aus brain/05-Referenzen/reddit-subreddit-subscriptions.md, pullt JSON-Endpoints mit Browser-Cookies aus /root/aria/reddit-cookies.txt (Netscape format wie yt-dlp). Filtert + speichert wie YouTube/RSS. """ from __future__ import annotations import sys as _sys _sys.path.insert(0, "/root/aria/lib") from aria_logging import get_logger as _get_logger _log = _get_logger("aria-akp-reddit-ingest") import http.cookiejar, json, os, re, sqlite3, sys, time, urllib.request, yaml from datetime import datetime, timezone, timedelta from pathlib import Path CONFIG_PATH = Path("/root/aria/brain/youtube/.config.yaml") SUB_LIST_PATH = Path("/root/aria/brain/05-Referenzen/reddit-subreddit-subscriptions.md") REDDIT_RAW_DIR = Path("/root/aria/brain/akp/raw/reddit") COOKIES_PATH = Path("/root/aria/reddit-cookies.txt") def load_config(): with CONFIG_PATH.open() as f: return yaml.safe_load(f) def db_conn(db_path): con = sqlite3.connect(db_path); con.row_factory = sqlite3.Row return con def parse_subs(memory_path): out = [] sub_re = re.compile(r"^-\s+r/(\w+)") if not memory_path.exists(): return [] for line in memory_path.read_text().splitlines(): m = sub_re.match(line.strip()) if m: out.append(m.group(1)) seen = set(); deduped = [] for s in out: if s in seen: continue seen.add(s); deduped.append(s) return deduped def build_opener(cookies_path: Path) -> urllib.request.OpenerDirector: """Build urllib opener with cookies + browser-like UA.""" jar = http.cookiejar.MozillaCookieJar() if cookies_path.exists(): try: jar.load(str(cookies_path), ignore_discard=True, ignore_expires=True) except Exception as e: print(f"[akp-reddit] cookie load failed: {e}", file=sys.stderr) handler = urllib.request.HTTPCookieProcessor(jar) opener = urllib.request.build_opener(handler) opener.addheaders = [ ("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15"), ("Accept", "application/json,text/plain,*/*"), ] return opener def fetch_subreddit(opener, sub: str, limit: int = 25) -> list[dict] | None: """GET reddit.com/r//new.json with cookies. Returns post-list or None.""" url = f"https://www.reddit.com/r/{sub}/new.json?limit={limit}&raw_json=1" try: with opener.open(url, timeout=20) as r: data = json.load(r) except Exception as e: print(f"[akp-reddit] r/{sub} fetch error: {e}", file=sys.stderr) return None children = data.get("data", {}).get("children", []) return [c["data"] for c in children if c.get("data")] def fetch_post_comments(opener, sub: str, post_id: str, top_n: int = 5) -> list[dict]: """Pull top-N comments for a post via reddit.com/r//comments/.json.""" url = f"https://www.reddit.com/r/{sub}/comments/{post_id}.json?limit={top_n}&sort=top&raw_json=1" try: with opener.open(url, timeout=20) as r: data = json.load(r) except Exception: return [] if not isinstance(data, list) or len(data) < 2: return [] comments = data[1].get("data", {}).get("children", []) out = [] for c in comments[:top_n]: if c.get("kind") != "t1": continue d = c.get("data", {}) body = d.get("body", "") if body and body != "[deleted]": out.append({"score": d.get("score", 0), "body": body}) return out def already_ingested(con, pid): return con.execute("SELECT 1 FROM ingested WHERE video_id=?", (pid,)).fetchone() is not None def cheap_filter(post, cfg, content) -> str | None: if post.get("score", 0) < 50: return "low_score" if post.get("num_comments", 0) < 10: return "low_comments" if not post.get("is_self") and not any(d in (post.get("url") or "") for d in ("arxiv.org", "github.com", "blog", "substack", "medium")): return "link_post_no_content_domain" if len(content.strip()) < 300: return "short_content" created = datetime.fromtimestamp(post.get("created_utc", 0), tz=timezone.utc) if datetime.now(timezone.utc) - created > timedelta(hours=72): return "too_old" blocklist = cfg["ingest"]["title_blocklist_regex"] if blocklist and re.search(blocklist, post.get("title", ""), re.IGNORECASE): return "title_blocklist" return None def build_content(post, comments): parts = [] if post.get("selftext"): parts.append(post["selftext"].strip()) if comments: parts.append("\n\n--- Top Comments ---\n") for c in comments: parts.append(f"[{c['score']} upvotes] {c['body'][:1200]}") return "\n\n".join(parts) def write_raw(post, sub, content): out_dir = REDDIT_RAW_DIR / sub out_dir.mkdir(parents=True, exist_ok=True) pid = post["id"] out_path = out_dir / f"{pid}.json" payload = { "video_id": f"reddit_{pid}", "channel_slug": sub, "channel_handle": f"r/{sub}", "title": post.get("title", ""), "url": f"https://www.reddit.com{post.get('permalink','')}", "external_url": post.get("url") if not post.get("is_self") else None, "upload_date": datetime.fromtimestamp(post.get("created_utc", 0)).strftime("%Y%m%d"), "published_at": datetime.fromtimestamp(post.get("created_utc", 0), tz=timezone.utc).isoformat(), "transcript": content, "transcript_chars": len(content), "ingested_at": datetime.now(timezone.utc).isoformat(), "source": "reddit", "yt_meta": { "score": post.get("score"), "upvote_ratio": post.get("upvote_ratio"), "num_comments": post.get("num_comments"), "author": post.get("author"), "is_self": post.get("is_self"), }, } out_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False)) return str(out_path) def store_ingested(con, pid, sub, title, url, raw_path, chars, pub_iso): con.execute( "INSERT OR REPLACE INTO ingested (video_id, channel, channel_handle, title, " "duration_seconds, published_at, url, transcript_chars, raw_path, source) " "VALUES (?,?,?,?,?,?,?,?,?,'reddit')", (f"reddit_{pid}", sub, f"r/{sub}", title[:300], None, pub_iso, url, chars, raw_path), ) con.commit() def store_skipped(con, pid, sub, title, url, reason): con.execute( "INSERT OR IGNORE INTO ingested (video_id, channel, title, url, skipped_reason, source) VALUES (?,?,?,?,?,'reddit')", (f"reddit_{pid}", sub, title[:200], url, reason), ) con.commit() def log_run_start(con): cur = con.execute("INSERT INTO run_log (stage, status) VALUES ('ingest-reddit', 'running')") con.commit() return cur.lastrowid def log_run_finish(con, run_id, processed, skipped, failed, notes=""): con.execute( "UPDATE run_log SET finished_at=datetime('now'), status='completed', " "items_processed=?, items_skipped=?, items_failed=?, notes=? WHERE id=?", (processed, skipped, failed, notes, run_id), ) con.commit() def main(argv): cfg = load_config() con = db_conn(cfg["paths"]["state_db"]) subs = parse_subs(SUB_LIST_PATH) if not subs: print("[akp-reddit] no subs"); return 0 if not COOKIES_PATH.exists(): print(f"[akp-reddit] {COOKIES_PATH} missing. Browser-Cookies needed (Plan B since Reddit-API rejected).", file=sys.stderr) return 0 opener = build_opener(COOKIES_PATH) limit = 25 if "--backfill" in argv: limit = 50 run_id = log_run_start(con) processed = 0; skipped = 0; failed = 0 notes_lines = [] for sub in subs: posts = fetch_subreddit(opener, sub, limit) if posts is None: notes_lines.append(f"r/{sub}: fetch_error") failed += 1 time.sleep(1) continue for post in posts: pid = post.get("id") if not pid: continue full_id = f"reddit_{pid}" if already_ingested(con, full_id): continue # Fetch comments (rate-limit-friendly: only for non-skipped posts ideally; # we'll do quick filter first via score/comments/age, then fetch comments) quick_reason = cheap_filter(post, cfg, post.get("selftext") or "x" * 400) if quick_reason and quick_reason in ("low_score","low_comments","too_old","title_blocklist","link_post_no_content_domain"): store_skipped(con, pid, sub, post.get("title",""), f"https://www.reddit.com{post.get('permalink','')}", quick_reason) skipped += 1 continue comments = fetch_post_comments(opener, sub, pid) content = build_content(post, comments) reason = cheap_filter(post, cfg, content) if reason: store_skipped(con, pid, sub, post.get("title",""), f"https://www.reddit.com{post.get('permalink','')}", reason) skipped += 1 continue try: raw_path = write_raw(post, sub, content) except Exception as e: failed += 1 notes_lines.append(f"r/{sub}/{pid}: write_err: {str(e)[:60]}") continue pub_iso = datetime.fromtimestamp(post.get("created_utc",0), tz=timezone.utc).isoformat() store_ingested(con, pid, sub, post.get("title",""), f"https://www.reddit.com{post.get('permalink','')}", raw_path, len(content), pub_iso) processed += 1 time.sleep(0.6) # rate-limit friendly time.sleep(1) log_run_finish(con, run_id, processed, skipped, failed, "; ".join(notes_lines)[:500]) print(f"[akp-reddit] processed={processed} skipped={skipped} failed={failed} subs={len(subs)}") if notes_lines: for n in notes_lines[:5]: print(f" NOTE: {n}") con.close() return 0 if __name__ == "__main__": _log.event("script_start") sys.exit(main(sys.argv[1:]))