1use std::collections::HashSet;
14use std::path::{Path, PathBuf};
15
16use anyhow::{Context, Result};
17use chrono::{DateTime, Duration, Months, NaiveDate, Utc};
18use clap::Parser;
19use futures::stream::{self, StreamExt};
20
21use crate::cli::transcript::format::CliFormat;
22use crate::transcript::error::TranscriptError;
23use crate::transcript::format::Format;
24use crate::transcript::source::{FetchOpts, TranscriptSource};
25use crate::transcript::sources::youtube::{metadata, Youtube};
26
27#[derive(Parser)]
30pub struct SyncCommand {
31 #[arg(required = true, num_args = 1..)]
35 pub channels: Vec<String>,
36
37 #[arg(long)]
40 pub out: PathBuf,
41
42 #[arg(long, default_value = "en")]
45 pub lang: String,
46
47 #[arg(long, value_enum, default_value_t = CliFormat::Srt)]
49 pub format: CliFormat,
50
51 #[arg(long)]
55 pub auto: bool,
56
57 #[arg(long)]
61 pub full: bool,
62
63 #[arg(long, value_name = "DATE")]
68 pub since: Option<String>,
69
70 #[arg(long, default_value_t = 4)]
72 pub concurrency: usize,
73
74 #[arg(long, value_name = "DATE_SPEC")]
80 pub refresh_metadata_older_than: Option<String>,
81
82 #[arg(long)]
84 pub dry_run: bool,
85}
86
87impl SyncCommand {
88 pub async fn execute(self) -> Result<()> {
90 let yt = Youtube::new()?;
91 self.sync_with(&yt).await
92 }
93
94 async fn sync_with(self, yt: &Youtube) -> Result<()> {
98 let plan = self.into_plan()?;
99 let report = run(&plan, yt).await;
100 report.print();
101 Ok(())
102 }
103
104 fn into_plan(self) -> Result<SyncPlan> {
106 let since = self
107 .since
108 .as_deref()
109 .map(parse_since)
110 .transpose()
111 .context("Invalid --since date")?;
112 let refresh_cutoff = self
113 .refresh_metadata_older_than
114 .as_deref()
115 .map(|s| parse_date_spec(s, Utc::now()))
116 .transpose()
117 .context("Invalid --refresh-metadata-older-than")?;
118 Ok(SyncPlan {
119 channels: self.channels,
120 out: self.out,
121 lang: self.lang,
122 format: self.format,
123 allow_auto: self.auto,
124 full: self.full,
125 since,
126 refresh_cutoff,
127 concurrency: self.concurrency.max(1),
128 dry_run: self.dry_run,
129 })
130 }
131}
132
133fn parse_since(s: &str) -> Result<DateTime<Utc>> {
136 if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
137 return Ok(dt.with_timezone(&Utc));
138 }
139 let date = NaiveDate::parse_from_str(s, "%Y-%m-%d")
140 .with_context(|| format!("expected YYYY-MM-DD or RFC 3339, got `{s}`"))?;
141 #[allow(clippy::expect_used)]
142 let naive = date
143 .and_hms_opt(0, 0, 0)
144 .expect("midnight is always a valid time");
145 Ok(DateTime::from_naive_utc_and_offset(naive, Utc))
146}
147
148fn parse_date_spec(s: &str, now: DateTime<Utc>) -> Result<DateTime<Utc>> {
157 let s = s.trim();
158 if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
159 return Ok(dt.with_timezone(&Utc));
160 }
161 if let Ok(date) = NaiveDate::parse_from_str(s, "%Y-%m-%d") {
162 #[allow(clippy::expect_used)]
163 let naive = date
164 .and_hms_opt(0, 0, 0)
165 .expect("midnight is always a valid time");
166 return Ok(DateTime::from_naive_utc_and_offset(naive, Utc));
167 }
168 parse_relative(s, now)
169 .with_context(|| format!("expected YYYY-MM-DD, RFC 3339, or `<N> <unit> ago`, got `{s}`"))
170}
171
172fn parse_relative(s: &str, now: DateTime<Utc>) -> Result<DateTime<Utc>> {
176 let lower = s.to_lowercase();
177 let parts: Vec<&str> = lower.split_whitespace().collect();
178 if parts.len() != 3 || parts[2] != "ago" {
179 anyhow::bail!("not a relative date spec");
180 }
181 let n: i64 = parts[0]
182 .parse()
183 .with_context(|| format!("relative count `{}` is not an integer", parts[0]))?;
184 if n < 0 {
185 anyhow::bail!("relative count must be non-negative");
186 }
187 let unit = parts[1].strip_suffix('s').unwrap_or(parts[1]);
189 let cutoff = match unit {
190 "minute" => now - Duration::minutes(n),
191 "hour" => now - Duration::hours(n),
192 "day" => now - Duration::days(n),
193 "week" => now - Duration::weeks(n),
194 "month" => now
195 .checked_sub_months(Months::new(n as u32))
196 .context("date underflow")?,
197 "year" => now
198 .checked_sub_months(Months::new(n as u32 * 12))
199 .context("date underflow")?,
200 other => anyhow::bail!("unknown relative unit `{other}`"),
201 };
202 Ok(cutoff)
203}
204
205struct SyncPlan {
208 channels: Vec<String>,
209 out: PathBuf,
210 lang: String,
211 format: CliFormat,
212 allow_auto: bool,
213 full: bool,
214 since: Option<DateTime<Utc>>,
215 refresh_cutoff: Option<DateTime<Utc>>,
219 concurrency: usize,
220 dry_run: bool,
221}
222
223#[derive(Debug, Default, PartialEq, Eq)]
225struct SyncReport {
226 synced: usize,
228 already_present: usize,
230 no_transcript: usize,
233 failed: usize,
235 would_fetch: usize,
237 metadata_synced: usize,
240 metadata_refreshed: usize,
243 metadata_failed: usize,
246 metadata_would_fetch: usize,
248 channel_errors: usize,
250}
251
252impl SyncReport {
253 fn print(&self) {
254 println!("\nSync complete:");
255 if self.would_fetch > 0 {
256 println!(" would fetch: {}", self.would_fetch);
257 }
258 if self.metadata_would_fetch > 0 {
259 println!(" would fetch meta: {}", self.metadata_would_fetch);
260 }
261 println!(" synced: {}", self.synced);
262 println!(" already present: {}", self.already_present);
263 println!(" no transcript: {}", self.no_transcript);
264 println!(" failed: {}", self.failed);
265 println!(" metadata synced: {}", self.metadata_synced);
266 println!(" metadata refreshed: {}", self.metadata_refreshed);
267 println!(" metadata failed: {}", self.metadata_failed);
268 if self.channel_errors > 0 {
269 println!(" channel errors: {}", self.channel_errors);
270 }
271 }
272}
273
274async fn run(plan: &SyncPlan, yt: &Youtube) -> SyncReport {
278 let mut report = SyncReport::default();
279 for channel in &plan.channels {
280 if let Err(e) = sync_channel(plan, yt, channel, &mut report).await {
281 eprintln!("channel `{channel}`: {e}");
282 report.channel_errors += 1;
283 }
284 }
285 report
286}
287
288async fn sync_channel(
292 plan: &SyncPlan,
293 yt: &Youtube,
294 channel: &str,
295 report: &mut SyncReport,
296) -> Result<(), TranscriptError> {
297 let channel_id = yt.resolve_channel_id(channel).await?;
298 let dir = plan.out.join(&channel_id);
299 if !plan.dry_run {
300 std::fs::create_dir_all(&dir)?;
301 }
302
303 let ids = enumerate(plan, yt, &channel_id).await?;
304 let ext = Format::from(plan.format).as_str();
305 let plan_result = plan_fetches(&ids, &dir, &plan.lang, ext, plan.full);
306 report.already_present += plan_result.already_present;
307
308 println!(
309 "channel {channel_id}: {} video(s), {} to fetch, {} already present",
310 ids.len(),
311 plan_result.to_fetch.len(),
312 plan_result.already_present,
313 );
314
315 if plan.dry_run {
316 for (id, path) in &plan_result.to_fetch {
317 println!(" would fetch {id} -> {}", path.display());
318 }
319 report.would_fetch += plan_result.to_fetch.len();
320
321 let mut candidate_ids = scan_synced_ids(&dir);
324 for (id, _) in &plan_result.to_fetch {
325 if !candidate_ids.contains(id) {
326 candidate_ids.push(id.clone());
327 }
328 }
329 let meta_items = plan_metadata(&dir, &candidate_ids, plan.refresh_cutoff);
330 for item in &meta_items {
331 println!(" would fetch meta {} -> {}", item.id, item.path.display());
332 }
333 report.metadata_would_fetch += meta_items.len();
334 return Ok(());
335 }
336
337 let opts = FetchOpts {
338 language: plan.lang.clone(),
339 allow_auto: plan.allow_auto,
340 translate_to: None,
341 };
342
343 let outcomes = stream::iter(plan_result.to_fetch)
344 .map(|(id, path)| {
345 let opts = &opts;
346 async move {
347 let outcome = fetch_and_write(yt, &id, opts, plan.format, &path).await;
348 (id, outcome)
349 }
350 })
351 .buffer_unordered(plan.concurrency)
352 .collect::<Vec<_>>()
353 .await;
354
355 for (id, outcome) in outcomes {
356 match outcome {
357 Ok(()) => report.synced += 1,
358 Err(e) if is_no_transcript(&e) => {
359 report.no_transcript += 1;
360 eprintln!(" skip {id}: {e}");
361 }
362 Err(e) => {
363 report.failed += 1;
364 eprintln!(" fail {id}: {e}");
365 }
366 }
367 }
368
369 sync_metadata(plan, yt, &dir, report).await;
370 Ok(())
371}
372
373async fn sync_metadata(plan: &SyncPlan, yt: &Youtube, dir: &Path, report: &mut SyncReport) {
383 let candidate_ids = scan_synced_ids(dir);
384 let meta_items = plan_metadata(dir, &candidate_ids, plan.refresh_cutoff);
385
386 let outcomes = stream::iter(meta_items)
387 .map(|item| async move {
388 let outcome = fetch_and_write_metadata(yt, &item.id, &item.path).await;
389 (item.reason, item.id, outcome)
390 })
391 .buffer_unordered(plan.concurrency)
392 .collect::<Vec<_>>()
393 .await;
394
395 for (reason, id, outcome) in outcomes {
396 match outcome {
397 Ok(()) => match reason {
398 MetaReason::Missing => report.metadata_synced += 1,
399 MetaReason::Stale => report.metadata_refreshed += 1,
400 },
401 Err(e) => {
402 report.metadata_failed += 1;
403 eprintln!(" meta fail {id}: {e}");
404 }
405 }
406 }
407}
408
409async fn enumerate(
412 plan: &SyncPlan,
413 yt: &Youtube,
414 channel_id: &str,
415) -> Result<Vec<String>, TranscriptError> {
416 if plan.full {
417 yt.all_channel_video_ids(channel_id).await
418 } else {
419 let entries = yt.recent_channel_videos(channel_id).await?;
420 Ok(entries
421 .into_iter()
422 .filter(|e| match (plan.since, e.published) {
423 (Some(since), Some(published)) => published >= since,
424 _ => true,
426 })
427 .map(|e| e.id)
428 .collect())
429 }
430}
431
432struct FetchPlan {
434 to_fetch: Vec<(String, PathBuf)>,
435 already_present: usize,
436}
437
438fn plan_fetches(ids: &[String], dir: &Path, lang: &str, ext: &str, full: bool) -> FetchPlan {
444 let mut to_fetch = Vec::new();
445 let mut already_present = 0;
446 for id in ids {
447 let path = dir.join(format!("{id}.{lang}.{ext}"));
448 if path.exists() {
449 already_present += 1;
450 if !full {
451 break;
452 }
453 } else {
454 to_fetch.push((id.clone(), path));
455 }
456 }
457 FetchPlan {
458 to_fetch,
459 already_present,
460 }
461}
462
463async fn fetch_and_write(
466 yt: &Youtube,
467 id: &str,
468 opts: &FetchOpts,
469 format: CliFormat,
470 path: &Path,
471) -> Result<(), TranscriptError> {
472 let transcript = yt.fetch(id, opts).await?;
473 let rendered = Format::from(format).render(&transcript)?;
474 write_atomic(path, &rendered)?;
475 Ok(())
476}
477
478fn write_atomic(path: &Path, contents: &str) -> std::io::Result<()> {
481 let file_name = path.file_name().and_then(|n| n.to_str()).unwrap_or("out");
482 let tmp = path.with_file_name(format!(".{file_name}.tmp"));
483 std::fs::write(&tmp, contents)?;
484 std::fs::rename(&tmp, path)
485}
486
487#[derive(Clone, Copy, Debug, PartialEq, Eq)]
490enum MetaReason {
491 Missing,
493 Stale,
496}
497
498struct MetaItem {
500 id: String,
501 path: PathBuf,
502 reason: MetaReason,
503}
504
505fn scan_synced_ids(dir: &Path) -> Vec<String> {
511 let mut ids = Vec::new();
512 let mut seen = HashSet::new();
513 let Ok(entries) = std::fs::read_dir(dir) else {
514 return ids;
515 };
516 for entry in entries.flatten() {
517 if !entry.file_type().is_ok_and(|t| t.is_file()) {
518 continue;
519 }
520 let name = entry.file_name();
521 let Some(name) = name.to_str() else { continue };
522 if name.starts_with('.') || name.ends_with(".tmp") || name.ends_with(".meta.yaml") {
523 continue;
524 }
525 let Some(id) = name.split('.').next().filter(|s| !s.is_empty()) else {
526 continue;
527 };
528 if seen.insert(id.to_string()) {
529 ids.push(id.to_string());
530 }
531 }
532 ids
533}
534
535fn plan_metadata(
541 dir: &Path,
542 candidate_ids: &[String],
543 refresh_cutoff: Option<DateTime<Utc>>,
544) -> Vec<MetaItem> {
545 let mut items = Vec::new();
546 for id in candidate_ids {
547 let path = dir.join(format!("{id}.meta.yaml"));
548 let reason = match std::fs::read_to_string(&path) {
549 Err(_) => Some(MetaReason::Missing),
551 Ok(contents) => match metadata::read_fetched_at(&contents) {
552 None => Some(MetaReason::Missing),
554 Some(fetched_at) => match refresh_cutoff {
555 Some(cutoff) if fetched_at < cutoff => Some(MetaReason::Stale),
556 _ => None,
557 },
558 },
559 };
560 if let Some(reason) = reason {
561 items.push(MetaItem {
562 id: id.clone(),
563 path,
564 reason,
565 });
566 }
567 }
568 items
569}
570
571async fn fetch_and_write_metadata(
575 yt: &Youtube,
576 id: &str,
577 path: &Path,
578) -> Result<(), TranscriptError> {
579 let meta = yt.fetch_video_metadata(id).await?;
580 let yaml = serde_yaml::to_string(&meta)
581 .map_err(|e| TranscriptError::ParseError(format!("serialise metadata sidecar: {e}")))?;
582 write_atomic(path, &yaml)?;
583 Ok(())
584}
585
586fn is_no_transcript(err: &TranscriptError) -> bool {
589 matches!(
590 err,
591 TranscriptError::PlayabilityRefused { .. }
592 | TranscriptError::LanguageNotFound { .. }
593 | TranscriptError::AutoCaptionsRequireOptIn(_)
594 )
595}
596
597#[cfg(test)]
598#[allow(clippy::unwrap_used, clippy::expect_used)]
599mod tests {
600 use super::*;
601 use clap::{CommandFactory, FromArgMatches};
602
603 fn parse(args: &[&str]) -> SyncCommand {
604 let cmd = SyncCommand::command().no_binary_name(true);
605 let matches = cmd.try_get_matches_from(args).unwrap();
606 SyncCommand::from_arg_matches(&matches).unwrap()
607 }
608
609 #[test]
610 fn sync_command_defaults() {
611 let cmd = parse(&["@handle", "--out", "/tmp/yt"]);
612 assert_eq!(cmd.channels, vec!["@handle".to_string()]);
613 assert_eq!(cmd.out, PathBuf::from("/tmp/yt"));
614 assert_eq!(cmd.lang, "en");
615 assert_eq!(cmd.format, CliFormat::Srt);
616 assert!(!cmd.auto);
617 assert!(!cmd.full);
618 assert_eq!(cmd.since, None);
619 assert_eq!(cmd.refresh_metadata_older_than, None);
620 assert_eq!(cmd.concurrency, 4);
621 assert!(!cmd.dry_run);
622 }
623
624 #[test]
625 fn sync_command_all_flags_and_multiple_channels() {
626 let cmd = parse(&[
627 "UC_x5XG1OV2P6uZZ5FSM9Ttw",
628 "@another",
629 "--out",
630 "/tmp/out",
631 "--lang",
632 "fr",
633 "--format",
634 "vtt",
635 "--auto",
636 "--full",
637 "--since",
638 "2024-01-01",
639 "--refresh-metadata-older-than",
640 "2 days ago",
641 "--concurrency",
642 "8",
643 "--dry-run",
644 ]);
645 assert_eq!(cmd.channels.len(), 2);
646 assert_eq!(cmd.lang, "fr");
647 assert_eq!(cmd.format, CliFormat::Vtt);
648 assert!(cmd.auto);
649 assert!(cmd.full);
650 assert_eq!(cmd.since.as_deref(), Some("2024-01-01"));
651 assert_eq!(
652 cmd.refresh_metadata_older_than.as_deref(),
653 Some("2 days ago")
654 );
655 assert_eq!(cmd.concurrency, 8);
656 assert!(cmd.dry_run);
657 }
658
659 #[test]
660 fn sync_command_requires_a_channel() {
661 let cmd = SyncCommand::command().no_binary_name(true);
662 assert!(cmd.try_get_matches_from(["--out", "/tmp"]).is_err());
663 }
664
665 #[test]
666 fn parse_since_accepts_date_and_rfc3339() {
667 let date = parse_since("2024-11-20").unwrap();
668 assert_eq!(date.to_rfc3339(), "2024-11-20T00:00:00+00:00");
669 let ts = parse_since("2024-11-20T12:30:00+00:00").unwrap();
670 assert_eq!(ts.to_rfc3339(), "2024-11-20T12:30:00+00:00");
671 }
672
673 #[test]
674 fn parse_since_rejects_garbage() {
675 assert!(parse_since("not-a-date").is_err());
676 }
677
678 fn now_fixed() -> DateTime<Utc> {
679 "2026-06-21T12:00:00Z".parse().unwrap()
680 }
681
682 #[test]
683 fn parse_date_spec_accepts_absolute_forms() {
684 let now = now_fixed();
685 assert_eq!(
686 parse_date_spec("2026-06-01", now).unwrap().to_rfc3339(),
687 "2026-06-01T00:00:00+00:00"
688 );
689 assert_eq!(
690 parse_date_spec("2026-06-01T12:00:00+10:00", now)
691 .unwrap()
692 .to_rfc3339(),
693 "2026-06-01T02:00:00+00:00"
694 );
695 }
696
697 #[test]
698 fn parse_date_spec_resolves_relative_units_against_now() {
699 let now = now_fixed();
700 assert_eq!(
701 parse_date_spec("30 minutes ago", now).unwrap().to_rfc3339(),
702 "2026-06-21T11:30:00+00:00"
703 );
704 assert_eq!(
705 parse_date_spec("2 hours ago", now).unwrap().to_rfc3339(),
706 "2026-06-21T10:00:00+00:00"
707 );
708 assert_eq!(
709 parse_date_spec("2 days ago", now).unwrap().to_rfc3339(),
710 "2026-06-19T12:00:00+00:00"
711 );
712 assert_eq!(
713 parse_date_spec("1 week ago", now).unwrap().to_rfc3339(),
714 "2026-06-14T12:00:00+00:00"
715 );
716 assert_eq!(
718 parse_date_spec("3 months ago", now).unwrap().to_rfc3339(),
719 "2026-03-21T12:00:00+00:00"
720 );
721 assert_eq!(
722 parse_date_spec("1 year ago", now).unwrap().to_rfc3339(),
723 "2025-06-21T12:00:00+00:00"
724 );
725 }
726
727 #[test]
728 fn parse_date_spec_accepts_singular_and_is_case_insensitive() {
729 let now = now_fixed();
730 assert_eq!(
731 parse_date_spec("1 Day Ago", now).unwrap().to_rfc3339(),
732 "2026-06-20T12:00:00+00:00"
733 );
734 }
735
736 #[test]
737 fn parse_date_spec_rejects_garbage() {
738 let now = now_fixed();
739 assert!(parse_date_spec("not-a-date", now).is_err());
740 assert!(parse_date_spec("2 fortnights ago", now).is_err());
741 assert!(parse_date_spec("yesterday", now).is_err());
742 assert!(parse_date_spec("-1 days ago", now).is_err());
743 assert!(parse_date_spec("2 days", now).is_err());
744 }
745
746 #[test]
747 fn into_plan_clamps_zero_concurrency_to_one() {
748 let cmd = parse(&["@h", "--out", "/tmp", "--concurrency", "0"]);
749 let plan = cmd.into_plan().unwrap();
750 assert_eq!(plan.concurrency, 1);
751 }
752
753 #[test]
754 fn plan_fetches_incremental_stops_at_first_present() {
755 let dir = tempfile::tempdir().unwrap();
756 let ids: Vec<String> = vec!["aaa".into(), "bbb".into(), "ccc".into()];
758 std::fs::write(dir.path().join("bbb.en.srt"), "x").unwrap();
759
760 let plan = plan_fetches(ids.as_slice(), dir.path(), "en", "srt", false);
761 assert_eq!(plan.already_present, 1);
763 assert_eq!(
764 plan.to_fetch
765 .iter()
766 .map(|(id, _)| id.clone())
767 .collect::<Vec<_>>(),
768 vec!["aaa".to_string()]
769 );
770 }
771
772 #[test]
773 fn plan_fetches_full_fills_gaps() {
774 let dir = tempfile::tempdir().unwrap();
775 let ids: Vec<String> = vec!["aaa".into(), "bbb".into(), "ccc".into()];
776 std::fs::write(dir.path().join("bbb.en.srt"), "x").unwrap();
777
778 let plan = plan_fetches(ids.as_slice(), dir.path(), "en", "srt", true);
779 assert_eq!(plan.already_present, 1);
781 assert_eq!(
782 plan.to_fetch
783 .iter()
784 .map(|(id, _)| id.clone())
785 .collect::<Vec<_>>(),
786 vec!["aaa".to_string(), "ccc".to_string()]
787 );
788 }
789
790 #[test]
791 fn plan_fetches_uses_lang_and_ext_in_path() {
792 let dir = tempfile::tempdir().unwrap();
793 let ids = vec!["zzz".to_string()];
794 let plan = plan_fetches(&ids, dir.path(), "fr", "vtt", false);
795 let (_, path) = &plan.to_fetch[0];
796 assert!(path.ends_with("zzz.fr.vtt"));
797 }
798
799 #[test]
800 fn write_atomic_writes_and_leaves_no_temp() {
801 let dir = tempfile::tempdir().unwrap();
802 let path = dir.path().join("vid.en.srt");
803 write_atomic(&path, "hello").unwrap();
804 assert_eq!(std::fs::read_to_string(&path).unwrap(), "hello");
805 assert!(!dir.path().join(".vid.en.srt.tmp").exists());
806 }
807
808 #[test]
809 fn is_no_transcript_classifies_skip_conditions() {
810 assert!(is_no_transcript(&TranscriptError::PlayabilityRefused {
811 status: "LOGIN_REQUIRED".into(),
812 reason: None,
813 }));
814 assert!(is_no_transcript(&TranscriptError::LanguageNotFound {
815 requested: "en".into(),
816 available: vec![],
817 }));
818 assert!(is_no_transcript(
819 &TranscriptError::AutoCaptionsRequireOptIn("en".into())
820 ));
821 assert!(!is_no_transcript(&TranscriptError::ParseError("x".into())));
822 }
823
824 use serde_json::Value;
827 use wiremock::matchers::{body_partial_json, method, path};
828 use wiremock::{Mock, MockServer, ResponseTemplate};
829
830 const CHANNEL_ID: &str = "UC_x5XG1OV2P6uZZ5FSM9Ttw";
831 const RSS_FEED: &str =
832 include_str!("../../../transcript/sources/youtube/fixtures/channel_rss.xml");
833 const PLAYER_RESPONSE: &str =
834 include_str!("../../../transcript/sources/youtube/fixtures/player_response_basic.json");
835 const PLAYER_RESPONSE_AGE_GATED: &str =
836 include_str!("../../../transcript/sources/youtube/fixtures/player_response_age_gated.json");
837 const TIMEDTEXT: &str =
838 include_str!("../../../transcript/sources/youtube/fixtures/timedtext_basic.json");
839 const WATCH_PAGE: &str = include_str!(
840 "../../../transcript/sources/youtube/fixtures/watch_page_with_visitor_data.html"
841 );
842 const BROWSE_PAGE1: &str =
843 include_str!("../../../transcript/sources/youtube/fixtures/browse_videos_page1.json");
844 const BROWSE_PAGE2: &str =
845 include_str!("../../../transcript/sources/youtube/fixtures/browse_videos_page2.json");
846 const WEB_METADATA: &str = include_str!(
847 "../../../transcript/sources/youtube/fixtures/player_response_web_metadata.json"
848 );
849
850 fn player_response_for(mock_uri: &str) -> String {
854 let mut value: Value = serde_json::from_str(PLAYER_RESPONSE).unwrap();
855 let tracks = value["captions"]["playerCaptionsTracklistRenderer"]["captionTracks"]
856 .as_array_mut()
857 .unwrap();
858 for track in tracks {
859 let lang = track["languageCode"].as_str().unwrap().to_string();
860 track["baseUrl"] = Value::String(format!("{mock_uri}/api/timedtext?lang={lang}"));
861 }
862 serde_json::to_string(&value).unwrap()
863 }
864
865 async fn mount_rss(server: &MockServer) {
866 Mock::given(method("GET"))
867 .and(path("/feeds/videos.xml"))
868 .respond_with(ResponseTemplate::new(200).set_body_string(RSS_FEED))
869 .mount(server)
870 .await;
871 }
872
873 async fn mount_watch(server: &MockServer) {
874 Mock::given(method("GET"))
875 .and(path("/watch"))
876 .respond_with(ResponseTemplate::new(200).set_body_string(WATCH_PAGE))
877 .mount(server)
878 .await;
879 }
880
881 async fn mount_player_body(server: &MockServer, body: String) {
884 Mock::given(method("POST"))
885 .and(path("/youtubei/v1/player"))
886 .respond_with(ResponseTemplate::new(200).set_body_string(body))
887 .mount(server)
888 .await;
889 Mock::given(method("GET"))
890 .and(path("/api/timedtext"))
891 .respond_with(ResponseTemplate::new(200).set_body_string(TIMEDTEXT))
892 .mount(server)
893 .await;
894 }
895
896 async fn mount_browse(server: &MockServer) {
898 Mock::given(method("POST"))
899 .and(path("/youtubei/v1/browse"))
900 .and(body_partial_json(
901 serde_json::json!({ "browseId": CHANNEL_ID }),
902 ))
903 .respond_with(ResponseTemplate::new(200).set_body_string(BROWSE_PAGE1))
904 .mount(server)
905 .await;
906 Mock::given(method("POST"))
907 .and(path("/youtubei/v1/browse"))
908 .and(body_partial_json(
909 serde_json::json!({ "continuation": "CONT_TOKEN_1" }),
910 ))
911 .respond_with(ResponseTemplate::new(200).set_body_string(BROWSE_PAGE2))
912 .mount(server)
913 .await;
914 }
915
916 async fn mock_youtube() -> (MockServer, Youtube) {
918 let server = MockServer::start().await;
919 mount_rss(&server).await;
920 mount_watch(&server).await;
921 mount_player_body(&server, player_response_for(&server.uri())).await;
922 let yt = Youtube::with_base_url(server.uri()).unwrap();
923 (server, yt)
924 }
925
926 async fn mount_player_split(server: &MockServer) {
932 Mock::given(method("POST"))
933 .and(path("/youtubei/v1/player"))
934 .and(body_partial_json(
935 serde_json::json!({ "context": { "client": { "clientName": "ANDROID_VR" } } }),
936 ))
937 .respond_with(
938 ResponseTemplate::new(200).set_body_string(player_response_for(&server.uri())),
939 )
940 .mount(server)
941 .await;
942 Mock::given(method("POST"))
943 .and(path("/youtubei/v1/player"))
944 .and(body_partial_json(
945 serde_json::json!({ "context": { "client": { "clientName": "WEB" } } }),
946 ))
947 .respond_with(ResponseTemplate::new(200).set_body_string(WEB_METADATA))
948 .mount(server)
949 .await;
950 Mock::given(method("GET"))
951 .and(path("/api/timedtext"))
952 .respond_with(ResponseTemplate::new(200).set_body_string(TIMEDTEXT))
953 .mount(server)
954 .await;
955 }
956
957 async fn mock_youtube_split() -> (MockServer, Youtube) {
960 let server = MockServer::start().await;
961 mount_rss(&server).await;
962 mount_watch(&server).await;
963 mount_player_split(&server).await;
964 let yt = Youtube::with_base_url(server.uri()).unwrap();
965 (server, yt)
966 }
967
968 fn channel_dir(root: &Path) -> PathBuf {
970 let dir = root.join(CHANNEL_ID);
971 std::fs::create_dir_all(&dir).unwrap();
972 dir
973 }
974
975 fn plan_for(out: PathBuf) -> SyncPlan {
976 SyncPlan {
977 channels: vec![CHANNEL_ID.to_string()],
978 out,
979 lang: "en".to_string(),
980 format: CliFormat::Srt,
981 allow_auto: false,
982 full: false,
983 since: None,
984 refresh_cutoff: None,
985 concurrency: 2,
986 dry_run: false,
987 }
988 }
989
990 #[tokio::test]
991 async fn run_writes_transcripts_then_skips_on_resync() {
992 let (_server, yt) = mock_youtube().await;
993 let dir = tempfile::tempdir().unwrap();
994 let plan = plan_for(dir.path().to_path_buf());
995
996 let report = run(&plan, &yt).await;
998 assert_eq!(report.synced, 3);
999 assert_eq!(report.already_present, 0);
1000 assert_eq!(report.failed, 0);
1001
1002 let channel_dir = dir.path().join(CHANNEL_ID);
1003 for id in ["aaaaaaaaaaa", "bbbbbbbbbbb", "ccccccccccc"] {
1004 assert!(channel_dir.join(format!("{id}.en.srt")).exists());
1005 }
1006
1007 let report2 = run(&plan, &yt).await;
1009 assert_eq!(report2.synced, 0);
1010 assert_eq!(report2.already_present, 1);
1011 }
1012
1013 #[tokio::test]
1014 async fn run_dry_run_writes_nothing() {
1015 let (_server, yt) = mock_youtube().await;
1016 let dir = tempfile::tempdir().unwrap();
1017 let mut plan = plan_for(dir.path().to_path_buf());
1018 plan.dry_run = true;
1019
1020 let report = run(&plan, &yt).await;
1021 assert_eq!(report.would_fetch, 3);
1022 assert_eq!(report.synced, 0);
1023 assert!(!dir.path().join(CHANNEL_ID).exists());
1025 }
1026
1027 #[tokio::test]
1028 async fn run_full_enumerates_via_browse() {
1029 let server = MockServer::start().await;
1031 mount_watch(&server).await;
1032 mount_browse(&server).await;
1033 mount_player_body(&server, player_response_for(&server.uri())).await;
1034 let yt = Youtube::with_base_url(server.uri()).unwrap();
1035
1036 let dir = tempfile::tempdir().unwrap();
1037 let mut plan = plan_for(dir.path().to_path_buf());
1038 plan.full = true;
1039
1040 let report = run(&plan, &yt).await;
1041 assert_eq!(report.synced, 3);
1042 let channel_dir = dir.path().join(CHANNEL_ID);
1043 for id in ["vid00000001", "vid00000002", "vid00000003"] {
1044 assert!(channel_dir.join(format!("{id}.en.srt")).exists());
1045 }
1046 }
1047
1048 #[tokio::test]
1049 async fn run_since_filters_out_older_entries() {
1050 let (_server, yt) = mock_youtube().await;
1053 let dir = tempfile::tempdir().unwrap();
1054 let mut plan = plan_for(dir.path().to_path_buf());
1055 plan.since = Some(parse_since("2024-11-21").unwrap());
1056
1057 let report = run(&plan, &yt).await;
1058 assert_eq!(report.synced, 1);
1059 assert!(dir
1060 .path()
1061 .join(CHANNEL_ID)
1062 .join("aaaaaaaaaaa.en.srt")
1063 .exists());
1064 }
1065
1066 #[tokio::test]
1067 async fn run_records_no_transcript_for_age_gated() {
1068 let server = MockServer::start().await;
1070 mount_rss(&server).await;
1071 mount_watch(&server).await;
1072 mount_player_body(&server, PLAYER_RESPONSE_AGE_GATED.to_string()).await;
1073 let yt = Youtube::with_base_url(server.uri()).unwrap();
1074
1075 let dir = tempfile::tempdir().unwrap();
1076 let report = run(&plan_for(dir.path().to_path_buf()), &yt).await;
1077 assert_eq!(report.no_transcript, 3);
1078 assert_eq!(report.synced, 0);
1079 assert_eq!(report.failed, 0);
1080 }
1081
1082 #[tokio::test]
1083 async fn run_records_failures_on_http_error() {
1084 let server = MockServer::start().await;
1086 mount_rss(&server).await;
1087 mount_watch(&server).await;
1088 Mock::given(method("POST"))
1089 .and(path("/youtubei/v1/player"))
1090 .respond_with(ResponseTemplate::new(500))
1091 .mount(&server)
1092 .await;
1093 let yt = Youtube::with_base_url(server.uri()).unwrap();
1094
1095 let dir = tempfile::tempdir().unwrap();
1096 let report = run(&plan_for(dir.path().to_path_buf()), &yt).await;
1097 assert_eq!(report.failed, 3);
1098 assert_eq!(report.synced, 0);
1099 }
1100
1101 #[tokio::test]
1102 async fn run_records_channel_error_when_unresolvable() {
1103 let server = MockServer::start().await;
1106 Mock::given(method("GET"))
1107 .and(path("/@ghost"))
1108 .respond_with(ResponseTemplate::new(200).set_body_string("<html>no id</html>"))
1109 .mount(&server)
1110 .await;
1111 let yt = Youtube::with_base_url(server.uri()).unwrap();
1112
1113 let dir = tempfile::tempdir().unwrap();
1114 let mut plan = plan_for(dir.path().to_path_buf());
1115 plan.channels = vec!["@ghost".to_string()];
1116
1117 let report = run(&plan, &yt).await;
1118 assert_eq!(report.channel_errors, 1);
1119 assert_eq!(report.synced, 0);
1120 }
1121
1122 #[tokio::test]
1123 async fn sync_with_drives_full_orchestration() {
1124 let (_server, yt) = mock_youtube().await;
1127 let dir = tempfile::tempdir().unwrap();
1128 let cmd = SyncCommand {
1129 channels: vec![CHANNEL_ID.to_string()],
1130 out: dir.path().to_path_buf(),
1131 lang: "en".to_string(),
1132 format: CliFormat::Srt,
1133 auto: false,
1134 full: false,
1135 since: None,
1136 refresh_metadata_older_than: None,
1137 concurrency: 2,
1138 dry_run: false,
1139 };
1140 cmd.sync_with(&yt).await.unwrap();
1141 assert!(dir
1142 .path()
1143 .join(CHANNEL_ID)
1144 .join("aaaaaaaaaaa.en.srt")
1145 .exists());
1146 }
1147
1148 #[tokio::test]
1149 async fn sync_with_surfaces_invalid_since() {
1150 let (_server, yt) = mock_youtube().await;
1152 let dir = tempfile::tempdir().unwrap();
1153 let cmd = SyncCommand {
1154 channels: vec![CHANNEL_ID.to_string()],
1155 out: dir.path().to_path_buf(),
1156 lang: "en".to_string(),
1157 format: CliFormat::Srt,
1158 auto: false,
1159 full: false,
1160 since: Some("not-a-date".to_string()),
1161 refresh_metadata_older_than: None,
1162 concurrency: 2,
1163 dry_run: false,
1164 };
1165 let err = cmd.sync_with(&yt).await.unwrap_err();
1166 assert!(err.to_string().contains("Invalid --since date"));
1167 }
1168
1169 #[tokio::test]
1172 async fn run_writes_metadata_sidecars_alongside_transcripts() {
1173 let (_server, yt) = mock_youtube_split().await;
1174 let dir = tempfile::tempdir().unwrap();
1175 let report = run(&plan_for(dir.path().to_path_buf()), &yt).await;
1176
1177 assert_eq!(report.synced, 3);
1178 assert_eq!(report.metadata_synced, 3);
1179 assert_eq!(report.metadata_refreshed, 0);
1180 assert_eq!(report.metadata_failed, 0);
1181
1182 let cdir = dir.path().join(CHANNEL_ID);
1183 for id in ["aaaaaaaaaaa", "bbbbbbbbbbb", "ccccccccccc"] {
1184 let sidecar = cdir.join(format!("{id}.meta.yaml"));
1185 assert!(sidecar.exists(), "missing sidecar for {id}");
1186 let body = std::fs::read_to_string(&sidecar).unwrap();
1187 assert!(body.contains("schema: 1"));
1189 assert!(body.contains("category: Music"));
1190 assert!(body.contains("like_count: 19148727"));
1191 assert!(body.contains("fetched_at:"));
1192 }
1193 }
1194
1195 #[tokio::test]
1196 async fn run_backfills_missing_sidecars_without_refetching_transcripts() {
1197 let (_server, yt) = mock_youtube_split().await;
1201 let dir = tempfile::tempdir().unwrap();
1202 let cdir = channel_dir(dir.path());
1203 for id in ["aaaaaaaaaaa", "bbbbbbbbbbb", "ccccccccccc"] {
1204 std::fs::write(cdir.join(format!("{id}.en.srt")), "x").unwrap();
1205 }
1206
1207 let report = run(&plan_for(dir.path().to_path_buf()), &yt).await;
1208
1209 assert_eq!(report.synced, 0, "transcripts already present");
1210 assert_eq!(report.metadata_synced, 3, "all sidecars backfilled");
1211 for id in ["aaaaaaaaaaa", "bbbbbbbbbbb", "ccccccccccc"] {
1212 assert!(cdir.join(format!("{id}.meta.yaml")).exists());
1213 }
1214 }
1215
1216 #[tokio::test]
1217 async fn run_refreshes_stale_sidecar_only_with_flag() {
1218 let (_server, yt) = mock_youtube_split().await;
1219 let dir = tempfile::tempdir().unwrap();
1220 let cdir = channel_dir(dir.path());
1221 std::fs::write(cdir.join("aaaaaaaaaaa.en.srt"), "x").unwrap();
1222 let sidecar = cdir.join("aaaaaaaaaaa.meta.yaml");
1223 let stale = "schema: 1\nvideo_id: aaaaaaaaaaa\ntitle: stale\n\
1224 is_live_content: false\nfetched_at: \"2000-01-01T00:00:00Z\"\n";
1225 std::fs::write(&sidecar, stale).unwrap();
1226
1227 let report = run(&plan_for(dir.path().to_path_buf()), &yt).await;
1229 assert_eq!(report.metadata_refreshed, 0);
1230 assert_eq!(report.metadata_synced, 0);
1231 assert_eq!(std::fs::read_to_string(&sidecar).unwrap(), stale);
1232
1233 let mut plan = plan_for(dir.path().to_path_buf());
1235 plan.refresh_cutoff = Some("2026-01-01T00:00:00Z".parse().unwrap());
1236 let report = run(&plan, &yt).await;
1237 assert_eq!(report.metadata_refreshed, 1);
1238 assert_eq!(report.metadata_synced, 0);
1239 let refreshed = std::fs::read_to_string(&sidecar).unwrap();
1240 assert_ne!(refreshed, stale);
1241 assert!(refreshed.contains("category: Music"));
1242 }
1243
1244 #[tokio::test]
1245 async fn run_treats_corrupt_sidecar_as_missing() {
1246 let (_server, yt) = mock_youtube_split().await;
1247 let dir = tempfile::tempdir().unwrap();
1248 let cdir = channel_dir(dir.path());
1249 std::fs::write(cdir.join("aaaaaaaaaaa.en.srt"), "x").unwrap();
1250 let sidecar = cdir.join("aaaaaaaaaaa.meta.yaml");
1251 std::fs::write(&sidecar, "schema: 1\nvideo_id: aaaaaaaaaaa\n").unwrap();
1253
1254 let report = run(&plan_for(dir.path().to_path_buf()), &yt).await;
1255
1256 assert_eq!(report.metadata_synced, 1);
1258 assert_eq!(report.metadata_refreshed, 0);
1259 assert!(std::fs::read_to_string(&sidecar)
1260 .unwrap()
1261 .contains("category: Music"));
1262 }
1263
1264 #[tokio::test]
1265 async fn run_metadata_failure_does_not_block_transcripts() {
1266 let server = MockServer::start().await;
1268 mount_rss(&server).await;
1269 mount_watch(&server).await;
1270 Mock::given(method("POST"))
1271 .and(path("/youtubei/v1/player"))
1272 .and(body_partial_json(
1273 serde_json::json!({ "context": { "client": { "clientName": "ANDROID_VR" } } }),
1274 ))
1275 .respond_with(
1276 ResponseTemplate::new(200).set_body_string(player_response_for(&server.uri())),
1277 )
1278 .mount(&server)
1279 .await;
1280 Mock::given(method("POST"))
1281 .and(path("/youtubei/v1/player"))
1282 .and(body_partial_json(
1283 serde_json::json!({ "context": { "client": { "clientName": "WEB" } } }),
1284 ))
1285 .respond_with(ResponseTemplate::new(500))
1286 .mount(&server)
1287 .await;
1288 Mock::given(method("GET"))
1289 .and(path("/api/timedtext"))
1290 .respond_with(ResponseTemplate::new(200).set_body_string(TIMEDTEXT))
1291 .mount(&server)
1292 .await;
1293 let yt = Youtube::with_base_url(server.uri()).unwrap();
1294
1295 let dir = tempfile::tempdir().unwrap();
1296 let report = run(&plan_for(dir.path().to_path_buf()), &yt).await;
1297
1298 assert_eq!(report.synced, 3, "transcripts unaffected");
1299 assert_eq!(report.failed, 0);
1300 assert_eq!(report.metadata_failed, 3);
1301 assert_eq!(report.metadata_synced, 0);
1302
1303 let cdir = dir.path().join(CHANNEL_ID);
1304 assert!(cdir.join("aaaaaaaaaaa.en.srt").exists());
1305 assert!(!cdir.join("aaaaaaaaaaa.meta.yaml").exists());
1306 }
1307
1308 #[tokio::test]
1309 async fn run_dry_run_counts_sidecars_but_writes_nothing() {
1310 let (_server, yt) = mock_youtube_split().await;
1311 let dir = tempfile::tempdir().unwrap();
1312 let mut plan = plan_for(dir.path().to_path_buf());
1313 plan.dry_run = true;
1314
1315 let report = run(&plan, &yt).await;
1316 assert_eq!(report.would_fetch, 3);
1317 assert_eq!(report.metadata_would_fetch, 3);
1318 assert_eq!(report.metadata_synced, 0);
1319 assert!(!dir.path().join(CHANNEL_ID).exists());
1320 }
1321
1322 #[test]
1323 fn scan_synced_ids_ignores_sidecars_and_temp_files() {
1324 let dir = tempfile::tempdir().unwrap();
1325 let p = dir.path();
1326 std::fs::write(p.join("aaaaaaaaaaa.en.srt"), "x").unwrap();
1327 std::fs::write(p.join("aaaaaaaaaaa.fr.vtt"), "x").unwrap(); std::fs::write(p.join("bbbbbbbbbbb.en.srt"), "x").unwrap();
1329 std::fs::write(p.join("bbbbbbbbbbb.meta.yaml"), "x").unwrap(); std::fs::write(p.join(".ccccccccccc.en.srt.tmp"), "x").unwrap(); let mut ids = scan_synced_ids(p);
1333 ids.sort();
1334 assert_eq!(
1335 ids,
1336 vec!["aaaaaaaaaaa".to_string(), "bbbbbbbbbbb".to_string()]
1337 );
1338 }
1339
1340 #[test]
1341 fn plan_metadata_classifies_missing_stale_and_fresh() {
1342 let dir = tempfile::tempdir().unwrap();
1343 let p = dir.path();
1344 std::fs::write(
1346 p.join("stale.meta.yaml"),
1347 "fetched_at: \"2000-01-01T00:00:00Z\"\n",
1348 )
1349 .unwrap();
1350 std::fs::write(
1351 p.join("fresh.meta.yaml"),
1352 "fetched_at: \"2026-12-31T00:00:00Z\"\n",
1353 )
1354 .unwrap();
1355 let cutoff: DateTime<Utc> = "2026-01-01T00:00:00Z".parse().unwrap();
1356 let ids = vec!["miss".to_string(), "stale".to_string(), "fresh".to_string()];
1357
1358 let items = plan_metadata(p, &ids, Some(cutoff));
1359 let by_id: std::collections::HashMap<_, _> =
1360 items.iter().map(|i| (i.id.as_str(), i.reason)).collect();
1361 assert_eq!(by_id.get("miss"), Some(&MetaReason::Missing));
1362 assert_eq!(by_id.get("stale"), Some(&MetaReason::Stale));
1363 assert_eq!(by_id.get("fresh"), None, "fresh sidecar is left alone");
1364
1365 let items = plan_metadata(p, &ids, None);
1367 assert_eq!(items.len(), 1);
1368 assert_eq!(items[0].id, "miss");
1369 }
1370
1371 #[test]
1372 fn report_print_covers_optional_branches() {
1373 let report = SyncReport {
1375 synced: 1,
1376 already_present: 2,
1377 no_transcript: 3,
1378 failed: 4,
1379 would_fetch: 5,
1380 metadata_synced: 7,
1381 metadata_refreshed: 8,
1382 metadata_failed: 9,
1383 metadata_would_fetch: 10,
1384 channel_errors: 6,
1385 };
1386 report.print();
1387 SyncReport::default().print();
1388 }
1389}