From cbee987f65174eecd6d6ada0725d0e4825db3ccc Mon Sep 17 00:00:00 2001 From: Jeffrey Sperling Date: Wed, 11 Mar 2026 19:09:06 -0700 Subject: [PATCH] Extract relevance_filter, add Bluesky/TruthSocial type hint + test coverage - Extract _relevance_filter from last30days.py closure to score.relevance_filter() for testability - Add BlueskyItem/TruthSocialItem to sort_items() type hint (was missing despite being in _ITEM_SOURCE_MAP) - Add tests: Bluesky/TruthSocial engagement scoring, sort_items mixed sources, relevance_filter behavior (threshold, minimum-result guarantee, missing attr), select_openai_model HTTP 401/403 error paths --- scripts/last30days.py | 29 +++----- scripts/lib/score.py | 20 +++++- tests/test_models.py | 21 ++++++ tests/test_score.py | 155 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 203 insertions(+), 22 deletions(-) diff --git a/scripts/last30days.py b/scripts/last30days.py index db99ccc..29365ad 100644 --- a/scripts/last30days.py +++ b/scripts/last30days.py @@ -1830,27 +1830,14 @@ def main(): deduped_web = websearch.dedupe_websearch(sorted_web) if sorted_web else [] # Post-retrieval relevance filter: drop low-relevance items per source - # Only filter when there are enough items (>3) to avoid empty results - def _relevance_filter(items, source_name, threshold=0.3): - """Filter items below relevance threshold with minimum-result guarantee.""" - if len(items) <= 3: - return items - passed = [i for i in items if getattr(i, 'relevance', 0.0) >= threshold] - if not passed: - # Keep top 3 by relevance if all filtered - print(f"[{source_name} WARNING] All results below relevance {threshold}, keeping top 3", file=sys.stderr) - by_rel = sorted(items, key=lambda x: getattr(x, 'relevance', 0.0), reverse=True) - return by_rel[:3] - return passed - - deduped_reddit = _relevance_filter(deduped_reddit, "REDDIT") - deduped_x = _relevance_filter(deduped_x, "X") - deduped_youtube = _relevance_filter(deduped_youtube, "YOUTUBE") - deduped_tiktok = _relevance_filter(deduped_tiktok, "TIKTOK") - deduped_ig = _relevance_filter(deduped_ig, "INSTAGRAM") - deduped_hn = _relevance_filter(deduped_hn, "HN") - deduped_bsky = _relevance_filter(deduped_bsky, "BLUESKY") - deduped_ts = _relevance_filter(deduped_ts, "TRUTHSOCIAL") + deduped_reddit = score.relevance_filter(deduped_reddit, "REDDIT") + deduped_x = score.relevance_filter(deduped_x, "X") + deduped_youtube = score.relevance_filter(deduped_youtube, "YOUTUBE") + deduped_tiktok = score.relevance_filter(deduped_tiktok, "TIKTOK") + deduped_ig = score.relevance_filter(deduped_ig, "INSTAGRAM") + deduped_hn = score.relevance_filter(deduped_hn, "HN") + deduped_bsky = score.relevance_filter(deduped_bsky, "BLUESKY") + deduped_ts = score.relevance_filter(deduped_ts, "TRUTHSOCIAL") # Cross-source linking: annotate items that discuss the same story dedupe.cross_source_link( diff --git a/scripts/lib/score.py b/scripts/lib/score.py index ffd108d..9ac8502 100644 --- a/scripts/lib/score.py +++ b/scripts/lib/score.py @@ -715,7 +715,7 @@ _ITEM_SOURCE_MAP = { _DEFAULT_TIEBREAKER = {"reddit": 0, "x": 1, "youtube": 2, "tiktok": 3, "instagram": 4, "hn": 5, "bluesky": 6, "truthsocial": 7, "polymarket": 8, "web": 9} -def sort_items(items: List[Union[schema.RedditItem, schema.XItem, schema.WebSearchItem, schema.YouTubeItem, schema.TikTokItem, schema.InstagramItem, schema.HackerNewsItem, schema.PolymarketItem]], query_type: QueryType = None) -> List: +def sort_items(items: List[Union[schema.RedditItem, schema.XItem, schema.WebSearchItem, schema.YouTubeItem, schema.TikTokItem, schema.InstagramItem, schema.HackerNewsItem, schema.BlueskyItem, schema.TruthSocialItem, schema.PolymarketItem]], query_type: QueryType = None) -> List: """Sort items by score (descending), then date, then source tiebreaker. Tiebreaker (tertiary sort key, after score and date): source priority @@ -749,3 +749,21 @@ def sort_items(items: List[Union[schema.RedditItem, schema.XItem, schema.WebSear return (score, date_key, source_priority, text) return sorted(items, key=sort_key) + + +def relevance_filter(items, source_name: str, threshold: float = 0.3): + """Filter items below relevance threshold with minimum-result guarantee. + + Items with no relevance attribute are treated as 0.0 (fail the filter). + If all items are below threshold, keeps the top 3 by relevance. + Lists with 3 or fewer items are returned unchanged. + """ + import sys + if len(items) <= 3: + return items + passed = [i for i in items if getattr(i, 'relevance', 0.0) >= threshold] + if not passed: + print(f"[{source_name} WARNING] All results below relevance {threshold}, keeping top 3", file=sys.stderr) + by_rel = sorted(items, key=lambda x: getattr(x, 'relevance', 0.0), reverse=True) + return by_rel[:3] + return passed diff --git a/tests/test_models.py b/tests/test_models.py index e68a08a..f093f27 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -140,6 +140,27 @@ class TestSelectOpenAIModel(unittest.TestCase): self.assertEqual(result, "gpt-4.1-mini") +class TestSelectOpenAIModelErrorPaths(unittest.TestCase): + def setUp(self): + from lib import cache + cache.MODEL_CACHE_FILE.unlink(missing_ok=True) + + def test_http_error_returns_fallback(self): + """HTTPError during model fetch should return fallback, not crash.""" + from unittest.mock import patch + from lib import http + with patch('lib.http.get', side_effect=http.HTTPError("Unauthorized", status_code=401)): + result = models.select_openai_model("bad-key", policy="auto") + self.assertEqual(result, models.OPENAI_FALLBACK_MODELS[0]) + + def test_http_403_returns_fallback(self): + from unittest.mock import patch + from lib import http + with patch('lib.http.get', side_effect=http.HTTPError("Forbidden", status_code=403)): + result = models.select_openai_model("bad-key", policy="auto") + self.assertEqual(result, models.OPENAI_FALLBACK_MODELS[0]) + + class TestSelectXAIModel(unittest.TestCase): def test_latest_policy(self): result = models.select_xai_model( diff --git a/tests/test_score.py b/tests/test_score.py index c5d010e..436258e 100644 --- a/tests/test_score.py +++ b/tests/test_score.py @@ -192,5 +192,160 @@ class TestInstagramEngagement(unittest.TestCase): ) +class TestBlueskyEngagement(unittest.TestCase): + """Tests for compute_bluesky_engagement_raw().""" + + def test_basic(self): + eng = schema.Engagement(likes=100, reposts=25, replies=15, quotes=5) + raw = score.compute_bluesky_engagement_raw(eng) + self.assertIsNotNone(raw) + self.assertGreater(raw, 0) + + def test_likes_dominate(self): + likes_heavy = schema.Engagement(likes=1000, reposts=0, replies=0, quotes=0) + reposts_heavy = schema.Engagement(likes=0, reposts=1000, replies=0, quotes=0) + self.assertGreater( + score.compute_bluesky_engagement_raw(likes_heavy), + score.compute_bluesky_engagement_raw(reposts_heavy), + ) + + def test_none_engagement(self): + self.assertIsNone(score.compute_bluesky_engagement_raw(None)) + + def test_no_likes_no_reposts(self): + eng = schema.Engagement(replies=10) + self.assertIsNone(score.compute_bluesky_engagement_raw(eng)) + + +class TestTruthSocialEngagement(unittest.TestCase): + """Tests for compute_truthsocial_engagement_raw().""" + + def test_basic(self): + eng = schema.Engagement(likes=100, reposts=25, replies=15) + raw = score.compute_truthsocial_engagement_raw(eng) + self.assertIsNotNone(raw) + self.assertGreater(raw, 0) + + def test_likes_dominate(self): + likes_heavy = schema.Engagement(likes=1000, reposts=0, replies=0) + reposts_heavy = schema.Engagement(likes=0, reposts=1000, replies=0) + self.assertGreater( + score.compute_truthsocial_engagement_raw(likes_heavy), + score.compute_truthsocial_engagement_raw(reposts_heavy), + ) + + def test_none_engagement(self): + self.assertIsNone(score.compute_truthsocial_engagement_raw(None)) + + +class TestScoreBlueskyItems(unittest.TestCase): + """Tests for score_bluesky_items().""" + + def test_scores_items(self): + items = [ + schema.BlueskyItem( + id="bsky1", text="Test", url="https://bsky.app/1", + author_handle="user.bsky.social", display_name="User", + engagement=schema.Engagement(likes=50, reposts=10, replies=5, quotes=2), + relevance=0.8, + ), + ] + result = score.score_bluesky_items(items) + self.assertEqual(len(result), 1) + self.assertGreater(result[0].score, 0) + + def test_empty_list(self): + self.assertEqual(score.score_bluesky_items([]), []) + + +class TestScoreTruthSocialItems(unittest.TestCase): + """Tests for score_truthsocial_items().""" + + def test_scores_items(self): + items = [ + schema.TruthSocialItem( + id="ts1", text="Test", url="https://truthsocial.com/1", + author_handle="@user", display_name="User", + engagement=schema.Engagement(likes=50, reposts=10, replies=5), + relevance=0.8, + ), + ] + result = score.score_truthsocial_items(items) + self.assertEqual(len(result), 1) + self.assertGreater(result[0].score, 0) + + def test_empty_list(self): + self.assertEqual(score.score_truthsocial_items([]), []) + + +class TestSortItemsMixedSources(unittest.TestCase): + """Test sort_items with Bluesky and TruthSocial items.""" + + def test_bluesky_item_sorts(self): + items = [ + schema.RedditItem(id="R1", title="Reddit", url="", subreddit="", score=30), + schema.BlueskyItem(id="B1", text="Bluesky", url="", author_handle="u.bsky.social", display_name="U", score=90), + ] + result = score.sort_items(items) + self.assertEqual(result[0].id, "B1") + + def test_truthsocial_item_sorts(self): + items = [ + schema.RedditItem(id="R1", title="Reddit", url="", subreddit="", score=30), + schema.TruthSocialItem(id="T1", text="TS", url="", author_handle="@u", display_name="U", score=90), + ] + result = score.sort_items(items) + self.assertEqual(result[0].id, "T1") + + +class TestRelevanceFilter(unittest.TestCase): + """Tests for relevance_filter().""" + + def _make_items(self, relevances): + """Helper: create RedditItems with given relevance values.""" + return [ + schema.RedditItem(id=f"R{i}", title=f"Item {i}", url="", subreddit="", relevance=r) + for i, r in enumerate(relevances) + ] + + def test_filters_below_threshold(self): + items = self._make_items([0.8, 0.1, 0.5, 0.2]) + result = score.relevance_filter(items, "TEST", threshold=0.3) + self.assertEqual(len(result), 2) + self.assertTrue(all(i.relevance >= 0.3 for i in result)) + + def test_small_list_unchanged(self): + items = self._make_items([0.1, 0.05, 0.02]) + result = score.relevance_filter(items, "TEST") + self.assertEqual(len(result), 3) + + def test_all_below_threshold_keeps_top_3(self): + items = self._make_items([0.1, 0.25, 0.05, 0.2, 0.15]) + result = score.relevance_filter(items, "TEST", threshold=0.3) + self.assertEqual(len(result), 3) + # Should be sorted by relevance: 0.25, 0.2, 0.15 + self.assertEqual(result[0].relevance, 0.25) + self.assertEqual(result[1].relevance, 0.2) + + def test_empty_list(self): + result = score.relevance_filter([], "TEST") + self.assertEqual(result, []) + + def test_items_without_relevance_attr_treated_as_zero(self): + """Objects lacking a relevance attribute get 0.0, failing the filter.""" + class BareItem: + def __init__(self, id): + self.id = id + items = [ + schema.RedditItem(id="R0", title="Has relevance", url="", subreddit="", relevance=0.8), + BareItem("B1"), + BareItem("B2"), + BareItem("B3"), + ] + result = score.relevance_filter(items, "TEST", threshold=0.3) + self.assertEqual(len(result), 1) + self.assertEqual(result[0].id, "R0") + + if __name__ == "__main__": unittest.main()