Merge pull request #340 from dzivkovi/fix/youtube-transcript-observability

fix(youtube): surface transcript-fetch ratio + add degraded nudge for stale yt-dlp
This commit is contained in:
Trevin Chow
2026-05-17 00:31:37 -07:00
committed by GitHub
5 changed files with 268 additions and 11 deletions
+70 -6
View File
@@ -45,26 +45,51 @@ def _is_youtube_active(config: dict, research_results: dict) -> bool:
return True
# Below this transcript-fetch ratio, YouTube is considered "degraded" rather
# than active. Picked at 50% so a single legitimate caption-disabled video in a
# multi-video result does not trip the nudge, but a stale-yt-dlp run that fails
# every transcript does. Tunable via DEGRADED_TRANSCRIPT_THRESHOLD env var if
# operators need to adjust without code changes.
DEFAULT_DEGRADED_TRANSCRIPT_THRESHOLD = 0.5
def _is_youtube_degraded(research_results: dict, threshold: float) -> bool:
"""YouTube is degraded when videos were returned but the transcript-fetch
ratio is below threshold. The canonical cause is a stale yt-dlp binary -
YouTube's caption format changes frequently and old binaries silently fail
every transcript while the search itself still succeeds.
"""
videos = int(research_results.get("youtube_videos_count") or 0)
transcripts = int(research_results.get("youtube_transcripts_count") or 0)
if videos <= 0:
return False
return (transcripts / videos) < threshold
def compute_quality_score(config: dict, research_results: dict) -> dict:
"""Compute research quality score based on 5 core sources.
Args:
config: Configuration dict from env.get_config()
research_results: Dict with keys like x_error, youtube_error,
reddit_error reflecting what happened this run.
reddit_error reflecting what happened this run. Optional keys
``youtube_videos_count`` and ``youtube_transcripts_count`` enable
degraded-YouTube detection (transcript-fetch ratio below threshold).
Returns:
{
"score_pct": 40-100,
"core_active": ["hn", "polymarket", ...],
"core_missing": ["x", "youtube"],
"core_errored": [], # configured but errored
"core_errored": [], # configured but errored at top level
"core_degraded": [], # configured and returned items but quality below threshold
"nudge_text": "..." or None if 100%
}
"""
core_active: List[str] = []
core_missing: List[str] = []
core_errored: List[str] = []
core_degraded: List[str] = []
# HN, Polymarket, and Reddit are always active
core_active.append("hn")
@@ -84,6 +109,13 @@ def compute_quality_score(config: dict, research_results: dict) -> dict:
yt_active = _is_youtube_active(config, research_results)
if yt_active:
core_active.append("youtube")
# Active means yt-dlp is installed and search did not error at the top
# level. But search-success + transcript-failure is the canonical
# stale-binary failure mode that the footer used to hide. Flag as
# degraded so the user gets an actionable nudge to update the binary.
threshold = float(config.get("DEGRADED_TRANSCRIPT_THRESHOLD") or DEFAULT_DEGRADED_TRANSCRIPT_THRESHOLD)
if _is_youtube_degraded(research_results, threshold):
core_degraded.append("youtube")
else:
core_missing.append("youtube")
# Check if configured but errored (yt-dlp installed but failed this run)
@@ -99,24 +131,41 @@ def compute_quality_score(config: dict, research_results: dict) -> dict:
has_sc = bool(config.get("SCRAPECREATORS_API_KEY"))
active_sources = research_results.get("active_sources") or []
nudge_text = _build_nudge_text(core_missing, core_errored, has_sc=has_sc, active_sources=active_sources) if core_missing else None
nudge_text = _build_nudge_text(
core_missing,
core_errored,
core_degraded,
research_results,
has_sc=has_sc,
active_sources=active_sources,
) if (core_missing or core_degraded) else None
return {
"score_pct": score_pct,
"core_active": core_active,
"core_missing": core_missing,
"core_errored": core_errored,
"core_degraded": core_degraded,
"nudge_text": nudge_text,
}
def _build_nudge_text(core_missing: List[str], core_errored: List[str], has_sc: bool = False, active_sources: list = None) -> str:
"""Build human-readable nudge text describing what was missed.
def _build_nudge_text(
core_missing: List[str],
core_errored: List[str],
core_degraded: List[str] = None,
research_results: dict = None,
has_sc: bool = False,
active_sources: list = None,
) -> str:
"""Build human-readable nudge text describing what was missed or degraded.
Prioritizes free suggestions. Optionally mentions bonus sources
(TikTok, Instagram, Threads, Pinterest) if ScrapeCreators key is configured.
"""
lines: List[str] = []
core_degraded = core_degraded or []
research_results = research_results or {}
# Describe what was missed
missed_parts: List[str] = []
@@ -129,7 +178,11 @@ def _build_nudge_text(core_missing: List[str], core_errored: List[str], has_sc:
active_count = 5 - len(core_missing)
lines.append(f"Research quality: {active_count}/5 core sources.")
lines.append(f"Missing: {', '.join(missed_parts)}.")
if missed_parts:
lines.append(f"Missing: {', '.join(missed_parts)}.")
if core_degraded:
degraded_labels = ", ".join(SOURCE_LABELS[s] for s in core_degraded)
lines.append(f"Degraded: {degraded_labels}.")
lines.append("")
# Free suggestions
@@ -159,6 +212,17 @@ def _build_nudge_text(core_missing: List[str], core_errored: List[str], has_sc:
"explanations on any topic. Install yt-dlp: brew install yt-dlp (free)"
)
if "youtube" in core_degraded:
videos = int(research_results.get("youtube_videos_count") or 0)
transcripts = int(research_results.get("youtube_transcripts_count") or 0)
free_suggestions.append(
f"YouTube returned {videos} videos but only {transcripts} transcripts "
"captured. The most common cause is a stale yt-dlp binary - YouTube's "
"caption format changes frequently and old binaries silently fail every "
"transcript. Update via your package manager: scoop update yt-dlp "
"(Windows), brew upgrade yt-dlp (macOS), or pip install -U yt-dlp."
)
# Mention bonus opt-in sources when SC key is present
if has_sc:
bonus_hints = []
+5 -4
View File
@@ -1285,15 +1285,16 @@ def _build_source_footer_lines(report: schema.Report) -> list[str]:
if total > 0:
total_str = f"{total:,}" if total >= 1000 else str(total)
parts.append(f"{total_str} {word}")
# YouTube: append "N with transcripts" instead of a third likes-based column.
# Transcripts are a more meaningful research-depth signal than likes.
# YouTube: always append "M/N with transcripts" so a zero-transcript run
# (typically caused by a stale yt-dlp binary) is visible at the conclusion
# surface. Hiding zero converts a problem signal into an absence; the very
# case that needs to be loud is the one previously omitted from the footer.
if source_key == "youtube":
with_transcripts = sum(
1 for it in items
if (it.metadata.get("transcript_highlights") or it.metadata.get("transcript_snippet"))
)
if with_transcripts > 0:
parts.append(f"{with_transcripts} with transcripts")
parts.append(f"{with_transcripts}/{len(items)} with transcripts")
stats = "".join(parts)
out.append(_footer_line_for_source(emoji, label, len(items), item_word, stats))