sawz/last30days-skill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
082efe03e3e8f5b2b4ed041fc7bce74abc13b82d
last30days-skill/tests/test_signals_v3.py
T
Matt Van Horn 082efe03e3 feat: surface YouTube + TikTok top comments alongside Reddit (#260) 
* feat(normalize): pass YouTube top_comments through with Reddit-compatible shape

_normalize_youtube silently dropped top_comments after enrich_with_comments
populated them, so the downstream signals/render/entity layers never saw
YouTube comments. Map likes->score and text->excerpt so the existing
Reddit-compatible readers Just Work.

Shared _remap_comments helper will be reused for TikTok in a later commit.

* feat(tiktok): fetch top comments via ScrapeCreators when opted in

Mirrors the youtube_comments pattern: new env.is_tiktok_comments_available
gate (requires SCRAPECREATORS_API_KEY + tiktok_comments in INCLUDE_SOURCES),
tiktok.enrich_with_comments ranks posts and fetches via
GET /v1/tiktok/video/comments. Vote field is digg_count; text and user.nickname
come across verbatim. Pipeline calls the enricher right after TikTok search
when the gate is open.

Comment-fetch errors never crash the pipeline — the enricher returns an
empty list on 4xx/5xx.

* feat(normalize): pass TikTok top_comments through with digg_count->score mapping

Instagram uses the same shortform normalizer and has no comment fetcher
today, so the key is harmlessly absent there — no Instagram regression.

* feat(signals): add YouTube + TikTok top-comment score to engagement formula

Mirrors Reddit's 10% top-comment slot. Without top_comments present, the
formula reduces to views-dominant weighting; with a high-signal comment,
the item gets a meaningful bump (log1p(10k) ~ 9.2, weighted 0.10 = ~0.92
on the engagement score).

Updated the existing dominant-weight and missing-fields tests to the new
weights (0.45/0.32/0.13 for YT, 0.45/0.27/0.18 for TT). Views still dominate.

* feat(render): source-aware thresholds and vote labels for top comments

10 upvotes on Reddit signals community interest; 10 likes on a viral
TikTok is noise. Introduce per-source minimums (reddit 10, youtube 50,
tiktok 500) and native vote labels ('upvotes' for Reddit, 'likes' for
YT/TT). First-pass numbers — tune after live observation.

* docs: generalize top-comment quoting to YouTube + TikTok, add tiktok_comments opt-in

Synthesis instructions previously called out Reddit top comments only.
Now cover Reddit/YouTube/TikTok uniformly with source-appropriate vote
labels (upvotes vs likes), and explicitly frame YT transcript highlights
and comments as complementary signals. README and setup-wizard copy
document the new tiktok_comments INCLUDE_SOURCES token.

---------

Co-authored-by: Matt Van Horn <455140+mvanhorn@users.noreply.github.com>
2026-04-15 08:26:06 -04:00

715 lines
28 KiB
Python
Raw Blame History

import math
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
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_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)
expected = (
0.50 * math.log1p(8000)
+ 0.30 * math.log1p(1500)
+ 0.20 * 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.20 * 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}")
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink