| 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(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 _discover( |
| 76 | topic: str, |
| 77 | depth: str, |
| 78 | subreddits: Optional[List[str]], |
| 79 | dedicated_subreddits: Optional[List[str]] = None, |
| 80 | ) -> List[Dict[str, Any]]: |
| 81 | # Dedicated lane: the entity's home subs are wholly on-topic. Pull |
| 82 | # top+hot+new (real scores from the listing) and mark them floor-exempt so |
| 83 | # an on-topic post whose title lacks the entity name is never dropped. |
| 84 | dedicated_posts: List[Dict[str, Any]] = [] |
| 85 | if dedicated_subreddits: |
| 86 | dedicated_posts = reddit_listing.fetch_listings( |
| 87 | dedicated_subreddits, depth=depth, query=topic, sorts=DEDICATED_SORTS |
| 88 | ) |
| 89 | for p in dedicated_posts: |
| 90 | p["dedicated"] = True |
| 91 | _log(f"Dedicated lane: {len(dedicated_posts)} posts from {dedicated_subreddits}") |
| 92 | |
| 93 | # search.json is permanently 403/429 keyless (no Tier 0). Discovery is RSS |
| 94 | # breadth (incl. global keyword search) + broad-sub listing partials for |
| 95 | # real upvote scores. |
| 96 | rss_posts = reddit_rss.search_rss(topic, depth=depth, subreddits=subreddits) |
| 97 | |
| 98 | if subreddits: |
| 99 | # Targeted run: the caller chose these subreddits, so their listing cards |
| 100 | # are on-topic — include them as scored discovery AND as a score source. |
| 101 | listing_posts = reddit_listing.fetch_listings(subreddits, depth=depth, query=topic) |
| 102 | score_source = listing_posts |
| 103 | else: |
| 104 | # Bare global run: subreddits derived from noisy RSS results are NOT |
| 105 | # reliably on-topic, so their listings are used ONLY to backfill scores |
| 106 | # onto the keyword-matched RSS posts — never merged as discovery, which |
| 107 | # would flood results with high-upvote but irrelevant posts. |
| 108 | listing_posts = [] |
| 109 | derived = _top_subreddits(rss_posts) |
| 110 | score_source = reddit_listing.fetch_listings(derived, depth=depth, query=topic) |
| 111 | _log( |
| 112 | f"Tier 1 (RSS) {len(rss_posts)} posts; " |
| 113 | f"{'listing discovery ' + str(len(listing_posts)) if subreddits else 'score-only'}; " |
| 114 | f"{len(score_source)} scored cards" |
| 115 | ) |
| 116 | |
| 117 | # Score lookup by post id, from the scored listing cards. |
| 118 | score_map: Dict[str, Dict[str, int]] = {} |
| 119 | for p in score_source: |
| 120 | pid = p.get("metadata", {}).get("post_id", "") |
| 121 | if pid: |
| 122 | score_map[pid] = {"score": p["score"], "num_comments": p["num_comments"]} |
| 123 | |
| 124 | # Merge: dedicated-sub posts first (floor-exempt), then scored broad listing |
| 125 | # posts (targeted only), then RSS breadth backfilled with real scores where |
| 126 | # the post appears in a listing. First writer wins the dedupe, so a thread |
| 127 | # in both the dedicated lane and a listing keeps its floor-exempt status. |
| 128 | merged: List[Dict[str, Any]] = [] |
| 129 | seen: set = set() |
| 130 | for p in dedicated_posts + listing_posts: |
| 131 | if p["url"] not in seen: |
| 132 | seen.add(p["url"]) |
| 133 | merged.append(p) |
| 134 | for p in rss_posts: |
| 135 | if p["url"] in seen: |
| 136 | continue |
| 137 | pid = reddit_listing._post_id(p["url"]) |
| 138 | if pid in score_map: |
| 139 | _apply_scores(p, score_map[pid]) |
| 140 | seen.add(p["url"]) |
| 141 | merged.append(p) |
| 142 | |
| 143 | # Backfill scores for RSS-only posts (no listing card scored them) from the |
| 144 | # free arctic-shift archive. Posts already scored by a listing keep that |
| 145 | # live score; arctic only fills the gap, and is best-effort (never raises). |
| 146 | need = [pid for p in merged |
| 147 | if not (p.get("engagement", {}).get("score")) |
| 148 | for pid in [reddit_listing._post_id(p["url"])] if pid] |
| 149 | if need: |
| 150 | scores = reddit_arctic.fetch_scores(need) |
| 151 | filled = 0 |
| 152 | for p in merged: |
| 153 | if p.get("engagement", {}).get("score"): |
| 154 | continue |
| 155 | pid = reddit_listing._post_id(p["url"]) |
| 156 | if pid in scores: |
| 157 | _apply_scores(p, scores[pid]) |
| 158 | filled += 1 |
| 159 | if filled: |
| 160 | _log(f"arctic-shift backfilled {filled} post scores") |
| 161 | return merged |
| 162 | |
| 163 | |
| 164 | def _enrich_one(post: Dict[str, Any]) -> Dict[str, Any]: |
| 165 | """Attach shreddit comments + real comment count. Never raises.""" |
| 166 | try: |
| 167 | data = reddit_shreddit.fetch_comments(post.get("url", "")) |
| 168 | if data.get("top_comments"): |
| 169 | post["top_comments"] = data["top_comments"] |
| 170 | if data.get("comment_insights"): |
| 171 | post["comment_insights"] = data["comment_insights"] |
| 172 | num = data.get("num_comments") |
| 173 | if num is not None: |
| 174 | post["num_comments"] = num |
| 175 | post.setdefault("engagement", {})["num_comments"] = num |
| 176 | except Exception: |
| 177 | pass # keep the post with whatever discovery gave us |
| 178 | return post |
| 179 | |
| 180 | |
| 181 | def _enrich(posts: List[Dict[str, Any]], depth: str) -> List[Dict[str, Any]]: |
| 182 | """Enrich the top N posts with comments under a total time budget.""" |
| 183 | limit = ENRICH_LIMITS.get(depth, ENRICH_LIMITS["default"]) |
| 184 | to_enrich = posts[:limit] |
| 185 | rest = posts[limit:] |
| 186 | if not to_enrich: |
| 187 | return posts |
| 188 | |
| 189 | result_map: Dict[int, Dict[str, Any]] = {} |
| 190 | try: |
| 191 | with ThreadPoolExecutor(max_workers=min(limit, MAX_ENRICH_WORKERS)) as executor: |
| 192 | futures = { |
| 193 | http.submit_with_context(executor, _enrich_one, post): i |
| 194 | for i, post in enumerate(to_enrich) |
| 195 | } |
| 196 | done, not_done = concurrent.futures.wait(futures, timeout=ENRICH_BUDGET) |
| 197 | for future in done: |
| 198 | idx = futures[future] |
| 199 | try: |
| 200 | result_map[idx] = future.result(timeout=0) |
| 201 | except Exception: |
| 202 | result_map[idx] = to_enrich[idx] |
| 203 | for future in not_done: |
| 204 | idx = futures[future] |
| 205 | result_map[idx] = to_enrich[idx] |
| 206 | future.cancel() |
| 207 | enriched = [result_map[i] for i in range(len(to_enrich))] |
| 208 | except Exception: |
| 209 | enriched = to_enrich |
| 210 | |
| 211 | return enriched + rest |
| 212 | |
| 213 | |
| 214 | def _slot_priority(topic: str, posts: List[Dict[str, Any]]) -> List[Dict[str, Any]]: |
| 215 | """Order posts for enrichment slots: entity-matching posts first. |
| 216 | |
| 217 | Comment slots (ENRICH_LIMITS) are scarce; spending them on high-upvote |
| 218 | posts that rerank later demotes as entity misses starves the on-topic |
| 219 | posts the user actually sees (2026-06-06 "OpenClaw vs Hermes" run: |
| 220 | 2,000+ upvote Gemma/GPU threads took every slot, then were demoted to |
| 221 | zero). Mirror rerank's demotion signal via the shared `_entity_grounded` |
| 222 | check (head token of the topic's stripped primary entity present in the |
| 223 | post text) so slots go to posts likely to survive final ranking — keying |
| 224 | on the same head token keeps the two paths from diverging. Falls back to |
| 225 | token-overlap relevance when the topic yields no usable primary entity. |
| 226 | Within each tier the incoming |
| 227 | (score-first) order is preserved. Never raises; on any failure the |
| 228 | incoming order is returned unchanged. |
| 229 | """ |
| 230 | try: |
| 231 | from . import relevance, rerank |
| 232 | |
| 233 | def _post_text(post: Dict[str, Any]) -> str: |
| 234 | return f"{post.get('title') or ''} {post.get('selftext') or ''}" |
| 235 | |
| 236 | entity = rerank._primary_entity(topic).lower() |
| 237 | if entity: |
| 238 | def _matches(post: Dict[str, Any]) -> bool: |
| 239 | return rerank._entity_grounded(_post_text(post), entity) |
| 240 | else: |
| 241 | prepared = relevance.PreparedQuery(topic) |
| 242 | |
| 243 | def _matches(post: Dict[str, Any]) -> bool: |
| 244 | return relevance.token_overlap_relevance(prepared, _post_text(post)) > 0.24 |
| 245 | |
| 246 | matches: List[Dict[str, Any]] = [] |
| 247 | misses: List[Dict[str, Any]] = [] |
| 248 | for post in posts: |
| 249 | (matches if _matches(post) else misses).append(post) |
| 250 | return matches + misses |
| 251 | except Exception: |
| 252 | return posts |
| 253 | |
| 254 | |
| 255 | def search_and_enrich( |
| 256 | topic: str, |
| 257 | from_date: str, |
| 258 | to_date: str, |
| 259 | depth: str = "default", |
| 260 | subreddits: Optional[List[str]] = None, |
| 261 | dedicated_subreddits: Optional[List[str]] = None, |
| 262 | ) -> List[Dict[str, Any]]: |
| 263 | """Full keyless Reddit pipeline: discover then enrich. |
| 264 | |
| 265 | Args: |
| 266 | topic: Search topic |
| 267 | from_date: Start date (YYYY-MM-DD) |
| 268 | to_date: End date (YYYY-MM-DD) |
| 269 | depth: 'quick', 'default', or 'deep' |
| 270 | subreddits: Optional pre-resolved broad/category subreddit names (no r/) |
| 271 | dedicated_subreddits: Optional entity-home subreddit names (no r/) pulled |
| 272 | in full (top+hot+new) and exempt from the relevance floor. |
| 273 | |
| 274 | Returns: |
| 275 | List of normalized item dicts matching the reddit_public output shape, |
| 276 | with top_comments/comment_insights attached on enriched posts. |
| 277 | Empty list when all keyless tiers fail (so SC backup can engage). |
| 278 | """ |
| 279 | posts = _discover(topic, depth, subreddits, dedicated_subreddits) |
| 280 | if not posts: |
| 281 | return [] |
| 282 | |
| 283 | # Date filter: keep posts in range or with unknown dates (mirrors reddit_public). |
| 284 | posts = [ |
| 285 | p for p in posts |
| 286 | if p.get("date") is None or (from_date <= p["date"] <= to_date) |
| 287 | ] |
| 288 | |
| 289 | # Relevance floor: strip zero-overlap posts (relevance exactly 0 = no |
| 290 | # title/body token match at all) when anything relevant remains, so |
| 291 | # backfilled high-upvote posts from popular subs can't bury on-topic RSS |
| 292 | # hits. Keep all only when nothing scored above zero. |
| 293 | before = len(posts) |
| 294 | # Dedicated-sub posts are floor-exempt: their whole subreddit is the topic, |
| 295 | # so an on-topic post whose title lacks the entity name must not be dropped. |
| 296 | on_topic = [p for p in posts if p.get("dedicated") or (p.get("relevance") or 0) >= RELEVANCE_FLOOR] |
| 297 | if len(on_topic) >= MIN_ON_TOPIC: |
| 298 | posts = on_topic |
| 299 | else: |
| 300 | nonzero = [p for p in posts if p.get("dedicated") or (p.get("relevance") or 0) > 0] |
| 301 | if nonzero: |
| 302 | posts = nonzero |
| 303 | if len(posts) < before: |
| 304 | _log(f"Relevance floor dropped {before - len(posts)} off-topic posts") |
| 305 | |
| 306 | # Provisional score-first order so enrichment-slot selection has a stable |
| 307 | # within-tier order to preserve. |
| 308 | posts.sort( |
| 309 | key=lambda p: ( |
| 310 | p.get("engagement", {}).get("score", 0) or 0, |
| 311 | p.get("relevance", 0) or 0, |
| 312 | p.get("date") or "", |
| 313 | ), |
| 314 | reverse=True, |
| 315 | ) |
| 316 | |
| 317 | # Enrichment slot selection is relevance-aware: entity-matching posts |
| 318 | # claim the scarce comment slots first (score order preserved within |
| 319 | # each tier). |
| 320 | posts = _enrich(_slot_priority(topic, posts), depth) |
| 321 | |
| 322 | # Final display order ranks relevance-first with a bounded engagement bonus, |
| 323 | # so an off-topic high-upvote post can't outrank an on-topic one in what the |
| 324 | # user sees. Enrichment above may have backfilled real comment counts. |
| 325 | posts.sort(key=_relevance_rank_key, reverse=True) |
| 326 | |
| 327 | for i, post in enumerate(posts): |
| 328 | post["id"] = f"R{i + 1}" |
| 329 | |
| 330 | return posts |
| 331 |