diff --git a/scripts/last30days.py b/scripts/last30days.py index 2532c25..420fc49 100644 --- a/scripts/last30days.py +++ b/scripts/last30days.py @@ -160,6 +160,7 @@ from lib import ( websearch, xai_x, youtube_yt, + query_type as qt, ) @@ -1692,14 +1693,19 @@ def main(): else: mode = sources + # Detect query type for source tiering and scoring adjustments + query_type = qt.detect_query_type(args.topic) + # Apply --search flag: restrict sources to the specified subset - search_do_hackernews = True - search_do_bluesky = has_bluesky - search_do_truthsocial = has_truthsocial - search_do_polymarket = True - search_run_youtube = has_ytdlp - search_run_tiktok = has_tiktok - search_run_instagram = has_instagram + # Source defaults are query-type-aware (Truth Social always opt-in, + # Bluesky only for query types where it adds signal) + search_do_hackernews = qt.is_source_enabled("hn", query_type) if not args.search else True + search_do_bluesky = has_bluesky and qt.is_source_enabled("bluesky", query_type) + search_do_truthsocial = False # Always opt-in (requires --search truthsocial) + search_do_polymarket = qt.is_source_enabled("polymarket", query_type) + search_run_youtube = has_ytdlp and qt.is_source_enabled("youtube", query_type) + search_run_tiktok = has_tiktok and qt.is_source_enabled("tiktok", query_type) + search_run_instagram = has_instagram and qt.is_source_enabled("instagram", query_type) search_run_xiaohongshu = has_xiaohongshu if args.search: search_sources = parse_search_flag(args.search) @@ -1795,19 +1801,19 @@ def main(): scored_bsky = score.score_bluesky_items(filtered_bsky) if filtered_bsky else [] scored_ts = score.score_truthsocial_items(filtered_ts) if filtered_ts else [] scored_pm = score.score_polymarket_items(filtered_pm) if filtered_pm else [] - scored_web = score.score_websearch_items(filtered_web) if filtered_web else [] + scored_web = score.score_websearch_items(filtered_web, query_type=query_type) if filtered_web else [] - # Sort items - sorted_reddit = score.sort_items(scored_reddit) - sorted_x = score.sort_items(scored_x) - sorted_youtube = score.sort_items(scored_youtube) if scored_youtube else [] - sorted_tiktok = score.sort_items(scored_tiktok) if scored_tiktok else [] - sorted_ig = score.sort_items(scored_ig) if scored_ig else [] - sorted_hn = score.sort_items(scored_hn) if scored_hn else [] - sorted_bsky = score.sort_items(scored_bsky) if scored_bsky else [] - sorted_ts = score.sort_items(scored_ts) if scored_ts else [] - sorted_pm = score.sort_items(scored_pm) if scored_pm else [] - sorted_web = score.sort_items(scored_web) if scored_web else [] + # Sort items (query-type-aware tiebreaker ordering) + sorted_reddit = score.sort_items(scored_reddit, query_type=query_type) + sorted_x = score.sort_items(scored_x, query_type=query_type) + sorted_youtube = score.sort_items(scored_youtube, query_type=query_type) if scored_youtube else [] + sorted_tiktok = score.sort_items(scored_tiktok, query_type=query_type) if scored_tiktok else [] + sorted_ig = score.sort_items(scored_ig, query_type=query_type) if scored_ig else [] + sorted_hn = score.sort_items(scored_hn, query_type=query_type) if scored_hn else [] + sorted_bsky = score.sort_items(scored_bsky, query_type=query_type) if scored_bsky else [] + sorted_ts = score.sort_items(scored_ts, query_type=query_type) if scored_ts else [] + sorted_pm = score.sort_items(scored_pm, query_type=query_type) if scored_pm else [] + sorted_web = score.sort_items(scored_web, query_type=query_type) if scored_web else [] # Dedupe items deduped_reddit = dedupe.dedupe_reddit(sorted_reddit) diff --git a/scripts/lib/query_type.py b/scripts/lib/query_type.py new file mode 100644 index 0000000..b542439 --- /dev/null +++ b/scripts/lib/query_type.py @@ -0,0 +1,109 @@ +"""Query type detection for source selection and scoring adjustments.""" + +import re +from typing import Literal + +QueryType = Literal["product", "concept", "opinion", "how_to", "comparison", "breaking_news", "prediction"] + +# Pattern-based classification (no LLM, no external deps) +_PRODUCT_PATTERNS = re.compile( + r"\b(price|pricing|cost|buy|purchase|deal|discount|subscription|plan|tier|free tier|alternative)\b", re.I +) +_CONCEPT_PATTERNS = re.compile( + r"\b(what is|what are|explain|definition|how does|how do|overview|introduction|guide to|primer)\b", re.I +) +_OPINION_PATTERNS = re.compile( + r"\b(worth it|thoughts on|opinion|review|experience with|recommend|should i|pros and cons|good or bad)\b", re.I +) +_HOWTO_PATTERNS = re.compile( + r"\b(how to|tutorial|step by step|setup|install|configure|deploy|migrate|implement|build a|create a)\b", re.I +) +_COMPARISON_PATTERNS = re.compile( + r"\b(vs\.?|versus|compared to|comparison|better than|or\b.*\bfor\b|difference between|switch from)\b", re.I +) +_BREAKING_PATTERNS = re.compile( + r"\b(latest|breaking|just announced|launched|released|new|update|news|happened|today|this week)\b", re.I +) +_PREDICTION_PATTERNS = re.compile( + r"\b(will|predict|forecast|odds|chance|probability|election|outcome|bet on|market for)\b", re.I +) + + +def detect_query_type(topic: str) -> QueryType: + """Classify a query into a type using pattern matching. + + Returns the most specific match. When multiple patterns match, + priority order is: comparison > how_to > product > opinion > prediction > concept > breaking_news. + """ + # Most specific first + if _COMPARISON_PATTERNS.search(topic): + return "comparison" + if _HOWTO_PATTERNS.search(topic): + return "how_to" + if _PRODUCT_PATTERNS.search(topic): + return "product" + if _OPINION_PATTERNS.search(topic): + return "opinion" + if _PREDICTION_PATTERNS.search(topic): + return "prediction" + if _CONCEPT_PATTERNS.search(topic): + return "concept" + if _BREAKING_PATTERNS.search(topic): + return "breaking_news" + + # Default: treat as breaking news (most common use case for "last 30 days") + return "breaking_news" + + +# Source tiering by query type. +# Tier 1: always run. Tier 2: run if available. Tier 3: opt-in only. +# Sources not listed are implicitly tier 3 (opt-in). +SOURCE_TIERS = { + "product": {"tier1": {"reddit", "x", "youtube"}, "tier2": {"web", "tiktok"}}, + "concept": {"tier1": {"reddit", "hn", "web"}, "tier2": {"youtube", "x"}}, + "opinion": {"tier1": {"reddit", "x"}, "tier2": {"youtube", "bluesky"}}, + "how_to": {"tier1": {"youtube", "reddit", "hn"}, "tier2": {"web"}}, + "comparison": {"tier1": {"reddit", "hn", "youtube"}, "tier2": {"x", "web"}}, + "breaking_news": {"tier1": {"x", "reddit", "web"}, "tier2": {"hn", "bluesky"}}, + "prediction": {"tier1": {"polymarket", "x", "reddit"}, "tier2": {"web"}}, +} + +# WebSearch penalty adjustment by query type. +# Concept/how_to queries benefit from authoritative web sources. +WEBSEARCH_PENALTY_BY_TYPE = { + "product": 15, # default: social discussion > blog posts + "concept": 0, # web docs are the best source + "opinion": 15, # social discussion > blog posts + "how_to": 5, # tutorials on web are valuable + "comparison": 10, # mix of social and web + "breaking_news": 10, # news sites are valuable + "prediction": 15, # social/market data > web articles +} + +# Tiebreaker priority overrides by query type. +# Maps source type name to priority (lower = higher priority). +TIEBREAKER_BY_TYPE = { + "product": {"reddit": 0, "x": 1, "youtube": 2, "tiktok": 3, "instagram": 4, "hn": 5, "web": 6, "polymarket": 7}, + "concept": {"hn": 0, "reddit": 1, "web": 2, "youtube": 3, "x": 4, "tiktok": 5, "instagram": 6, "polymarket": 7}, + "opinion": {"reddit": 0, "x": 1, "bluesky": 2, "youtube": 3, "hn": 4, "tiktok": 5, "web": 6, "polymarket": 7}, + "how_to": {"youtube": 0, "reddit": 1, "hn": 2, "web": 3, "x": 4, "tiktok": 5, "instagram": 6, "polymarket": 7}, + "comparison": {"reddit": 0, "hn": 1, "youtube": 2, "x": 3, "web": 4, "tiktok": 5, "instagram": 6, "polymarket": 7}, + "breaking_news": {"x": 0, "reddit": 1, "web": 2, "hn": 3, "bluesky": 4, "tiktok": 5, "youtube": 6, "polymarket": 7}, + "prediction": {"polymarket": 0, "x": 1, "reddit": 2, "web": 3, "hn": 4, "bluesky": 5, "youtube": 6, "tiktok": 7}, +} + + +def is_source_enabled(source: str, query_type: QueryType, explicitly_requested: bool = False) -> bool: + """Check if a source should run for a given query type. + + Tier 1 and Tier 2 sources are enabled. Tier 3 (unlisted) sources only run + if explicitly requested via --search flag. Truth Social is always opt-in. + """ + if source == "truthsocial": + return explicitly_requested + + if explicitly_requested: + return True + + tiers = SOURCE_TIERS.get(query_type, SOURCE_TIERS["breaking_news"]) + return source in tiers["tier1"] or source in tiers["tier2"] diff --git a/scripts/lib/score.py b/scripts/lib/score.py index f52bb04..33317b8 100644 --- a/scripts/lib/score.py +++ b/scripts/lib/score.py @@ -4,6 +4,7 @@ import math from typing import List, Optional, Union from . import dates, schema +from .query_type import QueryType, WEBSEARCH_PENALTY_BY_TYPE, TIEBREAKER_BY_TYPE # Score weights for Reddit/X (has engagement) WEIGHT_RELEVANCE = 0.45 @@ -642,19 +643,17 @@ def score_polymarket_items(items: List[schema.PolymarketItem]) -> List[schema.Po return items -def score_websearch_items(items: List[schema.WebSearchItem]) -> List[schema.WebSearchItem]: +def score_websearch_items(items: List[schema.WebSearchItem], query_type: QueryType = None) -> List[schema.WebSearchItem]: """Compute scores for WebSearch items WITHOUT engagement metrics. - Uses reweighted formula: 55% relevance + 45% recency - 15pt source penalty. - This ensures WebSearch items rank below comparable Reddit/X items. - - Date confidence adjustments: - - High confidence (URL-verified date): +10 bonus - - Med confidence (snippet-extracted date): no change - - Low confidence (no date signals): -20 penalty + Uses reweighted formula: 55% relevance + 45% recency - penalty. + Penalty varies by query type: concept queries get 0 penalty (web docs + are authoritative), while product/opinion queries get full 15pt penalty + (social discussion is more valuable). Args: items: List of WebSearch items + query_type: Query classification for penalty adjustment Returns: Items with updated scores @@ -682,8 +681,9 @@ def score_websearch_items(items: List[schema.WebSearchItem]) -> List[schema.WebS WEBSEARCH_WEIGHT_RECENCY * rec_score ) - # Apply source penalty (WebSearch < Reddit/X for same relevance/recency) - overall -= WEBSEARCH_SOURCE_PENALTY + # Apply source penalty (varies by query type) + penalty = WEBSEARCH_PENALTY_BY_TYPE.get(query_type, WEBSEARCH_SOURCE_PENALTY) if query_type else WEBSEARCH_SOURCE_PENALTY + overall -= penalty # Apply date confidence adjustments # High confidence (URL-verified): reward with bonus @@ -699,15 +699,33 @@ def score_websearch_items(items: List[schema.WebSearchItem]) -> List[schema.WebS return items -def sort_items(items: List[Union[schema.RedditItem, schema.XItem, schema.WebSearchItem, schema.YouTubeItem, schema.TikTokItem, schema.InstagramItem, schema.HackerNewsItem, schema.PolymarketItem]]) -> List: +_ITEM_SOURCE_MAP = { + schema.RedditItem: "reddit", + schema.XItem: "x", + schema.YouTubeItem: "youtube", + schema.TikTokItem: "tiktok", + schema.InstagramItem: "instagram", + schema.HackerNewsItem: "hn", + schema.PolymarketItem: "polymarket", +} +_DEFAULT_TIEBREAKER = {"reddit": 0, "x": 1, "youtube": 2, "tiktok": 3, "instagram": 4, "hn": 5, "polymarket": 6, "web": 7} + + +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: """Sort items by score (descending), then date, then source priority. + Source priority varies by query type: YouTube ranks first for how_to, + X ranks first for breaking_news, Polymarket ranks first for prediction. + Args: items: List of items to sort + query_type: Query classification for tiebreaker adjustment Returns: Sorted items """ + tiebreaker = TIEBREAKER_BY_TYPE.get(query_type, _DEFAULT_TIEBREAKER) if query_type else _DEFAULT_TIEBREAKER + def sort_key(item): # Primary: score descending (negate for descending) score = -item.score @@ -716,23 +734,9 @@ def sort_items(items: List[Union[schema.RedditItem, schema.XItem, schema.WebSear date = item.date or "0000-00-00" date_key = -int(date.replace("-", "")) - # Tertiary: source priority (Reddit > X > YouTube > TikTok > HN > Polymarket > WebSearch) - if isinstance(item, schema.RedditItem): - source_priority = 0 - elif isinstance(item, schema.XItem): - source_priority = 1 - elif isinstance(item, schema.YouTubeItem): - source_priority = 2 - elif isinstance(item, schema.TikTokItem): - source_priority = 3 - elif isinstance(item, schema.InstagramItem): - source_priority = 4 - elif isinstance(item, schema.HackerNewsItem): - source_priority = 5 - elif isinstance(item, schema.PolymarketItem): - source_priority = 6 - else: # WebSearchItem - source_priority = 7 + # Tertiary: query-type-aware source priority + source_name = _ITEM_SOURCE_MAP.get(type(item), "web") + source_priority = tiebreaker.get(source_name, 99) # Quaternary: title/text for stability text = getattr(item, "title", "") or getattr(item, "text", "") diff --git a/tests/test_query_type.py b/tests/test_query_type.py new file mode 100644 index 0000000..4b0f589 --- /dev/null +++ b/tests/test_query_type.py @@ -0,0 +1,142 @@ +"""Tests for query type detection and source tiering.""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent / "scripts")) + +from lib.query_type import ( + detect_query_type, + is_source_enabled, + WEBSEARCH_PENALTY_BY_TYPE, + TIEBREAKER_BY_TYPE, + SOURCE_TIERS, +) + + +class TestDetectQueryType(unittest.TestCase): + + def test_product_queries(self): + self.assertEqual(detect_query_type("cursor IDE pricing"), "product") + self.assertEqual(detect_query_type("is Claude Pro worth the cost"), "product") + self.assertEqual(detect_query_type("best free tier LLM API"), "product") + + def test_concept_queries(self): + self.assertEqual(detect_query_type("what is WebTransport"), "concept") + self.assertEqual(detect_query_type("explain React Server Components"), "concept") + self.assertEqual(detect_query_type("how does MCP protocol work"), "concept") + + def test_opinion_queries(self): + self.assertEqual(detect_query_type("is cursor worth it"), "opinion") + self.assertEqual(detect_query_type("thoughts on Claude Code"), "opinion") + self.assertEqual(detect_query_type("should i switch to Neovim"), "opinion") + + def test_howto_queries(self): + self.assertEqual(detect_query_type("how to deploy on Vercel"), "how_to") + self.assertEqual(detect_query_type("tutorial for building MCP servers"), "how_to") + self.assertEqual(detect_query_type("step by step Kubernetes setup"), "how_to") + + def test_comparison_queries(self): + self.assertEqual(detect_query_type("cursor vs windsurf"), "comparison") + self.assertEqual(detect_query_type("Claude compared to GPT-5"), "comparison") + self.assertEqual(detect_query_type("difference between React and Vue"), "comparison") + + def test_breaking_news_queries(self): + self.assertEqual(detect_query_type("latest AI funding rounds"), "breaking_news") + self.assertEqual(detect_query_type("OpenAI just announced GPT-6"), "breaking_news") + + def test_prediction_queries(self): + self.assertEqual(detect_query_type("will Trump win 2028 election"), "prediction") + self.assertEqual(detect_query_type("odds of Fed rate cut"), "prediction") + + def test_default_is_breaking_news(self): + self.assertEqual(detect_query_type("tariffs"), "breaking_news") + self.assertEqual(detect_query_type("AI agents"), "breaking_news") + + def test_comparison_beats_product(self): + """Comparison is more specific than product.""" + self.assertEqual(detect_query_type("cursor vs windsurf pricing"), "comparison") + + def test_howto_beats_concept(self): + """How-to is more specific than concept.""" + self.assertEqual(detect_query_type("how to explain transformers"), "how_to") + + +class TestIsSourceEnabled(unittest.TestCase): + + def test_truthsocial_always_opt_in(self): + for qt in ["product", "concept", "opinion", "breaking_news", "prediction"]: + self.assertFalse(is_source_enabled("truthsocial", qt)) + self.assertTrue(is_source_enabled("truthsocial", "breaking_news", explicitly_requested=True)) + + def test_tier1_sources_enabled(self): + self.assertTrue(is_source_enabled("reddit", "product")) + self.assertTrue(is_source_enabled("youtube", "how_to")) + self.assertTrue(is_source_enabled("polymarket", "prediction")) + self.assertTrue(is_source_enabled("x", "breaking_news")) + + def test_tier2_sources_enabled(self): + self.assertTrue(is_source_enabled("web", "product")) + self.assertTrue(is_source_enabled("bluesky", "opinion")) + + def test_tier3_sources_disabled_by_default(self): + self.assertFalse(is_source_enabled("instagram", "concept")) + self.assertFalse(is_source_enabled("tiktok", "comparison")) + self.assertFalse(is_source_enabled("bluesky", "product")) + + def test_explicit_request_overrides_tier(self): + self.assertTrue(is_source_enabled("instagram", "concept", explicitly_requested=True)) + self.assertTrue(is_source_enabled("tiktok", "comparison", explicitly_requested=True)) + + +class TestWebSearchPenalty(unittest.TestCase): + + def test_concept_has_zero_penalty(self): + self.assertEqual(WEBSEARCH_PENALTY_BY_TYPE["concept"], 0) + + def test_product_has_full_penalty(self): + self.assertEqual(WEBSEARCH_PENALTY_BY_TYPE["product"], 15) + + def test_howto_has_reduced_penalty(self): + self.assertLess(WEBSEARCH_PENALTY_BY_TYPE["how_to"], WEBSEARCH_PENALTY_BY_TYPE["product"]) + + def test_all_query_types_have_penalty(self): + for qt in ["product", "concept", "opinion", "how_to", "comparison", "breaking_news", "prediction"]: + self.assertIn(qt, WEBSEARCH_PENALTY_BY_TYPE) + + +class TestTiebreakerPriority(unittest.TestCase): + + def test_youtube_highest_for_howto(self): + self.assertEqual(TIEBREAKER_BY_TYPE["how_to"]["youtube"], 0) + + def test_x_highest_for_breaking_news(self): + self.assertEqual(TIEBREAKER_BY_TYPE["breaking_news"]["x"], 0) + + def test_polymarket_highest_for_prediction(self): + self.assertEqual(TIEBREAKER_BY_TYPE["prediction"]["polymarket"], 0) + + def test_hn_highest_for_concept(self): + self.assertEqual(TIEBREAKER_BY_TYPE["concept"]["hn"], 0) + + def test_all_query_types_have_tiebreakers(self): + for qt in ["product", "concept", "opinion", "how_to", "comparison", "breaking_news", "prediction"]: + self.assertIn(qt, TIEBREAKER_BY_TYPE) + + +class TestSourceTiers(unittest.TestCase): + + def test_all_query_types_have_tiers(self): + for qt in ["product", "concept", "opinion", "how_to", "comparison", "breaking_news", "prediction"]: + self.assertIn(qt, SOURCE_TIERS) + self.assertIn("tier1", SOURCE_TIERS[qt]) + self.assertIn("tier2", SOURCE_TIERS[qt]) + + def test_truthsocial_not_in_any_tier(self): + for qt, tiers in SOURCE_TIERS.items(): + self.assertNotIn("truthsocial", tiers["tier1"], f"truthsocial in tier1 for {qt}") + self.assertNotIn("truthsocial", tiers["tier2"], f"truthsocial in tier2 for {qt}") + + +if __name__ == "__main__": + unittest.main()