| 1 | """Reusable local scoring signals for v3 pipeline stages.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import math |
| 6 | from collections.abc import Iterable, Mapping |
| 7 | |
| 8 | from . import dates, relevance, schema |
| 9 | |
| 10 | # Editorial signal-to-noise scores. Grounding (Google Search) is 1.0 baseline; |
| 11 | # social platforms discounted for noise. |
| 12 | SOURCE_QUALITY = { |
| 13 | "xiaohongshu": 0.7, |
| 14 | "hackernews": 0.8, |
| 15 | "youtube": 0.85, |
| 16 | "digg": 0.85, |
| 17 | "arxiv": 0.9, |
| 18 | "techmeme": 0.85, |
| 19 | "trustpilot": 0.78, |
| 20 | # Verified-purchase reviews on a live aggregate rating: high-quality |
| 21 | # buyer evidence, a notch above Trustpilot's open review model. |
| 22 | "amazon": 0.8, |
| 23 | # Paid creative is first-party marketing copy: a reliable statement of |
| 24 | # what the brand is pushing, but not independent evidence about it, so it |
| 25 | # sits below the review-backed sources and above open social. |
| 26 | "meta_ads": 0.72, |
| 27 | "reddit": 0.6, |
| 28 | "x": 0.68, |
| 29 | "bluesky": 0.66, |
| 30 | "truthsocial": 0.6, |
| 31 | "polymarket": 0.5, |
| 32 | "instagram": 0.58, |
| 33 | "tiktok": 0.58, |
| 34 | "jobs": 0.72, |
| 35 | "corpus": 0.75, |
| 36 | } |
| 37 | |
| 38 | |
| 39 | def source_quality(source: str) -> float: |
| 40 | return SOURCE_QUALITY.get(source, 0.6) |
| 41 | |
| 42 | |
| 43 | def local_relevance( |
| 44 | item: schema.SourceItem, |
| 45 | ranking_query: "str | relevance.PreparedQuery", |
| 46 | ) -> float: |
| 47 | text = "\n".join( |
| 48 | part |
| 49 | for part in [item.title, item.body, item.snippet] |
| 50 | if part |
| 51 | ) |
| 52 | hashtags = item.metadata.get("hashtags") if isinstance(item.metadata, dict) else None |
| 53 | score = relevance.token_overlap_relevance(ranking_query, text, hashtags=hashtags) |
| 54 | |
| 55 | # High-engagement YouTube floor: official videos with millions of views |
| 56 | # often have titles that don't keyword-match the query (e.g., "YE - FATHER |
| 57 | # (feat. TRAVIS SCOTT)" doesn't match "kanye west"). The engagement signals |
| 58 | # say "this is important" even when text overlap is weak. |
| 59 | if item.source == "youtube" and (item.engagement.get("views") or 0) > 100_000: |
| 60 | score = max(score, 0.3) |
| 61 | |
| 62 | # Project-mode GitHub floor: items fetched via --github-repo are explicitly |
| 63 | # requested by the user and relevant by construction. Without this floor, |
| 64 | # repos with low token diversity (e.g., "openclaw/openclaw" -> 1 unique token) |
| 65 | # get pruned despite being the primary search target. |
| 66 | labels = item.metadata.get("labels", []) if isinstance(item.metadata, dict) else [] |
| 67 | if "project-mode" in labels: |
| 68 | score = max(score, 0.8) |
| 69 | |
| 70 | # Grounding-exempt floor (currently Amazon): the adapter already gated |
| 71 | # these against the model-supplied product keyword before creating them, |
| 72 | # so they are relevant by construction. Their text is marketing copy plus |
| 73 | # buyer reviews, which rarely repeats the topic phrasing -- a "Weber |
| 74 | # Grills" run surfaces a product named "Spirit E-325" whose reviews talk |
| 75 | # about searing, not about Weber. Without the floor, correctly-retrieved |
| 76 | # evidence gets pruned for failing a keyword match it was never going to |
| 77 | # win. Mirrors the project-mode GitHub floor above. |
| 78 | if isinstance(item.metadata, dict) and item.metadata.get("grounding_exempt"): |
| 79 | score = max(score, 0.8) |
| 80 | |
| 81 | return score |
| 82 | |
| 83 | |
| 84 | def freshness( |
| 85 | item: schema.SourceItem, |
| 86 | freshness_mode: str = "balanced_recent", |
| 87 | *, |
| 88 | reference_date: str | None = None, |
| 89 | max_days: int = 30, |
| 90 | ) -> int: |
| 91 | score = dates.recency_score( |
| 92 | item.published_at, |
| 93 | max_days=max_days, |
| 94 | reference_date=reference_date, |
| 95 | ) |
| 96 | if freshness_mode == "strict_recent": |
| 97 | return int(score) |
| 98 | if freshness_mode == "evergreen_ok": |
| 99 | return int((score * 0.6) + 40) |
| 100 | return int((score * 0.8) + 10) |
| 101 | |
| 102 | |
| 103 | def log1p_safe(value: float | int | None) -> float: |
| 104 | if value is None: |
| 105 | return 0.0 |
| 106 | try: |
| 107 | numeric = float(value) |
| 108 | except (TypeError, ValueError): |
| 109 | return 0.0 |
| 110 | if numeric <= 0: |
| 111 | return 0.0 |
| 112 | return math.log1p(numeric) |
| 113 | |
| 114 | |
| 115 | def _top_comment_score(item: schema.SourceItem) -> float: |
| 116 | comments = item.metadata.get("top_comments") or [] |
| 117 | if not comments or not isinstance(comments[0], dict): |
| 118 | return 0.0 |
| 119 | return log1p_safe(comments[0].get("score")) |
| 120 | |
| 121 | |
| 122 | # Per-platform log-reference for normalizing a top comment's vote count into a |
| 123 | # [0,1] signal. Reddit upvotes run in the hundreds-to-thousands; YouTube/TikTok |
| 124 | # likes run 10-600x higher (and the top end is display-abbreviated: "39K" is |
| 125 | # stored as 39000). A raw or single-scale log compare would let YouTube/TikTok |
| 126 | # dominate purely by platform scale, not by being funnier. Each value is the |
| 127 | # log1p of a "very high" top-comment count for that platform, so dividing a |
| 128 | # comment's log1p(score) by it yields a comparable cross-platform strength. |
| 129 | _VOTE_LOG_REFERENCE: dict[str, float] = { |
| 130 | "reddit": 7.6, # ~log1p(2000) |
| 131 | "hackernews": 6.2, # ~log1p(500) |
| 132 | "youtube": 10.3, # ~log1p(30000) |
| 133 | "tiktok": 10.3, # ~log1p(30000) |
| 134 | "instagram": 9.2, # ~log1p(10000) |
| 135 | "x": 9.2, # ~log1p(10000) |
| 136 | "bluesky": 9.2, # ~log1p(10000); like X/IG, not the Reddit default |
| 137 | } |
| 138 | _VOTE_LOG_REFERENCE_DEFAULT = 7.6 |
| 139 | |
| 140 | |
| 141 | def normalized_comment_vote(source: str, score: "float | int | None") -> float: |
| 142 | """Normalize a single comment's vote count to [0,1] within its platform. |
| 143 | |
| 144 | Same per-platform reference as ``top_comment_vote_signal`` so a 22k-like |
| 145 | TikTok comment and a 600-upvote Reddit comment rank on a comparable scale. |
| 146 | Used to rank the cross-candidate Top Community Comments block. |
| 147 | """ |
| 148 | base = log1p_safe(score) |
| 149 | if base <= 0.0: |
| 150 | return 0.0 |
| 151 | ref = _VOTE_LOG_REFERENCE.get(source, _VOTE_LOG_REFERENCE_DEFAULT) |
| 152 | return max(0.0, min(1.0, base / ref)) |
| 153 | |
| 154 | |
| 155 | def top_comment_vote_signal(candidate: schema.Candidate) -> float: |
| 156 | """Strength of a candidate's most-upvoted top comment, as [0,1]. |
| 157 | |
| 158 | Normalized *within the candidate's platform* (see ``_VOTE_LOG_REFERENCE``) |
| 159 | so a 22k-like TikTok comment and a 600-upvote Reddit comment land on a |
| 160 | comparable scale rather than letting raw counts dominate. Returns 0.0 when |
| 161 | no top comment carries votes. Used by the fun judge to amplify (never |
| 162 | drive) crowd-certified comments. |
| 163 | """ |
| 164 | best_log = 0.0 |
| 165 | for item in candidate.source_items: |
| 166 | comments = item.metadata.get("top_comments") or [] |
| 167 | for comment in comments[:3]: |
| 168 | if isinstance(comment, dict): |
| 169 | best_log = max(best_log, log1p_safe(comment.get("score"))) |
| 170 | if best_log <= 0.0: |
| 171 | return 0.0 |
| 172 | ref = _VOTE_LOG_REFERENCE.get(candidate.source, _VOTE_LOG_REFERENCE_DEFAULT) |
| 173 | return max(0.0, min(1.0, best_log / ref)) |
| 174 | |
| 175 | |
| 176 | # Per-source engagement weights: list of (field_name, weight) tuples. |
| 177 | # Reddit, YouTube, and TikTok use custom functions because they include |
| 178 | # a dedicated 10% top-comment-score slot (see _reddit_engagement, |
| 179 | # _youtube_engagement, _tiktok_engagement). |
| 180 | ENGAGEMENT_WEIGHTS: dict[str, list[tuple[str, float]]] = { |
| 181 | "x": [("likes", 0.55), ("reposts", 0.25), ("replies", 0.15), ("quotes", 0.05)], |
| 182 | "instagram": [("views", 0.50), ("likes", 0.30), ("comments", 0.20)], |
| 183 | "hackernews": [("points", 0.55), ("comments", 0.45)], |
| 184 | "bluesky": [("likes", 0.40), ("reposts", 0.30), ("replies", 0.20), ("quotes", 0.10)], |
| 185 | "truthsocial": [("likes", 0.45), ("reposts", 0.30), ("replies", 0.25)], |
| 186 | "polymarket": [("volume", 0.60), ("liquidity", 0.40)], |
| 187 | "digg": [("postCount", 0.40), ("uniqueAuthors", 0.30), ("rank_score", 0.30)], |
| 188 | "trustpilot": [("reviews", 1.0)], |
| 189 | "amazon": [("ratings", 1.0)], |
| 190 | # Meta publishes reach and spend only for political and issue ads, so a |
| 191 | # commercial creative has no audience signal at all. How many variants a |
| 192 | # brand cut of one creative is the closest honest proxy for how hard it is |
| 193 | # pushing that message. |
| 194 | "meta_ads": [("variants", 1.0)], |
| 195 | } |
| 196 | |
| 197 | |
| 198 | def _weighted_engagement(item: schema.SourceItem, weights: list[tuple[str, float]]) -> float | None: |
| 199 | values = [(log1p_safe(item.engagement.get(field)), weight) for field, weight in weights] |
| 200 | if not any(v for v, _ in values): |
| 201 | return None |
| 202 | return sum(v * w for v, w in values) |
| 203 | |
| 204 | |
| 205 | def _reddit_engagement(item: schema.SourceItem) -> float | None: |
| 206 | score = log1p_safe(item.engagement.get("score")) |
| 207 | comments = log1p_safe(item.engagement.get("num_comments")) |
| 208 | ratio = float(item.engagement.get("upvote_ratio") or 0.0) |
| 209 | top_comment = _top_comment_score(item) |
| 210 | if not any([score, comments, ratio, top_comment]): |
| 211 | return None |
| 212 | return (0.50 * score) + (0.35 * comments) + (0.05 * (ratio * 10.0)) + (0.10 * top_comment) |
| 213 | |
| 214 | |
| 215 | def _youtube_engagement(item: schema.SourceItem) -> float | None: |
| 216 | views = log1p_safe(item.engagement.get("views")) |
| 217 | likes = log1p_safe(item.engagement.get("likes")) |
| 218 | comments = log1p_safe(item.engagement.get("comments")) |
| 219 | top_comment = _top_comment_score(item) |
| 220 | if not any([views, likes, comments, top_comment]): |
| 221 | return None |
| 222 | # Mirrors Reddit: carve out 10% for top-comment signal, keep view-weight |
| 223 | # dominant. Without comments, the pre-change weights (0.50/0.35/0.15) |
| 224 | # still govern relative ordering. |
| 225 | return (0.45 * views) + (0.32 * likes) + (0.13 * comments) + (0.10 * top_comment) |
| 226 | |
| 227 | |
| 228 | def _tiktok_engagement(item: schema.SourceItem) -> float | None: |
| 229 | views = log1p_safe(item.engagement.get("views")) |
| 230 | likes = log1p_safe(item.engagement.get("likes")) |
| 231 | comments = log1p_safe(item.engagement.get("comments")) |
| 232 | top_comment = _top_comment_score(item) |
| 233 | if not any([views, likes, comments, top_comment]): |
| 234 | return None |
| 235 | return (0.45 * views) + (0.27 * likes) + (0.18 * comments) + (0.10 * top_comment) |
| 236 | |
| 237 | |
| 238 | def _instagram_engagement(item: schema.SourceItem) -> float | None: |
| 239 | # Mirrors _tiktok_engagement: reels are video-shaped, and a highly-liked top |
| 240 | # comment carves out 10% of the signal (via comment_like_count -> score) so |
| 241 | # crowd-loved IG comments lift their post's ranking like YouTube/TikTok. |
| 242 | views = log1p_safe(item.engagement.get("views")) |
| 243 | likes = log1p_safe(item.engagement.get("likes")) |
| 244 | comments = log1p_safe(item.engagement.get("comments")) |
| 245 | top_comment = _top_comment_score(item) |
| 246 | if not any([views, likes, comments, top_comment]): |
| 247 | return None |
| 248 | return (0.45 * views) + (0.27 * likes) + (0.18 * comments) + (0.10 * top_comment) |
| 249 | |
| 250 | |
| 251 | def _generic_engagement(item: schema.SourceItem) -> float | None: |
| 252 | if not item.engagement: |
| 253 | return None |
| 254 | values = [logged for v in item.engagement.values() if (logged := log1p_safe(v)) > 0] |
| 255 | if not values: |
| 256 | return None |
| 257 | return sum(values) / len(values) |
| 258 | |
| 259 | |
| 260 | def engagement_raw(item: schema.SourceItem) -> float | None: |
| 261 | if item.source == "reddit": |
| 262 | return _reddit_engagement(item) |
| 263 | if item.source == "youtube": |
| 264 | return _youtube_engagement(item) |
| 265 | if item.source == "tiktok": |
| 266 | return _tiktok_engagement(item) |
| 267 | if item.source == "instagram": |
| 268 | return _instagram_engagement(item) |
| 269 | weights = ENGAGEMENT_WEIGHTS.get(item.source) |
| 270 | if weights: |
| 271 | return _weighted_engagement(item, weights) |
| 272 | return _generic_engagement(item) |
| 273 | |
| 274 | |
| 275 | def normalize(values: list[float | None]) -> list[int | None]: |
| 276 | valid = [value for value in values if value is not None] |
| 277 | if not valid: |
| 278 | return [None for _ in values] |
| 279 | low = min(valid) |
| 280 | high = max(valid) |
| 281 | if math.isclose(low, high): |
| 282 | return [50 if value is not None else None for value in values] |
| 283 | return [ |
| 284 | None |
| 285 | if value is None |
| 286 | else int(((value - low) / (high - low)) * 100) |
| 287 | for value in values |
| 288 | ] |
| 289 | |
| 290 | |
| 291 | def annotate_stream( |
| 292 | items: list[schema.SourceItem], |
| 293 | ranking_query: "str | relevance.PreparedQuery", |
| 294 | freshness_mode: str, |
| 295 | reference_date: str | None = None, |
| 296 | max_days: int = 30, |
| 297 | ) -> list[schema.SourceItem]: |
| 298 | """Attach local scoring metadata and return items sorted by local_rank_score.""" |
| 299 | prepared_query = ranking_query if isinstance(ranking_query, relevance.PreparedQuery) else relevance.PreparedQuery(ranking_query) |
| 300 | engagement_scores = normalize([engagement_raw(item) for item in items]) |
| 301 | for item, eng_score in zip(items, engagement_scores, strict=True): |
| 302 | item.local_relevance = local_relevance(item, prepared_query) |
| 303 | item.freshness = freshness( |
| 304 | item, |
| 305 | freshness_mode, |
| 306 | reference_date=reference_date, |
| 307 | max_days=max_days, |
| 308 | ) |
| 309 | item.engagement_score = eng_score |
| 310 | item.source_quality = source_quality(item.source) |
| 311 | item.local_rank_score = ( |
| 312 | 0.65 * item.local_relevance |
| 313 | + 0.25 * (item.freshness / 100.0) |
| 314 | + 0.10 * ((eng_score or 0) / 100.0) |
| 315 | ) |
| 316 | return sorted(items, key=lambda item: item.local_rank_score or 0, reverse=True) |
| 317 | |
| 318 | |
| 319 | _SOCIAL_SOURCES = {"reddit", "x", "tiktok", "instagram", "bluesky", "truthsocial"} |
| 320 | |
| 321 | # Minimum view count for short-video platforms. Items below this floor |
| 322 | # are typically spam reposts or low-effort clips that add no unique signal. |
| 323 | _VIDEO_ENGAGEMENT_FLOOR_SOURCES = {"tiktok", "instagram"} |
| 324 | _VIDEO_ENGAGEMENT_FLOOR_VIEWS = 1000 |
| 325 | |
| 326 | |
| 327 | def _passes_engagement_floor(item: schema.SourceItem, sole_source: bool) -> bool: |
| 328 | """Check whether a TikTok/Instagram item meets the minimum view floor. |
| 329 | |
| 330 | Items from sources not in _VIDEO_ENGAGEMENT_FLOOR_SOURCES always pass. |
| 331 | If the item's source is the *only* source represented in the batch |
| 332 | (sole_source=True), all items pass so we never return an empty result |
| 333 | for a whole source. |
| 334 | """ |
| 335 | if item.source not in _VIDEO_ENGAGEMENT_FLOOR_SOURCES: |
| 336 | return True |
| 337 | if sole_source: |
| 338 | return True |
| 339 | views = item.engagement.get("views") or 0 if item.engagement else 0 |
| 340 | return views >= _VIDEO_ENGAGEMENT_FLOOR_VIEWS |
| 341 | |
| 342 | |
| 343 | def prune_low_relevance( |
| 344 | items: list[schema.SourceItem], |
| 345 | minimum: float = 0.15, |
| 346 | first_party_handles: Iterable[str] | None = None, |
| 347 | first_party_by_source: Mapping[str, Iterable[str]] | None = None, |
| 348 | ) -> list[schema.SourceItem]: |
| 349 | """Drop weak lexical matches when stronger evidence exists. |
| 350 | |
| 351 | Social-source items with genuinely zero engagement get a stricter |
| 352 | threshold because zero engagement on a social platform is a strong noise |
| 353 | signal. |
| 354 | |
| 355 | TikTok and Instagram items with fewer than 1000 views are pruned |
| 356 | (unless they are the only source represented in the batch). |
| 357 | |
| 358 | ``first_party_handles`` names accounts this run is explicitly searching |
| 359 | (the subject of the topic). Their own posts are exempt from the floor: a |
| 360 | post almost never contains its own author's name, so lexical relevance |
| 361 | scores it at or near zero no matter how on-topic it is. Without the |
| 362 | exemption a mixed batch loses them silently, because the ``filtered or |
| 363 | items`` rescue below only fires when *every* item fails. |
| 364 | |
| 365 | ``first_party_by_source`` scopes exemptions to one platform each (e.g. |
| 366 | ``--ig-creators`` names Instagram accounts). A creator flag must not |
| 367 | exempt a same-name account on a *different* platform: username reuse |
| 368 | across platforms is common, and an unrelated account would otherwise |
| 369 | bypass the relevance and engagement floors. |
| 370 | """ |
| 371 | sources_present = {item.source for item in items} |
| 372 | first_party = { |
| 373 | h.strip().lstrip("@").lower() |
| 374 | for h in (first_party_handles or ()) |
| 375 | if h and h.strip() |
| 376 | } |
| 377 | scoped_first_party = { |
| 378 | source: { |
| 379 | h.strip().lstrip("@").lower() |
| 380 | for h in handles |
| 381 | if h and h.strip() |
| 382 | } |
| 383 | for source, handles in (first_party_by_source or {}).items() |
| 384 | } |
| 385 | |
| 386 | def _is_first_party(item: schema.SourceItem) -> bool: |
| 387 | if not item.author: |
| 388 | return False |
| 389 | author = item.author.strip().lstrip("@").lower() |
| 390 | if author in first_party: |
| 391 | return True |
| 392 | return author in scoped_first_party.get(item.source, frozenset()) |
| 393 | |
| 394 | def passes(item: schema.SourceItem) -> bool: |
| 395 | # YouTube items with successfully extracted transcripts should not |
| 396 | # be pruned by title-only relevance scoring — the transcript content |
| 397 | # already proves substantive topical coverage. |
| 398 | if item.source == "youtube" and item.snippet: |
| 399 | return True |
| 400 | # Posts by an account this run is explicitly searching are evidence by |
| 401 | # provenance, not by lexical overlap with the topic. |
| 402 | if _is_first_party(item): |
| 403 | return True |
| 404 | rel = item.local_relevance if item.local_relevance is not None else 0.0 |
| 405 | if rel < minimum: |
| 406 | return False |
| 407 | # Key the stricter social gate on genuinely absent engagement, not on |
| 408 | # the normalized score: signals.normalize is min-max over the batch, so |
| 409 | # it maps the least-engaged item to exactly 0 even when that item has |
| 410 | # thousands of likes. |
| 411 | if item.source in _SOCIAL_SOURCES and not engagement_raw(item): |
| 412 | if rel < minimum * 1.5: |
| 413 | return False |
| 414 | sole_source = sources_present == {item.source} |
| 415 | if not _passes_engagement_floor(item, sole_source): |
| 416 | return False |
| 417 | return True |
| 418 | |
| 419 | filtered = [item for item in items if passes(item)] |
| 420 | return filtered or items |
| 421 |