From 943191a0ff19799aac269db306c4a5b14ac25a79 Mon Sep 17 00:00:00 2001 From: Aleksandr Bogomiakov Date: Fri, 14 Aug 2026 01:21:51 +0100 Subject: [PATCH] Added youtube import. Reworked Download Manager --- Cargo.lock | 2 + Cargo.toml | 6 +- Dockerfile | 13 +- README.md | 2 +- flake.nix | 3 + src/i18n/phrases.rs | 76 +- src/local_uploads.rs | 259 +++++ src/main.rs | 2 + src/metrics.rs | 26 +- src/music/mod.rs | 148 +++ src/player/mod.rs | 499 ++++++++- src/similarity.rs | 108 ++ src/youtube.rs | 1786 +++++++++++++++++++++++++++++++++ templates/player/modals.html | 301 +++++- templates/player/scripts.html | 476 ++++++++- templates/player/styles.html | 539 ++++++++++ 16 files changed, 4167 insertions(+), 79 deletions(-) create mode 100644 src/local_uploads.rs create mode 100644 src/youtube.rs diff --git a/Cargo.lock b/Cargo.lock index 3959df6..b1a7054 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1953,6 +1953,7 @@ dependencies = [ "futures-util", "id3", "image", + "libc", "librqbit", "md-5", "music-dht", @@ -1969,6 +1970,7 @@ dependencies = [ "symphonia", "tokio", "tokio-cron-scheduler", + "tokio-util", "tower", "tracing", "tracing-subscriber", diff --git a/Cargo.toml b/Cargo.toml index 078d27b..0636872 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "furumusic" -version = "0.10.2" +version = "0.10.3" edition = "2024" description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL" @@ -13,7 +13,9 @@ schemars = { version = "0.9", features = ["derive"] } serde = { version = "1", features = ["derive"] } openidconnect = "4.0" reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json", "stream"] } -tokio = { version = "1", features = ["sync", "fs", "io-util"] } +tokio = { version = "1", features = ["sync", "fs", "io-util", "process"] } +tokio-util = "0.7" +libc = "0.2" async-stream = "0.3" bytes = "1" tower = "0.5" diff --git a/Dockerfile b/Dockerfile index 5e4766a..7a3b644 100644 --- a/Dockerfile +++ b/Dockerfile @@ -14,14 +14,25 @@ COPY templates ./templates RUN cargo build --release +FROM denoland/deno:bin-2.8.3 AS deno + FROM debian:bookworm-slim +ARG YT_DLP_VERSION=2026.07.04 + RUN apt-get update \ - && apt-get install -y --no-install-recommends ca-certificates \ + && apt-get install -y --no-install-recommends \ + ca-certificates \ + ffmpeg \ + python3 \ + python3-pip \ + && pip3 install --break-system-packages --no-cache-dir --disable-pip-version-check \ + "yt-dlp[default]==${YT_DLP_VERSION}" \ && rm -rf /var/lib/apt/lists/* WORKDIR /data COPY --from=builder /app/target/release/furumusic /usr/local/bin/furumusic +COPY --from=deno /deno /usr/local/bin/deno EXPOSE 8000 CMD ["furumusic", "-l", "0.0.0.0:8000"] diff --git a/README.md b/README.md index 4346f8a..af7a047 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ catalog and media files; the browser is only the player. - one shared library with separate user accounts; - a responsive web player with artists, releases, search, queue, and playlists; -- direct file uploads and torrent or magnet imports; +- direct file uploads and imports from YouTube, torrent files, or magnet links; - optional AI-assisted recognition and normalization of metadata; - password login or OIDC/SSO with group-based access control; - optional federation without a central catalog or search service; diff --git a/flake.nix b/flake.nix index 706296b..2e74cb4 100644 --- a/flake.nix +++ b/flake.nix @@ -28,7 +28,10 @@ buildInputs = with pkgs; [ cacert + deno + ffmpeg-headless openssl + yt-dlp ] ++ lib.optionals stdenv.isDarwin [ libiconv ]; RUST_SRC_PATH = "${pkgs.rustPlatform.rustLibSrc}"; diff --git a/src/i18n/phrases.rs b/src/i18n/phrases.rs index e2240e3..72ceafe 100644 --- a/src/i18n/phrases.rs +++ b/src/i18n/phrases.rs @@ -389,9 +389,54 @@ translations! { player_live_releases: "Live releases" , "Концертные релизы"; player_soundtracks: "Soundtracks" , "Саундтреки"; - // Player torrent/history UI - player_torrent_manager: "Torrent manager" , "Торрент-менеджер"; - player_import_torrent: "Import torrent" , "Импортировать торрент"; + // Player download/history UI + player_torrent_manager: "Download Manager" , "Менеджер загрузок"; + player_import_torrent: "Open Download Manager" , "Открыть менеджер загрузок"; + player_youtube: "YouTube" , "YouTube"; + player_torrents: "Torrents" , "Торренты"; + player_files: "Files" , "Файлы"; + player_youtube_url: "Public video or playlist URL" , "Ссылка на открытое видео или плейлист"; + player_youtube_url_hint: "Public youtube.com, music.youtube.com and youtu.be links are supported. Each video is imported as one AI batch." , "Поддерживаются открытые ссылки youtube.com, music.youtube.com и youtu.be. Каждое видео импортируется одним ИИ-батчем."; + player_youtube_downloads: "YouTube downloads" , "Загрузки YouTube"; + player_no_youtube_downloads: "No YouTube downloads yet" , "Загрузок YouTube пока нет"; + player_youtube_parse: "Check link" , "Проверить ссылку"; + player_youtube_parsing: "Reading YouTube link..." , "Читаю ссылку YouTube..."; + player_youtube_preview_failed: "Could not read YouTube link" , "Не удалось прочитать ссылку YouTube"; + player_youtube_preview_title: "Choose videos to import" , "Выберите видео для импорта"; + player_youtube_select_all: "Select all" , "Отметить все"; + player_youtube_clear_selection: "Clear selection" , "Снять все"; + player_youtube_selected_count: "selected" , "выбрано"; + player_youtube_start_import: "Start import" , "Начать импорт"; + player_start_download: "Start download" , "Начать загрузку"; + player_retry_failed: "Retry failed" , "Повторить ошибки"; + player_download_steps: "Processing steps" , "Этапы обработки"; + player_chapters: "chapters" , "глав"; + player_youtube_video: "video" , "видео"; + player_youtube_playlist: "playlist" , "плейлист"; + player_youtube_items: "videos" , "видео"; + player_youtube_errors: "errors" , "ошибок"; + player_youtube_queued: "Queued" , "В очереди"; + player_youtube_resolving: "Reading link" , "Чтение ссылки"; + player_youtube_postprocessing: "FFmpeg processing" , "Обработка FFmpeg"; + player_youtube_awaiting_ai: "Waiting for AI" , "Ожидание ИИ"; + player_youtube_ai_processing: "AI processing" , "Обработка ИИ"; + player_youtube_needs_review: "Needs review" , "Требует проверки"; + player_youtube_complete_with_errors: "Completed with errors" , "Завершено с ошибками"; + player_youtube_skipped: "Already imported" , "Уже импортировано"; + player_youtube_cancelled: "Stopped" , "Остановлено"; + player_youtube_stop: "Stop import" , "Остановить импорт"; + player_youtube_stop_confirm: "Stop this YouTube import? Completed and already published audio will remain." , "Остановить этот импорт из YouTube? Готовое и уже переданное на обработку аудио останется."; + player_youtube_stopping: "Stopping YouTube import..." , "Останавливаю импорт из YouTube..."; + player_youtube_stopped: "YouTube import stopped." , "Импорт из YouTube остановлен."; + player_youtube_stop_failed: "Could not stop YouTube import" , "Не удалось остановить импорт из YouTube"; + player_youtube_starting: "Adding YouTube download..." , "Добавляю загрузку YouTube..."; + player_youtube_started: "YouTube download added." , "Загрузка YouTube добавлена."; + player_youtube_load_failed: "Could not load YouTube downloads" , "Не удалось загрузить список YouTube"; + player_youtube_start_failed: "Could not start YouTube download" , "Не удалось начать загрузку YouTube"; + player_youtube_retry_failed: "Could not retry YouTube download" , "Не удалось повторить загрузку YouTube"; + player_youtube_delete_failed: "Could not remove YouTube download" , "Не удалось удалить загрузку YouTube"; + player_youtube_delete_confirm: "Remove this YouTube download from history? Imported audio will remain." , "Удалить эту загрузку YouTube из истории? Импортированное аудио останется."; + player_remove_from_history: "Remove from history" , "Удалить из истории"; player_client_idle: "Client idle" , "Клиент простаивает"; player_active: "active" , "активно"; player_ai_idle: "AI idle" , "ИИ простаивает"; @@ -467,11 +512,24 @@ translations! { player_track_approved_imported: "Track approved and imported" , "Трек подтверждён и импортирован"; player_failed_update_selected_tracks: "Failed to update selected tracks" , "Не удалось обновить выбранные треки"; player_selected_tracks_updated: "Selected tracks updated" , "Выбранные треки обновлены"; - player_choose_saved_or_add_torrent: "Choose a saved item or upload new files." , "Выберите сохранённый элемент или загрузите новые файлы."; + player_choose_saved_or_add_torrent: "Choose a saved torrent or add a new one." , "Выберите сохранённый торрент или добавьте новый."; player_local_files: "Local audio files" , "Локальные аудиофайлы"; + player_file_uploads: "File uploads" , "Загрузки файлов"; + player_drop_audio_title: "Drop audio files here" , "Перетащите аудиофайлы сюда"; + player_drop_audio_hint: "or click to choose files" , "или нажмите, чтобы выбрать файлы"; + player_drop_audio_formats: "MP3, FLAC, WAV, M4A, OGG, Opus and AAC" , "MP3, FLAC, WAV, M4A, OGG, Opus и AAC"; + player_upload_selected_files: "Upload selected files" , "Загрузить выбранные файлы"; + player_upload_history: "Upload history" , "История загрузок"; + player_no_file_uploads: "No file uploads yet" , "Загрузок файлов пока нет"; + player_file_upload_load_failed: "Could not load file upload history" , "Не удалось загрузить историю файлов"; + player_remove_file_upload_confirm: "Remove this file upload from history? Imported audio will remain." , "Удалить эту загрузку файла из истории? Импортированное аудио останется."; + player_file_upload_history_removed: "File upload removed from history." , "Загрузка файла удалена из истории."; + player_file_upload_history_remove_failed: "Could not remove file upload from history" , "Не удалось удалить загрузку файла из истории"; + player_no_supported_audio_files: "Choose at least one supported audio file." , "Выберите хотя бы один поддерживаемый аудиофайл."; player_torrent_file: "Torrent file" , "Torrent-файл"; player_magnet_link: "Magnet link" , "Magnet-ссылка"; - player_upload_content: "Upload" , "Загрузить"; + player_upload_content: "Preview torrent" , "Проверить торрент"; + player_add_torrent: "Add torrent" , "Добавить торрент"; player_download_selected: "Download selected" , "Скачать выбранное"; player_pause_download: "Pause download" , "Поставить на паузу"; player_expand_all: "Expand all" , "Развернуть всё"; @@ -501,7 +559,7 @@ translations! { player_no_plays_yet: "No plays yet" , "Прослушиваний пока нет"; player_page: "Page" , "Страница"; player_of: "of" , "из"; - player_choose_torrent: "Choose local files, paste a magnet link, or choose a .torrent file." , "Выберите локальные файлы, вставьте magnet-ссылку или выберите .torrent файл."; + player_choose_torrent: "Paste a magnet link or choose a .torrent file." , "Вставьте magnet-ссылку или выберите .torrent файл."; player_uploading_files: "Uploading files..." , "Загружаю файлы..."; player_upload_complete: "Upload complete. Files are queued for processing." , "Загрузка завершена. Файлы поставлены в обработку."; player_upload_failed: "Upload failed" , "Загрузка не удалась"; @@ -511,8 +569,8 @@ translations! { player_all_files_selected: "All files are selected by default. Clear or adjust the tree before download." , "Все файлы выбраны по умолчанию. Перед скачиванием можно очистить или изменить выбор."; player_opening_saved_torrent: "Opening saved torrent..." , "Открываю сохранённый торрент..."; player_saved_torrent_opened: "Saved torrent opened. Adjust files or resume download." , "Сохранённый торрент открыт. Можно изменить файлы или продолжить скачивание."; - player_remove_torrent_confirm: "Remove this torrent from the client list? Downloaded files will stay on disk." , "Удалить этот торрент из списка клиента? Скачанные файлы останутся на диске."; - player_torrent_removed: "Torrent removed from the client list." , "Торрент удалён из списка клиента."; + player_remove_torrent_confirm: "Remove this torrent from history? Downloaded files will stay on disk." , "Удалить этот торрент из истории? Скачанные файлы останутся на диске."; + player_torrent_removed: "Torrent removed from history." , "Торрент удалён из истории."; player_select_one_file: "Select at least one file." , "Выберите хотя бы один файл."; player_starting_download: "Starting download..." , "Запускаю скачивание..."; player_download_started: "Download started. Files will move to inbox when complete." , "Скачивание началось. После завершения файлы будут перенесены во входящие."; @@ -523,6 +581,6 @@ translations! { player_pause_failed: "Pause failed" , "Не удалось поставить на паузу"; player_load_torrents_failed: "Could not load torrents" , "Не удалось загрузить торренты"; player_open_torrent_failed: "Could not open torrent" , "Не удалось открыть торрент"; - player_delete_torrent_failed: "Could not delete torrent" , "Не удалось удалить торрент"; + player_delete_torrent_failed: "Could not remove torrent from history" , "Не удалось удалить торрент из истории"; player_load_ai_queue_failed: "Could not load AI queue" , "Не удалось загрузить очередь ИИ"; } diff --git a/src/local_uploads.rs b/src/local_uploads.rs new file mode 100644 index 0000000..5538b54 --- /dev/null +++ b/src/local_uploads.rs @@ -0,0 +1,259 @@ +use std::collections::HashMap; + +use anyhow::{Context, bail}; +use serde::Serialize; +use sqlx::{FromRow, PgPool}; + +const LOCAL_UPLOAD_LIST_LIMIT: i64 = 100; + +#[derive(Debug, Clone, Serialize)] +pub struct LocalUploadDto { + pub id: String, + pub filename: String, + pub size_bytes: u64, + pub status: String, + pub error: Option, + pub created_at: String, + pub updated_at: String, + pub completed_at: Option, +} + +#[derive(Debug, Clone, FromRow)] +struct LocalUploadRow { + id: String, + user_id: i64, + filename: String, + size_bytes: i64, + status: String, + inbox_path: String, + error: Option, + created_at: String, + updated_at: String, + completed_at: Option, +} + +impl LocalUploadRow { + fn dto(&self) -> LocalUploadDto { + LocalUploadDto { + id: self.id.clone(), + filename: self.filename.clone(), + size_bytes: u64::try_from(self.size_bytes).unwrap_or(0), + status: self.status.clone(), + error: self.error.clone(), + created_at: self.created_at.clone(), + updated_at: self.updated_at.clone(), + completed_at: self.completed_at.clone(), + } + } +} + +pub async fn create( + pool: &PgPool, + id: &str, + user_id: i64, + filename: &str, + size_bytes: u64, + inbox_path: &str, +) -> anyhow::Result<()> { + let now = now_string(); + sqlx::query( + r#"INSERT INTO furumusic__local_upload + (id, user_id, filename, size_bytes, status, inbox_path, error, + created_at, updated_at, completed_at) + VALUES ($1, $2, $3, $4, 'uploading', $5, NULL, $6, $6, NULL)"#, + ) + .bind(id) + .bind(user_id) + .bind(filename) + .bind(i64::try_from(size_bytes).unwrap_or(i64::MAX)) + .bind(inbox_path) + .bind(now) + .execute(pool) + .await?; + Ok(()) +} + +pub async fn mark_queued(pool: &PgPool, id: &str, user_id: i64) -> anyhow::Result { + update_status(pool, id, user_id, "queued", None).await?; + load(pool, user_id, id).await.map(|row| row.dto()) +} + +pub async fn mark_failed(pool: &PgPool, id: &str, user_id: i64, error: &str) -> anyhow::Result<()> { + update_status(pool, id, user_id, "failed", Some(error)).await +} + +pub async fn list( + pool: &PgPool, + user_id: i64, + inbox_dir: &str, +) -> anyhow::Result> { + sync_statuses(pool, user_id, inbox_dir).await?; + let rows: Vec = sqlx::query_as( + r#"SELECT id, user_id, filename, size_bytes, status, inbox_path, error, + created_at, updated_at, completed_at + FROM furumusic__local_upload + WHERE user_id = $1 + ORDER BY created_at DESC, id DESC + LIMIT $2"#, + ) + .bind(user_id) + .bind(LOCAL_UPLOAD_LIST_LIMIT) + .fetch_all(pool) + .await?; + Ok(rows.iter().map(LocalUploadRow::dto).collect()) +} + +pub async fn remove(pool: &PgPool, user_id: i64, id: &str) -> anyhow::Result<()> { + let result = sqlx::query("DELETE FROM furumusic__local_upload WHERE id = $1 AND user_id = $2") + .bind(id) + .bind(user_id) + .execute(pool) + .await?; + if result.rows_affected() == 0 { + bail!("file upload history entry not found"); + } + Ok(()) +} + +async fn sync_statuses(pool: &PgPool, user_id: i64, inbox_dir: &str) -> anyhow::Result<()> { + let inbox_dir = inbox_dir.trim(); + if inbox_dir.is_empty() { + bail!("agent_inbox_dir is not configured"); + } + let inbox_root = crate::media_paths::resolve_config_path_buf(inbox_dir); + if !inbox_root.is_absolute() { + bail!("agent_inbox_dir must be an absolute path"); + } + + let rows: Vec = sqlx::query_as( + r#"SELECT id, user_id, filename, size_bytes, status, inbox_path, error, + created_at, updated_at, completed_at + FROM furumusic__local_upload + WHERE user_id = $1 AND status <> 'complete'"#, + ) + .bind(user_id) + .fetch_all(pool) + .await?; + if rows.is_empty() { + return Ok(()); + } + + let inbox_paths: Vec = rows.iter().map(|row| row.inbox_path.clone()).collect(); + let state_rows: Vec<(String, String, i64)> = sqlx::query_as( + r#"SELECT input_path, status::text, COUNT(*) + FROM furumusic__pending_review + WHERE input_path = ANY($1) + GROUP BY input_path, status"#, + ) + .bind(&inbox_paths) + .fetch_all(pool) + .await?; + let mut states_by_path: HashMap> = HashMap::new(); + for (input_path, status, total) in state_rows { + states_by_path + .entry(input_path) + .or_default() + .insert(status, total); + } + let error_rows: Vec<(String, String)> = sqlx::query_as( + r#"SELECT DISTINCT ON (input_path) input_path, error_message + FROM furumusic__pending_review + WHERE input_path = ANY($1) AND status = 'failed' + AND error_message IS NOT NULL + ORDER BY input_path, id DESC"#, + ) + .bind(&inbox_paths) + .fetch_all(pool) + .await?; + let errors_by_path: HashMap = error_rows.into_iter().collect(); + + for row in rows { + let counts = states_by_path + .get(&row.inbox_path) + .cloned() + .unwrap_or_default(); + let total: i64 = counts.values().sum(); + + let mut terminal_error = None; + let next = if total == 0 { + if matches!(row.status.as_str(), "uploading" | "failed" | "needs_review") { + continue; + } + let full_path = crate::media_paths::resolve_path_from_root(inbox_dir, &row.inbox_path); + if tokio::fs::try_exists(full_path).await.unwrap_or(false) { + "queued" + } else { + "complete" + } + } else if count(&counts, "processing") > 0 { + "ai_processing" + } else if count(&counts, "queued") > 0 { + "queued" + } else if count(&counts, "failed") > 0 { + terminal_error = errors_by_path.get(&row.inbox_path).cloned(); + "failed" + } else if count(&counts, "pending") > 0 || count(&counts, "rejected") > 0 { + "needs_review" + } else if count(&counts, "approved") > 0 || count(&counts, "auto_approved") > 0 { + "complete" + } else { + "queued" + }; + + if row.status != next || row.error != terminal_error { + update_status(pool, &row.id, row.user_id, next, terminal_error.as_deref()).await?; + } + } + Ok(()) +} + +async fn update_status( + pool: &PgPool, + id: &str, + user_id: i64, + status: &str, + error: Option<&str>, +) -> anyhow::Result<()> { + let now = now_string(); + let completed_at = + matches!(status, "complete" | "failed" | "needs_review").then(|| now.clone()); + sqlx::query( + r#"UPDATE furumusic__local_upload + SET status = $3, error = $4, updated_at = $5, completed_at = $6 + WHERE id = $1 AND user_id = $2"#, + ) + .bind(id) + .bind(user_id) + .bind(status) + .bind(error.map(trim_error)) + .bind(&now) + .bind(completed_at) + .execute(pool) + .await?; + Ok(()) +} + +async fn load(pool: &PgPool, user_id: i64, id: &str) -> anyhow::Result { + sqlx::query_as( + r#"SELECT id, user_id, filename, size_bytes, status, inbox_path, error, + created_at, updated_at, completed_at + FROM furumusic__local_upload WHERE id = $1 AND user_id = $2"#, + ) + .bind(id) + .bind(user_id) + .fetch_optional(pool) + .await? + .context("file upload history entry not found") +} + +fn count(counts: &HashMap, status: &str) -> i64 { + counts.get(status).copied().unwrap_or(0) +} + +fn trim_error(value: &str) -> String { + value.chars().take(4_000).collect() +} + +fn now_string() -> String { + chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string() +} diff --git a/src/main.rs b/src/main.rs index b0bc809..5b484f5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,6 +7,7 @@ mod federation; mod i18n; mod jobs; mod lastfm; +mod local_uploads; mod media_paths; mod metrics; mod music; @@ -16,6 +17,7 @@ mod scheduler; mod similarity; mod torrents; mod user; +mod youtube; use std::sync::Arc; diff --git a/src/metrics.rs b/src/metrics.rs index 157deb5..1003118 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -884,10 +884,18 @@ const KNOWN_HTTP_ROUTES: &[&str] = &[ "/api/player/lastfm/scrobble", "/api/player/agent-queue", "/api/player/offline/manifest", + "/api/player/youtube", + "/api/player/youtube/preview", + "/api/player/youtube/start", + "/api/player/youtube/{id}/retry", + "/api/player/youtube/{id}/cancel", + "/api/player/youtube/{id}", + "/api/player/uploads/local", + "/api/player/uploads/local/history", + "/api/player/uploads/local/history/{id}", "/api/player/torrents", "/api/player/torrents/session/{id}", "/api/player/torrents/preview", - "/api/player/uploads/local", "/api/player/uploads/tracks", "/api/player/uploads/tracks/{track_id}", "/api/player/uploads/bulk-tracks", @@ -951,6 +959,22 @@ mod tests { known_http_route("/share/release/42"), Some("/share/release/{id}") ); + assert_eq!( + known_http_route("/api/player/youtube/start"), + Some("/api/player/youtube/start") + ); + assert_eq!( + known_http_route("/api/player/youtube/job-42/retry"), + Some("/api/player/youtube/{id}/retry") + ); + assert_eq!( + known_http_route("/api/player/youtube/job-42/cancel"), + Some("/api/player/youtube/{id}/cancel") + ); + assert_eq!( + known_http_route("/api/player/uploads/local/history/upload-42"), + Some("/api/player/uploads/local/history/{id}") + ); } #[test] diff --git a/src/music/mod.rs b/src/music/mod.rs index c0962df..5821dcb 100644 --- a/src/music/mod.rs +++ b/src/music/mod.rs @@ -2569,6 +2569,152 @@ pub mod db_migrations { &[Operation::custom(add_similarity_routing_signature).build()]; } + // -- M0045: persistent YouTube download jobs ---------------------------- + + #[cot::db::migrations::migration_op] + async fn create_youtube_downloads( + ctx: migrations::MigrationContext<'_>, + ) -> cot::db::Result<()> { + ctx.db + .raw( + "CREATE TABLE IF NOT EXISTS furumusic__youtube_download ( + id VARCHAR(36) PRIMARY KEY, + user_id BIGINT NOT NULL, + source_url TEXT NOT NULL, + title TEXT NOT NULL, + source_kind VARCHAR(32) NOT NULL, + status VARCHAR(32) NOT NULL, + total_items INTEGER NOT NULL DEFAULT 0, + completed_items INTEGER NOT NULL DEFAULT 0, + failed_items INTEGER NOT NULL DEFAULT 0, + review_items INTEGER NOT NULL DEFAULT 0, + error TEXT, + created_at VARCHAR(32) NOT NULL, + updated_at VARCHAR(32) NOT NULL, + completed_at VARCHAR(32) + )", + ) + .await?; + ctx.db + .raw( + "CREATE TABLE IF NOT EXISTS furumusic__youtube_download_item ( + id VARCHAR(36) PRIMARY KEY, + job_id VARCHAR(36) NOT NULL REFERENCES furumusic__youtube_download(id) ON DELETE CASCADE, + source_id VARCHAR(128) NOT NULL, + source_url TEXT NOT NULL, + title TEXT NOT NULL, + playlist_index INTEGER NOT NULL, + status VARCHAR(32) NOT NULL, + progress_percent DOUBLE PRECISION NOT NULL DEFAULT 0, + downloaded_bytes BIGINT NOT NULL DEFAULT 0, + total_bytes BIGINT, + speed_bytes_per_sec BIGINT, + eta_seconds BIGINT, + chapter_count INTEGER NOT NULL DEFAULT 0, + audio_file_count INTEGER NOT NULL DEFAULT 0, + inbox_path TEXT, + error TEXT, + created_at VARCHAR(32) NOT NULL, + updated_at VARCHAR(32) NOT NULL, + completed_at VARCHAR(32), + UNIQUE(job_id, source_id) + )", + ) + .await?; + ctx.db + .raw( + "CREATE INDEX IF NOT EXISTS idx_youtube_download_user_updated + ON furumusic__youtube_download (user_id, updated_at DESC)", + ) + .await?; + ctx.db + .raw( + "CREATE INDEX IF NOT EXISTS idx_youtube_download_user_status + ON furumusic__youtube_download (user_id, status)", + ) + .await?; + ctx.db + .raw( + "CREATE INDEX IF NOT EXISTS idx_youtube_download_item_job_status + ON furumusic__youtube_download_item (job_id, status, playlist_index)", + ) + .await?; + ctx.db + .raw( + "CREATE INDEX IF NOT EXISTS idx_youtube_download_item_source + ON furumusic__youtube_download_item (source_id)", + ) + .await?; + Ok(()) + } + + #[derive(Debug, Copy, Clone)] + pub struct M0045CreateYouTubeDownloads; + + impl migrations::Migration for M0045CreateYouTubeDownloads { + const APP_NAME: &'static str = "furumusic"; + const MIGRATION_NAME: &'static str = "m_0045_create_youtube_downloads"; + const DEPENDENCIES: &'static [migrations::MigrationDependency] = + &[migrations::MigrationDependency::migration( + "furumusic", + "m_0044_add_similarity_routing_signature", + )]; + const OPERATIONS: &'static [Operation] = + &[Operation::custom(create_youtube_downloads).build()]; + } + + // -- M0046: persistent direct-file upload history ----------------------- + + #[cot::db::migrations::migration_op] + async fn create_local_upload_history( + ctx: migrations::MigrationContext<'_>, + ) -> cot::db::Result<()> { + ctx.db + .raw( + "CREATE TABLE IF NOT EXISTS furumusic__local_upload ( + id VARCHAR(36) PRIMARY KEY, + user_id BIGINT NOT NULL, + filename TEXT NOT NULL, + size_bytes BIGINT NOT NULL DEFAULT 0, + status VARCHAR(32) NOT NULL, + inbox_path TEXT NOT NULL, + error TEXT, + created_at VARCHAR(32) NOT NULL, + updated_at VARCHAR(32) NOT NULL, + completed_at VARCHAR(32) + )", + ) + .await?; + ctx.db + .raw( + "CREATE INDEX IF NOT EXISTS idx_local_upload_user_updated + ON furumusic__local_upload (user_id, updated_at DESC)", + ) + .await?; + ctx.db + .raw( + "CREATE INDEX IF NOT EXISTS idx_local_upload_user_status + ON furumusic__local_upload (user_id, status)", + ) + .await?; + Ok(()) + } + + #[derive(Debug, Copy, Clone)] + pub struct M0046CreateLocalUploadHistory; + + impl migrations::Migration for M0046CreateLocalUploadHistory { + const APP_NAME: &'static str = "furumusic"; + const MIGRATION_NAME: &'static str = "m_0046_create_local_upload_history"; + const DEPENDENCIES: &'static [migrations::MigrationDependency] = + &[migrations::MigrationDependency::migration( + "furumusic", + "m_0045_create_youtube_downloads", + )]; + const OPERATIONS: &'static [Operation] = + &[Operation::custom(create_local_upload_history).build()]; + } + pub const MIGRATIONS: &[&SyncDynMigration] = &[ &M0006CreateMediaFile, &M0007CreateArtist, @@ -2604,5 +2750,7 @@ pub mod db_migrations { &M0042RepairLegacyListenQualification, &M0043CreateSimilarityEmbeddings, &M0044AddSimilarityRoutingSignature, + &M0045CreateYouTubeDownloads, + &M0046CreateLocalUploadHistory, ]; } diff --git a/src/player/mod.rs b/src/player/mod.rs index 5d2d39a..afeb72d 100644 --- a/src/player/mod.rs +++ b/src/player/mod.rs @@ -21,8 +21,10 @@ use crate::auth; use crate::config::AppConfig; use crate::i18n::Translations; use crate::lastfm::{LastfmClient, LastfmCredentials, LastfmTrackPayload}; +use crate::local_uploads::LocalUploadDto; use crate::scheduler::SchedulerHandle; use crate::torrents::{TorrentPreviewRequest, TorrentService, TorrentStartRequest}; +use crate::youtube::{YouTubePreviewRequest, YouTubeService, YouTubeStartRequest}; mod dto; mod helpers; @@ -50,8 +52,7 @@ fn json_error(status: StatusCode, message: &str) -> cot::response::Response { #[derive(serde::Serialize)] struct LocalUploadResponse { ok: bool, - filename: String, - size: u64, + upload: LocalUploadDto, } const PLAYER_DEVICE_TTL_MS: i64 = 30_000; @@ -4830,6 +4831,7 @@ async fn local_upload_handler( session: Session, db: Database, config: AppConfig, + pool: &sqlx::PgPool, scheduler_handle: Arc>>, request: cot::request::Request, ) -> cot::Result> { @@ -4862,6 +4864,14 @@ async fn local_upload_handler( .filter(|value| !value.is_empty()) .unwrap_or_else(|| "upload.mp3".to_string()); let filename = sanitize_upload_filename(&original_name); + let upload_id_header = HeaderName::from_static("x-furumusic-upload-id"); + let upload_id = request + .headers() + .get(upload_id_header) + .and_then(|value| value.to_str().ok()) + .and_then(|value| uuid::Uuid::parse_str(value.trim()).ok()) + .unwrap_or_else(uuid::Uuid::new_v4) + .to_string(); let bytes = request .into_body() @@ -4878,14 +4888,53 @@ async fn local_upload_handler( let upload_dir = inbox_root .join("user_uploads") .join(user.id.to_string()) - .join(format!("local-{}", uuid::Uuid::new_v4())); - tokio::fs::create_dir_all(&upload_dir) - .await - .map_err(|err| cot::Error::internal(err.to_string()))?; + .join(format!("local-{upload_id}")); let destination = upload_dir.join(&filename); - tokio::fs::write(&destination, &bytes) - .await - .map_err(|err| cot::Error::internal(err.to_string()))?; + let Some(inbox_path) = + crate::media_paths::path_for_root(&inbox_root.to_string_lossy(), &destination) + else { + return Ok(json_error( + StatusCode::INTERNAL_SERVER_ERROR, + "upload destination escaped agent_inbox_dir", + )); + }; + if let Err(err) = crate::local_uploads::create( + pool, + &upload_id, + user.id, + &filename, + bytes.len() as u64, + &inbox_path, + ) + .await + { + return Ok(json_error(StatusCode::BAD_REQUEST, &err.to_string())); + } + + let temporary = upload_dir.join(format!(".{upload_id}.uploading")); + let write_result = async { + tokio::fs::create_dir_all(&upload_dir).await?; + tokio::fs::write(&temporary, &bytes).await?; + tokio::fs::rename(&temporary, &destination).await?; + Ok::<(), std::io::Error>(()) + } + .await; + if let Err(err) = write_result { + let message = format!("could not save uploaded file: {err}"); + let _ = tokio::fs::remove_dir_all(&upload_dir).await; + let _ = crate::local_uploads::mark_failed(pool, &upload_id, user.id, &message).await; + return Ok(json_error(StatusCode::INTERNAL_SERVER_ERROR, &message)); + } + + let upload = match crate::local_uploads::mark_queued(pool, &upload_id, user.id).await { + Ok(upload) => upload, + Err(err) => { + return Ok(json_error( + StatusCode::INTERNAL_SERVER_ERROR, + &err.to_string(), + )); + } + }; if let Some(handle) = scheduler_handle.get() { let handle = Arc::clone(handle); @@ -4896,12 +4945,39 @@ async fn local_upload_handler( }); } - Json(LocalUploadResponse { - ok: true, - filename, - size: bytes.len() as u64, - }) - .into_response() + Json(LocalUploadResponse { ok: true, upload }).into_response() +} + +async fn local_upload_history_handler( + auth_ctx: auth::AuthContext, + session: Session, + db: Database, + pool: &sqlx::PgPool, +) -> cot::Result { + let Some(user) = auth::get_request_user(&auth_ctx, &session, &db).await else { + return Ok(json_error(StatusCode::UNAUTHORIZED, "not authenticated")); + }; + let (config, _) = AppConfig::load_with_db(&db).await; + match crate::local_uploads::list(pool, user.id, &config.agent_inbox_dir).await { + Ok(items) => Json(items).into_response(), + Err(err) => Ok(json_error(StatusCode::BAD_REQUEST, &err.to_string())), + } +} + +async fn local_upload_history_remove_handler( + auth_ctx: auth::AuthContext, + session: Session, + db: Database, + pool: &sqlx::PgPool, + path: Path, +) -> cot::Result { + let Some(user) = auth::get_request_user(&auth_ctx, &session, &db).await else { + return Ok(json_error(StatusCode::UNAUTHORIZED, "not authenticated")); + }; + match crate::local_uploads::remove(pool, user.id, &path.0.id).await { + Ok(()) => Json(serde_json::json!({ "ok": true })).into_response(), + Err(err) => Ok(json_error(StatusCode::BAD_REQUEST, &err.to_string())), + } } fn sanitize_upload_filename(value: &str) -> String { @@ -8107,6 +8183,8 @@ impl App for PlayerApp { let pool: Arc> = Arc::new(tokio::sync::OnceCell::new()); let torrent_service: Arc>> = Arc::new(tokio::sync::OnceCell::new()); + let youtube_service: Arc>> = + Arc::new(tokio::sync::OnceCell::new()); let device_hub = Arc::clone(&self.device_hub); Router::with_urls([ @@ -8345,6 +8423,329 @@ impl App for PlayerApp { }, "player_agent_queue", ), + // -- YouTube downloads -- + Route::with_handler_and_name( + "/youtube/preview", + { + let youtube_service = Arc::clone(&youtube_service); + let scheduler_handle = Arc::clone(&self.scheduler_handle); + post( + move |auth_ctx: auth::AuthContext, + session: Session, + db: Database, + json: Json| { + let youtube_service = Arc::clone(&youtube_service); + let scheduler_handle = Arc::clone(&scheduler_handle); + async move { + if auth::get_request_user(&auth_ctx, &session, &db) + .await + .is_none() + { + return Ok(json_error( + StatusCode::UNAUTHORIZED, + "not authenticated", + )); + } + let service = youtube_service + .get_or_init(|| async { + Arc::new(YouTubeService::new(Arc::clone(&scheduler_handle))) + }) + .await; + match service.preview(json.0).await { + Ok(preview) => Json(preview).into_response(), + Err(err) => { + Ok(json_error(StatusCode::BAD_REQUEST, &err.to_string())) + } + } + } + }, + ) + }, + "player_youtube_preview", + ), + Route::with_handler_and_name( + "/youtube", + { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + let youtube_service = Arc::clone(&youtube_service); + let scheduler_handle = Arc::clone(&self.scheduler_handle); + get( + move |auth_ctx: auth::AuthContext, session: Session, db: Database| { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + let youtube_service = Arc::clone(&youtube_service); + let scheduler_handle = Arc::clone(&scheduler_handle); + async move { + let Some(user) = + auth::get_request_user(&auth_ctx, &session, &db).await + else { + return Ok(json_error( + StatusCode::UNAUTHORIZED, + "not authenticated", + )); + }; + let pg_pool = pool + .get_or_init(|| async { + sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .connect(&pool_config.database_url) + .await + .expect("player pool") + }) + .await; + let service = youtube_service + .get_or_init(|| async { + Arc::new(YouTubeService::new(Arc::clone(&scheduler_handle))) + }) + .await; + let (live_config, _) = AppConfig::load_with_db(&db).await; + match service + .list(pg_pool, user.id, &live_config.agent_inbox_dir) + .await + { + Ok(items) => Json(items).into_response(), + Err(err) => { + Ok(json_error(StatusCode::BAD_REQUEST, &err.to_string())) + } + } + } + }, + ) + }, + "player_youtube_list", + ), + Route::with_handler_and_name( + "/youtube/start", + { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + let youtube_service = Arc::clone(&youtube_service); + let scheduler_handle = Arc::clone(&self.scheduler_handle); + post( + move |auth_ctx: auth::AuthContext, + session: Session, + db: Database, + json: Json| { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + let youtube_service = Arc::clone(&youtube_service); + let scheduler_handle = Arc::clone(&scheduler_handle); + async move { + let Some(user) = + auth::get_request_user(&auth_ctx, &session, &db).await + else { + return Ok(json_error( + StatusCode::UNAUTHORIZED, + "not authenticated", + )); + }; + let pg_pool = pool + .get_or_init(|| async { + sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .connect(&pool_config.database_url) + .await + .expect("player pool") + }) + .await; + let service = youtube_service + .get_or_init(|| async { + Arc::new(YouTubeService::new(Arc::clone(&scheduler_handle))) + }) + .await; + let (live_config, _) = AppConfig::load_with_db(&db).await; + match service + .start( + pg_pool, + user.id, + json.0, + &live_config.agent_inbox_dir, + ) + .await + { + Ok(job) => Json(job).into_response(), + Err(err) => { + Ok(json_error(StatusCode::BAD_REQUEST, &err.to_string())) + } + } + } + }, + ) + }, + "player_youtube_start", + ), + Route::with_handler_and_name( + "/youtube/{id}/retry", + { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + let youtube_service = Arc::clone(&youtube_service); + let scheduler_handle = Arc::clone(&self.scheduler_handle); + post( + move |auth_ctx: auth::AuthContext, + session: Session, + db: Database, + path: Path| { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + let youtube_service = Arc::clone(&youtube_service); + let scheduler_handle = Arc::clone(&scheduler_handle); + async move { + let Some(user) = + auth::get_request_user(&auth_ctx, &session, &db).await + else { + return Ok(json_error( + StatusCode::UNAUTHORIZED, + "not authenticated", + )); + }; + let pg_pool = pool + .get_or_init(|| async { + sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .connect(&pool_config.database_url) + .await + .expect("player pool") + }) + .await; + let service = youtube_service + .get_or_init(|| async { + Arc::new(YouTubeService::new(Arc::clone(&scheduler_handle))) + }) + .await; + let (live_config, _) = AppConfig::load_with_db(&db).await; + match service + .retry( + pg_pool, + user.id, + &path.0.id, + &live_config.agent_inbox_dir, + ) + .await + { + Ok(job) => Json(job).into_response(), + Err(err) => { + Ok(json_error(StatusCode::BAD_REQUEST, &err.to_string())) + } + } + } + }, + ) + }, + "player_youtube_retry", + ), + Route::with_handler_and_name( + "/youtube/{id}/cancel", + { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + let youtube_service = Arc::clone(&youtube_service); + let scheduler_handle = Arc::clone(&self.scheduler_handle); + post( + move |auth_ctx: auth::AuthContext, + session: Session, + db: Database, + path: Path| { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + let youtube_service = Arc::clone(&youtube_service); + let scheduler_handle = Arc::clone(&scheduler_handle); + async move { + let Some(user) = + auth::get_request_user(&auth_ctx, &session, &db).await + else { + return Ok(json_error( + StatusCode::UNAUTHORIZED, + "not authenticated", + )); + }; + let pg_pool = pool + .get_or_init(|| async { + sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .connect(&pool_config.database_url) + .await + .expect("player pool") + }) + .await; + let service = youtube_service + .get_or_init(|| async { + Arc::new(YouTubeService::new(Arc::clone(&scheduler_handle))) + }) + .await; + match service.cancel(pg_pool, user.id, &path.0.id).await { + Ok(job) => Json(job).into_response(), + Err(err) => { + Ok(json_error(StatusCode::BAD_REQUEST, &err.to_string())) + } + } + } + }, + ) + }, + "player_youtube_cancel", + ), + Route::with_handler_and_name( + "/youtube/{id}", + { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + let youtube_service = Arc::clone(&youtube_service); + let scheduler_handle = Arc::clone(&self.scheduler_handle); + delete( + move |auth_ctx: auth::AuthContext, + session: Session, + db: Database, + path: Path| { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + let youtube_service = Arc::clone(&youtube_service); + let scheduler_handle = Arc::clone(&scheduler_handle); + async move { + let Some(user) = + auth::get_request_user(&auth_ctx, &session, &db).await + else { + return Ok(json_error( + StatusCode::UNAUTHORIZED, + "not authenticated", + )); + }; + let pg_pool = pool + .get_or_init(|| async { + sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .connect(&pool_config.database_url) + .await + .expect("player pool") + }) + .await; + let service = youtube_service + .get_or_init(|| async { + Arc::new(YouTubeService::new(Arc::clone(&scheduler_handle))) + }) + .await; + let (live_config, _) = AppConfig::load_with_db(&db).await; + match service + .remove( + pg_pool, + user.id, + &path.0.id, + &live_config.agent_inbox_dir, + ) + .await + { + Ok(()) => Json(serde_json::json!({ "ok": true })).into_response(), + Err(err) => { + Ok(json_error(StatusCode::BAD_REQUEST, &err.to_string())) + } + } + } + }, + ) + }, + "player_youtube_remove", + ), // -- Torrent import widget -- Route::with_handler_and_name( "/torrents", @@ -8546,20 +8947,34 @@ impl App for PlayerApp { Route::with_handler_and_name( "/uploads/local", { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); let scheduler_handle = Arc::clone(&self.scheduler_handle); post( move |auth_ctx: auth::AuthContext, session: Session, db: Database, request: cot::request::Request| { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); let scheduler_handle = Arc::clone(&scheduler_handle); async move { + let pg_pool = pool + .get_or_init(|| async { + sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .connect(&pool_config.database_url) + .await + .expect("player pool") + }) + .await; let (live_config, _) = AppConfig::load_with_db(&db).await; local_upload_handler( auth_ctx, session, db, live_config, + pg_pool, scheduler_handle, request, ) @@ -8570,6 +8985,60 @@ impl App for PlayerApp { }, "player_local_upload", ), + Route::with_handler_and_name( + "/uploads/local/history", + get({ + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + move |auth_ctx: auth::AuthContext, session: Session, db: Database| { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + async move { + let pg_pool = pool + .get_or_init(|| async { + sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .connect(&pool_config.database_url) + .await + .expect("player pool") + }) + .await; + local_upload_history_handler(auth_ctx, session, db, pg_pool).await + } + } + }), + "player_local_upload_history", + ), + Route::with_handler_and_name( + "/uploads/local/history/{id}", + delete({ + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + move |auth_ctx: auth::AuthContext, + session: Session, + db: Database, + path: Path| { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + async move { + let pg_pool = pool + .get_or_init(|| async { + sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .connect(&pool_config.database_url) + .await + .expect("player pool") + }) + .await; + local_upload_history_remove_handler( + auth_ctx, session, db, pg_pool, path, + ) + .await + } + } + }), + "player_local_upload_history_remove", + ), Route::with_handler_and_name( "/uploads/tracks", get({ diff --git a/src/similarity.rs b/src/similarity.rs index b0a410f..0d49469 100644 --- a/src/similarity.rs +++ b/src/similarity.rs @@ -7,6 +7,7 @@ use std::collections::{HashMap, HashSet}; use std::fs::File; use std::path::{Path, PathBuf}; +use std::process::Command; use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex, OnceLock, RwLock}; use std::time::{Duration, Instant}; @@ -1194,6 +1195,23 @@ fn decode_mono_window( path: &Path, start_seconds: f64, length_seconds: Option, +) -> Result> { + match decode_mono_window_native(path, start_seconds, length_seconds) { + Ok(samples) => Ok(samples), + Err(native_error) => decode_mono_window_ffmpeg(path, start_seconds, length_seconds) + .with_context(|| { + format!( + "native decoder failed for {} ({native_error:#}); FFmpeg fallback failed", + path.display() + ) + }), + } +} + +fn decode_mono_window_native( + path: &Path, + start_seconds: f64, + length_seconds: Option, ) -> Result> { let file = File::open(path).with_context(|| format!("opening {}", path.display()))?; let mut decoder = @@ -1229,6 +1247,64 @@ fn decode_mono_window( Ok(resample_sinc(&mono, source_rate, SAMPLE_RATE)) } +fn decode_mono_window_ffmpeg( + path: &Path, + start_seconds: f64, + length_seconds: Option, +) -> Result> { + let mut command = Command::new("ffmpeg"); + command.arg("-v").arg("error").arg("-nostdin"); + if start_seconds > 0.0 { + command.arg("-ss").arg(format!("{start_seconds:.6}")); + } + command.arg("-i").arg(path); + if let Some(length_seconds) = length_seconds { + command.arg("-t").arg(format!("{length_seconds:.6}")); + } + let output = command + .arg("-map") + .arg("0:a:0") + .arg("-vn") + .arg("-sn") + .arg("-dn") + .arg("-ac") + .arg("1") + .arg("-ar") + .arg(SAMPLE_RATE.to_string()) + .arg("-f") + .arg("f32le") + .arg("pipe:1") + .output() + .with_context(|| "starting FFmpeg; install FFmpeg to decode this audio format")?; + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + let detail = stderr + .lines() + .map(str::trim) + .rfind(|line| !line.is_empty()) + .unwrap_or("unknown FFmpeg error"); + anyhow::bail!("FFmpeg exited with {}: {detail}", output.status); + } + anyhow::ensure!( + output + .stdout + .len() + .is_multiple_of(std::mem::size_of::()), + "FFmpeg returned a truncated f32le stream" + ); + let samples: Vec = output + .stdout + .chunks_exact(std::mem::size_of::()) + .map(|bytes| f32::from_le_bytes(bytes.try_into().expect("four-byte sample"))) + .collect(); + anyhow::ensure!(!samples.is_empty(), "FFmpeg decoded track is empty"); + anyhow::ensure!( + samples.iter().all(|sample| sample.is_finite()), + "FFmpeg decoded non-finite samples" + ); + Ok(samples) +} + fn resample_sinc(input: &[f32], source_rate: usize, target_rate: usize) -> Vec { if input.len() < 2 || source_rate == 0 { return input.to_vec(); @@ -1458,4 +1534,36 @@ mod tests { assert_eq!(output.len(), 160); assert!(output.iter().all(|value| (*value - 0.25).abs() < 1e-6)); } + + #[test] + fn decodes_opus_with_ffmpeg_fallback() { + if Command::new("ffmpeg").arg("-version").output().is_err() { + return; + } + let path = std::env::temp_dir().join(format!( + "furumusic-similarity-{}.opus", + uuid::Uuid::new_v4() + )); + let generated = Command::new("ffmpeg") + .args([ + "-v", + "error", + "-f", + "lavfi", + "-i", + "sine=frequency=440:sample_rate=48000:duration=0.25", + "-c:a", + "libopus", + "-y", + ]) + .arg(&path) + .status() + .expect("start FFmpeg fixture generation"); + assert!(generated.success(), "generate Opus fixture"); + + let decoded = decode_mono_window(&path, 0.0, None).expect("decode Opus with fallback"); + let _ = std::fs::remove_file(&path); + assert!(decoded.len() >= SAMPLE_RATE / 5); + assert!(decoded.iter().all(|sample| sample.is_finite())); + } } diff --git a/src/youtube.rs b/src/youtube.rs new file mode 100644 index 0000000..68b3a62 --- /dev/null +++ b/src/youtube.rs @@ -0,0 +1,1786 @@ +use std::collections::{HashMap, HashSet, VecDeque}; +use std::path::{Path, PathBuf}; +use std::process::Stdio; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{Context, bail}; +use image::codecs::jpeg::JpegEncoder; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use sqlx::{FromRow, PgPool}; +use tokio::io::{AsyncBufReadExt, AsyncRead, BufReader}; +use tokio::process::Command; +use tokio::sync::{Mutex, OnceCell, Semaphore}; +use tokio_util::sync::CancellationToken; +use uuid::Uuid; + +use crate::scheduler::SchedulerHandle; + +const YOUTUBE_LIST_LIMIT: i64 = 100; +const RESOLVE_TIMEOUT: Duration = Duration::from_secs(180); +const MAX_ERROR_LEN: usize = 4_000; +const AUDIO_EXTENSIONS: &[&str] = &[ + "mp3", "flac", "ogg", "opus", "aac", "m4a", "wav", "ape", "wv", "wma", "tta", "aiff", "aif", +]; +const IMAGE_EXTENSIONS: &[&str] = &["jpg", "jpeg", "png", "webp", "bmp", "gif"]; + +#[derive(Debug, Deserialize)] +pub struct YouTubePreviewRequest { + pub url: String, +} + +#[derive(Debug, Deserialize)] +pub struct YouTubeStartRequest { + pub url: String, + pub selected_source_ids: Vec, +} + +#[derive(Debug, Clone, Serialize)] +pub struct YouTubePreviewItemDto { + pub source_id: String, + pub title: String, + pub playlist_index: i32, + pub selected_by_default: bool, +} + +#[derive(Debug, Clone, Serialize)] +pub struct YouTubePreviewDto { + pub source_url: String, + pub title: String, + pub source_kind: String, + pub items: Vec, +} + +#[derive(Debug, Clone, Serialize)] +pub struct YouTubeItemDto { + pub id: String, + pub source_id: String, + pub source_url: String, + pub title: String, + pub playlist_index: i32, + pub status: String, + pub progress_percent: f64, + pub downloaded_bytes: u64, + pub total_bytes: Option, + pub speed_bytes_per_sec: Option, + pub eta_seconds: Option, + pub chapter_count: i32, + pub audio_file_count: i32, + pub error: Option, + pub created_at: String, + pub updated_at: String, + pub completed_at: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct YouTubeJobDto { + pub id: String, + pub source_url: String, + pub title: String, + pub source_kind: String, + pub status: String, + pub total_items: i32, + pub completed_items: i32, + pub failed_items: i32, + pub review_items: i32, + pub error: Option, + pub created_at: String, + pub updated_at: String, + pub completed_at: Option, + pub items: Vec, +} + +#[derive(Debug, Clone, FromRow)] +struct YouTubeJobRow { + id: String, + user_id: i64, + source_url: String, + title: String, + source_kind: String, + status: String, + total_items: i32, + completed_items: i32, + failed_items: i32, + review_items: i32, + error: Option, + created_at: String, + updated_at: String, + completed_at: Option, +} + +#[derive(Debug, Clone, FromRow)] +struct YouTubeItemRow { + id: String, + job_id: String, + source_id: String, + source_url: String, + title: String, + playlist_index: i32, + status: String, + progress_percent: f64, + downloaded_bytes: i64, + total_bytes: Option, + speed_bytes_per_sec: Option, + eta_seconds: Option, + chapter_count: i32, + audio_file_count: i32, + inbox_path: Option, + error: Option, + created_at: String, + updated_at: String, + completed_at: Option, +} + +impl YouTubeItemRow { + fn dto(&self) -> YouTubeItemDto { + YouTubeItemDto { + id: self.id.clone(), + source_id: self.source_id.clone(), + source_url: self.source_url.clone(), + title: self.title.clone(), + playlist_index: self.playlist_index, + status: self.status.clone(), + progress_percent: self.progress_percent.clamp(0.0, 100.0), + downloaded_bytes: non_negative(self.downloaded_bytes), + total_bytes: self.total_bytes.map(non_negative), + speed_bytes_per_sec: self.speed_bytes_per_sec.map(non_negative), + eta_seconds: self.eta_seconds.map(non_negative), + chapter_count: self.chapter_count, + audio_file_count: self.audio_file_count, + error: self.error.clone(), + created_at: self.created_at.clone(), + updated_at: self.updated_at.clone(), + completed_at: self.completed_at.clone(), + } + } +} + +impl YouTubeJobRow { + fn dto(&self, items: Vec) -> YouTubeJobDto { + YouTubeJobDto { + id: self.id.clone(), + source_url: self.source_url.clone(), + title: self.title.clone(), + source_kind: self.source_kind.clone(), + status: self.status.clone(), + total_items: self.total_items, + completed_items: self.completed_items, + failed_items: self.failed_items, + review_items: self.review_items, + error: self.error.clone(), + created_at: self.created_at.clone(), + updated_at: self.updated_at.clone(), + completed_at: self.completed_at.clone(), + items, + } + } +} + +#[derive(Debug)] +struct ResolvedSource { + title: String, + kind: String, + items: Vec, +} + +#[derive(Debug)] +struct ResolvedItem { + source_id: String, + source_url: String, + title: String, + playlist_index: i32, +} + +#[derive(Debug)] +struct PreparedFolder { + inbox_path: Option, + chapter_count: i32, + audio_file_count: i32, + all_files_known: bool, +} + +pub struct YouTubeService { + running_jobs: Mutex>, + cancellations: Mutex>, + concurrency: Arc, + scheduler_handle: Arc>>, +} + +impl YouTubeService { + pub fn new(scheduler_handle: Arc>>) -> Self { + Self { + running_jobs: Mutex::new(HashSet::new()), + cancellations: Mutex::new(HashMap::new()), + concurrency: Arc::new(Semaphore::new(2)), + scheduler_handle, + } + } + + pub async fn preview( + &self, + request: YouTubePreviewRequest, + ) -> anyhow::Result { + let url = validate_youtube_url(&request.url)?; + let resolved = resolve_source(&url).await?; + let requested_video_id = requested_video_id(&url); + let select_requested_only = resolved.kind == "playlist" && requested_video_id.is_some(); + Ok(YouTubePreviewDto { + source_url: url, + title: resolved.title, + source_kind: resolved.kind, + items: resolved + .items + .into_iter() + .map(|item| { + let selected_by_default = !select_requested_only + || requested_video_id.as_deref() == Some(item.source_id.as_str()); + YouTubePreviewItemDto { + source_id: item.source_id, + title: item.title, + playlist_index: item.playlist_index, + selected_by_default, + } + }) + .collect(), + }) + } + + pub async fn start( + self: &Arc, + pool: &PgPool, + user_id: i64, + request: YouTubeStartRequest, + inbox_dir: &str, + ) -> anyhow::Result { + let url = validate_youtube_url(&request.url)?; + validate_inbox_dir(inbox_dir)?; + let selected: HashSet = request + .selected_source_ids + .into_iter() + .map(|id| id.trim().to_string()) + .filter(|id| !id.is_empty()) + .collect(); + if selected.is_empty() { + bail!("select at least one YouTube video to import"); + } + if selected.iter().any(|id| !valid_source_id(id)) { + bail!("YouTube selection contains an invalid video ID"); + } + + let resolved = resolve_source(&url).await?; + let selected_items: Vec = resolved + .items + .into_iter() + .filter(|item| selected.contains(&item.source_id)) + .collect(); + if selected_items.len() != selected.len() { + bail!("YouTube selection no longer matches the parsed link; preview it again"); + } + + let id = Uuid::new_v4().to_string(); + let now = now_string(); + let source_ids: Vec = selected_items + .iter() + .map(|item| item.source_id.clone()) + .collect(); + let already_imported = already_imported_source_ids(pool, user_id, &source_ids).await?; + let mut transaction = pool.begin().await?; + sqlx::query( + r#"INSERT INTO furumusic__youtube_download + (id, user_id, source_url, title, source_kind, status, + total_items, completed_items, failed_items, review_items, + error, created_at, updated_at, completed_at) + VALUES ($1, $2, $3, $4, $5, 'queued', $6, 0, 0, 0, + NULL, $7, $7, NULL)"#, + ) + .bind(&id) + .bind(user_id) + .bind(&url) + .bind(&resolved.title) + .bind(&resolved.kind) + .bind(i32::try_from(selected_items.len()).unwrap_or(i32::MAX)) + .bind(&now) + .execute(&mut *transaction) + .await?; + + for item in selected_items { + let is_already_imported = already_imported.contains(&item.source_id); + let item_id = Uuid::new_v4().to_string(); + let status = if is_already_imported { + "skipped" + } else { + "queued" + }; + let progress = if is_already_imported { 100.0 } else { 0.0 }; + let completed_at = is_already_imported.then(|| now.clone()); + sqlx::query( + r#"INSERT INTO furumusic__youtube_download_item + (id, job_id, source_id, source_url, title, playlist_index, + status, progress_percent, downloaded_bytes, total_bytes, + speed_bytes_per_sec, eta_seconds, chapter_count, + audio_file_count, inbox_path, error, created_at, updated_at, + completed_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 0, NULL, NULL, + NULL, 0, 0, NULL, NULL, $9, $9, $10)"#, + ) + .bind(&item_id) + .bind(&id) + .bind(&item.source_id) + .bind(&item.source_url) + .bind(&item.title) + .bind(item.playlist_index) + .bind(status) + .bind(progress) + .bind(&now) + .bind(completed_at) + .execute(&mut *transaction) + .await?; + } + transaction.commit().await?; + + self.spawn_job(pool.clone(), id.clone(), inbox_dir.to_string()) + .await; + load_job_dto(pool, user_id, &id).await + } + + pub async fn list( + self: &Arc, + pool: &PgPool, + user_id: i64, + inbox_dir: &str, + ) -> anyhow::Result> { + validate_inbox_dir(inbox_dir)?; + sync_ai_statuses(pool, user_id).await?; + + let resumable: Vec<(String, String)> = sqlx::query_as( + r#"SELECT id, status::text + FROM furumusic__youtube_download + WHERE user_id = $1 + AND status IN ('queued', 'resolving', 'downloading', 'postprocessing') + ORDER BY created_at"#, + ) + .bind(user_id) + .fetch_all(pool) + .await?; + for (id, _) in resumable { + self.spawn_job(pool.clone(), id, inbox_dir.to_string()) + .await; + } + + let ids: Vec = sqlx::query_scalar( + r#"SELECT id FROM furumusic__youtube_download + WHERE user_id = $1 ORDER BY created_at DESC, id DESC LIMIT $2"#, + ) + .bind(user_id) + .bind(YOUTUBE_LIST_LIMIT) + .fetch_all(pool) + .await?; + for id in &ids { + refresh_parent(pool, id).await?; + } + + let mut jobs = Vec::with_capacity(ids.len()); + for id in ids { + jobs.push(load_job_dto(pool, user_id, &id).await?); + } + Ok(jobs) + } + + pub async fn retry( + self: &Arc, + pool: &PgPool, + user_id: i64, + id: &str, + inbox_dir: &str, + ) -> anyhow::Result { + validate_inbox_dir(inbox_dir)?; + let job = load_job_row(pool, user_id, id).await?; + if is_active_job_status(&job.status) { + bail!("YouTube download is already active"); + } + + let now = now_string(); + sqlx::query( + r#"UPDATE furumusic__youtube_download_item + SET status = 'queued', progress_percent = 0, downloaded_bytes = 0, + total_bytes = NULL, speed_bytes_per_sec = NULL, eta_seconds = NULL, + error = NULL, completed_at = NULL, updated_at = $2 + WHERE job_id = $1 AND status = 'failed'"#, + ) + .bind(id) + .bind(&now) + .execute(pool) + .await?; + sqlx::query( + r#"UPDATE furumusic__youtube_download + SET status = 'queued', error = NULL, completed_at = NULL, updated_at = $2 + WHERE id = $1 AND user_id = $3"#, + ) + .bind(id) + .bind(&now) + .bind(user_id) + .execute(pool) + .await?; + + self.spawn_job(pool.clone(), id.to_string(), inbox_dir.to_string()) + .await; + load_job_dto(pool, user_id, id).await + } + + pub async fn cancel( + &self, + pool: &PgPool, + user_id: i64, + id: &str, + ) -> anyhow::Result { + let job = load_job_row(pool, user_id, id).await?; + if job.status == "cancelled" { + return load_job_dto(pool, user_id, id).await; + } + if !is_cancellable_job_status(&job.status) { + bail!("this YouTube import can no longer be stopped"); + } + + let now = now_string(); + let mut transaction = pool.begin().await?; + sqlx::query( + r#"UPDATE furumusic__youtube_download_item + SET status = 'cancelled', speed_bytes_per_sec = NULL, + eta_seconds = NULL, error = NULL, completed_at = $2, + updated_at = $2 + WHERE job_id = $1 + AND status IN ('queued', 'downloading', 'postprocessing')"#, + ) + .bind(id) + .bind(&now) + .execute(&mut *transaction) + .await?; + sqlx::query( + r#"UPDATE furumusic__youtube_download + SET status = 'cancelled', error = NULL, completed_at = $2, + updated_at = $2 + WHERE id = $1 AND user_id = $3"#, + ) + .bind(id) + .bind(&now) + .bind(user_id) + .execute(&mut *transaction) + .await?; + transaction.commit().await?; + + if let Some(token) = self.cancellations.lock().await.get(id).cloned() { + token.cancel(); + } + refresh_parent(pool, id).await?; + load_job_dto(pool, user_id, id).await + } + + pub async fn remove( + &self, + pool: &PgPool, + user_id: i64, + id: &str, + inbox_dir: &str, + ) -> anyhow::Result<()> { + let job = load_job_row(pool, user_id, id).await?; + if is_active_job_status(&job.status) || self.running_jobs.lock().await.contains(id) { + bail!("active YouTube downloads cannot be removed"); + } + + let result = + sqlx::query("DELETE FROM furumusic__youtube_download WHERE id = $1 AND user_id = $2") + .bind(id) + .bind(user_id) + .execute(pool) + .await?; + if result.rows_affected() == 0 { + bail!("YouTube download not found"); + } + + if let Ok(inbox_root) = validate_inbox_dir(inbox_dir) { + let staging = staging_job_root(&inbox_root, id); + if tokio::fs::try_exists(&staging).await.unwrap_or(false) { + let _ = tokio::fs::remove_dir_all(staging).await; + } + } + Ok(()) + } + + async fn spawn_job(self: &Arc, pool: PgPool, id: String, inbox_dir: String) { + { + let mut running = self.running_jobs.lock().await; + if !running.insert(id.clone()) { + return; + } + } + let cancel = CancellationToken::new(); + self.cancellations + .lock() + .await + .insert(id.clone(), cancel.clone()); + + let service = Arc::clone(self); + tokio::spawn(async move { + let permit = tokio::select! { + permit = Arc::clone(&service.concurrency).acquire_owned() => permit.ok(), + _ = cancel.cancelled() => None, + }; + let result = if let Some(permit) = permit { + let result = service.run_job(&pool, &id, &inbox_dir, &cancel).await; + drop(permit); + result + } else { + Ok(()) + }; + if let Err(err) = result + && !cancel.is_cancelled() + { + tracing::error!(job_id = %id, error = %err, "YouTube download job failed"); + let _ = fail_parent(&pool, &id, &err.to_string()).await; + } + if cancel.is_cancelled() + && let Ok(inbox_root) = validate_inbox_dir(&inbox_dir) + { + let _ = tokio::fs::remove_dir_all(staging_job_root(&inbox_root, &id)).await; + } + service.running_jobs.lock().await.remove(&id); + service.cancellations.lock().await.remove(&id); + }); + } + + async fn run_job( + &self, + pool: &PgPool, + id: &str, + inbox_dir: &str, + cancel: &CancellationToken, + ) -> anyhow::Result<()> { + if cancel.is_cancelled() { + return Ok(()); + } + let inbox_root = validate_inbox_dir(inbox_dir)?; + let job: YouTubeJobRow = sqlx::query_as( + r#"SELECT id, user_id, source_url, title, source_kind, status, + total_items, completed_items, failed_items, review_items, + error, created_at, updated_at, completed_at + FROM furumusic__youtube_download WHERE id = $1"#, + ) + .bind(id) + .fetch_one(pool) + .await?; + + let mut items = load_items(pool, id).await?; + if items.is_empty() { + if cancel.is_cancelled() { + return Ok(()); + } + set_parent_status(pool, id, "resolving", None).await?; + let resolved = resolve_source(&job.source_url).await?; + if cancel.is_cancelled() { + return Ok(()); + } + let now = now_string(); + sqlx::query( + r#"UPDATE furumusic__youtube_download + SET title = $2, source_kind = $3, total_items = $4, + status = 'queued', error = NULL, updated_at = $5 + WHERE id = $1"#, + ) + .bind(id) + .bind(&resolved.title) + .bind(&resolved.kind) + .bind(i32::try_from(resolved.items.len()).unwrap_or(i32::MAX)) + .bind(&now) + .execute(pool) + .await?; + + for item in resolved.items { + let already_imported = + source_already_imported(pool, job.user_id, id, &item.source_id).await?; + let item_id = Uuid::new_v4().to_string(); + let status = if already_imported { + "skipped" + } else { + "queued" + }; + let progress = if already_imported { 100.0 } else { 0.0 }; + let completed_at = already_imported.then(|| now.clone()); + sqlx::query( + r#"INSERT INTO furumusic__youtube_download_item + (id, job_id, source_id, source_url, title, playlist_index, + status, progress_percent, downloaded_bytes, total_bytes, + speed_bytes_per_sec, eta_seconds, chapter_count, + audio_file_count, inbox_path, error, created_at, updated_at, + completed_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 0, NULL, NULL, + NULL, 0, 0, NULL, NULL, $9, $9, $10)"#, + ) + .bind(&item_id) + .bind(id) + .bind(&item.source_id) + .bind(&item.source_url) + .bind(&item.title) + .bind(item.playlist_index) + .bind(status) + .bind(progress) + .bind(&now) + .bind(completed_at) + .execute(pool) + .await?; + } + items = load_items(pool, id).await?; + } + + for item in items { + if cancel.is_cancelled() { + break; + } + if !matches!( + item.status.as_str(), + "queued" | "downloading" | "postprocessing" + ) { + continue; + } + if let Err(err) = self + .process_item(pool, &job, &item, &inbox_root, cancel) + .await + { + if cancel.is_cancelled() { + break; + } + tracing::warn!( + job_id = %id, + item_id = %item.id, + source_id = %item.source_id, + error = %err, + "YouTube playlist item failed" + ); + fail_item(pool, &item.id, &err.to_string()).await?; + } + refresh_parent(pool, id).await?; + } + + self.trigger_discover().await; + let _ = tokio::fs::remove_dir(staging_job_root(&inbox_root, id)).await; + refresh_parent(pool, id).await?; + Ok(()) + } + + async fn process_item( + &self, + pool: &PgPool, + job: &YouTubeJobRow, + item: &YouTubeItemRow, + inbox_root: &Path, + cancel: &CancellationToken, + ) -> anyhow::Result<()> { + if cancel.is_cancelled() { + bail!("YouTube import cancelled"); + } + set_item_status(pool, &item.id, "downloading", None).await?; + set_parent_status(pool, &job.id, "downloading", None).await?; + + let stage = staging_item_root(inbox_root, &job.id, &item.id); + tokio::fs::create_dir_all(&stage).await?; + run_ytdlp_download(pool, &item.id, &item.source_url, &stage, cancel).await?; + + if cancel.is_cancelled() { + bail!("YouTube import cancelled"); + } + + set_item_status(pool, &item.id, "postprocessing", None).await?; + set_parent_status(pool, &job.id, "postprocessing", None).await?; + let prepared = + prepare_downloaded_folder(pool, inbox_root, job.user_id, item, &stage, cancel).await?; + + let now = now_string(); + if prepared.all_files_known { + sqlx::query( + r#"UPDATE furumusic__youtube_download_item + SET status = 'skipped', progress_percent = 100, + chapter_count = $2, audio_file_count = 0, + inbox_path = NULL, error = NULL, completed_at = $3, + updated_at = $3 WHERE id = $1"#, + ) + .bind(&item.id) + .bind(prepared.chapter_count) + .bind(&now) + .execute(pool) + .await?; + return Ok(()); + } + + let inbox_path = prepared + .inbox_path + .context("prepared YouTube folder has no inbox path")?; + sqlx::query( + r#"UPDATE furumusic__youtube_download_item + SET status = 'awaiting_ai', progress_percent = 100, + chapter_count = $2, audio_file_count = $3, inbox_path = $4, + error = NULL, completed_at = NULL, updated_at = $5 + WHERE id = $1"#, + ) + .bind(&item.id) + .bind(prepared.chapter_count) + .bind(prepared.audio_file_count) + .bind(&inbox_path) + .bind(&now) + .execute(pool) + .await?; + self.trigger_discover().await; + Ok(()) + } + + async fn trigger_discover(&self) { + if let Some(handle) = self.scheduler_handle.get() { + let handle = Arc::clone(handle); + tokio::spawn(async move { + if let Err(err) = handle.trigger_job_now("inbox_discover").await { + tracing::warn!( + "failed to trigger inbox_discover after YouTube download: {err}" + ); + } + }); + } + } +} + +async fn resolve_source(url: &str) -> anyhow::Result { + let mut command = base_ytdlp_command(); + command + .arg("--flat-playlist") + .arg("--dump-single-json") + .arg("--skip-download") + .arg("--ignore-errors") + .arg("--no-warnings") + .arg("--") + .arg(url) + .kill_on_drop(true); + + let output = tokio::time::timeout(RESOLVE_TIMEOUT, command.output()) + .await + .context("yt-dlp metadata resolution timed out")??; + let stdout = String::from_utf8_lossy(&output.stdout); + if stdout.trim().is_empty() { + let stderr = String::from_utf8_lossy(&output.stderr); + bail!("yt-dlp could not resolve URL: {}", useful_error(&stderr)); + } + let value: serde_json::Value = + serde_json::from_str(stdout.trim()).context("yt-dlp returned invalid metadata JSON")?; + if value.is_null() { + bail!("yt-dlp could not resolve a public YouTube video or playlist"); + } + + let parent_title = json_string(&value, "title").unwrap_or_else(|| "YouTube download".into()); + if let Some(entries) = value.get("entries").and_then(|v| v.as_array()) { + let mut items = Vec::new(); + let mut seen = HashSet::new(); + for (index, entry) in entries.iter().enumerate() { + let Some(source_id) = json_string(entry, "id").filter(|id| valid_source_id(id)) else { + continue; + }; + if !seen.insert(source_id.clone()) { + continue; + } + let title = json_string(entry, "title").unwrap_or_else(|| source_id.clone()); + items.push(ResolvedItem { + source_url: format!("https://www.youtube.com/watch?v={source_id}"), + source_id, + title, + playlist_index: i32::try_from(index + 1).unwrap_or(i32::MAX), + }); + } + if items.is_empty() { + bail!("the playlist contains no available public YouTube videos"); + } + return Ok(ResolvedSource { + title: parent_title, + kind: "playlist".into(), + items, + }); + } + + let source_id = json_string(&value, "id") + .filter(|id| valid_source_id(id)) + .context("yt-dlp metadata has no valid YouTube video ID")?; + Ok(ResolvedSource { + title: parent_title.clone(), + kind: "video".into(), + items: vec![ResolvedItem { + source_url: format!("https://www.youtube.com/watch?v={source_id}"), + source_id, + title: parent_title, + playlist_index: 1, + }], + }) +} + +fn base_ytdlp_command() -> Command { + let mut command = Command::new("yt-dlp"); + command + .arg("--no-config") + .arg("--no-cookies") + .arg("--no-cookies-from-browser") + .arg("--js-runtimes") + .arg("deno") + .stdin(Stdio::null()); + command +} + +async fn run_ytdlp_download( + pool: &PgPool, + item_id: &str, + url: &str, + stage: &Path, + cancel: &CancellationToken, +) -> anyhow::Result<()> { + let mut command = base_ytdlp_command(); + command + .arg("--no-playlist") + .arg("--continue") + .arg("--newline") + .arg("--no-colors") + .arg("--progress-delta") + .arg("1") + .arg("--progress-template") + .arg("download:YT_PROGRESS|%(progress.downloaded_bytes)s|%(progress.total_bytes)s|%(progress.total_bytes_estimate)s|%(progress.speed)s|%(progress.eta)s") + .arg("--retries") + .arg("10") + .arg("--fragment-retries") + .arg("10") + .arg("--retry-sleep") + .arg("exp=1:20") + .arg("--socket-timeout") + .arg("30") + .arg("--sleep-requests") + .arg("1") + .arg("-f") + .arg("bestaudio/best") + .arg("--extract-audio") + .arg("--audio-format") + .arg("best") + .arg("--embed-metadata") + .arg("--no-embed-info-json") + .arg("--split-chapters") + .arg("--write-thumbnail") + .arg("--convert-thumbnails") + .arg("jpg") + .arg("--write-info-json") + .arg("--no-write-comments") + .arg("--windows-filenames") + .arg("-P") + .arg(stage) + .arg("-o") + .arg("__source__.%(ext)s") + .arg("-o") + .arg("chapter:%(section_number)03d - %(section_title).180B.%(ext)s") + .arg("-o") + .arg("thumbnail:cover.%(ext)s") + .arg("-o") + .arg("infojson:metadata.%(ext)s") + .arg("--") + .arg(url) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + configure_process_group(&mut command); + + let mut child = command + .spawn() + .context("could not start yt-dlp; install yt-dlp, FFmpeg/FFprobe, yt-dlp-ejs and Deno")?; + let stdout = child + .stdout + .take() + .context("yt-dlp stdout is unavailable")?; + let stderr = child + .stderr + .take() + .context("yt-dlp stderr is unavailable")?; + let stdout_task = tokio::spawn(read_ytdlp_output(stdout, pool.clone(), item_id.to_string())); + let stderr_task = tokio::spawn(read_ytdlp_output(stderr, pool.clone(), item_id.to_string())); + + let exit = tokio::select! { + exit = child.wait() => exit?, + _ = cancel.cancelled() => { + terminate_process_tree(&mut child).await; + let _ = stdout_task.await; + let _ = stderr_task.await; + bail!("YouTube import cancelled"); + } + }; + let stdout_lines = stdout_task.await??; + let stderr_lines = stderr_task.await??; + if !exit.success() { + let details = if stderr_lines.is_empty() { + stdout_lines.join("\n") + } else { + stderr_lines.join("\n") + }; + bail!("yt-dlp failed: {}", useful_error(&details)); + } + Ok(()) +} + +#[cfg(unix)] +fn configure_process_group(command: &mut Command) { + use std::os::unix::process::CommandExt as _; + command.as_std_mut().process_group(0); +} + +#[cfg(not(unix))] +fn configure_process_group(_command: &mut Command) {} + +async fn terminate_process_tree(child: &mut tokio::process::Child) { + #[cfg(unix)] + if let Some(process_group) = child.id().and_then(|id| i32::try_from(id).ok()) { + // yt-dlp launches FFmpeg as a child. Both processes are placed in their + // own group so stopping an import does not leave FFmpeg running. + // SAFETY: `kill` receives a checked positive process-group ID negated + // according to POSIX; it does not dereference application memory. + unsafe { + libc::kill(-process_group, libc::SIGTERM); + } + tokio::time::sleep(Duration::from_millis(300)).await; + // SAFETY: same process-group ID and POSIX contract as above. + unsafe { + libc::kill(-process_group, libc::SIGKILL); + } + } + let _ = child.kill().await; + let _ = child.wait().await; +} + +async fn read_ytdlp_output( + reader: R, + pool: PgPool, + item_id: String, +) -> anyhow::Result> { + let mut lines = BufReader::new(reader).lines(); + let mut tail = VecDeque::with_capacity(30); + while let Some(line) = lines.next_line().await? { + if let Some(progress) = parse_progress_line(&line) { + let _ = persist_item_progress(&pool, &item_id, progress).await; + } else if is_postprocessing_line(&line) { + let _ = set_item_status(&pool, &item_id, "postprocessing", None).await; + } + if !line.trim().is_empty() { + if tail.len() == 30 { + tail.pop_front(); + } + tail.push_back(line); + } + } + Ok(tail.into_iter().collect()) +} + +#[derive(Debug, PartialEq)] +struct DownloadProgress { + downloaded: i64, + total: Option, + speed: Option, + eta: Option, + percent: f64, +} + +fn parse_progress_line(line: &str) -> Option { + let payload = line.trim().strip_prefix("YT_PROGRESS|")?; + let mut fields = payload.split('|'); + let downloaded = parse_number(fields.next()?)?; + let exact_total = parse_number(fields.next().unwrap_or("")); + let estimated_total = parse_number(fields.next().unwrap_or("")); + let total = exact_total.or(estimated_total).filter(|v| *v > 0); + let speed = parse_number(fields.next().unwrap_or("")); + let eta = parse_number(fields.next().unwrap_or("")); + let percent = total + .map(|total| downloaded as f64 / total as f64 * 100.0) + .unwrap_or(0.0) + .clamp(0.0, 100.0); + Some(DownloadProgress { + downloaded, + total, + speed, + eta, + percent, + }) +} + +fn parse_number(value: &str) -> Option { + let value = value.trim(); + if value.is_empty() || value.eq_ignore_ascii_case("NA") || value.eq_ignore_ascii_case("none") { + return None; + } + value + .parse::() + .ok() + .filter(|v| v.is_finite() && *v >= 0.0) + .map(|v| v.round() as i64) +} + +fn is_postprocessing_line(line: &str) -> bool { + let lower = line.to_ascii_lowercase(); + [ + "[extractaudio]", + "[splitchapters]", + "[thumbnailconvertor]", + "[metadata]", + "splitting video by chapters", + ] + .iter() + .any(|marker| lower.contains(marker)) +} + +async fn persist_item_progress( + pool: &PgPool, + item_id: &str, + progress: DownloadProgress, +) -> anyhow::Result<()> { + sqlx::query( + r#"UPDATE furumusic__youtube_download_item + SET progress_percent = $2, downloaded_bytes = $3, total_bytes = $4, + speed_bytes_per_sec = $5, eta_seconds = $6, updated_at = $7 + WHERE id = $1"#, + ) + .bind(item_id) + .bind(progress.percent) + .bind(progress.downloaded) + .bind(progress.total) + .bind(progress.speed) + .bind(progress.eta) + .bind(now_string()) + .execute(pool) + .await?; + Ok(()) +} + +async fn prepare_downloaded_folder( + pool: &PgPool, + inbox_root: &Path, + user_id: i64, + item: &YouTubeItemRow, + stage: &Path, + cancel: &CancellationToken, +) -> anyhow::Result { + if cancel.is_cancelled() { + bail!("YouTube import cancelled"); + } + let metadata = read_info_json(stage).await?; + let chapter_count = metadata + .as_ref() + .and_then(|value| value.get("chapters")) + .and_then(|value| value.as_array()) + .map(|chapters| i32::try_from(chapters.len()).unwrap_or(i32::MAX)) + .unwrap_or(0); + + let mut audio_files = find_audio_files(stage).await?; + if chapter_count > 0 { + for source in audio_files + .iter() + .filter(|path| file_name(path).starts_with("__source__.")) + { + tokio::fs::remove_file(source).await?; + } + } else if let Some(source) = audio_files + .iter() + .find(|path| file_name(path).starts_with("__source__.")) + .cloned() + { + let extension = source + .extension() + .and_then(|v| v.to_str()) + .unwrap_or("opus"); + let destination = stage.join(format!( + "{} [{}].{}", + sanitize_component(&item.title), + item.source_id, + extension + )); + if source != destination { + tokio::fs::rename(&source, &destination).await?; + } + } + + normalize_cover(stage).await; + cleanup_sidecars(stage).await?; + audio_files = find_audio_files(stage).await?; + if audio_files.is_empty() { + bail!("yt-dlp produced no supported audio files"); + } + + for audio in &audio_files { + let data = tokio::select! { + data = tokio::fs::read(audio) => data?, + _ = cancel.cancelled() => bail!("YouTube import cancelled"), + }; + let hash = format!("{:x}", Sha256::digest(&data)); + if crate::agent::rag::file_hash_exists(pool, &hash) + .await + .unwrap_or(false) + { + tokio::fs::remove_file(audio).await?; + } + } + audio_files = find_audio_files(stage).await?; + if audio_files.is_empty() { + tokio::fs::remove_dir_all(stage).await?; + return Ok(PreparedFolder { + inbox_path: None, + chapter_count, + audio_file_count: 0, + all_files_known: true, + }); + } + + if cancel.is_cancelled() { + bail!("YouTube import cancelled"); + } + let folder_name = format!("{} [{}]", sanitize_component(&item.title), item.source_id); + let destination = inbox_root + .join("user_uploads") + .join(user_id.to_string()) + .join(folder_name); + if let Some(parent) = destination.parent() { + tokio::fs::create_dir_all(parent).await?; + } + if tokio::fs::try_exists(&destination).await? { + let existing = find_audio_files(&destination).await?; + if existing.is_empty() { + bail!("YouTube inbox destination already exists without audio"); + } + tokio::fs::remove_dir_all(stage).await?; + audio_files = existing; + } else { + tokio::fs::rename(stage, &destination).await?; + } + + let inbox_root_text = inbox_root.to_string_lossy(); + let inbox_path = crate::media_paths::path_for_root(&inbox_root_text, &destination) + .context("YouTube destination escaped agent_inbox_dir")?; + Ok(PreparedFolder { + inbox_path: Some(inbox_path), + chapter_count, + audio_file_count: i32::try_from(audio_files.len()).unwrap_or(i32::MAX), + all_files_known: false, + }) +} + +async fn read_info_json(stage: &Path) -> anyhow::Result> { + let mut entries = tokio::fs::read_dir(stage).await?; + while let Some(entry) = entries.next_entry().await? { + if !entry.file_type().await?.is_file() { + continue; + } + let name = entry.file_name().to_string_lossy().to_string(); + if name.ends_with(".info.json") || name == "metadata.json" { + let data = tokio::fs::read(entry.path()).await?; + return Ok(Some( + serde_json::from_slice(&data).context("invalid yt-dlp info JSON")?, + )); + } + } + Ok(None) +} + +async fn find_audio_files(dir: &Path) -> anyhow::Result> { + let mut files = Vec::new(); + let mut entries = tokio::fs::read_dir(dir).await?; + while let Some(entry) = entries.next_entry().await? { + if entry.file_type().await?.is_file() && is_audio_path(&entry.path()) { + files.push(entry.path()); + } + } + files.sort(); + Ok(files) +} + +async fn normalize_cover(stage: &Path) { + let cover_path = stage.join("cover.jpg"); + if tokio::fs::try_exists(&cover_path).await.unwrap_or(false) { + return; + } + let Ok(mut entries) = tokio::fs::read_dir(stage).await else { + return; + }; + let mut candidate = None; + while let Ok(Some(entry)) = entries.next_entry().await { + let path = entry.path(); + let is_image = path + .extension() + .and_then(|v| v.to_str()) + .map(|ext| IMAGE_EXTENSIONS.contains(&ext.to_ascii_lowercase().as_str())) + .unwrap_or(false); + if is_image { + candidate = Some(path); + break; + } + } + let Some(candidate) = candidate else { + return; + }; + let Ok(data) = tokio::fs::read(&candidate).await else { + return; + }; + let encoded = tokio::task::spawn_blocking(move || -> anyhow::Result> { + let image = image::load_from_memory(&data)?.to_rgb8(); + let mut output = Vec::new(); + JpegEncoder::new_with_quality(&mut output, 90).encode( + &image, + image.width(), + image.height(), + image::ExtendedColorType::Rgb8, + )?; + Ok(output) + }) + .await; + if let Ok(Ok(encoded)) = encoded + && tokio::fs::write(&cover_path, encoded).await.is_ok() + && candidate != cover_path + { + let _ = tokio::fs::remove_file(candidate).await; + } +} + +async fn cleanup_sidecars(stage: &Path) -> anyhow::Result<()> { + let mut entries = tokio::fs::read_dir(stage).await?; + while let Some(entry) = entries.next_entry().await? { + if !entry.file_type().await?.is_file() { + continue; + } + let path = entry.path(); + if is_audio_path(&path) || file_name(&path).eq_ignore_ascii_case("cover.jpg") { + continue; + } + tokio::fs::remove_file(path).await?; + } + Ok(()) +} + +fn is_audio_path(path: &Path) -> bool { + path.extension() + .and_then(|value| value.to_str()) + .map(|value| AUDIO_EXTENSIONS.contains(&value.to_ascii_lowercase().as_str())) + .unwrap_or(false) +} + +async fn sync_ai_statuses(pool: &PgPool, user_id: i64) -> anyhow::Result<()> { + let items: Vec = sqlx::query_as( + r#"SELECT i.id, i.job_id, i.source_id, i.source_url, i.title, + i.playlist_index, i.status, i.progress_percent, + i.downloaded_bytes, i.total_bytes, i.speed_bytes_per_sec, + i.eta_seconds, i.chapter_count, i.audio_file_count, + i.inbox_path, i.error, i.created_at, i.updated_at, i.completed_at + FROM furumusic__youtube_download_item i + JOIN furumusic__youtube_download j ON j.id = i.job_id + WHERE j.user_id = $1 + AND i.status IN ('awaiting_ai', 'ai_processing', 'needs_review', 'ai_failed') + AND i.inbox_path IS NOT NULL"#, + ) + .bind(user_id) + .fetch_all(pool) + .await?; + + let mut touched_jobs = HashSet::new(); + for item in items { + let Some(prefix) = item.inbox_path.as_deref() else { + continue; + }; + let states: Vec<(String, i64)> = sqlx::query_as( + r#"SELECT status::text, COUNT(*) + FROM furumusic__pending_review + WHERE input_path = $1 + OR left(input_path, length($1) + 1) = $1 || '/' + GROUP BY status"#, + ) + .bind(prefix) + .fetch_all(pool) + .await?; + let counts: HashMap = states.into_iter().collect(); + let total: i64 = counts.values().sum(); + let expected = i64::from(item.audio_file_count.max(0)); + let active_processing = counts.get("processing").copied().unwrap_or(0) > 0; + let queued = counts.get("queued").copied().unwrap_or(0) > 0; + + let (next, error) = if total < expected || total == 0 { + ( + if active_processing { + "ai_processing" + } else { + "awaiting_ai" + }, + None, + ) + } else if active_processing { + ("ai_processing", None) + } else if queued { + ("awaiting_ai", None) + } else if counts.get("failed").copied().unwrap_or(0) > 0 { + let error: Option = sqlx::query_scalar( + r#"SELECT error_message FROM furumusic__pending_review + WHERE (input_path = $1 OR left(input_path, length($1) + 1) = $1 || '/') + AND status = 'failed' AND error_message IS NOT NULL + ORDER BY id DESC LIMIT 1"#, + ) + .bind(prefix) + .fetch_optional(pool) + .await? + .flatten(); + ("ai_failed", error) + } else if counts.get("pending").copied().unwrap_or(0) > 0 + || counts.get("rejected").copied().unwrap_or(0) > 0 + { + ("needs_review", None) + } else { + ("complete", None) + }; + + if item.status != next || item.error != error { + let now = now_string(); + let completed_at = + matches!(next, "complete" | "needs_review" | "ai_failed").then(|| now.clone()); + sqlx::query( + r#"UPDATE furumusic__youtube_download_item + SET status = $2, error = $3, completed_at = $4, updated_at = $5 + WHERE id = $1"#, + ) + .bind(&item.id) + .bind(next) + .bind(error) + .bind(completed_at) + .bind(&now) + .execute(pool) + .await?; + touched_jobs.insert(item.job_id); + } + } + for job_id in touched_jobs { + refresh_parent(pool, &job_id).await?; + } + Ok(()) +} + +async fn refresh_parent(pool: &PgPool, job_id: &str) -> anyhow::Result<()> { + let parent_status: Option = + sqlx::query_scalar("SELECT status::text FROM furumusic__youtube_download WHERE id = $1") + .bind(job_id) + .fetch_optional(pool) + .await?; + let states: Vec<(String, i64)> = sqlx::query_as( + r#"SELECT status::text, COUNT(*) + FROM furumusic__youtube_download_item WHERE job_id = $1 GROUP BY status"#, + ) + .bind(job_id) + .fetch_all(pool) + .await?; + if states.is_empty() { + return Ok(()); + } + let counts: HashMap = states.into_iter().collect(); + let count = |status: &str| counts.get(status).copied().unwrap_or(0); + let total: i64 = counts.values().sum(); + let failed = count("failed") + count("ai_failed"); + let review = count("needs_review"); + let completed = count("complete") + count("skipped") + review; + + let status = if parent_status.as_deref() == Some("cancelled") { + "cancelled" + } else if count("downloading") > 0 || count("queued") > 0 { + "downloading" + } else if count("postprocessing") > 0 { + "postprocessing" + } else if count("ai_processing") > 0 { + "ai_processing" + } else if count("awaiting_ai") > 0 { + "awaiting_ai" + } else if failed == total { + "failed" + } else if failed > 0 { + "complete_with_errors" + } else if review > 0 { + "needs_review" + } else { + "complete" + }; + let now = now_string(); + let completed_at = matches!( + status, + "complete" | "complete_with_errors" | "failed" | "needs_review" | "cancelled" + ) + .then(|| now.clone()); + sqlx::query( + r#"UPDATE furumusic__youtube_download + SET status = $2, total_items = $3, completed_items = $4, + failed_items = $5, review_items = $6, completed_at = $7, + updated_at = $8 WHERE id = $1"#, + ) + .bind(job_id) + .bind(status) + .bind(i32::try_from(total).unwrap_or(i32::MAX)) + .bind(i32::try_from(completed).unwrap_or(i32::MAX)) + .bind(i32::try_from(failed).unwrap_or(i32::MAX)) + .bind(i32::try_from(review).unwrap_or(i32::MAX)) + .bind(completed_at) + .bind(now) + .execute(pool) + .await?; + Ok(()) +} + +async fn source_already_imported( + pool: &PgPool, + user_id: i64, + current_job_id: &str, + source_id: &str, +) -> anyhow::Result { + let exists: bool = sqlx::query_scalar( + r#"SELECT EXISTS ( + SELECT 1 + FROM furumusic__youtube_download_item i + JOIN furumusic__youtube_download j ON j.id = i.job_id + WHERE j.user_id = $1 AND i.job_id <> $2 AND i.source_id = $3 + AND i.status IN ( + 'queued', 'downloading', 'postprocessing', 'awaiting_ai', + 'ai_processing', 'complete', 'needs_review', 'skipped' + ) + )"#, + ) + .bind(user_id) + .bind(current_job_id) + .bind(source_id) + .fetch_one(pool) + .await?; + Ok(exists) +} + +async fn already_imported_source_ids( + pool: &PgPool, + user_id: i64, + source_ids: &[String], +) -> anyhow::Result> { + let ids: Vec = sqlx::query_scalar( + r#"SELECT DISTINCT i.source_id + FROM furumusic__youtube_download_item i + JOIN furumusic__youtube_download j ON j.id = i.job_id + WHERE j.user_id = $1 AND i.source_id = ANY($2) + AND i.status IN ( + 'queued', 'downloading', 'postprocessing', 'awaiting_ai', + 'ai_processing', 'complete', 'needs_review', 'skipped' + )"#, + ) + .bind(user_id) + .bind(source_ids) + .fetch_all(pool) + .await?; + Ok(ids.into_iter().collect()) +} + +async fn load_job_dto(pool: &PgPool, user_id: i64, id: &str) -> anyhow::Result { + let row = load_job_row(pool, user_id, id).await?; + let items = load_items(pool, id) + .await? + .iter() + .map(YouTubeItemRow::dto) + .collect(); + Ok(row.dto(items)) +} + +async fn load_job_row(pool: &PgPool, user_id: i64, id: &str) -> anyhow::Result { + sqlx::query_as( + r#"SELECT id, user_id, source_url, title, source_kind, status, + total_items, completed_items, failed_items, review_items, + error, created_at, updated_at, completed_at + FROM furumusic__youtube_download WHERE id = $1 AND user_id = $2"#, + ) + .bind(id) + .bind(user_id) + .fetch_optional(pool) + .await? + .context("YouTube download not found") +} + +async fn load_items(pool: &PgPool, job_id: &str) -> anyhow::Result> { + Ok(sqlx::query_as( + r#"SELECT id, job_id, source_id, source_url, title, playlist_index, + status, progress_percent, downloaded_bytes, total_bytes, + speed_bytes_per_sec, eta_seconds, chapter_count, + audio_file_count, inbox_path, error, created_at, updated_at, + completed_at + FROM furumusic__youtube_download_item WHERE job_id = $1 + ORDER BY playlist_index, id"#, + ) + .bind(job_id) + .fetch_all(pool) + .await?) +} + +async fn set_parent_status( + pool: &PgPool, + id: &str, + status: &str, + error: Option<&str>, +) -> anyhow::Result<()> { + sqlx::query( + r#"UPDATE furumusic__youtube_download + SET status = $2, error = $3, updated_at = $4 + WHERE id = $1 AND status <> 'cancelled'"#, + ) + .bind(id) + .bind(status) + .bind(error.map(trim_error)) + .bind(now_string()) + .execute(pool) + .await?; + Ok(()) +} + +async fn set_item_status( + pool: &PgPool, + id: &str, + status: &str, + error: Option<&str>, +) -> anyhow::Result<()> { + sqlx::query( + r#"UPDATE furumusic__youtube_download_item + SET status = $2, error = $3, updated_at = $4 + WHERE id = $1 AND status <> 'cancelled'"#, + ) + .bind(id) + .bind(status) + .bind(error.map(trim_error)) + .bind(now_string()) + .execute(pool) + .await?; + Ok(()) +} + +async fn fail_item(pool: &PgPool, id: &str, error: &str) -> anyhow::Result<()> { + let now = now_string(); + sqlx::query( + r#"UPDATE furumusic__youtube_download_item + SET status = 'failed', error = $2, speed_bytes_per_sec = NULL, + eta_seconds = NULL, completed_at = $3, updated_at = $3 + WHERE id = $1"#, + ) + .bind(id) + .bind(trim_error(error)) + .bind(now) + .execute(pool) + .await?; + Ok(()) +} + +async fn fail_parent(pool: &PgPool, id: &str, error: &str) -> anyhow::Result<()> { + let now = now_string(); + sqlx::query( + r#"UPDATE furumusic__youtube_download + SET status = 'failed', error = $2, completed_at = $3, updated_at = $3 + WHERE id = $1"#, + ) + .bind(id) + .bind(trim_error(error)) + .bind(now) + .execute(pool) + .await?; + Ok(()) +} + +fn validate_youtube_url(value: &str) -> anyhow::Result { + let value = value.trim(); + if value.is_empty() { + bail!("YouTube URL is empty"); + } + let url = reqwest::Url::parse(value).context("invalid YouTube URL")?; + if !matches!(url.scheme(), "http" | "https") { + bail!("only HTTP and HTTPS YouTube URLs are supported"); + } + if !url.username().is_empty() || url.password().is_some() { + bail!("YouTube URL must not contain credentials"); + } + let host = url + .host_str() + .map(|host| host.trim_end_matches('.').to_ascii_lowercase()) + .context("YouTube URL has no host")?; + let allowed = host == "youtu.be" || host == "youtube.com" || host.ends_with(".youtube.com"); + if !allowed { + bail!("only youtube.com, music.youtube.com and youtu.be URLs are supported"); + } + Ok(url.to_string()) +} + +fn validate_inbox_dir(value: &str) -> anyhow::Result { + let value = value.trim(); + if value.is_empty() { + bail!("agent_inbox_dir is not configured"); + } + let path = crate::media_paths::resolve_config_path_buf(value); + if !path.is_absolute() { + bail!("agent_inbox_dir must be an absolute path"); + } + Ok(path) +} + +fn staging_job_root(inbox_root: &Path, job_id: &str) -> PathBuf { + inbox_root.join(".downloads").join("youtube").join(job_id) +} + +fn staging_item_root(inbox_root: &Path, job_id: &str, item_id: &str) -> PathBuf { + staging_job_root(inbox_root, job_id).join(item_id) +} + +fn sanitize_component(value: &str) -> String { + let value: String = value + .chars() + .take(160) + .map(|character| match character { + '/' | '\\' | ':' | '*' | '?' | '"' | '<' | '>' | '|' => '_', + character if character.is_control() => '_', + character => character, + }) + .collect(); + let value = value.trim().trim_matches('.').trim(); + if value.is_empty() { + "YouTube audio".to_string() + } else { + value.to_string() + } +} + +fn valid_source_id(value: &str) -> bool { + !value.is_empty() + && value.len() <= 128 + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) +} + +fn requested_video_id(value: &str) -> Option { + let url = reqwest::Url::parse(value).ok()?; + let host = url.host_str()?.trim_end_matches('.').to_ascii_lowercase(); + let candidate = if host == "youtu.be" { + url.path_segments()? + .find(|segment| !segment.is_empty()) + .map(str::to_string) + } else if url.path() == "/watch" { + url.query_pairs() + .find_map(|(key, value)| (key == "v").then_some(value.into_owned())) + } else { + let mut segments = url.path_segments()?; + match (segments.next(), segments.next()) { + (Some("embed" | "live" | "shorts"), Some(id)) => Some(id.to_string()), + _ => None, + } + }?; + valid_source_id(&candidate).then_some(candidate) +} + +fn json_string(value: &serde_json::Value, key: &str) -> Option { + value + .get(key) + .and_then(|value| value.as_str()) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_string) +} + +fn file_name(path: &Path) -> String { + path.file_name() + .and_then(|value| value.to_str()) + .unwrap_or_default() + .to_string() +} + +fn is_active_job_status(status: &str) -> bool { + matches!( + status, + "queued" | "resolving" | "downloading" | "postprocessing" | "awaiting_ai" | "ai_processing" + ) +} + +fn is_cancellable_job_status(status: &str) -> bool { + matches!( + status, + "queued" | "resolving" | "downloading" | "postprocessing" + ) +} + +fn useful_error(value: &str) -> String { + let lines: Vec<&str> = value + .lines() + .map(str::trim) + .filter(|line| !line.is_empty()) + .collect(); + trim_error(lines.last().copied().unwrap_or("unknown error")) +} + +fn trim_error(value: &str) -> String { + value.chars().take(MAX_ERROR_LEN).collect() +} + +fn non_negative(value: i64) -> u64 { + u64::try_from(value).unwrap_or(0) +} + +fn now_string() -> String { + chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn validates_supported_youtube_hosts() { + assert!(validate_youtube_url("https://youtube.com/watch?v=abc").is_ok()); + assert!(validate_youtube_url("https://www.youtube.com/playlist?list=abc").is_ok()); + assert!(validate_youtube_url("https://music.youtube.com/watch?v=abc").is_ok()); + assert!(validate_youtube_url("https://youtu.be/abc").is_ok()); + } + + #[test] + fn rejects_non_youtube_and_lookalike_hosts() { + assert!(validate_youtube_url("https://example.com/video").is_err()); + assert!(validate_youtube_url("https://youtube.com.example.com/video").is_err()); + assert!(validate_youtube_url("file:///etc/passwd").is_err()); + assert!(validate_youtube_url("https://user@youtube.com/video").is_err()); + } + + #[test] + fn parses_machine_readable_progress() { + let progress = parse_progress_line("YT_PROGRESS|500|1000|NA|250|2").unwrap(); + assert_eq!( + progress, + DownloadProgress { + downloaded: 500, + total: Some(1000), + speed: Some(250), + eta: Some(2), + percent: 50.0, + } + ); + } + + #[test] + fn sanitizes_download_folder_names() { + assert_eq!( + sanitize_component(" Album: Live?/Test "), + "Album_ Live__Test" + ); + assert_eq!(sanitize_component("..."), "YouTube audio"); + } + + #[test] + fn extracts_explicit_video_from_playlist_links() { + assert_eq!( + requested_video_id("https://www.youtube.com/watch?v=qZ4PNyZGSJ8&list=RDqZ4PNyZGSJ8") + .as_deref(), + Some("qZ4PNyZGSJ8") + ); + assert_eq!( + requested_video_id("https://youtu.be/qZ4PNyZGSJ8?list=RDqZ4PNyZGSJ8").as_deref(), + Some("qZ4PNyZGSJ8") + ); + assert_eq!( + requested_video_id("https://youtube.com/playlist?list=PL123"), + None + ); + } +} diff --git a/templates/player/modals.html b/templates/player/modals.html index d5b92e3..6d81fb3 100644 --- a/templates/player/modals.html +++ b/templates/player/modals.html @@ -85,7 +85,7 @@ - +