| 1 | """Reusable local scoring signals for v3 pipeline stages.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import math |
| 6 | |
| 7 | from . import dates, relevance, schema |
| 8 | |
| 9 | # Editorial signal-to-noise scores. Grounding (Google Search) is 1.0 baseline; |
| 10 | # social platforms discounted for noise. |
| 11 | SOURCE_QUALITY = { |
| 12 | "xiaohongshu": 0.7, |
| 13 | "hackernews": 0.8, |
| 14 | "youtube": 0.85, |
| 15 | "digg": 0.85, |
| 16 | "arxiv": 0.9, |
| 17 | "techmeme": 0.85, |
| 18 | "trustpilot": 0.78, |
| 19 | "reddit": 0.6, |
| 20 | "x": 0.68, |
| 21 | "bluesky": 0.66, |
| 22 | "truthsocial": 0.6, |
| 23 | "polymarket": 0.5, |
| 24 | "instagram": 0.58, |
| 25 | "tiktok": 0.58, |
| 26 | "jobs": 0.72, |
| 27 | "corpus": 0.75, |
| 28 | } |
| 29 | |
| 30 | |
| 31 | def source_quality(source: str) -> float: |
| 32 | return SOURCE_QUALITY.get(source, 0.6) |
| 33 | |
| 34 | |
| 35 | def local_relevance( |
| 36 | item: schema.SourceItem, |
| 37 | ranking_query: "str | relevance.PreparedQuery", |
| 38 | ) -> float: |
| 39 | text = "\n".join( |
| 40 | part |
| 41 | for part in [item.title, item.body, item.snippet] |
| 42 | if part |
| 43 | ) |
| 44 | hashtags = item.metadata.get("hashtags") if isinstance(item.metadata, dict) else None |
| 45 | score = relevance.token_overlap_relevance(ranking_query, text, hashtags=hashtags) |
| 46 | |
| 47 | # High-engagement YouTube floor: official videos with millions of views |
| 48 | # often have titles that don't keyword-match the query (e.g., "YE - FATHER |
| 49 | # (feat. TRAVIS SCOTT)" doesn't match "kanye west"). The engagement signals |
| 50 | # say "this is important" even when text overlap is weak. |
| 51 | if item.source == "youtube" and (item.engagement.get("views") or 0) > 100_000: |
| 52 | score = max(score, 0.3) |
| 53 | |
| 54 | # Project-mode GitHub floor: items fetched via --github-repo are explicitly |
| 55 | # requested by the user and relevant by construction. Without this floor, |
| 56 | # repos with low token diversity (e.g., "openclaw/openclaw" -> 1 unique token) |
| 57 | # get pruned despite being the primary search target. |
| 58 | labels = item.metadata.get("labels", []) if isinstance(item.metadata, dict) else [] |
| 59 | if "project-mode" in labels: |
| 60 | score = max(score, 0.8) |
| 61 | |
| 62 | return score |
| 63 | |
| 64 | |
| 65 | def freshness( |
| 66 | item: schema.SourceItem, |
| 67 | freshness_mode: str = "balanced_recent", |
| 68 | *, |
| 69 | reference_date: str | None = None, |
| 70 | max_days: int = 30, |
| 71 | ) -> int: |
| 72 | score = dates.recency_score( |
| 73 | item.published_at, |
| 74 | max_days=max_days, |
| 75 | reference_date=reference_date, |
| 76 | ) |
| 77 | if freshness_mode == "strict_recent": |
| 78 | return int(score) |
| 79 | if freshness_mode == "evergreen_ok": |
| 80 | return int((score * 0.6) + 40) |
| 81 | return int((score * 0.8) + 10) |
| 82 | |
| 83 | |
| 84 | def log1p_safe(value: float | int | None) -> float: |
| 85 | if value is None: |
| 86 | return 0.0 |
| 87 | try: |
| 88 | numeric = float(value) |
| 89 | except (TypeError, ValueError): |
| 90 | return 0.0 |
| 91 | if numeric <= 0: |
| 92 | return 0.0 |
| 93 | return math.log1p(numeric) |
| 94 | |
| 95 | |
| 96 | def _top_comment_score(item: schema.SourceItem) -> float: |
| 97 | comments = item.metadata.get("top_comments") or [] |
| 98 | if not comments or not isinstance(comments[0], dict): |
| 99 | return 0.0 |
| 100 | return log1p_safe(comments[0].get("score")) |
| 101 | |
| 102 | |
| 103 | # Per-platform log-reference for normalizing a top comment's vote count into a |
| 104 | # [0,1] signal. Reddit upvotes run in the hundreds-to-thousands; YouTube/TikTok |
| 105 | # likes run 10-600x higher (and the top end is display-abbreviated: "39K" is |
| 106 | # stored as 39000). A raw or single-scale log compare would let YouTube/TikTok |
| 107 | # dominate purely by platform scale, not by being funnier. Each value is the |
| 108 | # log1p of a "very high" top-comment count for that platform, so dividing a |
| 109 | # comment's log1p(score) by it yields a comparable cross-platform strength. |
| 110 | _VOTE_LOG_REFERENCE: dict[str, float] = { |
| 111 | "reddit": 7.6, # ~log1p(2000) |
| 112 | "hackernews": 6.2, # ~log1p(500) |
| 113 | "youtube": 10.3, # ~log1p(30000) |
| 114 | "tiktok": 10.3, # ~log1p(30000) |
| 115 | "instagram": 9.2, # ~log1p(10000) |
| 116 | "x": 9.2, # ~log1p(10000) |
| 117 | "bluesky": 9.2, # ~log1p(10000); like X/IG, not the Reddit default |
| 118 | } |
| 119 | _VOTE_LOG_REFERENCE_DEFAULT = 7.6 |
| 120 | |
| 121 | |
| 122 | def normalized_comment_vote(source: str, score: "float | int | None") -> float: |
| 123 | """Normalize a single comment's vote count to [0,1] within its platform. |
| 124 | |
| 125 | Same per-platform reference as ``top_comment_vote_signal`` so a 22k-like |
| 126 | TikTok comment and a 600-upvote Reddit comment rank on a comparable scale. |
| 127 | Used to rank the cross-candidate Top Community Comments block. |
| 128 | """ |
| 129 | base = log1p_safe(score) |
| 130 | if base <= 0.0: |
| 131 | return 0.0 |
| 132 | ref = _VOTE_LOG_REFERENCE.get(source, _VOTE_LOG_REFERENCE_DEFAULT) |
| 133 | return max(0.0, min(1.0, base / ref)) |
| 134 | |
| 135 | |
| 136 | def top_comment_vote_signal(candidate: schema.Candidate) -> float: |
| 137 | """Strength of a candidate's most-upvoted top comment, as [0,1]. |
| 138 | |
| 139 | Normalized *within the candidate's platform* (see ``_VOTE_LOG_REFERENCE``) |
| 140 | so a 22k-like TikTok comment and a 600-upvote Reddit comment land on a |
| 141 | comparable scale rather than letting raw counts dominate. Returns 0.0 when |
| 142 | no top comment carries votes. Used by the fun judge to amplify (never |
| 143 | drive) crowd-certified comments. |
| 144 | """ |
| 145 | best_log = 0.0 |
| 146 | for item in candidate.source_items: |
| 147 | comments = item.metadata.get("top_comments") or [] |
| 148 | for comment in comments[:3]: |
| 149 | if isinstance(comment, dict): |
| 150 | best_log = max(best_log, log1p_safe(comment.get("score"))) |
| 151 | if best_log <= 0.0: |
| 152 | return 0.0 |
| 153 | ref = _VOTE_LOG_REFERENCE.get(candidate.source, _VOTE_LOG_REFERENCE_DEFAULT) |
| 154 | return max(0.0, min(1.0, best_log / ref)) |
| 155 | |
| 156 | |
| 157 | # Per-source engagement weights: list of (field_name, weight) tuples. |
| 158 | # Reddit, YouTube, and TikTok use custom functions because they include |
| 159 | # a dedicated 10% top-comment-score slot (see _reddit_engagement, |
| 160 | # _youtube_engagement, _tiktok_engagement). |
| 161 | ENGAGEMENT_WEIGHTS: dict[str, list[tuple[str, float]]] = { |
| 162 | "x": [("likes", 0.55), ("reposts", 0.25), ("replies", 0.15), ("quotes", 0.05)], |
| 163 | "instagram": [("views", 0.50), ("likes", 0.30), ("comments", 0.20)], |
| 164 | "hackernews": [("points", 0.55), ("comments", 0.45)], |
| 165 | "bluesky": [("likes", 0.40), ("reposts", 0.30), ("replies", 0.20), ("quotes", 0.10)], |
| 166 | "truthsocial": [("likes", 0.45), ("reposts", 0.30), ("replies", 0.25)], |
| 167 | "polymarket": [("volume", 0.60), ("liquidity", 0.40)], |
| 168 | "digg": [("postCount", 0.40), ("uniqueAuthors", 0.30), ("rank_score", 0.30)], |
| 169 | "trustpilot": [("reviews", 1.0)], |
| 170 | } |
| 171 | |
| 172 | |
| 173 | def _weighted_engagement(item: schema.SourceItem, weights: list[tuple[str, float]]) -> float | None: |
| 174 | values = [(log1p_safe(item.engagement.get(field)), weight) for field, weight in weights] |
| 175 | if not any(v for v, _ in values): |
| 176 | return None |
| 177 | return sum(v * w for v, w in values) |
| 178 | |
| 179 | |
| 180 | def _reddit_engagement(item: schema.SourceItem) -> float | None: |
| 181 | score = log1p_safe(item.engagement.get("score")) |
| 182 | comments = log1p_safe(item.engagement.get("num_comments")) |
| 183 | ratio = float(item.engagement.get("upvote_ratio") or 0.0) |
| 184 | top_comment = _top_comment_score(item) |
| 185 | if not any([score, comments, ratio, top_comment]): |
| 186 | return None |
| 187 | return (0.50 * score) + (0.35 * comments) + (0.05 * (ratio * 10.0)) + (0.10 * top_comment) |
| 188 | |
| 189 | |
| 190 | def _youtube_engagement(item: schema.SourceItem) -> float | None: |
| 191 | views = log1p_safe(item.engagement.get("views")) |
| 192 | likes = log1p_safe(item.engagement.get("likes")) |
| 193 | comments = log1p_safe(item.engagement.get("comments")) |
| 194 | top_comment = _top_comment_score(item) |
| 195 | if not any([views, likes, comments, top_comment]): |
| 196 | return None |
| 197 | # Mirrors Reddit: carve out 10% for top-comment signal, keep view-weight |
| 198 | # dominant. Without comments, the pre-change weights (0.50/0.35/0.15) |
| 199 | # still govern relative ordering. |
| 200 | return (0.45 * views) + (0.32 * likes) + (0.13 * comments) + (0.10 * top_comment) |
| 201 | |
| 202 | |
| 203 | def _tiktok_engagement(item: schema.SourceItem) -> float | None: |
| 204 | views = log1p_safe(item.engagement.get("views")) |
| 205 | likes = log1p_safe(item.engagement.get("likes")) |
| 206 | comments = log1p_safe(item.engagement.get("comments")) |
| 207 | top_comment = _top_comment_score(item) |
| 208 | if not any([views, likes, comments, top_comment]): |
| 209 | return None |
| 210 | return (0.45 * views) + (0.27 * likes) + (0.18 * comments) + (0.10 * top_comment) |
| 211 | |
| 212 | |
| 213 | def _instagram_engagement(item: schema.SourceItem) -> float | None: |
| 214 | # Mirrors _tiktok_engagement: reels are video-shaped, and a highly-liked top |
| 215 | # comment carves out 10% of the signal (via comment_like_count -> score) so |
| 216 | # crowd-loved IG comments lift their post's ranking like YouTube/TikTok. |
| 217 | views = log1p_safe(item.engagement.get("views")) |
| 218 | likes = log1p_safe(item.engagement.get("likes")) |
| 219 | comments = log1p_safe(item.engagement.get("comments")) |
| 220 | top_comment = _top_comment_score(item) |
| 221 | if not any([views, likes, comments, top_comment]): |
| 222 | return None |
| 223 | return (0.45 * views) + (0.27 * likes) + (0.18 * comments) + (0.10 * top_comment) |
| 224 | |
| 225 | |
| 226 | def _generic_engagement(item: schema.SourceItem) -> float | None: |
| 227 | if not item.engagement: |
| 228 | return None |
| 229 | values = [logged for v in item.engagement.values() if (logged := log1p_safe(v)) > 0] |
| 230 | if not values: |
| 231 | return None |
| 232 | return sum(values) / len(values) |
| 233 | |
| 234 | |
| 235 | def engagement_raw(item: schema.SourceItem) -> float | None: |
| 236 | if item.source == "reddit": |
| 237 | return _reddit_engagement(item) |
| 238 | if item.source == "youtube": |
| 239 | return _youtube_engagement(item) |
| 240 | if item.source == "tiktok": |
| 241 | return _tiktok_engagement(item) |
| 242 | if item.source == "instagram": |
| 243 | return _instagram_engagement(item) |
| 244 | weights = ENGAGEMENT_WEIGHTS.get(item.source) |
| 245 | if weights: |
| 246 | return _weighted_engagement(item, weights) |
| 247 | return _generic_engagement(item) |
| 248 | |
| 249 | |
| 250 | def normalize(values: list[float | None]) -> list[int | None]: |
| 251 | valid = [value for value in values if value is not None] |
| 252 | if not valid: |
| 253 | return [None for _ in values] |
| 254 | low = min(valid) |
| 255 | high = max(valid) |
| 256 | if math.isclose(low, high): |
| 257 | return [50 if value is not None else None for value in values] |
| 258 | return [ |
| 259 | None |
| 260 | if value is None |
| 261 | else int(((value - low) / (high - low)) * 100) |
| 262 | for value in values |
| 263 | ] |
| 264 | |
| 265 | |
| 266 | def annotate_stream( |
| 267 | items: list[schema.SourceItem], |
| 268 | ranking_query: "str | relevance.PreparedQuery", |
| 269 | freshness_mode: str, |
| 270 | reference_date: str | None = None, |
| 271 | max_days: int = 30, |
| 272 | ) -> list[schema.SourceItem]: |
| 273 | """Attach local scoring metadata and return items sorted by local_rank_score.""" |
| 274 | prepared_query = ranking_query if isinstance(ranking_query, relevance.PreparedQuery) else relevance.PreparedQuery(ranking_query) |
| 275 | engagement_scores = normalize([engagement_raw(item) for item in items]) |
| 276 | for item, eng_score in zip(items, engagement_scores, strict=True): |
| 277 | item.local_relevance = local_relevance(item, prepared_query) |
| 278 | item.freshness = freshness( |
| 279 | item, |
| 280 | freshness_mode, |
| 281 | reference_date=reference_date, |
| 282 | max_days=max_days, |
| 283 | ) |
| 284 | item.engagement_score = eng_score |
| 285 | item.source_quality = source_quality(item.source) |
| 286 | item.local_rank_score = ( |
| 287 | 0.65 * item.local_relevance |
| 288 | + 0.25 * (item.freshness / 100.0) |
| 289 | + 0.10 * ((eng_score or 0) / 100.0) |
| 290 | ) |
| 291 | return sorted(items, key=lambda item: item.local_rank_score or 0, reverse=True) |
| 292 | |
| 293 | |
| 294 | _SOCIAL_SOURCES = {"reddit", "x", "tiktok", "instagram", "bluesky", "truthsocial"} |
| 295 | |
| 296 | # Minimum view count for short-video platforms. Items below this floor |
| 297 | # are typically spam reposts or low-effort clips that add no unique signal. |
| 298 | _VIDEO_ENGAGEMENT_FLOOR_SOURCES = {"tiktok", "instagram"} |
| 299 | _VIDEO_ENGAGEMENT_FLOOR_VIEWS = 1000 |
| 300 | |
| 301 | |
| 302 | def _passes_engagement_floor(item: schema.SourceItem, sole_source: bool) -> bool: |
| 303 | """Check whether a TikTok/Instagram item meets the minimum view floor. |
| 304 | |
| 305 | Items from sources not in _VIDEO_ENGAGEMENT_FLOOR_SOURCES always pass. |
| 306 | If the item's source is the *only* source represented in the batch |
| 307 | (sole_source=True), all items pass so we never return an empty result |
| 308 | for a whole source. |
| 309 | """ |
| 310 | if item.source not in _VIDEO_ENGAGEMENT_FLOOR_SOURCES: |
| 311 | return True |
| 312 | if sole_source: |
| 313 | return True |
| 314 | views = item.engagement.get("views") or 0 if item.engagement else 0 |
| 315 | return views >= _VIDEO_ENGAGEMENT_FLOOR_VIEWS |
| 316 | |
| 317 | |
| 318 | def prune_low_relevance( |
| 319 | items: list[schema.SourceItem], |
| 320 | minimum: float = 0.15, |
| 321 | ) -> list[schema.SourceItem]: |
| 322 | """Drop weak lexical matches when stronger evidence exists. |
| 323 | |
| 324 | Social-source items with zero engagement get a stricter threshold |
| 325 | because zero engagement on a social platform is a strong noise signal. |
| 326 | |
| 327 | TikTok and Instagram items with fewer than 1000 views are pruned |
| 328 | (unless they are the only source represented in the batch). |
| 329 | """ |
| 330 | sources_present = {item.source for item in items} |
| 331 | |
| 332 | def passes(item: schema.SourceItem) -> bool: |
| 333 | # YouTube items with successfully extracted transcripts should not |
| 334 | # be pruned by title-only relevance scoring — the transcript content |
| 335 | # already proves substantive topical coverage. |
| 336 | if item.source == "youtube" and item.snippet: |
| 337 | return True |
| 338 | rel = item.local_relevance if item.local_relevance is not None else 0.0 |
| 339 | if rel < minimum: |
| 340 | return False |
| 341 | if item.source in _SOCIAL_SOURCES and (item.engagement_score is None or item.engagement_score == 0): |
| 342 | if rel < minimum * 1.5: |
| 343 | return False |
| 344 | sole_source = sources_present == {item.source} |
| 345 | if not _passes_engagement_floor(item, sole_source): |
| 346 | return False |
| 347 | return True |
| 348 | |
| 349 | filtered = [item for item in items if passes(item)] |
| 350 | return filtered or items |
| 351 |