AndrianBalanescu commited on
Commit
58cb127
Β·
1 Parent(s): cdafb70

fix(pipeline): stop reporting success over broken harvest stages

Browse files

The weekly harvest reported "COMPLETED SUCCESSFULLY" while transcription had
been producing nothing for weeks: 8 of the 12 recorded runs transcribed zero
videos, and 15 of 16 runs printed the success banner anyway.

- transcribe: return attempted/processed/failed counts and exit 1 when work
was attempted but nothing was produced, so a total failure is visible.
- transcribe: the configured endpoint pointed at homelab:8086, where nothing
listens; Whisper is served via Caddy at /api/v1/whisper/v1/audio/
transcriptions, and requires the multipart part to be named "file".
- transcribe: select the real downloaded media file. The old check accepted
any non-".part" file, which matched yt-dlp's ".ytdl" metadata JSON and made
Whisper answer HTTP 500 "Invalid data".
- transcribe: drop --impersonate (combined with browser cookies it removes
audio-only formats, so the fallback downloaded a 475 MB 1080p video) and
normalize everything to 16 kHz mono WAV with ffmpeg.
- transcribe: enforce max_duration_sec, which was a dead parameter, so
hour-long clips were downloaded in full.
- pipeline: add set -o pipefail. Every stage pipes into tee, so a failing
stage was invisible to set -e. Treat a transcription outage as a loud
non-fatal warning, and add a freshness health check that exits non-zero
when the corpus, transcripts, or digests are stale or missing.
- setup_crontab: invoke the shell script with bash. Cron ran it through
python3, which crashed with SyntaxError on every run.

moldovan-qwen/scripts/run_automated_pipeline.sh CHANGED
@@ -8,6 +8,10 @@
8
  # 4. Generates new viral short-form video scripts (TikTok/Reels)
9
 
10
  set -e
 
 
 
 
11
 
12
  SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
13
  BASE_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
@@ -39,7 +43,13 @@ fi
39
 
40
  # Step 2: Transcribe new videos via homelab Whisper
41
  echo "[2/6] Transcribing new videos via homelab Whisper..." | tee -a "$LOG_FILE"
42
- python3 "${SCRIPT_DIR}/transcribe_moldovan_videos.py" --limit 3 --corpus "${BASE_DIR}/data/youtube_corpus/moldovan_youtube_corpus.jsonl" --output-dir "${BASE_DIR}/data/transcripts" | tee -a "$LOG_FILE"
 
 
 
 
 
 
43
 
44
  # Step 3: Synthesize SFT Dataset
45
  echo "[3/6] Synthesizing SFT training dataset..." | tee -a "$LOG_FILE"
@@ -57,6 +67,40 @@ python3 "${SCRIPT_DIR}/generate_moldovan_video_scripts.py" --output-dir "${BASE_
57
  echo "[6/6] Synthesizing Daily News Digest with Edge-TTS ro-RO Voiceover..." | tee -a "$LOG_FILE"
58
  python3 "${BASE_DIR}/engine/synthesize_daily_news.py" | tee -a "$LOG_FILE"
59
 
60
- echo "============================================================" | tee -a "$LOG_FILE"
61
- echo "πŸŽ‰ PIPELINE CYCLE COMPLETED SUCCESSFULLY: $(date -u)" | tee -a "$LOG_FILE"
62
- echo "============================================================" | tee -a "$LOG_FILE"
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
8
  # 4. Generates new viral short-form video scripts (TikTok/Reels)
9
 
10
  set -e
11
+ # A failing left-hand command in a pipeline is invisible to `set -e` without
12
+ # pipefail: every stage below pipes into `tee`, so a broken stage used to leave
13
+ # the pipeline reporting "COMPLETED SUCCESSFULLY" anyway.
14
+ set -o pipefail
15
 
16
  SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
17
  BASE_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
 
43
 
44
  # Step 2: Transcribe new videos via homelab Whisper
45
  echo "[2/6] Transcribing new videos via homelab Whisper..." | tee -a "$LOG_FILE"
46
+ # Transcription depends on an external service (Whisper on the homelab). Its
47
+ # outage must NOT abort the other five stages, but it must be loud: this stage
48
+ # previously printed "Transcribed 0 videos" and exit 0, which hid a long outage.
49
+ if ! python3 "${SCRIPT_DIR}/transcribe_moldovan_videos.py" --limit 3 --corpus "${BASE_DIR}/data/youtube_corpus/moldovan_youtube_corpus.jsonl" --output-dir "${BASE_DIR}/data/transcripts" 2>&1 | tee -a "$LOG_FILE"; then
50
+ echo "[!] WARNING: transcription stage failed (Whisper unreachable or downloads blocked)." | tee -a "$LOG_FILE"
51
+ echo "[!] Continuing with remaining stages; corpus/transcripts may be stale." | tee -a "$LOG_FILE"
52
+ fi
53
 
54
  # Step 3: Synthesize SFT Dataset
55
  echo "[3/6] Synthesizing SFT training dataset..." | tee -a "$LOG_FILE"
 
67
  echo "[6/6] Synthesizing Daily News Digest with Edge-TTS ro-RO Voiceover..." | tee -a "$LOG_FILE"
68
  python3 "${BASE_DIR}/engine/synthesize_daily_news.py" | tee -a "$LOG_FILE"
69
 
70
+ # ─── Freshness health check ───────────────────────────────────────────────────
71
+ # The weekly job previously reported success even when stages produced nothing,
72
+ # so staleness went unnoticed for weeks (transcripts had been empty since
73
+ # mid-August). Surface the age of every key artifact, and fail loudly if the
74
+ # corpus or the news digest did not advance in this run.
75
+ echo "[health] Artifact freshness:" | tee -a "$LOG_FILE"
76
+ HEALTH_FAIL=0
77
+ _report_age() {
78
+ local label="$1" path="$2" max_days="$3"
79
+ if [ ! -e "$path" ]; then
80
+ echo " [!] ${label}: MISSING (${path})" | tee -a "$LOG_FILE"
81
+ HEALTH_FAIL=1
82
+ return
83
+ fi
84
+ local age_days
85
+ age_days=$(( ( $(date +%s) - $(stat -f %m "$path" 2>/dev/null || stat -c %Y "$path") ) / 86400 ))
86
+ if [ "$age_days" -gt "$max_days" ]; then
87
+ echo " [!] ${label}: STALE (${age_days}d > ${max_days}d)" | tee -a "$LOG_FILE"
88
+ HEALTH_FAIL=1
89
+ else
90
+ echo " [βœ“] ${label}: ${age_days}d old" | tee -a "$LOG_FILE"
91
+ fi
92
+ }
93
+ _report_age "youtube corpus" "${BASE_DIR}/data/youtube_corpus/moldovan_youtube_corpus.jsonl" 10
94
+ _report_age "transcripts dir" "${BASE_DIR}/data/transcripts" 10
95
+ _report_age "news digest dir" "${BASE_DIR}/data/generated_digests" 3
96
+
97
+ if [ "$HEALTH_FAIL" -eq 0 ]; then
98
+ echo "============================================================" | tee -a "$LOG_FILE"
99
+ echo "πŸŽ‰ PIPELINE CYCLE COMPLETED SUCCESSFULLY: $(date -u)" | tee -a "$LOG_FILE"
100
+ echo "============================================================" | tee -a "$LOG_FILE"
101
+ else
102
+ echo "============================================================" | tee -a "$LOG_FILE"
103
+ echo "⚠️ PIPELINE COMPLETED WITH STALE/MISSING ARTIFACTS: $(date -u)" | tee -a "$LOG_FILE"
104
+ echo "============================================================" | tee -a "$LOG_FILE"
105
+ exit 1
106
+ fi
moldovan-qwen/scripts/setup_crontab.sh CHANGED
@@ -20,7 +20,7 @@ export HF_TOKEN="${HF_TOKEN:-$(printenv HF_TOKEN)}"
20
  ALBUM_SCHEDULE="0 3 * * * /usr/bin/python3 ${ALBUM_SCRIPT} >> ${LOG_DIR}/daily_album_cron.log 2>&1"
21
 
22
  # 2. General pipeline harvesting & datasets (Every Sunday at 02:00 AM)
23
- PIPELINE_SCHEDULE="0 2 * * 0 ${PIPELINE_SCRIPT} >> ${LOG_DIR}/cron_pipeline.log 2>&1"
24
 
25
  echo "============================================================"
26
  echo "πŸ‡²πŸ‡© MOLDOVAN AI AUTOMATED SCHEDULER SETUP"
 
20
  ALBUM_SCHEDULE="0 3 * * * /usr/bin/python3 ${ALBUM_SCRIPT} >> ${LOG_DIR}/daily_album_cron.log 2>&1"
21
 
22
  # 2. General pipeline harvesting & datasets (Every Sunday at 02:00 AM)
23
+ PIPELINE_SCHEDULE="0 2 * * 0 . \$HOME/.flow_cron_env && /bin/bash ${PIPELINE_SCRIPT} >> ${LOG_DIR}/cron_pipeline.log 2>&1"
24
 
25
  echo "============================================================"
26
  echo "πŸ‡²πŸ‡© MOLDOVAN AI AUTOMATED SCHEDULER SETUP"
moldovan-qwen/scripts/transcribe_moldovan_videos.py CHANGED
@@ -8,10 +8,11 @@ using the Whisper STT service on the homelab (now accessible via Tailscale).
8
  Pipeline:
9
  1. Read video IDs from the YouTube corpus (moldovan_youtube_corpus.jsonl)
10
  2. Download audio with yt-dlp (works even though subtitles are bot-walled)
11
- 3. Transcribe with Whisper at http://homelab:8086/v1/transcribe
12
  4. Save transcripts to data/transcripts/
13
 
14
- Whisper endpoint: POST http://homelab:8086/v1/transcribe (multipart audio file)
 
15
  """
16
 
17
  import os
@@ -23,7 +24,16 @@ import urllib.request
23
  import urllib.error
24
  from typing import List, Dict, Optional
25
 
26
- WHISPER_URL = os.environ.get("WHISPER_URL", "http://homelab:8086/v1/transcribe")
 
 
 
 
 
 
 
 
 
27
  CORPUS_PATH = "moldovan-qwen/data/youtube_corpus/moldovan_youtube_corpus.jsonl"
28
  TRANSCRIPT_DIR = "moldovan-qwen/data/transcripts"
29
 
@@ -51,33 +61,113 @@ def get_existing_transcripts(transcript_dir: str) -> set:
51
  existing.add(fname[:-5])
52
  return existing
53
 
54
- def download_audio(video_id: str, work_dir: str) -> Optional[str]:
55
- """Download audio for a video using yt-dlp. Returns path to audio file."""
56
- out_template = os.path.join(work_dir, f"{video_id}.%(ext)s")
 
 
 
 
 
 
 
 
57
  cmd = [
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
58
  "yt-dlp",
59
- "--impersonate", "chrome",
60
  "--cookies-from-browser", "chrome",
61
- "-f", "bestaudio[ext=m4a]/bestaudio",
 
 
 
62
  "-o", out_template,
63
  f"https://www.youtube.com/watch?v={video_id}",
64
  ]
65
  try:
66
  result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
67
  if result.returncode != 0:
68
- print(f" [!] yt-dlp failed for {video_id}: {result.stderr[-200:]}")
69
- return None
70
- # Find the downloaded file
71
- for fname in os.listdir(work_dir):
72
- if fname.startswith(video_id) and not fname.endswith(".part"):
73
- return os.path.join(work_dir, fname)
74
- return None
75
  except subprocess.TimeoutExpired:
76
  print(f" [!] yt-dlp timed out for {video_id}")
77
- return None
78
  except Exception as e:
79
  print(f" [!] yt-dlp error for {video_id}: {e}")
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
80
  return None
 
 
81
 
82
  def transcribe_audio(audio_path: str, timeout: int = 600) -> Optional[Dict]:
83
  """Send audio to homelab Whisper for transcription."""
@@ -87,10 +177,12 @@ def transcribe_audio(audio_path: str, timeout: int = 600) -> Optional[Dict]:
87
  with open(audio_path, "rb") as f:
88
  audio_data = f.read()
89
 
90
- # Build multipart body
 
 
91
  body = b""
92
  body += f"--{boundary}\r\n".encode()
93
- body += b'Content-Disposition: form-data; name="audio"; filename="audio.wav"\r\n'
94
  body += b"Content-Type: audio/wav\r\n\r\n"
95
  body += audio_data
96
  body += f"\r\n--{boundary}--\r\n".encode()
@@ -114,8 +206,14 @@ def transcribe_audio(audio_path: str, timeout: int = 600) -> Optional[Dict]:
114
 
115
  def transcribe_videos(limit: int = 5, max_duration_sec: int = 600,
116
  corpus_path: str = CORPUS_PATH,
117
- transcript_dir: str = TRANSCRIPT_DIR):
118
- """Transcribe a batch of videos from the corpus."""
 
 
 
 
 
 
119
  os.makedirs(transcript_dir, exist_ok=True)
120
  os.makedirs("/tmp/moldovan_audio", exist_ok=True)
121
 
@@ -128,7 +226,7 @@ def transcribe_videos(limit: int = 5, max_duration_sec: int = 600,
128
 
129
  if not pending:
130
  print("[βœ“] All videos already transcribed!")
131
- return
132
 
133
  # PRIORITIZE SPOKEN CONTENT over music:
134
  # Podcasts, news, interviews, satire, vlogs, diaspora = spoken (transcribe first)
@@ -137,31 +235,41 @@ def transcribe_videos(limit: int = 5, max_duration_sec: int = 600,
137
  spoken = [v for v in pending if v.get("category") in SPOKEN_CATEGORIES]
138
  music = [v for v in pending if v.get("category") not in SPOKEN_CATEGORIES]
139
 
140
- # Sort spoken content by views (most popular first), then fall back to music
141
- spoken.sort(key=lambda v: v.get("view_count", 0), reverse=True)
142
- music.sort(key=lambda v: v.get("view_count", 0), reverse=True)
 
 
 
 
 
143
  pending = spoken + music
144
 
145
  print(f"[*] Prioritizing spoken content: {len(spoken)} spoken + {len(music)} music candidates.")
146
 
147
  processed = 0
 
 
148
  for video in pending[:limit]:
149
  vid = video["video_id"]
150
  title = video.get("title", "")[:60]
151
  cat = video.get("category", "general")
152
  print(f"\n[{processed+1}/{min(limit, len(pending))}] Transcribing [{cat.upper()}]: {title}")
153
  print(f" Video: {vid} | Views: {video.get('view_count', 0)}")
 
154
 
155
  # Download audio
156
- audio_path = download_audio(vid, "/tmp/moldovan_audio")
157
  if not audio_path:
158
  print(" [!] Could not download audio. Skipping.")
 
159
  continue
160
 
161
  # Transcribe
162
  result = transcribe_audio(audio_path)
163
  if not result:
164
  print(" [!] Transcription failed. Skipping.")
 
165
  continue
166
 
167
  detected_lang = result.get("language", "unknown")
@@ -172,6 +280,7 @@ def transcribe_videos(limit: int = 5, max_duration_sec: int = 600,
172
  expected_langs = {"ro", "ru", "mo", "en"}
173
  if detected_lang not in expected_langs:
174
  print(f" [!] Skipping β€” detected language '{detected_lang}' (likely music/audio, not spoken content).")
 
175
  try:
176
  os.remove(audio_path)
177
  except Exception:
@@ -206,16 +315,32 @@ def transcribe_videos(limit: int = 5, max_duration_sec: int = 600,
206
 
207
  time.sleep(1) # Gentle delay
208
 
209
- print(f"\n[βœ“] Done! Transcribed {processed} videos.")
 
210
 
211
  def main():
212
  parser = argparse.ArgumentParser(description="Transcribe Moldovan YouTube videos via homelab Whisper")
213
  parser.add_argument("--limit", type=int, default=5, help="Max videos to transcribe")
214
  parser.add_argument("--corpus", default=CORPUS_PATH, help="Path to corpus JSONL")
215
  parser.add_argument("--output-dir", default=TRANSCRIPT_DIR, help="Transcript output dir")
 
 
216
  args = parser.parse_args()
217
 
218
- transcribe_videos(limit=args.limit, corpus_path=args.corpus, transcript_dir=args.output_dir)
 
 
 
 
 
 
 
 
 
 
 
 
 
219
 
220
  if __name__ == "__main__":
221
- main()
 
8
  Pipeline:
9
  1. Read video IDs from the YouTube corpus (moldovan_youtube_corpus.jsonl)
10
  2. Download audio with yt-dlp (works even though subtitles are bot-walled)
11
+ 3. Transcribe with Whisper via the homelab Caddy proxy
12
  4. Save transcripts to data/transcripts/
13
 
14
+ Whisper endpoint: POST http://homelab/api/v1/whisper/v1/audio/transcriptions
15
+ (proxied by Caddy, which strips /api/v1/whisper; multipart field must be "file")
16
  """
17
 
18
  import os
 
24
  import urllib.error
25
  from typing import List, Dict, Optional
26
 
27
+ # Whisper is exposed through the homelab Caddy reverse proxy at
28
+ # /api/v1/whisper/* (prefix stripped, forwarded to faster-whisper:8000). The
29
+ # faster-whisper server implements the OpenAI-compatible route
30
+ # /v1/audio/transcriptions and requires the multipart field to be named "file".
31
+ # The previous value (http://homelab:8086/v1/transcribe) pointed at a port that
32
+ # nothing listens on, which is why transcription silently produced nothing.
33
+ WHISPER_URL = os.environ.get(
34
+ "WHISPER_URL",
35
+ "http://homelab/api/v1/whisper/v1/audio/transcriptions",
36
+ )
37
  CORPUS_PATH = "moldovan-qwen/data/youtube_corpus/moldovan_youtube_corpus.jsonl"
38
  TRANSCRIPT_DIR = "moldovan-qwen/data/transcripts"
39
 
 
61
  existing.add(fname[:-5])
62
  return existing
63
 
64
+ def _normalize_audio(src: str) -> Optional[str]:
65
+ """Downmix any media file to 16 kHz mono WAV for Whisper.
66
+
67
+ yt-dlp cannot always supply an audio-only stream (availability of the
68
+ audio renditions fluctuates per request and per video), so it sometimes
69
+ returns a full video container. Sending that to Whisper wastes bandwidth and
70
+ GPU memory and, on a busy GPU, fails with CUDA out of memory. Extracting a
71
+ small mono track locally keeps uploads tiny and transcription reliable.
72
+ Returns the original path if ffmpeg is unavailable or extraction fails.
73
+ """
74
+ out_path = os.path.splitext(src)[0] + ".16k.wav"
75
  cmd = [
76
+ "ffmpeg", "-nostdin", "-y", "-loglevel", "error",
77
+ "-i", src,
78
+ "-vn", "-ac", "1", "-ar", "16000", "-f", "wav",
79
+ out_path,
80
+ ]
81
+ try:
82
+ result = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
83
+ if result.returncode == 0 and os.path.isfile(out_path) \
84
+ and os.path.getsize(out_path) > 0:
85
+ return out_path
86
+ print(f" [!] ffmpeg normalize failed: {result.stderr[-160:]}")
87
+ except FileNotFoundError:
88
+ print(" [!] ffmpeg not found; sending original media to Whisper")
89
+ except subprocess.TimeoutExpired:
90
+ print(" [!] ffmpeg normalize timed out")
91
+ except Exception as e:
92
+ print(f" [!] ffmpeg normalize error: {e}")
93
+ return src
94
+
95
+
96
+ def _run_ytdlp(video_id: str, out_template: str, fmt: str,
97
+ max_duration_sec: int) -> bool:
98
+ """Run one yt-dlp attempt with a specific format selector."""
99
+ cmd = [
100
+ # NOTE: do NOT combine --impersonate with --cookies-from-browser. That
101
+ # combination makes audio-only formats disappear from yt-dlp's listing
102
+ # for some videos; the fallback then downloads a full 1080p VIDEO
103
+ # (measured: 475 MB for a 35-minute clip) which Whisper then fails on
104
+ # with CUDA out of memory. Cookies alone are sufficient.
105
  "yt-dlp",
 
106
  "--cookies-from-browser", "chrome",
107
+ "-f", fmt,
108
+ "--no-playlist",
109
+ # Do not download clips longer than the caller's limit.
110
+ "--match-filter", f"duration < {max_duration_sec}",
111
  "-o", out_template,
112
  f"https://www.youtube.com/watch?v={video_id}",
113
  ]
114
  try:
115
  result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
116
  if result.returncode != 0:
117
+ tail = (result.stderr or result.stdout or "").strip()[-160:]
118
+ print(f" [!] yt-dlp failed ({fmt}): {tail}")
119
+ return False
120
+ return True
 
 
 
121
  except subprocess.TimeoutExpired:
122
  print(f" [!] yt-dlp timed out for {video_id}")
123
+ return False
124
  except Exception as e:
125
  print(f" [!] yt-dlp error for {video_id}: {e}")
126
+ return False
127
+
128
+
129
+ def download_audio(video_id: str, work_dir: str,
130
+ max_duration_sec: int = 600) -> Optional[str]:
131
+ """Download audio for a video using yt-dlp, normalized to 16 kHz mono WAV.
132
+
133
+ Format availability on YouTube is not stable: `bestaudio` can report
134
+ "Requested format is not available" for a video on one request and succeed
135
+ on the next, and `bestaudio/best` silently degrades to a full video stream.
136
+ So we try progressively permissive selectors, then normalize whatever we get
137
+ to a small mono WAV. Returns None only if every attempt fails.
138
+ """
139
+ out_template = os.path.join(work_dir, f"{video_id}.%(ext)s")
140
+
141
+ # Ordered from most to least specific. `best` is last because it may include
142
+ # video; normalization below makes that acceptable rather than fatal.
143
+ for fmt in ("bestaudio", "bestaudio/best", "best"):
144
+ if _run_ytdlp(video_id, out_template, fmt, max_duration_sec):
145
+ media = _pick_downloaded_media(video_id, work_dir)
146
+ if media:
147
+ return _normalize_audio(media)
148
+ print(f" [!] yt-dlp reported success but no media found ({fmt})")
149
+ return None
150
+
151
+
152
+ def _pick_downloaded_media(video_id: str, work_dir: str) -> Optional[str]:
153
+ """Pick the real downloaded media file, never a sidecar or partial."""
154
+ media_exts = {".m4a", ".webm", ".mp4", ".opus", ".mp3", ".wav", ".aac",
155
+ ".ogg", ".mkv", ".flac"}
156
+ candidates = []
157
+ for fname in os.listdir(work_dir):
158
+ if not fname.startswith(video_id):
159
+ continue
160
+ if ".part" in fname or fname.endswith(".ytdl"):
161
+ continue
162
+ if os.path.splitext(fname)[1].lower() not in media_exts:
163
+ continue
164
+ full = os.path.join(work_dir, fname)
165
+ if os.path.isfile(full) and os.path.getsize(full) > 0:
166
+ candidates.append(full)
167
+ if not candidates:
168
  return None
169
+ return max(candidates, key=lambda p: os.path.getmtime(p))
170
+
171
 
172
  def transcribe_audio(audio_path: str, timeout: int = 600) -> Optional[Dict]:
173
  """Send audio to homelab Whisper for transcription."""
 
177
  with open(audio_path, "rb") as f:
178
  audio_data = f.read()
179
 
180
+ # Build multipart body. The faster-whisper server requires the file part to
181
+ # be named "file" (OpenAI-compatible schema); it was previously sent as
182
+ # "audio", which the server rejected.
183
  body = b""
184
  body += f"--{boundary}\r\n".encode()
185
+ body += b'Content-Disposition: form-data; name="file"; filename="audio.wav"\r\n'
186
  body += b"Content-Type: audio/wav\r\n\r\n"
187
  body += audio_data
188
  body += f"\r\n--{boundary}--\r\n".encode()
 
206
 
207
  def transcribe_videos(limit: int = 5, max_duration_sec: int = 600,
208
  corpus_path: str = CORPUS_PATH,
209
+ transcript_dir: str = TRANSCRIPT_DIR) -> dict:
210
+ """Transcribe a batch of videos from the corpus.
211
+
212
+ Returns a summary dict ``{attempted, processed, failed}`` so callers (and the
213
+ weekly pipeline) can tell a healthy no-op from a total failure. Previously
214
+ the function printed "Done! Transcribed 0 videos." and exited 0 even when
215
+ every download failed, which made a fully broken harvest look like success.
216
+ """
217
  os.makedirs(transcript_dir, exist_ok=True)
218
  os.makedirs("/tmp/moldovan_audio", exist_ok=True)
219
 
 
226
 
227
  if not pending:
228
  print("[βœ“] All videos already transcribed!")
229
+ return {"attempted": 0, "processed": 0, "failed": 0}
230
 
231
  # PRIORITIZE SPOKEN CONTENT over music:
232
  # Podcasts, news, interviews, satire, vlogs, diaspora = spoken (transcribe first)
 
235
  spoken = [v for v in pending if v.get("category") in SPOKEN_CATEGORIES]
236
  music = [v for v in pending if v.get("category") not in SPOKEN_CATEGORIES]
237
 
238
+ # Sort spoken content by recency first (freshest news/interviews feed the
239
+ # current topical corpus), using views only to break ties within a day.
240
+ # Ranking by all-time views transcribed decade-old viral clips instead of
241
+ # this week's content.
242
+ spoken.sort(key=lambda v: (v.get("published_at", ""), v.get("view_count", 0)),
243
+ reverse=True)
244
+ music.sort(key=lambda v: (v.get("published_at", ""), v.get("view_count", 0)),
245
+ reverse=True)
246
  pending = spoken + music
247
 
248
  print(f"[*] Prioritizing spoken content: {len(spoken)} spoken + {len(music)} music candidates.")
249
 
250
  processed = 0
251
+ failed = 0
252
+ attempted = 0
253
  for video in pending[:limit]:
254
  vid = video["video_id"]
255
  title = video.get("title", "")[:60]
256
  cat = video.get("category", "general")
257
  print(f"\n[{processed+1}/{min(limit, len(pending))}] Transcribing [{cat.upper()}]: {title}")
258
  print(f" Video: {vid} | Views: {video.get('view_count', 0)}")
259
+ attempted += 1
260
 
261
  # Download audio
262
+ audio_path = download_audio(vid, "/tmp/moldovan_audio", max_duration_sec)
263
  if not audio_path:
264
  print(" [!] Could not download audio. Skipping.")
265
+ failed += 1
266
  continue
267
 
268
  # Transcribe
269
  result = transcribe_audio(audio_path)
270
  if not result:
271
  print(" [!] Transcription failed. Skipping.")
272
+ failed += 1
273
  continue
274
 
275
  detected_lang = result.get("language", "unknown")
 
280
  expected_langs = {"ro", "ru", "mo", "en"}
281
  if detected_lang not in expected_langs:
282
  print(f" [!] Skipping β€” detected language '{detected_lang}' (likely music/audio, not spoken content).")
283
+ failed += 1
284
  try:
285
  os.remove(audio_path)
286
  except Exception:
 
315
 
316
  time.sleep(1) # Gentle delay
317
 
318
+ print(f"\n[βœ“] Done! Transcribed {processed}/{attempted} videos (failed: {failed}).")
319
+ return {"attempted": attempted, "processed": processed, "failed": failed}
320
 
321
  def main():
322
  parser = argparse.ArgumentParser(description="Transcribe Moldovan YouTube videos via homelab Whisper")
323
  parser.add_argument("--limit", type=int, default=5, help="Max videos to transcribe")
324
  parser.add_argument("--corpus", default=CORPUS_PATH, help="Path to corpus JSONL")
325
  parser.add_argument("--output-dir", default=TRANSCRIPT_DIR, help="Transcript output dir")
326
+ parser.add_argument("--allow-zero", action="store_true",
327
+ help="Exit 0 even if nothing was transcribed (e.g. nothing pending)")
328
  args = parser.parse_args()
329
 
330
+ summary = transcribe_videos(limit=args.limit, corpus_path=args.corpus,
331
+ transcript_dir=args.output_dir)
332
+ if not summary:
333
+ return 0
334
+
335
+ # A total failure of the batch must surface as a non-zero exit so the weekly
336
+ # pipeline stops reporting "COMPLETED SUCCESSFULLY" while transcription is
337
+ # silently broken (that masking hid a long-running Whisper outage).
338
+ if summary["attempted"] > 0 and summary["processed"] == 0 and not args.allow_zero:
339
+ print("[!] ERROR: attempted transcription but produced no transcripts.")
340
+ print("[!] Check yt-dlp downloads and the Whisper endpoint (WHISPER_URL).")
341
+ return 1
342
+ return 0
343
+
344
 
345
  if __name__ == "__main__":
346
+ raise SystemExit(main())
tests/unit/test_pipeline_failure_honesty.py ADDED
@@ -0,0 +1,257 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """Regression contracts for the weekly harvesting pipeline's failure honesty.
2
+
3
+ Two silent-failure bugs made a fully broken harvest look healthy:
4
+
5
+ 1. ``transcribe_moldovan_videos.py`` printed "Done! Transcribed 0 videos." and
6
+ exited 0 when every download or transcription failed. The weekly pipeline
7
+ therefore reported "COMPLETED SUCCESSFULLY" while transcription had been
8
+ producing nothing (measured: 8 of 12 recorded runs transcribed 0 videos).
9
+
10
+ 2. ``run_automated_pipeline.sh`` used ``set -e`` without ``pipefail``, and every
11
+ stage pipes into ``tee``. A failing stage on the left of a pipe was invisible
12
+ to ``set -e``, so any broken stage still ended in a success banner.
13
+
14
+ 3. The yt-dlp format selector ``bestaudio[ext=m4a]/bestaudio`` failed with
15
+ "Requested format is not available" whenever the only audio-only streams were
16
+ webm/opus, silently yielding 0 downloads.
17
+
18
+ All tests are offline: network calls are stubbed, nothing is downloaded.
19
+ """
20
+
21
+ from __future__ import annotations
22
+
23
+ import importlib.util
24
+ import sys
25
+ from pathlib import Path
26
+
27
+ import pytest
28
+
29
+ REPO_ROOT = Path(__file__).resolve().parents[2]
30
+ TRANSCRIBE = REPO_ROOT / "moldovan-qwen" / "scripts" / "transcribe_moldovan_videos.py"
31
+ PIPELINE = REPO_ROOT / "moldovan-qwen" / "scripts" / "run_automated_pipeline.sh"
32
+
33
+
34
+ def _load_transcribe():
35
+ spec = importlib.util.spec_from_file_location("transcribe_mod", TRANSCRIBE)
36
+ mod = importlib.util.module_from_spec(spec)
37
+ spec.loader.exec_module(mod)
38
+ return mod
39
+
40
+
41
+ def _run_main(mod, argv):
42
+ old = sys.argv
43
+ sys.argv = ["transcribe_moldovan_videos.py", *argv]
44
+ try:
45
+ return mod.main()
46
+ finally:
47
+ sys.argv = old
48
+
49
+
50
+ def test_total_failure_exits_nonzero(monkeypatch):
51
+ """Attempting work and producing nothing must fail loudly, not exit 0."""
52
+ mod = _load_transcribe()
53
+ monkeypatch.setattr(
54
+ mod, "transcribe_videos",
55
+ lambda **kwargs: {"attempted": 3, "processed": 0, "failed": 3},
56
+ )
57
+
58
+ assert _run_main(mod, ["--limit", "3"]) == 1
59
+
60
+
61
+ def test_partial_success_exits_zero(monkeypatch):
62
+ """One good transcript means the batch is not a total failure."""
63
+ mod = _load_transcribe()
64
+ monkeypatch.setattr(
65
+ mod, "transcribe_videos",
66
+ lambda **kwargs: {"attempted": 3, "processed": 1, "failed": 2},
67
+ )
68
+
69
+ assert _run_main(mod, ["--limit", "3"]) == 0
70
+
71
+
72
+ def test_nothing_pending_is_a_healthy_noop(monkeypatch):
73
+ """Zero attempts is not a failure: nothing needed transcribing."""
74
+ mod = _load_transcribe()
75
+ monkeypatch.setattr(
76
+ mod, "transcribe_videos",
77
+ lambda **kwargs: {"attempted": 0, "processed": 0, "failed": 0},
78
+ )
79
+
80
+ assert _run_main(mod, ["--limit", "3"]) == 0
81
+
82
+
83
+ def test_allow_zero_override_keeps_exit_zero(monkeypatch):
84
+ """An explicit override must still be able to tolerate a total failure."""
85
+ mod = _load_transcribe()
86
+ monkeypatch.setattr(
87
+ mod, "transcribe_videos",
88
+ lambda **kwargs: {"attempted": 2, "processed": 0, "failed": 2},
89
+ )
90
+
91
+ assert _run_main(mod, ["--allow-zero"]) == 0
92
+
93
+
94
+ def test_transcribe_videos_returns_a_counts_summary(tmp_path, monkeypatch):
95
+ """The summary contract callers depend on: attempted/processed/failed."""
96
+ mod = _load_transcribe()
97
+ monkeypatch.setattr(mod, "CORPUS_PATH", str(tmp_path / "none.jsonl"))
98
+ monkeypatch.setattr(mod, "TRANSCRIPT_DIR", str(tmp_path / "out"))
99
+
100
+ summary = mod.transcribe_videos(
101
+ limit=1,
102
+ corpus_path=str(tmp_path / "none.jsonl"),
103
+ transcript_dir=str(tmp_path / "out"),
104
+ )
105
+
106
+ assert summary == {"attempted": 0, "processed": 0, "failed": 0}
107
+
108
+
109
+ def test_download_audio_uses_a_fallback_chain_not_a_strict_selector(tmp_path, monkeypatch):
110
+ """Format availability fluctuates, so a single strict selector is unsafe.
111
+
112
+ `bestaudio[ext=m4a]/bestaudio` failed with "Requested format is not
113
+ available" on webm/opus-only audio, and plain `bestaudio` was observed to
114
+ fail intermittently on the same video. The download path must try several
115
+ selectors and must not require an m4a container.
116
+ """
117
+ mod = _load_transcribe()
118
+ attempts = []
119
+
120
+ class _Result:
121
+ returncode = 0
122
+ stderr = ""
123
+
124
+ def fake_run(cmd, **kwargs):
125
+ attempts.append(cmd)
126
+ return _Result()
127
+
128
+ monkeypatch.setattr(mod.subprocess, "run", fake_run)
129
+ # The normalized WAV is what download_audio should end up returning.
130
+ (tmp_path / "vid123.16k.wav").write_bytes(b"wav")
131
+
132
+ mod.download_audio("vid123", str(tmp_path))
133
+
134
+ fmts = [c[c.index("-f") + 1] for c in attempts if "-f" in c]
135
+ assert fmts, "no yt-dlp format selector was used"
136
+ assert not any("[ext=m4a]" in f for f in fmts), (
137
+ "strict m4a-only selector reintroduced"
138
+ )
139
+ assert "bestaudio" in fmts[0], f"first attempt should prefer audio: {fmts}"
140
+
141
+
142
+ def test_download_audio_normalizes_to_small_mono_wav(tmp_path, monkeypatch):
143
+ """A video container from yt-dlp must be downmixed before it reaches Whisper.
144
+
145
+ yt-dlp sometimes has no audio-only rendition available and returns a full
146
+ video; uploading that wasted bandwidth and blew up the GPU (CUDA OOM on a
147
+ 475 MB 1080p file). ffmpeg normalizes to 16 kHz mono first.
148
+ """
149
+ mod = _load_transcribe()
150
+ calls = []
151
+
152
+ class _Result:
153
+ returncode = 0
154
+ stderr = ""
155
+
156
+ def fake_run(cmd, **kwargs):
157
+ calls.append(cmd)
158
+ # Materialize the normalized output so the helper sees a real file.
159
+ if cmd and cmd[0] == "ffmpeg":
160
+ open(cmd[-1], "wb").write(b"RIFF")
161
+ return _Result()
162
+
163
+ monkeypatch.setattr(mod.subprocess, "run", fake_run)
164
+ (tmp_path / "vidABC.mp4").write_bytes(b"\x00" * 2048)
165
+
166
+ out = mod.download_audio("vidABC", str(tmp_path))
167
+
168
+ ffmpeg_calls = [c for c in calls if c and c[0] == "ffmpeg"]
169
+ assert ffmpeg_calls, "no ffmpeg normalization step ran"
170
+ ff = ffmpeg_calls[0]
171
+ assert "-vn" in ff, "video stream must be dropped"
172
+ assert "16000" in ff, "must resample to 16 kHz"
173
+ assert "1" in ff, "must downmix to mono"
174
+ assert out is not None and out.endswith(".16k.wav"), out
175
+
176
+
177
+ def test_pipeline_enables_pipefail():
178
+ """`set -e` alone cannot see a failure on the left of a pipe into `tee`."""
179
+ text = PIPELINE.read_text(encoding="utf-8")
180
+ assert "set -e" in text
181
+ assert "set -o pipefail" in text or "set -euo pipefail" in text, (
182
+ "pipeline lacks pipefail: a broken stage would still report success"
183
+ )
184
+
185
+
186
+ def test_pipeline_treats_transcription_outage_as_nonfatal_but_loud():
187
+ """Whisper is external, so its outage must warn, not abort all stages."""
188
+ text = PIPELINE.read_text(encoding="utf-8")
189
+ marker = "transcribe_moldovan_videos.py"
190
+ assert marker in text
191
+ # The transcribe stage must be wrapped so its failure is handled explicitly.
192
+ assert "WARNING: transcription stage failed" in text, (
193
+ "transcription failure is not surfaced as a warning"
194
+ )
195
+
196
+
197
+ # ─── Whisper endpoint wiring (found broken in audit 3) ──────────────────────
198
+ # The configured endpoint pointed at http://homelab:8086/v1/transcribe, a port
199
+ # nothing listens on, and the multipart part was named "audio" while the server
200
+ # requires "file". Both made every transcription attempt fail. The correct route
201
+ # is the homelab Caddy proxy: /api/v1/whisper/v1/audio/transcriptions.
202
+
203
+ def test_whisper_url_points_at_a_real_route():
204
+ """The default endpoint must be the verified Caddy-proxied route."""
205
+ mod = _load_transcribe()
206
+ url = mod.WHISPER_URL
207
+
208
+ assert ":8086" not in url, (
209
+ "stale endpoint: nothing listens on homelab:8086 (verified curl 000)"
210
+ )
211
+ assert "/v1/transcribe" not in url, (
212
+ "stale route: faster-whisper serves /v1/audio/transcriptions"
213
+ )
214
+ assert url.endswith("/v1/audio/transcriptions"), url
215
+
216
+
217
+ def test_whisper_multipart_field_is_named_file():
218
+ """faster-whisper requires the file part to be named 'file'."""
219
+ src = TRANSCRIBE.read_text(encoding="utf-8")
220
+ assert 'name="file"' in src, (
221
+ "multipart part must be named 'file'; 'audio' is rejected by the server"
222
+ )
223
+ assert 'name="audio"' not in src, "stale multipart field name reintroduced"
224
+
225
+
226
+ def test_whisper_url_is_overridable_by_env(monkeypatch):
227
+ """Deployments must be able to point at another Whisper instance."""
228
+ monkeypatch.setenv("WHISPER_URL", "http://example.test/v1/audio/transcriptions")
229
+ mod = _load_transcribe()
230
+ assert mod.WHISPER_URL == "http://example.test/v1/audio/transcriptions"
231
+
232
+
233
+ def test_pipeline_has_a_freshness_health_check():
234
+ """Success must depend on artifacts actually advancing.
235
+
236
+ Transcripts had been empty since mid-August while the weekly job still
237
+ printed a success banner, so staleness went unnoticed for weeks. The
238
+ pipeline must inspect artifact age and refuse to claim success when material
239
+ is stale or missing.
240
+ """
241
+ text = PIPELINE.read_text(encoding="utf-8")
242
+
243
+ assert "[health]" in text, "no freshness health section"
244
+ assert "STALE" in text, "staleness is never reported"
245
+ assert "PIPELINE COMPLETED WITH STALE/MISSING ARTIFACTS" in text, (
246
+ "pipeline can still claim success with stale artifacts"
247
+ )
248
+ # Every key artifact the farm consumes must be covered.
249
+ for artifact in ("youtube_corpus", "transcripts", "generated_digests"):
250
+ assert artifact in text, f"{artifact} not covered by the health check"
251
+
252
+
253
+ def test_pipeline_fails_when_health_check_fails():
254
+ """The staleness verdict must change the exit code, not just the message."""
255
+ text = PIPELINE.read_text(encoding="utf-8")
256
+ assert "HEALTH_FAIL" in text
257
+ assert "exit 1" in text, "no non-zero exit for a stale/missing artifact"