| 1 | """Keyless Reddit pipeline: free discovery + comment enrichment. |
| 2 | |
| 3 | ``search.json`` is permanently 403/429 keyless, so it is not used. Discovery |
| 4 | runs on the surfaces that still serve data without a key, then enrichment runs |
| 5 | on whatever was discovered: |
| 6 | |
| 7 | Dedicated lane entity-home subreddits (e.g. r/Kanye) pulled in full via the |
| 8 | shreddit listing partials (top+hot+new, real scores), kept |
| 9 | whole — floor-exempt — because the sub IS the topic. |
| 10 | RSS lane reddit_rss breadth (incl. global keyword search) + broad-sub |
| 11 | listing partials for real upvote scores. Relevance-floored. |
| 12 | Enrichment shreddit comment + count enrichment (reddit_shreddit) for the |
| 13 | top-ranked posts (author + score + text + permalink). |
| 14 | |
| 15 | Returns ``[]`` (never raises) so ``pipeline.py`` can fall through to the |
| 16 | ScrapeCreators backup when every keyless lane comes up empty. |
| 17 | """ |
| 18 | |
| 19 | import concurrent.futures |
| 20 | import math |
| 21 | import sys |
| 22 | from concurrent.futures import ThreadPoolExecutor |
| 23 | from typing import Any, Dict, List, Optional |
| 24 | |
| 25 | from collections import Counter |
| 26 | |
| 27 | from . import http |
| 28 | from . import reddit_rss, reddit_shreddit, reddit_listing, reddit_arctic |
| 29 | # Scores are backfilled from popular derived subreddits, so an engagement-first |
| 30 | # final sort buries on-topic RSS hits under viral off-topic posts. A relevance |
| 31 | # floor + relevance-first final ranking keeps the section on-topic. Thresholds |
| 32 | # are shared with the keyed path (reddit.py) via relevance.py. |
| 33 | from .relevance import RELEVANCE_FLOOR, MIN_ON_TOPIC |
| 34 | |
| 35 | ENRICH_LIMITS = reddit_shreddit.ENRICH_LIMITS |
| 36 | ENRICH_BUDGET = 45 # seconds total across all enrichment threads |
| 37 | MAX_ENRICH_WORKERS = 4 |
| 38 | MAX_DERIVED_SUBS = 5 # subreddits derived from RSS results for score backfill |
| 39 | # Dedicated subreddits (the entity's home, e.g. r/Kanye for "Kanye West") are |
| 40 | # wholly on-topic, so pull top+hot+new — the top-of-month listing alone misses |
| 41 | # fresh threads — and keep every item (floor-exempt). |
| 42 | DEDICATED_SORTS = ["top", "hot", "new"] |
| 43 | |
| 44 | |
| 45 | def _relevance_rank_key(post: Dict[str, Any]) -> float: |
| 46 | """Rank by relevance first, with a bounded engagement bonus as tiebreaker. |
| 47 | |
| 48 | Mirrors reddit.py: the log-scaled bonus (capped at 0.25) orders |
| 49 | similarly-relevant posts by discussion volume but is too small to lift an |
| 50 | off-topic post (relevance ~0) above an on-topic one. |
| 51 | """ |
| 52 | eng = post.get("engagement", {}) |
| 53 | total = (eng.get("score", 0) or 0) + (eng.get("num_comments", 0) or 0) |
| 54 | return (post.get("relevance") or 0.0) + min(0.25, math.log10(max(0, total) + 1) / 20.0) |
| 55 | |
| 56 | |
| 57 | def _log(msg: str) -> None: |
| 58 | sys.stderr.write(f"[RedditKeyless] {msg}\n") |
| 59 | sys.stderr.flush() |
| 60 | |
| 61 | |
| 62 | def _top_subreddits(posts: List[Dict[str, Any]], limit: int = MAX_DERIVED_SUBS) -> List[str]: |
| 63 | """Most frequent subreddits across discovered posts (for score backfill).""" |
| 64 | counts = Counter(p.get("subreddit", "") for p in posts if p.get("subreddit")) |
| 65 | return [sub for sub, _ in counts.most_common(limit)] |
| 66 | |
| 67 | |
| 68 | def _apply_scores(post: Dict[str, Any], scored: Dict[str, int]) -> None: |
| 69 | post["score"] = scored["score"] |
| 70 | post["num_comments"] = scored["num_comments"] |
| 71 | post.setdefault("engagement", {})["score"] = scored["score"] |
| 72 | post["engagement"]["num_comments"] = scored["num_comments"] |
| 73 | |
| 74 | |
| 75 | def _scored_listings( |
| 76 | subreddits: List[str], |
| 77 | depth: str = "default", |
| 78 | query: str = "", |
| 79 | sorts: Optional[List[str]] = None, |
| 80 | ) -> List[Dict[str, Any]]: |
| 81 | """Scored subreddit listings: shreddit partials, arctic-shift supplement. |
| 82 | |
| 83 | The shreddit ``community-more-posts`` partials 403 from datacenter IPs |
| 84 | (and any host Reddit decides to block). Shreddit is tried first; arctic- |
| 85 | shift supplements with any posts shreddit missed. Individual sort lanes |
| 86 | can fail silently (shreddit's ``fetch_listings`` flattens results without |
| 87 | exposing per-sort status), so arctic is called for all requested subreddits |
| 88 | and merged via deduplication. This ensures fresh posts sought through |
| 89 | ``hot`` or ``new`` are recovered even when only ``top`` succeeded. Never |
| 90 | raises. |
| 91 | """ |
| 92 | posts = reddit_listing.fetch_listings(subreddits, depth=depth, query=query, sorts=sorts) |
| 93 | |
| 94 | # Supplement with arctic for all requested subreddits. Shreddit's per-sort |
| 95 | # success/failure is opaque, so arctic provides coverage for any failed |
| 96 | # sort lanes (e.g., hot/new failing while top succeeded). Deduplication |
| 97 | # ensures no redundant posts when shreddit fully succeeded. |
| 98 | if subreddits: |
| 99 | try: |
| 100 | arctic_posts = reddit_arctic.fetch_listings( |
| 101 | subreddits, depth=depth, query=query, sorts=sorts |
| 102 | ) |
| 103 | except Exception as exc: # the fallback must never break the pipeline |
| 104 | _log(f"arctic-shift listing supplement failed: {exc}") |
| 105 | arctic_posts = [] |
| 106 | if arctic_posts: |
| 107 | # Merge and dedupe by URL — shreddit posts take priority. |
| 108 | seen = {p["url"] for p in posts} |
| 109 | added = 0 |
| 110 | for p in arctic_posts: |
| 111 | if p["url"] not in seen: |
| 112 | seen.add(p["url"]) |
| 113 | posts.append(p) |
| 114 | added += 1 |
| 115 | if added: |
| 116 | _log(f"arctic-shift supplement: {added} new posts from {len(arctic_posts)} arctic results") |
| 117 | return posts |
| 118 | |
| 119 | |
| 120 | def _discover( |
| 121 | topic: str, |
| 122 | depth: str, |
| 123 | subreddits: Optional[List[str]], |
| 124 | dedicated_subreddits: Optional[List[str]] = None, |
| 125 | ) -> List[Dict[str, Any]]: |
| 126 | # Dedicated lane: the entity's home subs are wholly on-topic. Pull |
| 127 | # top+hot+new (real scores from the listing) and mark them floor-exempt so |
| 128 | # an on-topic post whose title lacks the entity name is never dropped. |
| 129 | dedicated_posts: List[Dict[str, Any]] = [] |
| 130 | if dedicated_subreddits: |
| 131 | dedicated_posts = _scored_listings( |
| 132 | dedicated_subreddits, depth=depth, query=topic, sorts=DEDICATED_SORTS |
| 133 | ) |
| 134 | for p in dedicated_posts: |
| 135 | p["dedicated"] = True |
| 136 | _log(f"Dedicated lane: {len(dedicated_posts)} posts from {dedicated_subreddits}") |
| 137 | |
| 138 | # search.json is permanently 403/429 keyless (no Tier 0). Discovery is RSS |
| 139 | # breadth (incl. global keyword search) + broad-sub listing partials for |
| 140 | # real upvote scores. |
| 141 | rss_posts = reddit_rss.search_rss(topic, depth=depth, subreddits=subreddits) |
| 142 | |
| 143 | if subreddits: |
| 144 | # Targeted run: the caller chose these subreddits, so their listing cards |
| 145 | # are on-topic — include them as scored discovery AND as a score source. |
| 146 | listing_posts = _scored_listings(subreddits, depth=depth, query=topic) |
| 147 | score_source = listing_posts |
| 148 | else: |
| 149 | # Bare global run: subreddits derived from noisy RSS results are NOT |
| 150 | # reliably on-topic, so their listings are used ONLY to backfill scores |
| 151 | # onto the keyword-matched RSS posts — never merged as discovery, which |
| 152 | # would flood results with high-upvote but irrelevant posts. |
| 153 | listing_posts = [] |
| 154 | derived = _top_subreddits(rss_posts) |
| 155 | score_source = _scored_listings(derived, depth=depth, query=topic) |
| 156 | _log( |
| 157 | f"Tier 1 (RSS) {len(rss_posts)} posts; " |
| 158 | f"{'listing discovery ' + str(len(listing_posts)) if subreddits else 'score-only'}; " |
| 159 | f"{len(score_source)} scored cards" |
| 160 | ) |
| 161 | |
| 162 | # Score lookup by post id, from the scored listing cards. |
| 163 | score_map: Dict[str, Dict[str, int]] = {} |
| 164 | for p in score_source: |
| 165 | pid = p.get("metadata", {}).get("post_id", "") |
| 166 | if pid: |
| 167 | score_map[pid] = {"score": p["score"], "num_comments": p["num_comments"]} |
| 168 | |
| 169 | # Merge: dedicated-sub posts first (floor-exempt), then scored broad listing |
| 170 | # posts (targeted only), then RSS breadth backfilled with real scores where |
| 171 | # the post appears in a listing. First writer wins the dedupe, so a thread |
| 172 | # in both the dedicated lane and a listing keeps its floor-exempt status. |
| 173 | merged: List[Dict[str, Any]] = [] |
| 174 | seen: set = set() |
| 175 | for p in dedicated_posts + listing_posts: |
| 176 | if p["url"] not in seen: |
| 177 | seen.add(p["url"]) |
| 178 | merged.append(p) |
| 179 | for p in rss_posts: |
| 180 | if p["url"] in seen: |
| 181 | continue |
| 182 | pid = reddit_listing._post_id(p["url"]) |
| 183 | if pid in score_map: |
| 184 | _apply_scores(p, score_map[pid]) |
| 185 | seen.add(p["url"]) |
| 186 | merged.append(p) |
| 187 | |
| 188 | # Backfill scores for RSS-only posts (no listing card scored them) from the |
| 189 | # free arctic-shift archive. Posts already scored by a listing keep that |
| 190 | # live score; arctic only fills the gap, and is best-effort (never raises). |
| 191 | need = [pid for p in merged |
| 192 | if not (p.get("engagement", {}).get("score")) |
| 193 | for pid in [reddit_listing._post_id(p["url"])] if pid] |
| 194 | if need: |
| 195 | scores = reddit_arctic.fetch_scores(need) |
| 196 | filled = 0 |
| 197 | for p in merged: |
| 198 | if p.get("engagement", {}).get("score"): |
| 199 | continue |
| 200 | pid = reddit_listing._post_id(p["url"]) |
| 201 | if pid in scores: |
| 202 | _apply_scores(p, scores[pid]) |
| 203 | filled += 1 |
| 204 | if filled: |
| 205 | _log(f"arctic-shift backfilled {filled} post scores") |
| 206 | return merged |
| 207 | |
| 208 | |
| 209 | def _enrich_one(post: Dict[str, Any]) -> Dict[str, Any]: |
| 210 | """Attach shreddit comments + real comment count. Never raises.""" |
| 211 | try: |
| 212 | data = reddit_shreddit.fetch_comments(post.get("url", "")) |
| 213 | if data.get("top_comments"): |
| 214 | post["top_comments"] = data["top_comments"] |
| 215 | if data.get("comment_insights"): |
| 216 | post["comment_insights"] = data["comment_insights"] |
| 217 | num = data.get("num_comments") |
| 218 | if num is not None: |
| 219 | post["num_comments"] = num |
| 220 | post.setdefault("engagement", {})["num_comments"] = num |
| 221 | except Exception: |
| 222 | pass # keep the post with whatever discovery gave us |
| 223 | return post |
| 224 | |
| 225 | |
| 226 | def _enrich(posts: List[Dict[str, Any]], depth: str) -> List[Dict[str, Any]]: |
| 227 | """Enrich the top N posts with comments under a total time budget.""" |
| 228 | limit = ENRICH_LIMITS.get(depth, ENRICH_LIMITS["default"]) |
| 229 | to_enrich = posts[:limit] |
| 230 | rest = posts[limit:] |
| 231 | if not to_enrich: |
| 232 | return posts |
| 233 | |
| 234 | result_map: Dict[int, Dict[str, Any]] = {} |
| 235 | try: |
| 236 | with ThreadPoolExecutor(max_workers=min(limit, MAX_ENRICH_WORKERS)) as executor: |
| 237 | futures = { |
| 238 | http.submit_with_context(executor, _enrich_one, post): i |
| 239 | for i, post in enumerate(to_enrich) |
| 240 | } |
| 241 | # The budget covers the fetches; the allowance covers the shared |
| 242 | # bucket's queue (other lanes, other entities in compare mode). |
| 243 | done, not_done = concurrent.futures.wait( |
| 244 | futures, |
| 245 | timeout=ENRICH_BUDGET + http.reddit_keyless_wait_allowance(len(to_enrich)), |
| 246 | ) |
| 247 | for future in done: |
| 248 | idx = futures[future] |
| 249 | try: |
| 250 | result_map[idx] = future.result(timeout=0) |
| 251 | except Exception: |
| 252 | result_map[idx] = to_enrich[idx] |
| 253 | for future in not_done: |
| 254 | idx = futures[future] |
| 255 | result_map[idx] = to_enrich[idx] |
| 256 | future.cancel() |
| 257 | enriched = [result_map[i] for i in range(len(to_enrich))] |
| 258 | except Exception: |
| 259 | enriched = to_enrich |
| 260 | |
| 261 | return enriched + rest |
| 262 | |
| 263 | |
| 264 | def _by_comments(posts: List[Dict[str, Any]]) -> List[Dict[str, Any]]: |
| 265 | """Stable-sort posts by comment count descending for enrichment slots. |
| 266 | |
| 267 | Ties (equal counts, including unknown counts treated as 0) preserve the |
| 268 | incoming order, which search_and_enrich's provisional score-first sort |
| 269 | establishes. Mirrors _relevance_rank_key's `or 0` guard so a present-but- |
| 270 | None count is treated as 0 rather than raising. |
| 271 | """ |
| 272 | def _comment_count(post: Dict[str, Any]) -> int: |
| 273 | eng = post.get("engagement") or {} |
| 274 | return eng.get("num_comments") or post.get("num_comments") or 0 |
| 275 | |
| 276 | return sorted(posts, key=_comment_count, reverse=True) |
| 277 | |
| 278 | |
| 279 | def _slot_priority(topic: str, posts: List[Dict[str, Any]]) -> List[Dict[str, Any]]: |
| 280 | """Order posts for enrichment slots: entity-matching posts first. |
| 281 | |
| 282 | Comment slots (ENRICH_LIMITS) are scarce; spending them on high-upvote |
| 283 | posts that rerank later demotes as entity misses starves the on-topic |
| 284 | posts the user actually sees (2026-06-06 "OpenClaw vs Hermes" run: |
| 285 | 2,000+ upvote Gemma/GPU threads took every slot, then were demoted to |
| 286 | zero). Mirror rerank's demotion signal via the shared `_entity_grounded` |
| 287 | check (head token of the topic's stripped primary entity present in the |
| 288 | post text) so slots go to posts likely to survive final ranking — keying |
| 289 | on the same head token keeps the two paths from diverging. Falls back to |
| 290 | token-overlap relevance when the topic yields no usable primary entity. |
| 291 | Within each tier posts are ordered by comment count descending (stable: |
| 292 | equal or unknown counts preserve the incoming score-first order), so the |
| 293 | scarce slots go to the threads with the most discussion rather than to |
| 294 | near-empty threads that merely ranked higher by score. Never raises; on |
| 295 | any failure the incoming order is returned unchanged. |
| 296 | """ |
| 297 | try: |
| 298 | from . import relevance, rerank |
| 299 | |
| 300 | def _post_text(post: Dict[str, Any]) -> str: |
| 301 | return f"{post.get('title') or ''} {post.get('selftext') or ''}" |
| 302 | |
| 303 | entity = rerank._primary_entity(topic).lower() |
| 304 | if entity: |
| 305 | def _matches(post: Dict[str, Any]) -> bool: |
| 306 | return rerank._entity_grounded(_post_text(post), entity) |
| 307 | else: |
| 308 | prepared = relevance.PreparedQuery(topic) |
| 309 | |
| 310 | def _matches(post: Dict[str, Any]) -> bool: |
| 311 | return relevance.token_overlap_relevance(prepared, _post_text(post)) > 0.24 |
| 312 | |
| 313 | matches: List[Dict[str, Any]] = [] |
| 314 | misses: List[Dict[str, Any]] = [] |
| 315 | for post in posts: |
| 316 | (matches if _matches(post) else misses).append(post) |
| 317 | return _by_comments(matches) + _by_comments(misses) |
| 318 | except Exception: |
| 319 | return posts |
| 320 | |
| 321 | |
| 322 | def search_and_enrich( |
| 323 | topic: str, |
| 324 | from_date: str, |
| 325 | to_date: str, |
| 326 | depth: str = "default", |
| 327 | subreddits: Optional[List[str]] = None, |
| 328 | dedicated_subreddits: Optional[List[str]] = None, |
| 329 | ) -> List[Dict[str, Any]]: |
| 330 | """Full keyless Reddit pipeline: discover then enrich. |
| 331 | |
| 332 | Args: |
| 333 | topic: Search topic |
| 334 | from_date: Start date (YYYY-MM-DD) |
| 335 | to_date: End date (YYYY-MM-DD) |
| 336 | depth: 'quick', 'default', or 'deep' |
| 337 | subreddits: Optional pre-resolved broad/category subreddit names (no r/) |
| 338 | dedicated_subreddits: Optional entity-home subreddit names (no r/) pulled |
| 339 | in full (top+hot+new) and exempt from the relevance floor. |
| 340 | |
| 341 | Returns: |
| 342 | List of normalized item dicts matching the reddit_public output shape, |
| 343 | with top_comments/comment_insights attached on enriched posts. |
| 344 | Empty list when all keyless tiers fail (so SC backup can engage). |
| 345 | """ |
| 346 | posts = _discover(topic, depth, subreddits, dedicated_subreddits) |
| 347 | if not posts: |
| 348 | return [] |
| 349 | |
| 350 | # Date filter: keep posts in range or with unknown dates (mirrors reddit_public). |
| 351 | posts = [ |
| 352 | p for p in posts |
| 353 | if p.get("date") is None or (from_date <= p["date"] <= to_date) |
| 354 | ] |
| 355 | |
| 356 | # Relevance floor: strip zero-overlap posts (relevance exactly 0 = no |
| 357 | # title/body token match at all) when anything relevant remains, so |
| 358 | # backfilled high-upvote posts from popular subs can't bury on-topic RSS |
| 359 | # hits. Keep all only when nothing scored above zero. |
| 360 | before = len(posts) |
| 361 | # Dedicated-sub posts are floor-exempt: their whole subreddit is the topic, |
| 362 | # so an on-topic post whose title lacks the entity name must not be dropped. |
| 363 | on_topic = [p for p in posts if p.get("dedicated") or (p.get("relevance") or 0) >= RELEVANCE_FLOOR] |
| 364 | if len(on_topic) >= MIN_ON_TOPIC: |
| 365 | posts = on_topic |
| 366 | else: |
| 367 | nonzero = [p for p in posts if p.get("dedicated") or (p.get("relevance") or 0) > 0] |
| 368 | if nonzero: |
| 369 | posts = nonzero |
| 370 | if len(posts) < before: |
| 371 | _log(f"Relevance floor dropped {before - len(posts)} off-topic posts") |
| 372 | |
| 373 | # Provisional score-first order so enrichment-slot selection has a stable |
| 374 | # within-tier tiebreak order to preserve: within each entity tier, slots go |
| 375 | # to the most-commented threads first, and equal counts keep score order. |
| 376 | posts.sort( |
| 377 | key=lambda p: ( |
| 378 | p.get("engagement", {}).get("score", 0) or 0, |
| 379 | p.get("relevance", 0) or 0, |
| 380 | p.get("date") or "", |
| 381 | ), |
| 382 | reverse=True, |
| 383 | ) |
| 384 | |
| 385 | # Enrichment slot selection is comment-aware within entity tiers: |
| 386 | # entity-matching posts claim the scarce comment slots first, and within |
| 387 | # each tier the most-commented threads get slots first (score order is the |
| 388 | # stable tiebreak for equal counts). |
| 389 | posts = _enrich(_slot_priority(topic, posts), depth) |
| 390 | |
| 391 | # Final display order ranks relevance-first with a bounded engagement bonus, |
| 392 | # so an off-topic high-upvote post can't outrank an on-topic one in what the |
| 393 | # user sees. Enrichment above may have backfilled real comment counts. |
| 394 | posts.sort(key=_relevance_rank_key, reverse=True) |
| 395 | |
| 396 | for i, post in enumerate(posts): |
| 397 | post["id"] = f"R{i + 1}" |
| 398 | |
| 399 | return posts |
| 400 |