sawz/last30days-skill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #65 from j-sperling/feat/search-quality-consolidation

Browse Source 
Consolidate query/relevance modules and improve search quality
This commit is contained in:
Matt Van Horn
2026-03-14 07:31:24 -07:00
parent 3830274e11 058c4e1899
commit 52a22f5cb1
32 changed files with 1978 additions and 538 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/evaluate_search_quality.py
+641
View File
@@ -0,0 +1,641 @@
#!/usr/bin/env python3
"""Run local search-quality evaluations across fixed topics.
This is an optional local gate, not a required CI job. It compares a baseline
revision against a candidate checkout, computes deterministic regression
metrics, and optionally calls Gemini as a judge for graded relevance labels.
"""
from __future__ import annotations
import argparse
import json
import math
import os
import shlex
import shutil
import subprocess
import sys
import tempfile
import textwrap
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
sys.path.insert(0, str(Path(__file__).parent))
from lib import env as envlib
REPO_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_TOPICS: List[Tuple[str, str]] = [
("nano banana pro prompting", "product"),
("codex vs claude code", "comparison"),
("anthropic odds", "prediction"),
("kanye west", "breaking_news"),
("remotion animations for Claude Code", "how_to"),
]
DEFAULT_SEARCH = "reddit,x,youtube,hn,polymarket"
SOURCE_KEYS = [
"reddit",
"x",
"youtube",
"tiktok",
"instagram",
"hackernews",
"bluesky",
"truthsocial",
"polymarket",
"websearch",
]
DEFAULT_JUDGE_MODEL = "gemini-3-pro-preview"
GEMINI_API_URL = "https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={api_key}"
def slugify(topic: str) -> str:
return "".join(c.lower() if c.isalnum() else "-" for c in topic).strip("-")
def path_without_node(path_value: str) -> str:
parts = []
for entry in path_value.split(os.pathsep):
if not entry:
continue
if (Path(entry) / "node").exists():
continue
parts.append(entry)
return os.pathsep.join(parts)
def write_exec_wrapper(path: Path, target: str, fixed_args: List[str]) -> None:
quoted_target = shlex.quote(target)
quoted_args = " ".join(shlex.quote(arg) for arg in fixed_args)
path.write_text(f"#!/bin/sh\nexec {quoted_target} {quoted_args} \"$@\"\n")
path.chmod(0o755)
def create_eval_tool_path(eval_home: Path, base_path: str) -> str:
"""Create safe wrapper binaries for local evaluation subprocesses."""
bin_dir = eval_home / "bin"
bin_dir.mkdir(parents=True, exist_ok=True)
real_ytdlp = shutil.which("yt-dlp")
if real_ytdlp:
write_exec_wrapper(
bin_dir / "yt-dlp",
real_ytdlp,
["--ignore-config", "--no-cookies-from-browser"],
)
if not base_path:
return str(bin_dir)
return os.pathsep.join([str(bin_dir), base_path])
def stable_item_key(source: str, item: Dict[str, Any]) -> str:
url = str(item.get("url") or "").strip()
if url:
return url
item_id = str(item.get("id") or "").strip()
text = item_text(source, item)
return f"{source}:{item_id}:{text[:120]}"
def item_text(source: str, item: Dict[str, Any]) -> str:
if source in {"x", "bluesky", "truthsocial"}:
return str(item.get("text") or "").strip()
if source == "polymarket":
return str(item.get("question") or item.get("title") or "").strip()
return str(item.get("title") or "").strip()
def build_ranked_items(report: Dict[str, Any], per_source_limit: int) -> List[Dict[str, Any]]:
ranked: List[Dict[str, Any]] = []
for source in SOURCE_KEYS:
items = list(report.get(source) or [])[:per_source_limit]
for item in items:
ranked.append({
"source": source,
"key": stable_item_key(source, item),
"url": str(item.get("url") or "").strip(),
"text": item_text(source, item),
"score": float(item.get("score") or 0),
"relevance": float(item.get("relevance") or 0),
"date": item.get("date"),
})
ranked.sort(key=lambda item: (-item["score"], item["source"], item["key"]))
return ranked
def url_sets_by_source(report: Dict[str, Any]) -> Dict[str, set[str]]:
result: Dict[str, set[str]] = {}
for source in SOURCE_KEYS:
items = report.get(source) or []
urls = {
stable_item_key(source, item)
for item in items
}
result[source] = urls
return result
def jaccard(left: Iterable[str], right: Iterable[str]) -> float:
left_set = set(left)
right_set = set(right)
if not left_set and not right_set:
return 1.0
union = left_set | right_set
if not union:
return 1.0
return len(left_set & right_set) / len(union)
def retention(left: Iterable[str], right: Iterable[str]) -> float:
left_set = set(left)
right_set = set(right)
if not left_set:
return 1.0
return len(left_set & right_set) / len(left_set)
def precision_at_k(ranking: List[Dict[str, Any]], judgments: Dict[str, int], k: int) -> float:
top = ranking[:k]
if not top:
return 0.0
hits = sum(1 for item in top if judgments.get(item["key"], 0) >= 2)
return hits / len(top)
def ndcg_at_k(
ranking: List[Dict[str, Any]],
judgments: Dict[str, int],
k: int,
judged_pool: Optional[List[Dict[str, Any]]] = None,
) -> float:
top = ranking[:k]
if not top:
return 0.0
def dcg(grades: List[int]) -> float:
total = 0.0
for index, grade in enumerate(grades, start=1):
total += (2**grade - 1) / math.log2(index + 1)
return total
actual = [judgments.get(item["key"], 0) for item in top]
ideal_candidates = judged_pool or ranking
ideal = sorted(
(judgments.get(item["key"], 0) for item in ideal_candidates),
reverse=True,
)[:len(top)]
ideal_score = dcg(ideal)
if ideal_score == 0:
return 0.0
return dcg(actual) / ideal_score
def source_coverage_recall(
ranking: List[Dict[str, Any]],
judged_pool: List[Dict[str, Any]],
judgments: Dict[str, int],
) -> float:
good_sources = {item["source"] for item in judged_pool if judgments.get(item["key"], 0) >= 2}
if not good_sources:
return 1.0
hit_sources = {
item["source"]
for item in ranking
if judgments.get(item["key"], 0) >= 2
}
return len(hit_sources & good_sources) / len(good_sources)
def create_eval_env(include_web: bool) -> Tuple[Dict[str, str], Path]:
config = envlib.get_config()
eval_home = Path(tempfile.mkdtemp(prefix="last30days-eval-home-"))
(eval_home / ".config").mkdir(parents=True, exist_ok=True)
safe_path = create_eval_tool_path(
eval_home,
path_without_node(os.environ.get("PATH", "")),
)
passthrough = {
"HOME": str(eval_home),
"XDG_CONFIG_HOME": str(eval_home / ".config"),
"PATH": safe_path,
"LANG": os.environ.get("LANG", "en_US.UTF-8"),
"LC_ALL": os.environ.get("LC_ALL", ""),
"TMPDIR": os.environ.get("TMPDIR", ""),
"PYTHONUTF8": "1",
"LAST30DAYS_CONFIG_DIR": "",
"BIRD_DISABLE_BROWSER_COOKIES": "1",
"LAST30DAYS_DISABLE_BROWSER_COOKIES": "1",
}
for key in ("OPENAI_API_KEY", "XAI_API_KEY", "SCRAPECREATORS_API_KEY"):
value = config.get(key)
if value:
passthrough[key] = value
if include_web:
for key in ("PARALLEL_API_KEY", "BRAVE_API_KEY", "OPENROUTER_API_KEY"):
value = config.get(key)
if value:
passthrough[key] = value
return passthrough, eval_home
def run_last30days(
repo_dir: Path,
topic: str,
*,
search: str,
timeout_seconds: int,
include_web: bool,
env: Dict[str, str],
) -> Tuple[Dict[str, Any], str]:
cmd = [
sys.executable,
"scripts/last30days.py",
topic,
"--emit",
"json",
"--search",
search,
"--timeout",
str(timeout_seconds),
]
if not include_web:
cmd.append("--no-native-web")
result = subprocess.run(
cmd,
cwd=repo_dir,
env=env,
capture_output=True,
text=True,
timeout=timeout_seconds + 30,
check=False,
)
if result.returncode != 0:
raise RuntimeError(
f"{repo_dir.name} failed for '{topic}' with exit {result.returncode}\n{result.stderr.strip()}"
)
return json.loads(result.stdout), result.stderr
def create_worktree(rev: str) -> Path:
worktree_dir = Path(tempfile.mkdtemp(prefix="last30days-eval-"))
subprocess.run(
["git", "worktree", "add", "--detach", str(worktree_dir), rev],
cwd=REPO_ROOT,
check=True,
capture_output=True,
text=True,
)
return worktree_dir
def remove_worktree(path: Path) -> None:
subprocess.run(
["git", "worktree", "remove", "--force", str(path)],
cwd=REPO_ROOT,
check=False,
capture_output=True,
text=True,
)
shutil.rmtree(path, ignore_errors=True)
def extract_gemini_text(payload: Dict[str, Any]) -> str:
for candidate in payload.get("candidates", []):
content = candidate.get("content") or {}
for part in content.get("parts", []):
text = part.get("text")
if text:
return text
raise ValueError("Gemini response did not contain text")
def resolve_google_judge_api_key(config: Dict[str, Any]) -> Optional[str]:
"""Resolve the local canonical Google API key name.
This workspace conventionally uses GOOGLE_API_KEY. We also accept the
more Gemini-specific aliases for portability.
"""
return (
os.environ.get("GOOGLE_API_KEY")
or config.get("GOOGLE_API_KEY")
or os.environ.get("GEMINI_API_KEY")
or config.get("GEMINI_API_KEY")
or os.environ.get("GOOGLE_GENAI_API_KEY")
or config.get("GOOGLE_GENAI_API_KEY")
)
def call_gemini_judge(api_key: str, model: str, prompt: str) -> Dict[str, Any]:
body = {
"contents": [{"parts": [{"text": prompt}]}],
"generationConfig": {
"temperature": 0,
"responseMimeType": "application/json",
},
}
url = GEMINI_API_URL.format(model=model, api_key=api_key)
request = Request(
url,
data=json.dumps(body).encode("utf-8"),
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urlopen(request, timeout=120) as response:
payload = json.loads(response.read().decode("utf-8"))
except HTTPError as exc:
detail = exc.read().decode("utf-8", errors="replace")
raise RuntimeError(f"Gemini HTTP {exc.code}: {detail}") from exc
except URLError as exc:
raise RuntimeError(f"Gemini request failed: {exc}") from exc
return json.loads(extract_gemini_text(payload))
def build_judge_prompt(
*,
topic: str,
query_type: str,
items: List[Dict[str, Any]],
) -> str:
item_lines = []
for item in items:
item_lines.append(
"\n".join([
f"- id: {item['key']}",
f" source: {item['source']}",
f" title: {item['text'][:220]}",
f" url: {item['url']}",
f" date: {item.get('date') or 'unknown'}",
])
)
joined = "\n".join(item_lines)
return textwrap.dedent(
f"""
Judge search-result relevance for a last-30-days research tool.
Topic: {topic}
Query type: {query_type}
Score each item on this 0-3 scale:
- 0 = off-topic or clearly bad
- 1 = weak or tangential
- 2 = relevant and useful
- 3 = highly relevant, one of the best results
Focus on actual user intent, not just token overlap. Penalize items that
only match generic words like "odds", "review", or "tips" without
matching the real entity or subject. Favor items that would genuinely
help answer the topic in the context of recent discussion.
Return strict JSON with this shape:
{{
"judgments": [
{{"id": "ITEM_ID", "grade": 0, "reason": "short reason"}}
]
}}
Items:
{joined}
"""
).strip()
def get_judgments(
*,
output_dir: Path,
slug: str,
topic: str,
query_type: str,
items: List[Dict[str, Any]],
judge_model: str,
gemini_api_key: Optional[str],
) -> Dict[str, int]:
cache_file = output_dir / "judgments" / f"{slug}.json"
cache_file.parent.mkdir(parents=True, exist_ok=True)
if cache_file.exists():
cached = json.loads(cache_file.read_text())
return {entry["id"]: int(entry["grade"]) for entry in cached.get("judgments", [])}
if not gemini_api_key:
return {}
prompt = build_judge_prompt(topic=topic, query_type=query_type, items=items)
payload = call_gemini_judge(gemini_api_key, judge_model, prompt)
cache_file.write_text(json.dumps(payload, indent=2))
return {entry["id"]: int(entry["grade"]) for entry in payload.get("judgments", [])}
def summarize_topic(
*,
topic: str,
query_type: str,
baseline_report: Dict[str, Any],
candidate_report: Dict[str, Any],
judged_pool: List[Dict[str, Any]],
judgments: Dict[str, int],
per_source_limit: int,
) -> Dict[str, Any]:
baseline_ranked = build_ranked_items(baseline_report, per_source_limit)
candidate_ranked = build_ranked_items(candidate_report, per_source_limit)
baseline_sets = url_sets_by_source(baseline_report)
candidate_sets = url_sets_by_source(candidate_report)
metrics = {
"topic": topic,
"query_type": query_type,
"baseline": {
"precision_at_5": precision_at_k(baseline_ranked, judgments, 5),
"ndcg_at_5": ndcg_at_k(baseline_ranked, judgments, 5, judged_pool),
"source_coverage_recall": source_coverage_recall(baseline_ranked, judged_pool, judgments),
},
"candidate": {
"precision_at_5": precision_at_k(candidate_ranked, judgments, 5),
"ndcg_at_5": ndcg_at_k(candidate_ranked, judgments, 5, judged_pool),
"source_coverage_recall": source_coverage_recall(candidate_ranked, judged_pool, judgments),
},
"stability": {
"overall_jaccard": jaccard(
set().union(*baseline_sets.values()),
set().union(*candidate_sets.values()),
),
"overall_retention_vs_baseline": retention(
set().union(*baseline_sets.values()),
set().union(*candidate_sets.values()),
),
"per_source": {
source: {
"baseline_count": len(baseline_sets[source]),
"candidate_count": len(candidate_sets[source]),
"jaccard": jaccard(baseline_sets[source], candidate_sets[source]),
"retention_vs_baseline": retention(baseline_sets[source], candidate_sets[source]),
}
for source in SOURCE_KEYS
},
},
}
return metrics
def write_markdown_summary(
output_dir: Path,
baseline_label: str,
candidate_label: str,
topic_summaries: List[Dict[str, Any]],
) -> None:
lines = [
f"# Search Quality Evaluation",
"",
f"- Baseline: `{baseline_label}`",
f"- Candidate: `{candidate_label}`",
f"- Generated: {datetime.now().isoformat(timespec='seconds')}",
"",
"## Topic Metrics",
"",
"| Topic | Base P@5 | Cand P@5 | Base nDCG@5 | Cand nDCG@5 | Jaccard | Retention |",
"|---|---:|---:|---:|---:|---:|---:|",
]
for summary in topic_summaries:
lines.append(
"| {topic} | {bp:.2f} | {cp:.2f} | {bn:.2f} | {cn:.2f} | {jac:.2f} | {ret:.2f} |".format(
topic=summary["topic"],
bp=summary["baseline"]["precision_at_5"],
cp=summary["candidate"]["precision_at_5"],
bn=summary["baseline"]["ndcg_at_5"],
cn=summary["candidate"]["ndcg_at_5"],
jac=summary["stability"]["overall_jaccard"],
ret=summary["stability"]["overall_retention_vs_baseline"],
)
)
lines.append("")
lines.append("## Notes")
lines.append("")
lines.append("- `Precision@5` and `nDCG@5` depend on the judged union pool, not a full gold corpus.")
lines.append("- `Source coverage recall` measures whether a run surfaced at least one judged-good result from the good sources in the judged pool.")
lines.append("- `Jaccard` and `retention` are stability guards against baseline drift, not truth metrics.")
(output_dir / "summary.md").write_text("\n".join(lines))
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Evaluate last30days search quality locally")
parser.add_argument("--baseline-rev", default="origin/main", help="Git revision for the baseline run")
parser.add_argument("--candidate-rev", default=None, help="Optional git revision for the candidate run")
parser.add_argument("--no-default-topics", action="store_true", help="Do not include the built-in 5-topic suite")
parser.add_argument("--topic", action="append", default=[], help="Extra topic to evaluate (repeatable)")
parser.add_argument("--search", default=DEFAULT_SEARCH, help="Comma-separated sources passed to --search")
parser.add_argument("--timeout", type=int, default=180, help="Per-topic timeout passed to last30days")
parser.add_argument("--per-source-limit", type=int, default=5, help="Items per source to judge")
parser.add_argument("--include-web", action="store_true", help="Include web-search keys and native web backends")
parser.add_argument("--judge-model", default=None, help="Gemini judge model override")
parser.add_argument("--judge-provider", choices=["auto", "gemini", "none"], default="auto")
parser.add_argument("--keep-worktrees", action="store_true", help="Leave temporary baseline/candidate worktrees on disk")
parser.add_argument("--output-dir", default=None, help="Output directory (default: docs/test-results/search-quality-<timestamp>)")
return parser.parse_args()
def main() -> int:
args = parse_args()
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
output_dir = Path(args.output_dir) if args.output_dir else REPO_ROOT / "docs" / "test-results" / f"search-quality-{timestamp}"
output_dir.mkdir(parents=True, exist_ok=True)
topics = [] if args.no_default_topics else list(DEFAULT_TOPICS)
topics.extend((topic, "custom") for topic in args.topic)
if not topics:
raise SystemExit("No topics configured. Use the default suite or pass --topic.")
judge_config = envlib.get_config()
judge_provider = args.judge_provider
gemini_api_key = resolve_google_judge_api_key(judge_config)
judge_model = args.judge_model or judge_config.get("GEMINI_MODEL") or DEFAULT_JUDGE_MODEL
if judge_provider == "auto":
judge_provider = "gemini" if gemini_api_key else "none"
if judge_provider == "none":
gemini_api_key = None
eval_env, eval_home = create_eval_env(include_web=args.include_web)
baseline_dir = create_worktree(args.baseline_rev)
candidate_dir = create_worktree(args.candidate_rev) if args.candidate_rev else REPO_ROOT
baseline_label = args.baseline_rev
candidate_label = args.candidate_rev or "working-tree"
topic_summaries: List[Dict[str, Any]] = []
try:
for topic, query_type in topics:
slug = slugify(topic)
baseline_report, baseline_stderr = run_last30days(
baseline_dir,
topic,
search=args.search,
timeout_seconds=args.timeout,
include_web=args.include_web,
env=eval_env,
)
candidate_report, candidate_stderr = run_last30days(
candidate_dir,
topic,
search=args.search,
timeout_seconds=args.timeout,
include_web=args.include_web,
env=eval_env,
)
topic_dir = output_dir / slug
topic_dir.mkdir(parents=True, exist_ok=True)
(topic_dir / "baseline.json").write_text(json.dumps(baseline_report, indent=2))
(topic_dir / "candidate.json").write_text(json.dumps(candidate_report, indent=2))
(topic_dir / "baseline.stderr.txt").write_text(baseline_stderr)
(topic_dir / "candidate.stderr.txt").write_text(candidate_stderr)
baseline_ranked = build_ranked_items(baseline_report, args.per_source_limit)
candidate_ranked = build_ranked_items(candidate_report, args.per_source_limit)
union_map = {item["key"]: item for item in baseline_ranked + candidate_ranked}
judgments = get_judgments(
output_dir=output_dir,
slug=slug,
topic=topic,
query_type=query_type,
items=list(union_map.values()),
judge_model=judge_model,
gemini_api_key=gemini_api_key,
)
summary = summarize_topic(
topic=topic,
query_type=query_type,
baseline_report=baseline_report,
candidate_report=candidate_report,
judged_pool=list(union_map.values()),
judgments=judgments,
per_source_limit=args.per_source_limit,
)
topic_summaries.append(summary)
payload = {
"baseline": baseline_label,
"candidate": candidate_label,
"judge_provider": judge_provider,
"judge_model": judge_model if gemini_api_key else None,
"topics": topic_summaries,
}
(output_dir / "summary.json").write_text(json.dumps(payload, indent=2))
write_markdown_summary(output_dir, baseline_label, candidate_label, topic_summaries)
print(output_dir)
return 0
finally:
if not args.keep_worktrees:
remove_worktree(baseline_dir)
if args.candidate_rev:
remove_worktree(candidate_dir)
shutil.rmtree(eval_home, ignore_errors=True)
if __name__ == "__main__":
raise SystemExit(main())
scripts/last30days.py
+12 -8
View File
@@ -353,7 +353,7 @@ def _search_x(
raw_response = {"error": str(e)}
x_error = f"{type(e).__name__}: {e}"
x_items = bird_x.parse_bird_response(raw_response or {})
x_items = bird_x.parse_bird_response(raw_response or {}, query=topic)
# Check for error in response (Bird returns list on success, dict on error)
if raw_response and isinstance(raw_response, dict) and raw_response.get("error") and not x_error:
@@ -508,7 +508,7 @@ def _search_hackernews(
except Exception as e:
return [], f"{type(e).__name__}: {e}"
hn_items = hackernews.parse_hackernews_response(response)
hn_items = hackernews.parse_hackernews_response(response, query=topic)
if response.get("error"):
hn_error = response["error"]
@@ -1829,12 +1829,16 @@ def main():
deduped_pm = dedupe.dedupe_polymarket(sorted_pm) if sorted_pm else []
deduped_web = websearch.dedupe_websearch(sorted_web) if sorted_web else []
# Minimum result guarantee: if all Reddit results were filtered out but
# we had raw results, keep top 3 by relevance regardless of score
if not deduped_reddit and normalized_reddit:
print("[REDDIT WARNING] All results scored below threshold, keeping top 3 by relevance", file=sys.stderr)
by_relevance = sorted(normalized_reddit, key=lambda item: item.relevance, reverse=True)
deduped_reddit = by_relevance[:3]
# Post-retrieval relevance filter: drop low-relevance items per source
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")
deduped_pm = score.relevance_filter(deduped_pm, "POLYMARKET") if deduped_pm else []
# Cross-source linking: annotate items that discuss the same story
dedupe.cross_source_link(
scripts/lib/bird_x.py
+24 -55
View File
@@ -14,6 +14,8 @@ from pathlib import Path
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple
from .relevance import token_overlap_relevance as _compute_relevance
# Path to the vendored bird-search wrapper
_BIRD_SEARCH_MJS = Path(__file__).parent / "vendor" / "bird-search" / "bird-search.mjs"
@@ -63,56 +65,10 @@ def _extract_core_subject(topic: str) -> str:
X search is literal keyword AND matching — all words must appear.
Aggressively strip question/meta/research words to keep only the
core product/concept name (2-3 words max).
core product/concept name (max 5 words).
"""
text = topic.lower().strip()
# Phase 1: Strip multi-word prefixes (longest first)
prefixes = [
'what are the best', 'what is the best', 'what are the latest',
'what are people saying about', 'what do people think about',
'how do i use', 'how to use', 'how to',
'what are', 'what is', 'tips for', 'best practices for',
]
for p in prefixes:
if text.startswith(p + ' '):
text = text[len(p):].strip()
break
# Phase 2: Strip multi-word suffixes
suffixes = [
'best practices', 'use cases', 'prompt techniques',
'prompting techniques', 'prompting tips',
]
for s in suffixes:
if text.endswith(' ' + s):
text = text[:-len(s)].strip()
break
# Phase 3: Filter individual noise words
_noise = {
# Question/filler words
'a', 'an', 'the', 'is', 'are', 'was', 'were', 'and', 'or',
'of', 'in', 'on', 'for', 'with', 'about', 'to',
'people', 'saying', 'think', 'said', 'lately',
# Research/meta descriptors
'best', 'top', 'good', 'great', 'awesome', 'killer',
'latest', 'new', 'news', 'update', 'updates',
'trendiest', 'trending', 'hottest', 'hot', 'popular', 'viral',
'practices', 'features', 'guide', 'tutorial',
'recommendations', 'advice', 'review', 'reviews',
'usecases', 'examples', 'comparison', 'versus', 'vs',
'plugin', 'plugins', 'skill', 'skills', 'tool', 'tools',
# Prompting meta words
'prompt', 'prompts', 'prompting', 'techniques', 'tips',
'tricks', 'methods', 'strategies', 'approaches',
# Action words
'using', 'uses', 'use',
}
words = text.split()
result = [w for w in words if w not in _noise]
return ' '.join(result[:3]) or topic.lower().strip() # Max 3 words
from .query import extract_core_subject
return extract_core_subject(topic, max_words=5, strip_suffixes=True)
def is_bird_installed() -> bool:
@@ -291,16 +247,28 @@ def search_x(
response = _run_bird_search(query, count, timeout)
# Check if we got results
items = parse_bird_response(response)
items = parse_bird_response(response, query=core_topic)
# Retry with fewer keywords if 0 results and query has 3+ words
# Retry with OR groups for multi-word queries (X supports OR operator)
core_words = core_topic.split()
if not items and len(core_words) >= 2:
from .query import extract_compound_terms
compounds = extract_compound_terms(topic)
if compounds:
# Build OR-group query: ("multi-agent" OR "agent simulation") since:DATE
or_parts = ' OR '.join(f'"{t}"' for t in compounds[:3])
_log(f"0 results for '{core_topic}', retrying with OR groups: {or_parts}")
query = f"({or_parts}) since:{from_date}"
response = _run_bird_search(query, count, timeout)
items = parse_bird_response(response, query=core_topic)
# Retry with fewer keywords if still 0 results and query has 3+ words
if not items and len(core_words) > 2:
shorter = ' '.join(core_words[:2])
_log(f"0 results for '{core_topic}', retrying with '{shorter}'")
query = f"{shorter} since:{from_date}"
response = _run_bird_search(query, count, timeout)
items = parse_bird_response(response)
items = parse_bird_response(response, query=core_topic)
# Last-chance retry: use strongest remaining token (often the product name)
if not items and core_words:
@@ -388,7 +356,7 @@ def search_handles(
continue
response = json.loads(output)
items = parse_bird_response(response)
items = parse_bird_response(response, query=core_topic)
all_items.extend(items)
except json.JSONDecodeError:
@@ -399,11 +367,12 @@ def search_handles(
return all_items
def parse_bird_response(response: Dict[str, Any]) -> List[Dict[str, Any]]:
def parse_bird_response(response: Dict[str, Any], query: str = "") -> List[Dict[str, Any]]:
"""Parse Bird response to match xai_x output format.
Args:
response: Raw Bird JSON response
query: Original search query for relevance scoring
Returns:
List of normalized item dicts matching xai_x.parse_x_response() format.
@@ -481,7 +450,7 @@ def parse_bird_response(response: Dict[str, Any]) -> List[Dict[str, Any]]:
"date": date,
"engagement": engagement if any(v is not None for v in engagement.values()) else None,
"why_relevant": "", # Bird doesn't provide relevance explanations
"relevance": 0.7, # Default relevance, let score.py re-rank
"relevance": _compute_relevance(query, str(tweet.get("text", ""))) if query else 0.7,
}
items.append(item)
scripts/lib/bluesky.py
+4 -16
View File
@@ -67,26 +67,14 @@ def _create_session(handle: str, app_password: str) -> Optional[str]:
def _extract_core_subject(topic: str) -> str:
"""Extract core subject from verbose query for Bluesky search."""
text = topic.lower().strip()
prefixes = [
'what are the best', 'what is the best', 'what are the latest',
'what are people saying about', 'what do people think about',
'how do i use', 'how to use', 'how to',
'what are', 'what is', 'tips for', 'best practices for',
]
for p in prefixes:
if text.startswith(p + ' '):
text = text[len(p):].strip()
noise = {
from .query import extract_core_subject
_BSKY_NOISE = frozenset({
'best', 'top', 'good', 'great', 'awesome',
'latest', 'new', 'news', 'update', 'updates',
'trending', 'hottest', 'popular', 'viral',
'practices', 'features', 'recommendations', 'advice',
}
words = text.split()
filtered = [w for w in words if w not in noise]
result = ' '.join(filtered) if filtered else text
return result.rstrip('?!.')
})
return extract_core_subject(topic, noise=_BSKY_NOISE)
def _parse_date(item: Dict[str, Any]) -> Optional[str]:
scripts/lib/env.py
+4
View File
@@ -243,10 +243,14 @@ def get_config() -> Dict[str, Any]:
keys = [
('XAI_API_KEY', None),
('GOOGLE_API_KEY', None),
('GEMINI_API_KEY', None),
('GOOGLE_GENAI_API_KEY', None),
('OPENROUTER_API_KEY', None),
('PARALLEL_API_KEY', None),
('BRAVE_API_KEY', None),
('XIAOHONGSHU_API_BASE', None),
('GEMINI_MODEL', None),
('OPENAI_MODEL_POLICY', 'auto'),
('OPENAI_MODEL_PIN', None),
('XAI_MODEL_POLICY', 'latest'),
scripts/lib/hackernews.py
+21 -8
View File
@@ -12,6 +12,8 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any, Dict, List, Optional
from . import http
from .query import extract_core_subject
from .relevance import token_overlap_relevance
ALGOLIA_SEARCH_URL = "https://hn.algolia.com/api/v1/search"
ALGOLIA_SEARCH_BY_DATE_URL = "https://hn.algolia.com/api/v1/search_by_date"
@@ -84,13 +86,17 @@ def search_hackernews(
from_ts = _date_to_unix(from_date)
to_ts = _date_to_unix(to_date) + 86400 # Include the end date
_log(f"Searching for '{topic}' (since {from_date}, count={count})")
# Use extracted core subject instead of raw topic for cleaner Algolia matching
core = extract_core_subject(topic)
_log(f"Searching for '{core}' (raw: '{topic}', since {from_date}, count={count})")
# Use relevance-sorted search (better for topic matching)
# Use relevance-sorted search with minimum engagement filter.
# NOTE: restrictSearchableAttributes=title omitted intentionally — it would
# miss Ask HN/Show HN threads where the topic appears in the body.
params = {
"query": topic,
"query": core,
"tags": "story",
"numericFilters": f"created_at_i>{from_ts},created_at_i<{to_ts}",
"numericFilters": f"created_at_i>{from_ts},created_at_i<{to_ts},points>2",
"hitsPerPage": str(count),
}
@@ -111,9 +117,13 @@ def search_hackernews(
return response
def parse_hackernews_response(response: Dict[str, Any]) -> List[Dict[str, Any]]:
def parse_hackernews_response(response: Dict[str, Any], query: str = "") -> List[Dict[str, Any]]:
"""Parse Algolia response into normalized item dicts.
Args:
response: Algolia search response
query: Original search query for token-overlap relevance scoring
Returns:
List of item dicts ready for normalization.
"""
@@ -134,11 +144,14 @@ def parse_hackernews_response(response: Dict[str, Any]) -> List[Dict[str, Any]]:
article_url = hit.get("url") or ""
hn_url = f"https://news.ycombinator.com/item?id={object_id}"
# Relevance: Algolia rank position gives a base, engagement boosts it
# Position 0 = most relevant from Algolia
# Relevance: blend Algolia rank with token-overlap content matching
rank_score = max(0.3, 1.0 - (i * 0.02)) # 1.0 -> 0.3 over 35 items
engagement_boost = min(0.2, math.log1p(points) / 40)
relevance = min(1.0, rank_score * 0.7 + engagement_boost + 0.1)
if query:
content_score = token_overlap_relevance(query, hit.get("title", ""))
relevance = min(1.0, 0.6 * rank_score + 0.4 * content_score + engagement_boost)
else:
relevance = min(1.0, rank_score * 0.7 + engagement_boost + 0.1)
items.append({
"object_id": object_id,
scripts/lib/instagram.py
+33 -105
View File
@@ -17,6 +17,8 @@ try:
except ImportError:
_requests = None
from . import http
SCRAPECREATORS_BASE = "https://api.scrapecreators.com"
# Depth configurations: how many results to fetch / captions to extract
@@ -29,93 +31,13 @@ DEPTH_CONFIG = {
# Max words to keep from each caption
CAPTION_MAX_WORDS = 500
# Stopwords for relevance computation (shared with tiktok.py pattern)
STOPWORDS = frozenset({
'the', 'a', 'an', 'to', 'for', 'how', 'is', 'in', 'of', 'on',
'and', 'with', 'from', 'by', 'at', 'this', 'that', 'it', 'my',
'your', 'i', 'me', 'we', 'you', 'what', 'are', 'do', 'can',
'its', 'be', 'or', 'not', 'no', 'so', 'if', 'but', 'about',
'all', 'just', 'get', 'has', 'have', 'was', 'will',
})
# Synonym groups for relevance scoring
SYNONYMS = {
'hip': {'rap', 'hiphop'},
'hop': {'rap', 'hiphop'},
'rap': {'hip', 'hop', 'hiphop'},
'hiphop': {'rap', 'hip', 'hop'},
'js': {'javascript'},
'javascript': {'js'},
'ts': {'typescript'},
'typescript': {'ts'},
'ai': {'artificial', 'intelligence'},
'ml': {'machine', 'learning'},
'react': {'reactjs'},
'reactjs': {'react'},
}
def _tokenize(text: str) -> Set[str]:
"""Lowercase, strip punctuation, remove stopwords, drop single-char tokens."""
words = re.sub(r'[^\w\s]', ' ', text.lower()).split()
tokens = {w for w in words if w not in STOPWORDS and len(w) > 1}
expanded = set(tokens)
for t in tokens:
if t in SYNONYMS:
expanded.update(SYNONYMS[t])
return expanded
def _compute_relevance(query: str, text: str, hashtags: List[str] = None) -> float:
"""Compute relevance as ratio of query tokens found in text + hashtags.
Uses ratio overlap (intersection / query_length). Hashtags provide
an Instagram-specific relevance boost. Floors at 0.1.
"""
q_tokens = _tokenize(query)
# Combine text and hashtags for matching
combined = text
if hashtags:
combined = f"{text} {' '.join(hashtags)}"
t_tokens = _tokenize(combined)
# Split concatenated hashtags (e.g., "claudecode" -> "claude", "code")
if hashtags:
for tag in hashtags:
tag_lower = tag.lower()
for qt in q_tokens:
if qt in tag_lower and qt != tag_lower:
t_tokens.add(qt)
if not q_tokens:
return 0.5 # Neutral fallback
overlap = len(q_tokens & t_tokens)
ratio = overlap / len(q_tokens)
return max(0.1, min(1.0, ratio))
from .relevance import token_overlap_relevance as _compute_relevance
def _extract_core_subject(topic: str) -> str:
"""Extract core subject from verbose query for Instagram search.
Strips meta/research words to keep only the core product/concept name.
"""
text = topic.lower().strip()
# Strip multi-word prefixes
prefixes = [
'what are the best', 'what is the best', 'what are the latest',
'what are people saying about', 'what do people think about',
'how do i use', 'how to use', 'how to',
'what are', 'what is', 'tips for', 'best practices for',
]
for p in prefixes:
if text.startswith(p + ' '):
text = text[len(p):].strip()
# Strip individual noise words
noise = {
"""Extract core subject from verbose query for Instagram search."""
from .query import extract_core_subject
_INSTAGRAM_NOISE = frozenset({
'best', 'top', 'good', 'great', 'awesome', 'killer',
'latest', 'new', 'news', 'update', 'updates',
'trending', 'hottest', 'popular', 'viral',
@@ -123,12 +45,8 @@ def _extract_core_subject(topic: str) -> str:
'recommendations', 'advice',
'prompt', 'prompts', 'prompting',
'methods', 'strategies', 'approaches',
}
words = text.split()
filtered = [w for w in words if w not in noise]
result = ' '.join(filtered) if filtered else text
return result.rstrip('?!.')
})
return extract_core_subject(topic, noise=_INSTAGRAM_NOISE)
def _log(msg: str):
@@ -207,26 +125,36 @@ def search_instagram(
if not token:
return {"items": [], "error": "No SCRAPECREATORS_API_KEY configured"}
if not _requests:
return {"items": [], "error": "requests library not installed"}
config = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"])
core_topic = _extract_core_subject(topic)
_log(f"Searching Instagram for '{core_topic}' (depth={depth}, count={config['results_per_page']})")
try:
resp = _requests.get(
f"{SCRAPECREATORS_BASE}/v2/instagram/reels/search",
params={"query": core_topic},
headers=_sc_headers(token),
timeout=30,
)
resp.raise_for_status()
data = resp.json()
except Exception as e:
_log(f"ScrapeCreators error: {e}")
return {"items": [], "error": f"{type(e).__name__}: {e}"}
if not _requests:
_log("requests library not installed, falling back to urllib")
try:
from urllib.parse import urlencode
params = urlencode({"query": core_topic})
url = f"{SCRAPECREATORS_BASE}/v2/instagram/reels/search?{params}"
headers = _sc_headers(token)
headers["User-Agent"] = http.USER_AGENT
data = http.get(url, headers=headers, timeout=30, retries=2)
except Exception as e:
_log(f"ScrapeCreators error (urllib): {e}")
return {"items": [], "error": f"{type(e).__name__}: {e}"}
else:
try:
resp = _requests.get(
f"{SCRAPECREATORS_BASE}/v2/instagram/reels/search",
params={"query": core_topic},
headers=_sc_headers(token),
timeout=30,
)
resp.raise_for_status()
data = resp.json()
except Exception as e:
_log(f"ScrapeCreators error: {e}")
return {"items": [], "error": f"{type(e).__name__}: {e}"}
# Items are in the 'reels' array (ScrapeCreators v2 response)
raw_items = data.get("reels") or data.get("items") or data.get("data") or []
scripts/lib/models.py
+4
View File
@@ -53,6 +53,10 @@ def is_search_capable_model(model_id: str) -> bool:
Includes mini variants (same structured extraction quality, lower cost).
Excludes: nano (no web_search), gpt-4o-mini (no domain filtering),
chat/codex/pro/preview/turbo/search (specialized variants).
Note: gpt-5 with reasoning effort="minimal" does NOT support web_search
(per OpenAI docs). We never set reasoning params — our usage is pure
tool invocation + JSON extraction — so gpt-5 is safe to include here.
"""
model_lower = model_id.lower()
scripts/lib/polymarket.py
+47 -26
View File
@@ -13,6 +13,8 @@ from typing import Any, Dict, List, Optional
from urllib.parse import quote_plus, urlencode
from . import http
from .query_type import detect_query_type
from .relevance import LOW_SIGNAL_QUERY_TOKENS, token_overlap_relevance
GAMMA_SEARCH_URL = "https://gamma-api.polymarket.com/public-search"
@@ -73,7 +75,7 @@ def _expand_queries(topic: str) -> List[str]:
words = core.split()
if len(words) >= 2:
for word in words:
if len(word) > 1: # skip single-char words
if len(word) > 1 and word.lower() not in LOW_SIGNAL_QUERY_TOKENS:
queries.append(word)
# Add the full topic if different from core
@@ -314,8 +316,8 @@ def _shorten_question(question: str) -> str:
def _compute_text_similarity(topic: str, title: str, outcomes: List[str] = None) -> float:
"""Score how well the event title (or outcome names) match the search topic.
Returns 0.0-1.0. Title substring match gets 1.0, outcome match gets 0.85/0.7,
title token overlap gets proportional score.
Returns 0.0-1.0. Exact title phrase match gets 1.0. Otherwise we reuse the
shared query-centric relevance scorer and take the best title/outcome match.
"""
core = _extract_core_subject(topic).lower()
title_lower = title.lower()
@@ -326,27 +328,45 @@ def _compute_text_similarity(topic: str, title: str, outcomes: List[str] = None)
if core in title_lower:
return 1.0
# Check if topic appears in any outcome name (bidirectional)
query_type = detect_query_type(topic)
title_score = token_overlap_relevance(core, title)
best_score = title_score
if outcomes:
core_tokens = set(core.split())
best_outcome_score = 0.0
for outcome_name in outcomes:
outcome_lower = outcome_name.lower()
# Bidirectional: "arizona" in "arizona basketball" OR "arizona basketball" contains "arizona"
if core in outcome_lower or outcome_lower in core:
best_outcome_score = max(best_outcome_score, 0.85)
elif core_tokens & set(outcome_lower.split()):
best_outcome_score = max(best_outcome_score, 0.7)
if best_outcome_score > 0:
return best_outcome_score
outcome_score = token_overlap_relevance(core, outcome_name)
if _strong_phrase_match(core, outcome_lower):
outcome_score = max(outcome_score, 0.92 if len(outcome_lower.split()) >= 2 else 0.88)
if title_score < 0.3:
outcome_cap = 0.55 if query_type == "prediction" else 0.24
outcome_score = min(outcome_cap, outcome_score)
else:
outcome_score = max(title_score, 0.75 * title_score + 0.25 * outcome_score)
best_score = max(best_score, outcome_score)
# Token overlap fallback against title
topic_tokens = set(core.split())
title_tokens = set(title_lower.split())
if not topic_tokens:
return 0.5
overlap = len(topic_tokens & title_tokens)
return overlap / len(topic_tokens)
return round(best_score, 2)
def _strong_phrase_match(core: str, candidate: str) -> bool:
"""Require real token matches, not accidental short substrings.
This prevents binary outcomes like "No" from matching "nano" or similar
short-string accidents.
"""
candidate = " ".join(re.sub(r"[^\w\s]", " ", candidate.lower()).split())
core = " ".join(re.sub(r"[^\w\s]", " ", core.lower()).split())
if not candidate or not core:
return False
candidate_tokens = candidate.split()
core_tokens = set(core.split())
if len(candidate_tokens) >= 2:
return candidate in core or core in candidate
token = candidate_tokens[0]
return len(token) > 2 and token in core_tokens
def _safe_float(val, default=0.0) -> float:
@@ -484,7 +504,8 @@ def parse_polymarket_response(response: Dict[str, Any], topic: str = "") -> List
except (IndexError, TypeError):
end_date = None
# Quality-signal relevance (replaces position-based decay)
# Semantic relevance should dominate. Market quality should refine
# relevant matches, not rescue unrelated high-liquidity events.
text_score = _compute_text_similarity(topic, title, all_outcome_names) if topic else 0.5
# Volume signal: log-scaled monthly volume (most stable signal)
@@ -504,13 +525,13 @@ def parse_polymarket_response(response: Dict[str, Any], topic: str = "") -> List
# Competitive bonus: markets near 50/50 are more interesting
competitive_score = event_competitive
relevance = min(1.0, (
0.30 * text_score +
0.30 * vol_score +
0.15 * liq_score +
market_quality = (
0.50 * vol_score +
0.25 * liq_score +
0.15 * movement_score +
0.10 * competitive_score
))
)
relevance = min(1.0, text_score * (0.75 + 0.25 * market_quality))
# Surface the topic-matching outcome to the front before truncating
if topic and outcome_prices:
scripts/lib/query.py
+117
View File
@@ -0,0 +1,117 @@
"""Shared query preprocessing utilities: noise-word stripping, core subject
extraction, and compound term detection. Used by all search modules."""
import re
from typing import FrozenSet, List, Optional, Set
# Common multi-word prefixes stripped from all queries (identical across modules)
PREFIXES = [
'what are the best', 'what is the best', 'what are the latest',
'what are people saying about', 'what do people think about',
'how do i use', 'how to use', 'how to',
'what are', 'what is', 'tips for', 'best practices for',
]
# Multi-word suffixes (used by bird_x)
SUFFIXES = [
'best practices', 'use cases', 'prompt techniques',
'prompting techniques', 'prompting tips',
]
# Base noise words shared across most modules
NOISE_WORDS = frozenset({
# Articles/prepositions/conjunctions
'a', 'an', 'the', 'is', 'are', 'was', 'were', 'and', 'or',
'of', 'in', 'on', 'for', 'with', 'about', 'to',
# Question words
'how', 'what', 'which', 'who', 'why', 'when', 'where',
'does', 'should', 'could', 'would',
# Research/meta descriptors
'best', 'top', 'good', 'great', 'awesome', 'killer',
'latest', 'new', 'news', 'update', 'updates',
'trendiest', 'trending', 'hottest', 'hot', 'popular', 'viral',
'practices', 'features', 'guide', 'tutorial',
'recommendations', 'advice', 'review', 'reviews',
'usecases', 'examples', 'comparison', 'versus', 'vs',
'plugin', 'plugins', 'skill', 'skills', 'tool', 'tools',
# Prompting meta words
'prompt', 'prompts', 'prompting', 'techniques', 'tips',
'tricks', 'methods', 'strategies', 'approaches',
# Action words
'using', 'uses', 'use',
# Misc filler
'people', 'saying', 'think', 'said', 'lately',
})
def extract_core_subject(
topic: str,
*,
noise: Optional[FrozenSet[str]] = None,
max_words: Optional[int] = None,
strip_suffixes: bool = False,
) -> str:
"""Extract core subject from a verbose search query.
Strips common question/meta prefixes and noise words to produce a
compact search-friendly query. Platforms customize via parameters.
Args:
topic: Raw user query
noise: Override noise word set (default: NOISE_WORDS)
max_words: Cap result to N words (default: no cap)
strip_suffixes: Also strip trailing multi-word suffixes (bird_x uses this)
Returns:
Cleaned query string
"""
text = topic.lower().strip()
if not text:
return text
# Phase 1: Strip multi-word prefixes (longest first, stop after first match)
for p in PREFIXES:
if text.startswith(p + ' '):
text = text[len(p):].strip()
break
# Phase 2: Strip multi-word suffixes (opt-in)
if strip_suffixes:
for s in SUFFIXES:
if text.endswith(' ' + s):
text = text[:-len(s)].strip()
break
# Phase 3: Filter individual noise words
noise_set = noise if noise is not None else NOISE_WORDS
words = text.split()
filtered = [w for w in words if w not in noise_set]
# Apply word cap if requested
if max_words is not None and filtered:
filtered = filtered[:max_words]
result = ' '.join(filtered) if filtered else text
return result.rstrip('?!.') if not max_words else (result or topic.lower().strip())
def extract_compound_terms(topic: str) -> List[str]:
"""Detect multi-word terms that should be quoted in search queries.
Identifies:
- Hyphenated terms: "multi-agent", "vc-backed"
- Title-cased multi-word names: "Claude Code", "React Native"
Returns list of terms suitable for quoting (e.g., '"multi-agent"').
"""
terms: List[str] = []
# Hyphenated terms
for match in re.finditer(r'\b\w+-\w+(?:-\w+)*\b', topic):
terms.append(match.group())
# Title-cased sequences (2+ capitalized words in a row)
for match in re.finditer(r'(?:[A-Z][a-z]+\s+){1,}[A-Z][a-z]+', topic):
terms.append(match.group())
return terms
scripts/lib/query_type.py
+2 -2
View File
@@ -7,7 +7,7 @@ QueryType = Literal["product", "concept", "opinion", "how_to", "comparison", "br
# 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
r"\b(price|pricing|cost|buy|purchase|deal|discount|subscription|plan|tier|free tier|alternative|prompt|prompts|prompting|template|templates)\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
@@ -16,7 +16,7 @@ _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|prompting|prompts?|best practices|tips|examples|animation|animations)\b",
r"\b(how to|tutorial|step by step|setup|install|configure|deploy|migrate|implement|build a|create a|prompting|prompts?|best practices|tips|examples|animation|animations|video workflow|render pipeline)\b",
re.I,
)
_COMPARISON_PATTERNS = re.compile(
scripts/lib/reddit.py
+42 -28
View File
@@ -48,7 +48,11 @@ DEPTH_CONFIG = {
},
}
# Stopwords for query extraction
from .query import extract_core_subject as _query_extract
from .query_type import detect_query_type
from .relevance import token_overlap_relevance
# Reddit-specific noise words (preserves original smaller set)
NOISE_WORDS = frozenset({
'best', 'top', 'good', 'great', 'awesome', 'killer',
'latest', 'new', 'news', 'update', 'updates',
@@ -82,24 +86,7 @@ def _extract_core_subject(topic: str) -> str:
Strips meta/research words to keep only the core product/concept name.
"""
text = topic.lower().strip()
# Strip multi-word prefixes
prefixes = [
'what are the best', 'what is the best', 'what are the latest',
'what are people saying about', 'what do people think about',
'how do i use', 'how to use', 'how to',
'what are', 'what is', 'tips for', 'best practices for',
]
for p in prefixes:
if text.startswith(p + ' '):
text = text[len(p):].strip()
words = text.split()
filtered = [w for w in words if w not in NOISE_WORDS]
result = ' '.join(filtered) if filtered else text
return result.rstrip('?!.')
return _query_extract(topic, noise=NOISE_WORDS)
def expand_reddit_queries(topic: str, depth: str) -> List[str]:
@@ -121,10 +108,14 @@ def expand_reddit_queries(topic: str, depth: str) -> List[str]:
if core.lower() != original_clean.lower() and len(original_clean.split()) <= 8:
queries.append(original_clean)
if depth in ("default", "deep"):
# Opinion/review variants help mostly for product and opinion queries.
# They contaminate broader searches like predictions or breaking news.
qtype = detect_query_type(topic)
if depth in ("default", "deep") and qtype in ("product", "opinion"):
queries.append(f"{core} worth it OR thoughts OR review")
if depth == "deep":
# Problem/bug variants are useful for tool workflows, not generic news.
if depth == "deep" and qtype in ("product", "opinion", "how_to"):
queries.append(f"{core} issues OR problems OR bug OR broken")
return queries
@@ -199,7 +190,7 @@ def _parse_date(created_utc) -> Optional[str]:
return None
def _normalize_post(post: Dict[str, Any], idx: int, source_label: str = "global") -> Dict[str, Any]:
def _normalize_post(post: Dict[str, Any], idx: int, source_label: str = "global", query: str = "") -> Dict[str, Any]:
"""Normalize a ScrapeCreators Reddit post to our internal format."""
permalink = post.get("permalink", "")
url = f"https://www.reddit.com{permalink}" if permalink else post.get("url", "")
@@ -208,10 +199,17 @@ def _normalize_post(post: Dict[str, Any], idx: int, source_label: str = "global"
if url and "reddit.com" not in url:
url = ""
title = str(post.get("title", "")).strip()
selftext = str(post.get("selftext", ""))
# Score the title first, then let the body provide limited support.
# This keeps long selftexts from overpowering the visible topic signal.
relevance = _compute_post_relevance(query, title, selftext) if query else 0.7
return {
"id": f"R{idx}",
"reddit_id": post.get("id", ""),
"title": str(post.get("title", "")).strip(),
"title": title,
"url": url,
"subreddit": str(post.get("subreddit", "")).strip(),
"date": _parse_date(post.get("created_utc")),
@@ -220,12 +218,28 @@ def _normalize_post(post: Dict[str, Any], idx: int, source_label: str = "global"
"num_comments": post.get("num_comments", 0),
"upvote_ratio": post.get("upvote_ratio"),
},
"relevance": 0.7,
"relevance": relevance,
"why_relevant": f"Reddit {source_label} search",
"selftext": str(post.get("selftext", ""))[:500],
}
def _compute_post_relevance(query: str, title: str, selftext: str) -> float:
"""Compute Reddit relevance with title-first weighting.
Title should carry most of the weight because it is the visible summary the
user sees. Selftext can lift a marginal match, but it should not rescue a
weak or ambiguous title into the top ranks.
"""
title_score = token_overlap_relevance(query, title)
if not selftext.strip():
return title_score
body_score = token_overlap_relevance(query, selftext)
support_score = max(title_score, body_score)
return round(0.75 * title_score + 0.25 * support_score, 2)
def _global_search(
query: str,
token: str,
@@ -431,23 +445,23 @@ def search_reddit(
_log(f" -> {len(posts)} results")
all_raw_posts.extend(posts)
# Normalize all posts
# Normalize all posts (with query for relevance scoring)
core = _extract_core_subject(topic)
all_items = []
for i, post in enumerate(all_raw_posts):
item = _normalize_post(post, i + 1, "global")
item = _normalize_post(post, i + 1, "global", query=core)
all_items.append(item)
# === Phase 3: Subreddit Discovery + Targeted Search ===
discovered_subs = discover_subreddits(all_raw_posts, topic=topic, max_subs=config["subreddit_searches"])
_log(f"Discovered subreddits: {discovered_subs}")
core = _extract_core_subject(topic)
for sub in discovered_subs[:config["subreddit_searches"]]:
_log(f"Subreddit search: r/{sub} for '{core}'")
sub_posts = _subreddit_search(sub, core, token, sort="relevance", timeframe=timeframe)
_log(f" -> {len(sub_posts)} results from r/{sub}")
for j, post in enumerate(sub_posts):
item = _normalize_post(post, len(all_items) + j + 1, f"r/{sub}")
item = _normalize_post(post, len(all_items) + j + 1, f"r/{sub}", query=core)
all_items.append(item)
# === Phase 4: Deduplicate ===
scripts/lib/relevance.py
+148
View File
@@ -0,0 +1,148 @@
"""Shared token-overlap relevance scoring for search result ranking.
The score is intentionally query-centric:
- exact phrase matches should score very high
- partial matches should pay a meaningful penalty
- matches on generic words alone ("odds", "review") should not pass as relevant
"""
import re
from typing import List, Optional, Set
# Stopwords for relevance computation (common English words that dilute token overlap)
STOPWORDS = frozenset({
'the', 'a', 'an', 'to', 'for', 'how', 'is', 'in', 'of', 'on',
'and', 'with', 'from', 'by', 'at', 'this', 'that', 'it', 'my',
'your', 'i', 'me', 'we', 'you', 'what', 'are', 'do', 'can',
'its', 'be', 'or', 'not', 'no', 'so', 'if', 'but', 'about',
'all', 'just', 'get', 'has', 'have', 'was', 'will',
})
# Synonym groups for relevance scoring (bidirectional expansion)
# Superset of all platform-specific synonym dicts
SYNONYMS = {
'hip': {'rap', 'hiphop'},
'hop': {'rap', 'hiphop'},
'rap': {'hip', 'hop', 'hiphop'},
'hiphop': {'rap', 'hip', 'hop'},
'js': {'javascript'},
'javascript': {'js'},
'ts': {'typescript'},
'typescript': {'ts'},
'ai': {'artificial', 'intelligence'},
'ml': {'machine', 'learning'},
'react': {'reactjs'},
'reactjs': {'react'},
'svelte': {'sveltejs'},
'sveltejs': {'svelte'},
'vue': {'vuejs'},
'vuejs': {'vue'},
}
# Generic query words that should not carry relevance on their own.
# They still help when paired with stronger entity/topic matches.
LOW_SIGNAL_QUERY_TOKENS = frozenset({
'advice', 'animation', 'animations', 'best', 'chance', 'chances',
'code', 'compare', 'comparison', 'differences', 'explain', 'guide',
'guides', 'how', 'latest', 'news', 'odds', 'opinion', 'opinions',
'prediction', 'predictions', 'probability', 'probabilities', 'prompt',
'prompting', 'prompts', 'rate', 'review', 'reviews', 'thoughts',
'tip', 'tips', 'tutorial', 'tutorials', 'update', 'updates', 'use',
'using', 'versus', 'vs', 'worth',
})
def tokenize(text: str) -> Set[str]:
"""Lowercase, strip punctuation, remove stopwords, drop single-char tokens.
Expands tokens with synonyms for better cross-domain matching.
"""
words = re.sub(r'[^\w\s]', ' ', text.lower()).split()
tokens = {w for w in words if w not in STOPWORDS and len(w) > 1}
expanded = set(tokens)
for t in tokens:
if t in SYNONYMS:
expanded.update(SYNONYMS[t])
return expanded
def _normalize_phrase(text: str) -> str:
"""Normalize text for phrase containment checks."""
return ' '.join(re.sub(r'[^\w\s]', ' ', text.lower()).split())
def token_overlap_relevance(
query: str,
text: str,
hashtags: Optional[List[str]] = None,
) -> float:
"""Compute a query-centric relevance score between 0.0 and 1.0.
The score combines:
- query coverage
- informative-token coverage
- a small precision term to penalize extra noise
- an exact phrase bonus
Generic tokens alone are capped below the post-retrieval 0.3 threshold.
Args:
query: Search query
text: Content text to match against
hashtags: Optional list of hashtags (TikTok/Instagram). Concatenated
hashtags are split to match query tokens (e.g. "claudecode" matches "claude").
Returns:
Float between 0.0 and 1.0 (0.5 for empty queries)
"""
q_tokens = tokenize(query)
# Combine text and hashtags for matching
combined = text
if hashtags:
combined = f"{text} {' '.join(hashtags)}"
t_tokens = tokenize(combined)
# Split concatenated hashtags (e.g., "claudecode" -> matches "claude", "code")
if hashtags:
for tag in hashtags:
tag_lower = tag.lower()
for qt in q_tokens:
if qt in tag_lower and qt != tag_lower:
t_tokens.add(qt)
if not q_tokens:
return 0.5 # Neutral fallback for empty/stopword-only queries
overlap_tokens = q_tokens & t_tokens
overlap = len(overlap_tokens)
if overlap == 0:
return 0.0
informative_q_tokens = {t for t in q_tokens if t not in LOW_SIGNAL_QUERY_TOKENS}
if not informative_q_tokens:
informative_q_tokens = q_tokens
coverage = overlap / len(q_tokens)
informative_overlap = len(informative_q_tokens & t_tokens) / len(informative_q_tokens)
precision_denominator = min(len(t_tokens), len(q_tokens) + 4) or 1
precision = overlap / precision_denominator
phrase_bonus = 0.0
normalized_query = _normalize_phrase(query)
normalized_text = _normalize_phrase(combined)
if normalized_query and normalized_query in normalized_text:
phrase_bonus = 0.12 if len(normalized_query.split()) > 1 else 0.16
base = (
0.55 * (coverage ** 1.35) +
0.25 * informative_overlap +
0.20 * precision
)
# If we only matched generic query words, keep the score below the
# normal relevance filter threshold so these do not survive by default.
if informative_q_tokens and not (informative_q_tokens & t_tokens):
return round(min(0.24, base), 2)
return round(min(1.0, base + phrase_bonus), 2)
scripts/lib/score.py
+28 -4
View File
@@ -11,6 +11,12 @@ WEIGHT_RELEVANCE = 0.45
WEIGHT_RECENCY = 0.25
WEIGHT_ENGAGEMENT = 0.30
# Polymarket needs stronger semantic weighting because volume/liquidity already
# show up as engagement and lightly influence parse-time relevance.
PM_WEIGHT_RELEVANCE = 0.60
PM_WEIGHT_RECENCY = 0.20
PM_WEIGHT_ENGAGEMENT = 0.20
# WebSearch weights (no engagement data available)
WEBSEARCH_WEIGHT_RELEVANCE = 0.55
WEBSEARCH_WEIGHT_RECENCY = 0.45
@@ -632,9 +638,9 @@ def score_polymarket_items(items: List[schema.PolymarketItem]) -> List[schema.Po
)
overall = (
WEIGHT_RELEVANCE * rel_score +
WEIGHT_RECENCY * rec_score +
WEIGHT_ENGAGEMENT * eng_score
PM_WEIGHT_RELEVANCE * rel_score +
PM_WEIGHT_RECENCY * rec_score +
PM_WEIGHT_ENGAGEMENT * eng_score
)
if eng_raw[i] is None:
@@ -715,7 +721,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 +755,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
scripts/lib/scrapecreators_x.py
+6 -55
View File
@@ -7,10 +7,9 @@ Requires SCRAPECREATORS_API_KEY in config.
API docs: https://scrapecreators.com/docs
"""
import re
import sys
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Set
from typing import Any, Dict, List, Optional
try:
import requests as _requests
@@ -25,67 +24,19 @@ DEPTH_CONFIG = {
"deep": {"results_per_page": 40},
}
STOPWORDS = frozenset({
'the', 'a', 'an', 'to', 'for', 'how', 'is', 'in', 'of', 'on',
'and', 'with', 'from', 'by', 'at', 'this', 'that', 'it', 'my',
'your', 'i', 'me', 'we', 'you', 'what', 'are', 'do', 'can',
'its', 'be', 'or', 'not', 'no', 'so', 'if', 'but', 'about',
'all', 'just', 'get', 'has', 'have', 'was', 'will',
})
SYNONYMS = {
'js': {'javascript'}, 'javascript': {'js'},
'ts': {'typescript'}, 'typescript': {'ts'},
'ai': {'artificial', 'intelligence'},
'ml': {'machine', 'learning'},
'react': {'reactjs'}, 'reactjs': {'react'},
}
def _tokenize(text: str) -> Set[str]:
"""Lowercase, strip punctuation, remove stopwords, drop single-char tokens."""
words = re.sub(r'[^\w\s]', ' ', text.lower()).split()
tokens = {w for w in words if w not in STOPWORDS and len(w) > 1}
expanded = set(tokens)
for t in tokens:
if t in SYNONYMS:
expanded.update(SYNONYMS[t])
return expanded
def _compute_relevance(query: str, text: str) -> float:
"""Compute relevance as ratio of query tokens found in text. Floors at 0.1."""
q_tokens = _tokenize(query)
t_tokens = _tokenize(text)
if not q_tokens:
return 0.5
overlap = len(q_tokens & t_tokens)
ratio = overlap / len(q_tokens)
return max(0.1, min(1.0, ratio))
from .relevance import token_overlap_relevance as _compute_relevance
def _extract_core_subject(topic: str) -> str:
"""Extract core subject from verbose query for Twitter search."""
text = topic.lower().strip()
prefixes = [
'what are the best', 'what is the best', 'what are the latest',
'what are people saying about', 'what do people think about',
'how do i use', 'how to use', 'how to',
'what are', 'what is', 'tips for', 'best practices for',
]
for p in prefixes:
if text.startswith(p + ' '):
text = text[len(p):].strip()
noise = {
from .query import extract_core_subject
_SC_X_NOISE = frozenset({
'best', 'top', 'good', 'great', 'awesome',
'latest', 'new', 'news', 'update', 'updates',
'trending', 'hottest', 'popular', 'viral',
'practices', 'features', 'recommendations', 'advice',
}
words = text.split()
filtered = [w for w in words if w not in noise]
result = ' '.join(filtered) if filtered else text
return result.rstrip('?!.')
})
return extract_core_subject(topic, noise=_SC_X_NOISE)
def _log(msg: str):
scripts/lib/tiktok.py
+33 -105
View File
@@ -17,6 +17,8 @@ try:
except ImportError:
_requests = None
from . import http
SCRAPECREATORS_BASE = "https://api.scrapecreators.com/v1/tiktok"
# Depth configurations: how many results to fetch / captions to extract
@@ -29,93 +31,13 @@ DEPTH_CONFIG = {
# Max words to keep from each caption
CAPTION_MAX_WORDS = 500
# Stopwords for relevance computation (shared with youtube_yt.py pattern)
STOPWORDS = frozenset({
'the', 'a', 'an', 'to', 'for', 'how', 'is', 'in', 'of', 'on',
'and', 'with', 'from', 'by', 'at', 'this', 'that', 'it', 'my',
'your', 'i', 'me', 'we', 'you', 'what', 'are', 'do', 'can',
'its', 'be', 'or', 'not', 'no', 'so', 'if', 'but', 'about',
'all', 'just', 'get', 'has', 'have', 'was', 'will',
})
# Synonym groups for relevance scoring
SYNONYMS = {
'hip': {'rap', 'hiphop'},
'hop': {'rap', 'hiphop'},
'rap': {'hip', 'hop', 'hiphop'},
'hiphop': {'rap', 'hip', 'hop'},
'js': {'javascript'},
'javascript': {'js'},
'ts': {'typescript'},
'typescript': {'ts'},
'ai': {'artificial', 'intelligence'},
'ml': {'machine', 'learning'},
'react': {'reactjs'},
'reactjs': {'react'},
}
def _tokenize(text: str) -> Set[str]:
"""Lowercase, strip punctuation, remove stopwords, drop single-char tokens."""
words = re.sub(r'[^\w\s]', ' ', text.lower()).split()
tokens = {w for w in words if w not in STOPWORDS and len(w) > 1}
expanded = set(tokens)
for t in tokens:
if t in SYNONYMS:
expanded.update(SYNONYMS[t])
return expanded
def _compute_relevance(query: str, text: str, hashtags: List[str] = None) -> float:
"""Compute relevance as ratio of query tokens found in text + hashtags.
Uses ratio overlap (intersection / query_length). Hashtags provide
a TikTok-specific relevance boost. Floors at 0.1.
"""
q_tokens = _tokenize(query)
# Combine text and hashtags for matching
combined = text
if hashtags:
combined = f"{text} {' '.join(hashtags)}"
t_tokens = _tokenize(combined)
# Split concatenated hashtags (e.g., "claudecode" -> "claude", "code")
if hashtags:
for tag in hashtags:
tag_lower = tag.lower()
for qt in q_tokens:
if qt in tag_lower and qt != tag_lower:
t_tokens.add(qt)
if not q_tokens:
return 0.5 # Neutral fallback
overlap = len(q_tokens & t_tokens)
ratio = overlap / len(q_tokens)
return max(0.1, min(1.0, ratio))
from .relevance import token_overlap_relevance as _compute_relevance
def _extract_core_subject(topic: str) -> str:
"""Extract core subject from verbose query for TikTok search.
Strips meta/research words to keep only the core product/concept name.
"""
text = topic.lower().strip()
# Strip multi-word prefixes
prefixes = [
'what are the best', 'what is the best', 'what are the latest',
'what are people saying about', 'what do people think about',
'how do i use', 'how to use', 'how to',
'what are', 'what is', 'tips for', 'best practices for',
]
for p in prefixes:
if text.startswith(p + ' '):
text = text[len(p):].strip()
# Strip individual noise words
noise = {
"""Extract core subject from verbose query for TikTok search."""
from .query import extract_core_subject
_TIKTOK_NOISE = frozenset({
'best', 'top', 'good', 'great', 'awesome', 'killer',
'latest', 'new', 'news', 'update', 'updates',
'trending', 'hottest', 'popular', 'viral',
@@ -123,12 +45,8 @@ def _extract_core_subject(topic: str) -> str:
'recommendations', 'advice',
'prompt', 'prompts', 'prompting',
'methods', 'strategies', 'approaches',
}
words = text.split()
filtered = [w for w in words if w not in noise]
result = ' '.join(filtered) if filtered else text
return result.rstrip('?!.')
})
return extract_core_subject(topic, noise=_TIKTOK_NOISE)
def _log(msg: str):
@@ -204,26 +122,36 @@ def search_tiktok(
if not token:
return {"items": [], "error": "No SCRAPECREATORS_API_KEY configured"}
if not _requests:
return {"items": [], "error": "requests library not installed"}
config = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"])
core_topic = _extract_core_subject(topic)
_log(f"Searching TikTok for '{core_topic}' (depth={depth}, count={config['results_per_page']})")
try:
resp = _requests.get(
f"{SCRAPECREATORS_BASE}/search/keyword",
params={"query": core_topic, "sort_by": "relevance"},
headers=_sc_headers(token),
timeout=30,
)
resp.raise_for_status()
data = resp.json()
except Exception as e:
_log(f"ScrapeCreators error: {e}")
return {"items": [], "error": f"{type(e).__name__}: {e}"}
if not _requests:
_log("requests library not installed, falling back to urllib")
try:
from urllib.parse import urlencode
params = urlencode({"query": core_topic, "sort_by": "relevance"})
url = f"{SCRAPECREATORS_BASE}/search/keyword?{params}"
headers = _sc_headers(token)
headers["User-Agent"] = http.USER_AGENT
data = http.get(url, headers=headers, timeout=30, retries=2)
except Exception as e:
_log(f"ScrapeCreators error (urllib): {e}")
return {"items": [], "error": f"{type(e).__name__}: {e}"}
else:
try:
resp = _requests.get(
f"{SCRAPECREATORS_BASE}/search/keyword",
params={"query": core_topic, "sort_by": "relevance"},
headers=_sc_headers(token),
timeout=30,
)
resp.raise_for_status()
data = resp.json()
except Exception as e:
_log(f"ScrapeCreators error: {e}")
return {"items": [], "error": f"{type(e).__name__}: {e}"}
# Items are nested under aweme_info
raw_entries = data.get("search_item_list") or data.get("data") or []
scripts/lib/youtube_yt.py
+11 -87
View File
@@ -35,65 +35,7 @@ TRANSCRIPT_LIMITS = {
# Max words to keep from each transcript
TRANSCRIPT_MAX_WORDS = 500
# Stopwords for relevance computation (common English words that dilute token overlap)
STOPWORDS = frozenset({
'the', 'a', 'an', 'to', 'for', 'how', 'is', 'in', 'of', 'on',
'and', 'with', 'from', 'by', 'at', 'this', 'that', 'it', 'my',
'your', 'i', 'me', 'we', 'you', 'what', 'are', 'do', 'can',
'its', 'be', 'or', 'not', 'no', 'so', 'if', 'but', 'about',
'all', 'just', 'get', 'has', 'have', 'was', 'will',
})
# Synonym groups for relevance scoring (bidirectional expansion)
SYNONYMS = {
'hip': {'rap', 'hiphop'},
'hop': {'rap', 'hiphop'},
'rap': {'hip', 'hop', 'hiphop'},
'hiphop': {'rap', 'hip', 'hop'},
'js': {'javascript'},
'javascript': {'js'},
'ts': {'typescript'},
'typescript': {'ts'},
'ai': {'artificial', 'intelligence'},
'ml': {'machine', 'learning'},
'react': {'reactjs'},
'reactjs': {'react'},
'svelte': {'sveltejs'},
'sveltejs': {'svelte'},
'vue': {'vuejs'},
'vuejs': {'vue'},
}
def _tokenize(text: str) -> Set[str]:
"""Lowercase, strip punctuation, remove stopwords, drop single-char tokens.
Expands tokens with synonyms for better cross-domain matching."""
words = re.sub(r'[^\w\s]', ' ', text.lower()).split()
tokens = {w for w in words if w not in STOPWORDS and len(w) > 1}
# Expand synonyms
expanded = set(tokens)
for t in tokens:
if t in SYNONYMS:
expanded.update(SYNONYMS[t])
return expanded
def _compute_relevance(query: str, title: str) -> float:
"""Compute relevance as ratio of query tokens found in title.
Uses ratio overlap (intersection / query_length) so short queries
score higher when fully represented in the title. Floors at 0.1.
"""
q_tokens = _tokenize(query)
t_tokens = _tokenize(title)
if not q_tokens:
return 0.5 # Neutral fallback for empty/stopword-only queries
overlap = len(q_tokens & t_tokens)
ratio = overlap / len(q_tokens)
return max(0.1, min(1.0, ratio))
from .relevance import token_overlap_relevance as _compute_relevance
def _log(msg: str):
@@ -110,26 +52,12 @@ def is_ytdlp_installed() -> bool:
def _extract_core_subject(topic: str) -> str:
"""Extract core subject from verbose query for YouTube search.
Strips meta/research words to keep only the core product/concept name,
similar to bird_x.py's approach.
NOTE: 'tips', 'tricks', 'tutorial', 'guide', 'review', 'reviews'
are intentionally KEPT — they're YouTube content types that improve search.
"""
text = topic.lower().strip()
# Strip multi-word prefixes
prefixes = [
'what are the best', 'what is the best', 'what are the latest',
'what are people saying about', 'what do people think about',
'how do i use', 'how to use', 'how to',
'what are', 'what is', 'tips for', 'best practices for',
]
for p in prefixes:
if text.startswith(p + ' '):
text = text[len(p):].strip()
# Strip individual noise words
# NOTE: 'tips', 'tricks', 'tutorial', 'guide', 'review', 'reviews'
# are intentionally KEPT — they're YouTube content types that improve search
noise = {
from .query import extract_core_subject
# YouTube-specific noise set: smaller than default, keeps content-type words
_YT_NOISE = frozenset({
'best', 'top', 'good', 'great', 'awesome', 'killer',
'latest', 'new', 'news', 'update', 'updates',
'trending', 'hottest', 'popular', 'viral',
@@ -137,12 +65,8 @@ def _extract_core_subject(topic: str) -> str:
'recommendations', 'advice',
'prompt', 'prompts', 'prompting',
'methods', 'strategies', 'approaches',
}
words = text.split()
filtered = [w for w in words if w not in noise]
result = ' '.join(filtered) if filtered else text
return result.rstrip('?!.')
})
return extract_core_subject(topic, noise=_YT_NOISE)
def search_youtube(
@@ -171,9 +95,9 @@ def search_youtube(
_log(f"Searching YouTube for '{core_topic}' (since {from_date}, count={count})")
# yt-dlp search with full metadata (no --flat-playlist so dates are real).
# No --dateafter — we filter by date in Python with a soft fallback,
# because YouTube search returns relevance-sorted results and strict date
# filtering returns 0 for evergreen topics like "thumbnail tips".
# NOTE: --dateafter intentionally omitted — YouTube search returns
# relevance-sorted results and strict date filtering returns 0 for
# evergreen topics. Python soft filter (below) handles date filtering.
cmd = [
"yt-dlp",
"--ignore-config",