github-featured/mvanhorn--last30days-skill
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity
Files
main
mvanhorn--last30days-skill/tests/test_signals_v3.py
T
Matt Van Horn 9e00fe5b6d feat: Instagram comments + comments-by-default + rank-based comment diversity (#751) 
* feat(instagram): wire Instagram comments via ScrapeCreators

U1: instagram.enrich_with_comments + _fetch_post_comments hit
GET /v2/instagram/post/comments, sorted by comment_like_count;
env.is_instagram_comments_available gate (key + instagram_comments in
INCLUDE_SOURCES); pipeline enriches the Instagram source when available.
Mirrors the TikTok-comments implementation.

* feat(instagram): full vote-weighting participation for IG comments

U2: comment_like_count is remapped to the shared score field; a dedicated
_instagram_engagement mirrors _tiktok_engagement with the 0.10 top-comment
carve-out so highly-liked IG comments lift item ranking; render gains the
instagram vote label (likes) + min-score threshold. _VOTE_LOG_REFERENCE
already had instagram. IG comment votes now weight ranking/Best-Takes/render
like YouTube/TikTok.

* feat(render): rank-based cross-platform comment diversity

U3: _render_top_comments interleaves comments round-robin by within-platform
rank (every platform's #1, then every #2, then #3) instead of a global vote-
magnitude sort. Top-3-of-each-platform outranks 4th-of-any and each platform's
#1 is guaranteed a slot -- a viral platform can no longer sweep the list. The
cross-platform list drops the per-platform absolute floor (min_score=0) so a
less-watched video's killer low-vote comment surfaces too; the per-candidate
card still applies the floor. Vote strength only orders within a platform.

* feat(nux): comments-by-default Step 5 tier copy

U4: recommended tier is now posts AND top comments for TikTok + Instagram plus
YouTube comments (INCLUDE_SOURCES=tiktok,instagram,youtube_comments,
tiktok_comments,instagram_comments); Everything adds Threads + Pinterest. No
posts-only tier. Operator-approved copy; mirrored in the non-modal 5b flow.

* feat(nux): accurate ScrapeCreators offer copy + fix stale docstrings

U5: Step 4 offer states comments are on by default (posts AND top comments +
YouTube comments) and describes the real auto-enrichment (Reddit public+SC
merged, YouTube search backstop) instead of vague 'backup'. Fixes the
is_youtube_comments_available docstring that claimed comments are excluded
from the Recommended tier. (Xiaohongshu was never labeled ScrapeCreators in
the docs; no change needed there.)

* docs: comments-by-default tier + instagram_comments in AGENTS/CONFIGURATION

U6: AGENTS.md onboarding note now describes comments-on-by-default (Recommended
tier), IG-comments parity, and the rank-based diversity selection.
CONFIGURATION.md documents instagram_comments/tiktok_comments INCLUDE_SOURCES
keys and the new default string; corrects the youtube_comments row and the
Step 5 tier prose.

---------

Co-authored-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com>
2026-07-04 08:28:14 -07:00

818 lines
33 KiB
Python
Raw Permalink Blame History

import math
import unittest
from lib import schema, signals
from lib.hackernews import parse_hackernews_response
class SignalsV3Tests(unittest.TestCase):
def test_reddit_engagement_uses_source_specific_formula(self):
item = schema.SourceItem(
item_id="r1",
source="reddit",
title="Title",
body="Body",
url="https://example.com",
engagement={"score": 99, "num_comments": 20, "upvote_ratio": 0.8},
metadata={"top_comments": [{"score": 10}]},
)
expected = (
0.50 * math.log1p(99)
+ 0.35 * math.log1p(20)
+ 0.05 * (0.8 * 10.0)
+ 0.10 * math.log1p(10)
)
self.assertAlmostEqual(expected, signals.engagement_raw(item))
def test_youtube_engagement_adds_top_comment_slot(self):
with_comment = schema.SourceItem(
item_id="yt1",
source="youtube",
title="Title",
body="Body",
url="https://youtube.com/watch?v=a",
engagement={"views": 10000, "likes": 500, "comments": 30},
metadata={"top_comments": [{"score": 500}]},
)
without = schema.SourceItem(
item_id="yt2",
source="youtube",
title="Title",
body="Body",
url="https://youtube.com/watch?v=b",
engagement={"views": 10000, "likes": 500, "comments": 30},
metadata={"top_comments": []},
)
with_score = signals.engagement_raw(with_comment)
without_score = signals.engagement_raw(without)
self.assertIsNotNone(with_score)
self.assertIsNotNone(without_score)
self.assertGreater(with_score, without_score)
expected = (
0.45 * math.log1p(10000)
+ 0.32 * math.log1p(500)
+ 0.13 * math.log1p(30)
+ 0.10 * math.log1p(500)
)
self.assertAlmostEqual(expected, with_score, places=6)
def test_youtube_engagement_empty_returns_none(self):
item = schema.SourceItem(
item_id="yt-empty",
source="youtube",
title="Title",
body="Body",
url="https://youtube.com/watch?v=e",
engagement={},
metadata={"top_comments": []},
)
self.assertIsNone(signals.engagement_raw(item))
def test_tiktok_engagement_adds_top_comment_slot(self):
item = schema.SourceItem(
item_id="tt1",
source="tiktok",
title="Title",
body="Body",
url="https://tiktok.com/@u/video/1",
engagement={"views": 100000, "likes": 5000, "comments": 500},
metadata={"top_comments": [{"score": 1200}]},
)
expected = (
0.45 * math.log1p(100000)
+ 0.27 * math.log1p(5000)
+ 0.18 * math.log1p(500)
+ 0.10 * math.log1p(1200)
)
self.assertAlmostEqual(expected, signals.engagement_raw(item), places=6)
def test_instagram_engagement_adds_top_comment_slot(self):
"""U2: IG gets the same 0.10 top-comment carve-out as TikTok, so a
highly-liked IG comment lifts its post's ranking."""
item = schema.SourceItem(
item_id="ig1",
source="instagram",
title="Title",
body="Body",
url="https://www.instagram.com/reel/ABC/",
engagement={"views": 100000, "likes": 5000, "comments": 500},
metadata={"top_comments": [{"score": 1200}]},
)
expected = (
0.45 * math.log1p(100000)
+ 0.27 * math.log1p(5000)
+ 0.18 * math.log1p(500)
+ 0.10 * math.log1p(1200)
)
self.assertAlmostEqual(expected, signals.engagement_raw(item), places=6)
def test_instagram_comment_vote_uses_instagram_reference(self):
"""U2: normalized_comment_vote uses the instagram reference, not the default."""
strength = signals.normalized_comment_vote("instagram", 5000)
self.assertGreater(strength, 0.0)
self.assertLessEqual(strength, 1.0)
def test_youtube_ranking_promotes_viral_comment_thread(self):
"""A moderately-viewed YouTube video with a 10k-like comment should
outrank a slightly-higher-viewed video with no high-signal comments."""
viral_comment = schema.SourceItem(
item_id="yt-with-viral-comment",
source="youtube",
title="Deploy to Fly.io",
body="Deploy to Fly.io walkthrough",
url="https://youtube.com/watch?v=x",
published_at="2026-03-15",
engagement={"views": 5000, "likes": 200, "comments": 50},
metadata={"top_comments": [{"score": 10000}]},
)
higher_views = schema.SourceItem(
item_id="yt-higher-views-no-comment",
source="youtube",
title="Deploy to Fly.io",
body="Deploy to Fly.io walkthrough",
url="https://youtube.com/watch?v=y",
published_at="2026-03-15",
engagement={"views": 8000, "likes": 300, "comments": 60},
metadata={"top_comments": []},
)
ranked = signals.annotate_stream(
[higher_views, viral_comment],
ranking_query="How do I deploy on Fly.io?",
freshness_mode="balanced_recent",
)
self.assertEqual("yt-with-viral-comment", ranked[0].item_id)
def test_polymarket_engagement_uses_market_fields(self):
item = schema.SourceItem(
item_id="pm1",
source="polymarket",
title="Title",
body="Body",
url="https://example.com",
engagement={"volume": 1000, "liquidity": 250},
)
expected = (0.60 * math.log1p(1000)) + (0.40 * math.log1p(250))
self.assertAlmostEqual(expected, signals.engagement_raw(item))
def test_grounding_uses_generic_fallback(self):
item = schema.SourceItem(
item_id="g1",
source="grounding",
title="Title",
body="Body",
url="https://example.com",
engagement={"shares": 10, "reads": 100},
)
expected = (math.log1p(10) + math.log1p(100)) / 2
self.assertAlmostEqual(expected, signals.engagement_raw(item))
def test_annotate_stream_sorts_by_source_specific_reddit_engagement(self):
higher = schema.SourceItem(
item_id="r-high",
source="reddit",
title="High signal",
body="claude code skill",
url="https://example.com/high",
published_at="2026-03-15",
engagement={"score": 120, "num_comments": 40, "upvote_ratio": 0.9},
metadata={"top_comments": [{"score": 15}]},
)
lower = schema.SourceItem(
item_id="r-low",
source="reddit",
title="Lower signal",
body="claude code skill",
url="https://example.com/low",
published_at="2026-03-15",
engagement={"score": 4, "num_comments": 1, "upvote_ratio": 0.5},
metadata={"top_comments": [{"score": 1}]},
)
ranked = signals.annotate_stream(
[lower, higher],
ranking_query="What recent evidence matters for claude code skill?",
freshness_mode="balanced_recent",
)
self.assertEqual(["r-high", "r-low"], [item.item_id for item in ranked])
def test_local_relevance_dominates_over_high_engagement_noise(self):
relevant = schema.SourceItem(
item_id="relevant",
source="reddit",
title="Deploy to Fly.io with MCP in 60 seconds",
body="Deploy to Fly.io guide with concrete steps.",
url="https://example.com/relevant",
published_at="2026-03-15",
engagement={"score": 2, "num_comments": 0, "upvote_ratio": 0.8},
metadata={"top_comments": []},
)
noisy = schema.SourceItem(
item_id="noisy",
source="reddit",
title="BATTLEFIELD 6 GAME UPDATE 1.2.2.0",
body="Patch notes and gameplay discussion.",
url="https://example.com/noisy",
published_at="2026-03-15",
engagement={"score": 5000, "num_comments": 1200, "upvote_ratio": 0.95},
metadata={"top_comments": [{"score": 400}]},
)
ranked = signals.annotate_stream(
[noisy, relevant],
ranking_query="How do I deploy on Fly.io?",
freshness_mode="evergreen_ok",
)
self.assertEqual("relevant", ranked[0].item_id)
def test_prune_low_relevance_keeps_stronger_matches(self):
strong = schema.SourceItem(
item_id="strong",
source="reddit",
title="Deploy to Fly.io",
body="Step-by-step Fly.io deploy guide.",
url="https://example.com/strong",
local_relevance=0.3,
)
weak = schema.SourceItem(
item_id="weak",
source="reddit",
title="Battlefield update",
body="Patch notes.",
url="https://example.com/weak",
local_relevance=0.0,
)
pruned = signals.prune_low_relevance([strong, weak], minimum=0.1)
self.assertEqual(["strong"], [item.item_id for item in pruned])
def test_prune_low_relevance_falls_back_when_all_are_weak(self):
weak = schema.SourceItem(
item_id="weak",
source="reddit",
title="Generic post",
body="Generic body.",
url="https://example.com/weak",
metadata={"local_relevance": 0.02},
)
pruned = signals.prune_low_relevance([weak], minimum=0.1)
self.assertEqual(["weak"], [item.item_id for item in pruned])
# -- Iteration 1: HN engagement bug --
def test_hackernews_parse_emits_comments_key(self):
"""parse_hackernews_response must emit 'comments' (not 'num_comments')."""
response = {
"hits": [
{
"objectID": "123",
"title": "Show HN: Something Cool",
"url": "https://example.com",
"author": "pg",
"points": 150,
"num_comments": 45,
"created_at_i": 1710720000,
},
],
}
items = parse_hackernews_response(response, query="something cool")
self.assertIn("comments", items[0]["engagement"])
self.assertNotIn("num_comments", items[0]["engagement"])
self.assertEqual(items[0]["engagement"]["comments"], 45)
def test_hackernews_engagement_raw_uses_both_fields(self):
"""engagement_raw for HN must weight both points and comments."""
item = schema.SourceItem(
item_id="hn1",
source="hackernews",
title="Show HN: Something",
body="Description",
url="https://example.com",
engagement={"points": 150, "comments": 45},
)
expected = 0.55 * math.log1p(150) + 0.45 * math.log1p(45)
result = signals.engagement_raw(item)
self.assertIsNotNone(result)
self.assertAlmostEqual(expected, result)
# Verify comments actually contributed (not just points)
points_only = 0.55 * math.log1p(150)
self.assertGreater(result, points_only)
# -- Iteration 4: Missing engagement formula tests --
def test_x_engagement_dominant_weight(self):
"""X: likes at 0.55 should dominate over quotes at 0.05."""
item = schema.SourceItem(
item_id="x1", source="x", title="T", body="B",
url="https://example.com",
engagement={"likes": 100, "reposts": 100, "replies": 100, "quotes": 100},
)
result = signals.engagement_raw(item)
self.assertIsNotNone(result)
expected = (
0.55 * math.log1p(100)
+ 0.25 * math.log1p(100)
+ 0.15 * math.log1p(100)
+ 0.05 * math.log1p(100)
)
self.assertAlmostEqual(expected, result)
def test_x_engagement_all_zero_returns_none(self):
item = schema.SourceItem(
item_id="x2", source="x", title="T", body="B",
url="https://example.com",
engagement={"likes": 0, "reposts": 0, "replies": 0, "quotes": 0},
)
self.assertIsNone(signals.engagement_raw(item))
def test_x_engagement_missing_fields(self):
"""Missing fields default to 0, no crash."""
item = schema.SourceItem(
item_id="x3", source="x", title="T", body="B",
url="https://example.com",
engagement={"likes": 50},
)
result = signals.engagement_raw(item)
self.assertIsNotNone(result)
expected = 0.55 * math.log1p(50)
self.assertAlmostEqual(expected, result)
def test_youtube_engagement_dominant_weight(self):
"""YouTube: views at 0.45 should dominate. With no top-comment data,
the remaining 0.90 of weight is split views/likes/comments 0.45/0.32/0.13."""
item = schema.SourceItem(
item_id="yt1", source="youtube", title="T", body="B",
url="https://example.com",
engagement={"views": 10000, "likes": 500, "comments": 80},
)
result = signals.engagement_raw(item)
self.assertIsNotNone(result)
expected = (
0.45 * math.log1p(10000)
+ 0.32 * math.log1p(500)
+ 0.13 * math.log1p(80)
)
self.assertAlmostEqual(expected, result)
def test_youtube_engagement_all_zero_returns_none(self):
item = schema.SourceItem(
item_id="yt2", source="youtube", title="T", body="B",
url="https://example.com",
engagement={"views": 0, "likes": 0, "comments": 0},
)
self.assertIsNone(signals.engagement_raw(item))
def test_youtube_engagement_missing_fields(self):
item = schema.SourceItem(
item_id="yt3", source="youtube", title="T", body="B",
url="https://example.com",
engagement={"views": 5000},
)
result = signals.engagement_raw(item)
self.assertIsNotNone(result)
expected = 0.45 * math.log1p(5000)
self.assertAlmostEqual(expected, result)
def test_tiktok_engagement_dominant_weight(self):
item = schema.SourceItem(
item_id="tt1", source="tiktok", title="T", body="B",
url="https://example.com",
engagement={"views": 50000, "likes": 3000, "comments": 200},
)
result = signals.engagement_raw(item)
self.assertIsNotNone(result)
expected = (
0.45 * math.log1p(50000)
+ 0.27 * math.log1p(3000)
+ 0.18 * math.log1p(200)
)
self.assertAlmostEqual(expected, result)
def test_tiktok_engagement_all_zero_returns_none(self):
item = schema.SourceItem(
item_id="tt2", source="tiktok", title="T", body="B",
url="https://example.com",
engagement={"views": 0, "likes": 0, "comments": 0},
)
self.assertIsNone(signals.engagement_raw(item))
def test_tiktok_engagement_missing_fields(self):
item = schema.SourceItem(
item_id="tt3", source="tiktok", title="T", body="B",
url="https://example.com",
engagement={"likes": 1000},
)
result = signals.engagement_raw(item)
self.assertIsNotNone(result)
expected = 0.27 * math.log1p(1000)
self.assertAlmostEqual(expected, result)
def test_instagram_engagement_dominant_weight(self):
item = schema.SourceItem(
item_id="ig1", source="instagram", title="T", body="B",
url="https://example.com",
engagement={"views": 8000, "likes": 1500, "comments": 100},
)
result = signals.engagement_raw(item)
self.assertIsNotNone(result)
# U2: IG now uses _instagram_engagement (video-shaped, with a 0.10
# top-comment carve-out); no top comment here so that term is 0.
expected = (
0.45 * math.log1p(8000)
+ 0.27 * math.log1p(1500)
+ 0.18 * math.log1p(100)
)
self.assertAlmostEqual(expected, result)
def test_instagram_engagement_all_zero_returns_none(self):
item = schema.SourceItem(
item_id="ig2", source="instagram", title="T", body="B",
url="https://example.com",
engagement={"views": 0, "likes": 0, "comments": 0},
)
self.assertIsNone(signals.engagement_raw(item))
def test_instagram_engagement_missing_fields(self):
item = schema.SourceItem(
item_id="ig3", source="instagram", title="T", body="B",
url="https://example.com",
engagement={"comments": 50},
)
result = signals.engagement_raw(item)
self.assertIsNotNone(result)
expected = 0.18 * math.log1p(50)
self.assertAlmostEqual(expected, result)
def test_hackernews_engagement_all_zero_returns_none(self):
item = schema.SourceItem(
item_id="hn2", source="hackernews", title="T", body="B",
url="https://example.com",
engagement={"points": 0, "comments": 0},
)
self.assertIsNone(signals.engagement_raw(item))
def test_hackernews_engagement_missing_fields(self):
item = schema.SourceItem(
item_id="hn3", source="hackernews", title="T", body="B",
url="https://example.com",
engagement={"points": 75},
)
result = signals.engagement_raw(item)
self.assertIsNotNone(result)
expected = 0.55 * math.log1p(75)
self.assertAlmostEqual(expected, result)
def test_bluesky_engagement_dominant_weight(self):
"""Bluesky: likes at 0.40 should dominate over quotes at 0.10."""
item = schema.SourceItem(
item_id="bs1", source="bluesky", title="T", body="B",
url="https://example.com",
engagement={"likes": 200, "reposts": 50, "replies": 30, "quotes": 10},
)
result = signals.engagement_raw(item)
self.assertIsNotNone(result)
expected = (
0.40 * math.log1p(200)
+ 0.30 * math.log1p(50)
+ 0.20 * math.log1p(30)
+ 0.10 * math.log1p(10)
)
self.assertAlmostEqual(expected, result)
def test_bluesky_engagement_all_zero_returns_none(self):
item = schema.SourceItem(
item_id="bs2", source="bluesky", title="T", body="B",
url="https://example.com",
engagement={"likes": 0, "reposts": 0, "replies": 0, "quotes": 0},
)
self.assertIsNone(signals.engagement_raw(item))
def test_bluesky_engagement_missing_fields(self):
item = schema.SourceItem(
item_id="bs3", source="bluesky", title="T", body="B",
url="https://example.com",
engagement={"likes": 100, "replies": 20},
)
result = signals.engagement_raw(item)
self.assertIsNotNone(result)
expected = 0.40 * math.log1p(100) + 0.20 * math.log1p(20)
self.assertAlmostEqual(expected, result)
def test_truthsocial_engagement_dominant_weight(self):
"""Truth Social: likes at 0.45 should dominate over replies at 0.25."""
item = schema.SourceItem(
item_id="ts1", source="truthsocial", title="T", body="B",
url="https://example.com",
engagement={"likes": 500, "reposts": 100, "replies": 50},
)
result = signals.engagement_raw(item)
self.assertIsNotNone(result)
expected = (
0.45 * math.log1p(500)
+ 0.30 * math.log1p(100)
+ 0.25 * math.log1p(50)
)
self.assertAlmostEqual(expected, result)
def test_truthsocial_engagement_all_zero_returns_none(self):
item = schema.SourceItem(
item_id="ts2", source="truthsocial", title="T", body="B",
url="https://example.com",
engagement={"likes": 0, "reposts": 0, "replies": 0},
)
self.assertIsNone(signals.engagement_raw(item))
def test_truthsocial_engagement_missing_fields(self):
item = schema.SourceItem(
item_id="ts3", source="truthsocial", title="T", body="B",
url="https://example.com",
engagement={"reposts": 80},
)
result = signals.engagement_raw(item)
self.assertIsNotNone(result)
expected = 0.30 * math.log1p(80)
self.assertAlmostEqual(expected, result)
# -- Fix 5: Rebalance engagement weight --
def test_engagement_weight_meaningful_for_social_ranking(self):
"""Engagement must have enough weight to differentiate otherwise-equal items."""
high_engagement = schema.SourceItem(
item_id="viral",
source="x",
title="Trending topic discussion",
body="Popular social post",
url="https://example.com/viral",
published_at="2026-03-15",
engagement={"likes": 50000, "reposts": 5000, "replies": 2000, "quotes": 500},
)
low_engagement = schema.SourceItem(
item_id="quiet",
source="x",
title="Trending topic discussion",
body="Popular social post",
url="https://example.com/quiet",
published_at="2026-03-15",
engagement={"likes": 10, "reposts": 1, "replies": 0, "quotes": 0},
)
ranked = signals.annotate_stream(
[low_engagement, high_engagement],
ranking_query="trending topic discussion",
freshness_mode="balanced_recent",
)
high_score = ranked[0].local_rank_score
low_score = ranked[1].local_rank_score
gap = high_score - low_score
# With 10% engagement weight, the gap should be >= 0.06
# With 5% weight, gap would be ~0.04
self.assertGreaterEqual(gap, 0.06,
f"Engagement gap should be >= 0.06 with 10% weight, got {gap:.4f}")
# -- Fix 4: Lower prune threshold for social media --
def test_prune_keeps_social_items_above_003(self):
"""Social media items with low but non-trivial relevance should survive pruning."""
social = schema.SourceItem(
item_id="social",
source="x",
title="Viral tweet about topic",
body="Short social post",
url="https://example.com/social",
metadata={"local_relevance": 0.05},
)
strong = schema.SourceItem(
item_id="strong",
source="grounding",
title="Detailed article about topic",
body="In-depth analysis",
url="https://example.com/strong",
metadata={"local_relevance": 0.4},
)
pruned = signals.prune_low_relevance([strong, social])
ids = [item.item_id for item in pruned]
self.assertIn("social", ids, "Item with relevance 0.05 should survive pruning")
self.assertIn("strong", ids)
# -- Unit 3: YouTube high-engagement relevance floor --
def test_youtube_high_engagement_gets_relevance_floor(self):
"""YouTube video with >100K views gets at least 0.3 relevance even with low text overlap."""
item = schema.SourceItem(
item_id="yt-official",
source="youtube",
title="YE - FATHER (feat. TRAVIS SCOTT)",
body="Official music video",
url="https://youtube.com/watch?v=abc",
engagement={"views": 8_000_000, "likes": 422_000, "comments": 5000},
)
rel = signals.local_relevance(item, "kanye west")
self.assertGreaterEqual(rel, 0.3, f"High-engagement YouTube should get >= 0.3 relevance, got {rel}")
def test_youtube_low_engagement_no_floor(self):
"""YouTube video with <100K views does NOT get the relevance floor."""
item = schema.SourceItem(
item_id="yt-small",
source="youtube",
title="Random unrelated video title",
body="Nothing relevant here",
url="https://youtube.com/watch?v=xyz",
engagement={"views": 500, "likes": 10, "comments": 1},
)
rel = signals.local_relevance(item, "kanye west")
self.assertLess(rel, 0.3, f"Low-engagement YouTube should not get floor, got {rel}")
def test_non_youtube_high_engagement_no_floor(self):
"""Non-YouTube items with high engagement do NOT get the YouTube floor."""
item = schema.SourceItem(
item_id="reddit-viral",
source="reddit",
title="Completely unrelated post",
body="Nothing about the topic",
url="https://reddit.com/r/test",
engagement={"score": 50000, "num_comments": 3000},
)
rel = signals.local_relevance(item, "kanye west")
self.assertLess(rel, 0.3, f"Non-YouTube item should not get YouTube floor, got {rel}")
# -- Unit 8: Engagement floor for TikTok/Instagram --
def test_tiktok_below_1000_views_pruned(self):
"""TikTok items with <1000 views should be pruned when other sources exist."""
spam = schema.SourceItem(
item_id="tt-spam", source="tiktok", title="AI news clip", body="Generic",
url="https://tiktok.com/spam",
local_relevance=0.4, engagement={"views": 500, "likes": 10, "comments": 1},
)
good = schema.SourceItem(
item_id="r-good", source="reddit", title="Good discussion", body="Quality",
url="https://reddit.com/good",
local_relevance=0.5, engagement_score=50,
)
pruned = signals.prune_low_relevance([good, spam])
ids = [item.item_id for item in pruned]
self.assertNotIn("tt-spam", ids, "TikTok with 500 views should be pruned")
self.assertIn("r-good", ids)
def test_instagram_below_1000_views_pruned(self):
"""Instagram items with <1000 views should be pruned when other sources exist."""
spam = schema.SourceItem(
item_id="ig-spam", source="instagram", title="Repost clip", body="Generic",
url="https://instagram.com/spam",
local_relevance=0.4, engagement={"views": 200, "likes": 5, "comments": 0},
)
good = schema.SourceItem(
item_id="x-good", source="x", title="Good tweet", body="Quality",
url="https://x.com/good",
local_relevance=0.5, engagement_score=50,
)
pruned = signals.prune_low_relevance([good, spam])
ids = [item.item_id for item in pruned]
self.assertNotIn("ig-spam", ids, "Instagram with 200 views should be pruned")
def test_tiktok_above_1000_views_kept(self):
"""TikTok items with >=1000 views should survive pruning."""
good_tt = schema.SourceItem(
item_id="tt-good", source="tiktok", title="Popular clip", body="Relevant",
url="https://tiktok.com/good",
local_relevance=0.4, engagement={"views": 5000, "likes": 200, "comments": 30},
)
other = schema.SourceItem(
item_id="r-other", source="reddit", title="Reddit post", body="Relevant",
url="https://reddit.com/other",
local_relevance=0.5, engagement_score=50,
)
pruned = signals.prune_low_relevance([other, good_tt])
ids = [item.item_id for item in pruned]
self.assertIn("tt-good", ids, "TikTok with 5000 views should be kept")
def test_tiktok_sole_source_not_pruned(self):
"""When TikTok is the only source, low-view items should NOT be pruned."""
items = [
schema.SourceItem(
item_id=f"tt-{i}", source="tiktok", title=f"Clip {i}", body="Content",
url=f"https://tiktok.com/{i}",
local_relevance=0.4, engagement={"views": 300, "likes": 5, "comments": 0},
)
for i in range(3)
]
pruned = signals.prune_low_relevance(items)
self.assertEqual(len(pruned), 3, "Sole-source TikTok items should all survive")
def test_non_video_sources_unaffected_by_floor(self):
"""Reddit/X items should not be affected by the video engagement floor."""
low_eng_x = schema.SourceItem(
item_id="x-low", source="x", title="Tweet", body="Topic discussion",
url="https://x.com/low",
local_relevance=0.5, engagement={"likes": 2, "reposts": 0},
engagement_score=5,
)
other = schema.SourceItem(
item_id="r-other", source="reddit", title="Post", body="Topic",
url="https://reddit.com/other",
local_relevance=0.5, engagement_score=50,
)
pruned = signals.prune_low_relevance([other, low_eng_x])
ids = [item.item_id for item in pruned]
self.assertIn("x-low", ids, "X items should not be affected by video floor")
def test_aspiresnippets_scenario(self):
"""@aspiresnippets scenario: 5 TikTok items with 200-700 views all pruned."""
spam_items = [
schema.SourceItem(
item_id=f"aspire-{i}", source="tiktok", title=f"AI news {i}", body="Generic clip",
url=f"https://tiktok.com/aspire/{i}",
local_relevance=0.3, engagement={"views": 200 + i * 100, "likes": 5, "comments": 0},
)
for i in range(5)
]
good = schema.SourceItem(
item_id="good-yt", source="youtube", title="In-depth analysis", body="Quality content",
url="https://youtube.com/good",
local_relevance=0.6, engagement_score=70,
)
pruned = signals.prune_low_relevance([good] + spam_items)
aspire_ids = [item.item_id for item in pruned if item.item_id.startswith("aspire")]
self.assertEqual(len(aspire_ids), 0, f"All @aspiresnippets items should be pruned, got {aspire_ids}")
# -- Fix 468: YouTube items with transcripts survive relevance pruning --
def test_youtube_with_transcript_survives_pruning_even_with_low_relevance(self):
"""A YouTube item with a non-empty snippet (transcript) should not be
pruned even if its title-only relevance is below the threshold."""
has_transcript = schema.SourceItem(
item_id="yt-transcript",
source="youtube",
title="Short title",
body="Short body",
url="https://youtube.com/watch?v=abc",
snippet="This is a detailed transcript about the topic with substantive discussion...",
local_relevance=0.05,
)
strong = schema.SourceItem(
item_id="yt-strong",
source="youtube",
title="Strong video",
body="Detailed analysis of the topic",
url="https://youtube.com/watch?v=strong",
snippet="Detailed transcript content about the topic",
local_relevance=0.6,
)
pruned = signals.prune_low_relevance([strong, has_transcript], minimum=0.15)
ids = [item.item_id for item in pruned]
self.assertIn("yt-transcript", ids,
"YouTube item with transcript should survive pruning")
self.assertIn("yt-strong", ids, "Strong item should survive")
def test_youtube_without_transcript_is_pruned_normally(self):
"""A YouTube item with no transcript (empty snippet) and low relevance
should still be pruned when stronger items exist."""
no_transcript = schema.SourceItem(
item_id="yt-no-transcript",
source="youtube",
title="Short title",
body="Short body",
url="https://youtube.com/watch?v=xyz",
snippet="",
local_relevance=0.05,
)
strong = schema.SourceItem(
item_id="yt-strong",
source="youtube",
title="Strong video",
body="Detailed analysis of the topic",
url="https://youtube.com/watch?v=strong",
snippet="Detailed transcript content about the topic",
local_relevance=0.6,
)
pruned = signals.prune_low_relevance([strong, no_transcript], minimum=0.15)
ids = [item.item_id for item in pruned]
self.assertIn("yt-strong", ids, "Strong item should survive")
self.assertNotIn("yt-no-transcript", ids,
"YouTube item without transcript should be pruned normally")
def test_youtube_transcript_exemption_does_not_affect_other_sources(self):
"""Non-YouTube items with low relevance are still pruned even if they
have a non-empty snippet (the exemption is YouTube-specific)."""
reddit_with_snippet = schema.SourceItem(
item_id="reddit-snippet",
source="reddit",
title="Short title",
body="Short body",
url="https://reddit.com/r/test",
snippet="Some snippet content",
local_relevance=0.05,
)
strong = schema.SourceItem(
item_id="strong",
source="reddit",
title="Strong post",
body="Detailed analysis of the topic",
url="https://reddit.com/r/strong",
local_relevance=0.5,
)
pruned = signals.prune_low_relevance([strong, reddit_with_snippet], minimum=0.15)
ids = [item.item_id for item in pruned]
self.assertIn("strong", ids, "Strong item should survive")
self.assertNotIn("reddit-snippet", ids,
"Non-YouTube items should still be pruned by relevance threshold")
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink