diff --git a/crates/biliup-cli/src/server/common/upload.rs b/crates/biliup-cli/src/server/common/upload.rs index c431c4b..ba13cd4 100644 --- a/crates/biliup-cli/src/server/common/upload.rs +++ b/crates/biliup-cli/src/server/common/upload.rs @@ -5,7 +5,9 @@ use crate::server::core::downloader::SegmentInfo; use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::context::{Context, Stage, Worker, WorkerStatus}; use crate::server::infrastructure::models::InsertFileItem; -use crate::server::infrastructure::models::hook_step::process_video; +use crate::server::infrastructure::models::hook_step::{ + HookStep, process_video, process_video_paths, +}; use crate::server::infrastructure::models::upload_streamer::UploadStreamer; use async_channel::Receiver; use biliup::bilibili::{BiliBili, ResponseData, Studio, Video}; @@ -52,8 +54,15 @@ where let upload_context = initialize_upload_context(&ctx.config(), &ctx.stateless_client(), upload_config).await?; - // 2. 流水线处理视频上传 - let uploaded_videos = pipeline_upload_videos(rx, &upload_context).await?; + // 2. 流水线处理视频上传（segment_processor 在每段上传前执行；用于 Remux 等 + // 在原地改写路径的预处理） + let segment_processors: Vec = ctx + .live_streamer() + .segment_processor + .clone() + .unwrap_or_default(); + let uploaded_videos = + pipeline_upload_videos(rx, &upload_context, &segment_processors).await?; // 3. 提交到B站 if !uploaded_videos.videos.is_empty() { @@ -122,27 +131,46 @@ async fn get_upload_line(client: &reqwest::Client, line: &str) -> AppResult( rx: Inspect, F>, context: &UploadContext, + segment_processors: &[HookStep], ) -> AppResult where F: FnMut(&SegmentInfo), { - // let mut desc_v2 = Vec::new(); - // for credit in context.upload_config.desc_v2_credit { - // desc_v2.push(Credit { - // type_id: credit.type_id, - // raw_text: credit.raw_text, - // biz_id: credit.biz_id, - // }); - // } - let mut uploaded = UploadedVideos::default(); pin!(rx); // 流式处理后续事件 while let Some(event) = rx.next().await { - let video = upload_single_file(&event.prev_file_path, context).await?; - uploaded.videos.push(video); - uploaded.paths.push(event.prev_file_path); - // 失败的文件不加入路径列表，避免后处理出错 + // segment_processor 在上传前对路径列表做就地转换（如 Remux .ts→.mp4）。 + // 单段失败（典型场景：磁盘满让 ffmpeg remux 写头失败）不应拖死整批—— + // 否则已成功上传的段也无法到达 submit + postprocessor，本地 `rm` 不触发， + // 文件越堆越多，磁盘进一步紧张，形成正反馈。 + let mut paths: Vec = vec![event.prev_file_path.clone()]; + if !segment_processors.is_empty() + && let Err(e) = process_video_paths(&mut paths, segment_processors).await + { + error!( + file = ?event.prev_file_path, + "segment_processor failed, skipping segment: {:?}", e + ); + continue; + } + let upload_path = paths + .first() + .cloned() + .unwrap_or_else(|| event.prev_file_path.clone()); + match upload_single_file(&upload_path, context).await { + Ok(video) => { + uploaded.videos.push(video); + // postprocessor 看到的是 segment_processor 转换后的路径 + uploaded.paths.push(upload_path); + } + Err(e) => { + error!( + file = ?upload_path, + "upload_single_file failed, skipping segment: {:?}", e + ); + } + } } Ok(uploaded) diff --git a/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs b/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs index 30af9fc..dfd6680 100644 --- a/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs +++ b/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs @@ -2,11 +2,12 @@ use crate::server::errors::{AppError, AppResult}; use error_stack::{ResultExt, bail}; use serde::{Deserialize, Serialize}; use std::borrow::Cow; -use std::path::Path; +use std::path::{Path, PathBuf}; +use std::time::Duration; use tokio::fs; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::process::Command; -use tracing::{error, info}; +use tracing::{error, info, warn}; /// 钩子步骤枚举：支持多种操作格式 /// 既支持 key-value 形式（如 {run: "..."}），也支持纯字符串（如 "rm"） @@ -17,6 +18,12 @@ pub enum HookStep { Run { run: String }, /// 移动文件格式：{mv: "target_dir"} Move { mv: String }, + /// ts→mp4 无重编码 remux 格式：{remux: "mp4"} + /// 解决 B 站对 .ts 直传时常见的 "时间戳跳变" 转码失败。 + /// 用 ffmpeg `-c copy -fflags +genpts+igndts -bsf:a aac_adtstoasc -movflags +faststart`， + /// 重新生成 PTS。仅当输入扩展名为 .ts/.m2ts 时生效；其它格式跳过。 + /// 路径列表会被原地替换为新生成的 .mp4，原文件保留（postprocessor 中再 "rm"）。 + Remux { remux: String }, /// 删除文件格式："rm" Remove(String), } @@ -44,6 +51,13 @@ impl HookStep { // 移动文件到指定目录 self.move_file(video_paths, mv).await?; } + HookStep::Remux { remux } => { + bail!(AppError::Custom(format!( + "remux step needs the path-mutating execute_paths API; \ + called via the legacy execute path with target={}", + remux + ))); + } HookStep::Remove(cmd) if cmd == "rm" => { // 删除文件 HookStep::remove_file(video_paths).await?; @@ -56,6 +70,110 @@ impl HookStep { Ok(()) } + /// Path-mutating variant: lets steps swap entries in the working set + /// (e.g. Remux replaces .ts with the newly-created .mp4). + pub async fn execute_paths(&self, video_paths: &mut Vec) -> AppResult<()> { + match self { + HookStep::Remux { remux } => { + let target = remux.to_lowercase(); + if target != "mp4" { + bail!(AppError::Custom(format!( + "remux: only target=\"mp4\" supported, got {}", + remux + ))); + } + for p in video_paths.iter_mut() { + let ext = p + .extension() + .and_then(|s| s.to_str()) + .map(|s| s.to_lowercase()) + .unwrap_or_default(); + if ext != "ts" && ext != "m2ts" { + info!("remux: skipping non-ts file {}", p.display()); + continue; + } + let mp4 = p.with_extension("mp4"); + if mp4.exists() && tokio::fs::metadata(&mp4).await + .map(|m| m.len() > 0) + .unwrap_or(false) + { + info!("remux: reusing existing {}", mp4.display()); + *p = mp4; + continue; + } + Self::ffmpeg_remux_to_mp4(p, &mp4).await?; + *p = mp4; + } + } + other => { + // For non-path-mutating steps, fall back to the read-only API. + let refs: Vec<&Path> = video_paths.iter().map(|p| p.as_path()).collect(); + other.execute(&refs).await?; + } + } + Ok(()) + } + + async fn ffmpeg_remux_to_mp4(src: &Path, dst: &Path) -> AppResult<()> { + info!("remux ts→mp4: {} → {}", src.display(), dst.display()); + let started = std::time::Instant::now(); + let status = Command::new("ffmpeg") + .args([ + "-hide_banner", + "-loglevel", + "warning", + "-y", + "-fflags", + "+genpts+igndts", + "-i", + ]) + .arg(src) + .args([ + "-c", + "copy", + "-bsf:a", + "aac_adtstoasc", + "-movflags", + "+faststart", + "-avoid_negative_ts", + "make_zero", + ]) + .arg(dst) + .kill_on_drop(true) + .status() + .await + .change_context(AppError::Custom("failed to spawn ffmpeg".into()))?; + if !status.success() { + // Clean up partial output so a retry restarts cleanly. + let _ = tokio::fs::remove_file(dst).await; + bail!(AppError::Custom(format!( + "ffmpeg remux failed (status {:?}) for {}", + status, + src.display() + ))); + } + let meta = tokio::fs::metadata(dst).await.change_context_lazy(|| { + AppError::Custom(format!("remux output missing: {}", dst.display())) + })?; + if meta.len() == 0 { + let _ = tokio::fs::remove_file(dst).await; + bail!(AppError::Custom(format!( + "ffmpeg produced empty mp4: {}", + dst.display() + ))); + } + info!( + "remux done {} ({:.1} MiB) in {:.1}s", + dst.display(), + meta.len() as f64 / 1048576.0, + started.elapsed().as_secs_f64() + ); + // Suppress unused-import warning for Duration on platforms that don't + // need it (kept for future timeout work). + let _ = Duration::from_secs(0); + Ok(()) + } + /// 执行钩子步骤操作 /// /// # 参数 @@ -264,6 +382,85 @@ pub async fn process_video(video_path: &[&Path], processors: &[HookStep]) -> App Ok(()) } +/// Path-mutating processor pipeline. After each step the path list reflects any +/// in-place transforms (e.g. Remux replaces .ts with .mp4). Callers can read the +/// final paths to find the actual files to upload / postprocess. +pub async fn process_video_paths( + video_paths: &mut Vec, + processors: &[HookStep], +) -> AppResult<()> { + info!("Starting video processing (path-mutating)..."); + for processor in processors { + processor.execute_paths(video_paths).await?; + } + info!("Video processing completed"); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn deserialize_yaml_variants() { + let yaml = r#" +- run: "echo hi" +- mv: "/dst" +- remux: "mp4" +- "rm" +"#; + let steps: Vec = serde_yaml::from_str(yaml).unwrap(); + assert_eq!(steps.len(), 4); + assert!(matches!(steps[0], HookStep::Run { .. })); + assert!(matches!(steps[1], HookStep::Move { .. })); + assert!(matches!(steps[2], HookStep::Remux { .. })); + assert!(matches!(steps[3], HookStep::Remove(_))); + } + + #[test] + fn deserialize_json_variants() { + let json = r#"[ + {"run": "echo hi"}, + {"mv": "/dst"}, + {"remux": "mp4"}, + "rm" + ]"#; + let steps: Vec = serde_json::from_str(json).unwrap(); + assert_eq!(steps.len(), 4); + assert!(matches!(steps[2], HookStep::Remux { .. })); + } + + #[tokio::test] + async fn remux_skips_non_ts() { + let dir = tempfile::tempdir().unwrap(); + let p = dir.path().join("foo.mp4"); + std::fs::write(&p, b"not a real mp4 but we shouldn't touch it").unwrap(); + let mut paths = vec![p.clone()]; + let step = HookStep::Remux { remux: "mp4".into() }; + step.execute_paths(&mut paths).await.unwrap(); + // Already .mp4 → unchanged. + assert_eq!(paths[0], p); + } + + /// 复现 pipeline_upload_videos 的 fault-tolerance 触发条件： + /// segment_processor 在 ffmpeg 退非 0 时返回 Err。生产事故就是这一步因磁盘满 + /// 而失败，调用方必须能从 Err 恢复（log + continue 而不是 ? 早退）。 + #[tokio::test] + async fn remux_fails_when_ffmpeg_cannot_read_input() { + let dir = tempfile::tempdir().unwrap(); + let bogus = dir.path().join("does_not_exist.ts"); + let mut paths = vec![bogus.clone()]; + let step = HookStep::Remux { remux: "mp4".into() }; + let result = step.execute_paths(&mut paths).await; + assert!( + result.is_err(), + "ffmpeg 读不到输入应返回 Err，调用方才能在 pipeline 里做 log+continue" + ); + // 失败时不要把破损的 mp4 留在磁盘上（idempotent retry 前提） + assert!(!bogus.with_extension("mp4").exists()); + } +} + pub async fn process(input: &[u8], processors: &Option>) { if let Some(hooks) = processors { // 依次执行每个处理器步骤