import os import subprocess import json import logging import random import glob as glob_module from datetime import datetime from typing import Dict, Any, Optional, Tuple, List logger = logging.getLogger(__name__) # Base directory for audio assets (relative to project root) # 兼容本地（backend/services/ → 3层上溯到项目根）和 HF Spaces（services/ → 2层上溯到 /app） def _find_assets_dir() -> str: here = os.path.abspath(__file__) candidates = [ os.path.join(os.path.dirname(os.path.dirname(here)), "assets"), # HF Spaces: /app/assets os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(here))), "assets"), # 本地: 项目根/assets ] for path in candidates: if os.path.isdir(path): return path return candidates[-1] _ASSETS_DIR = _find_assets_dir() def _find_cjk_font() -> str: """Find a suitable CJK font for text overlay.""" candidates = [ "/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc", "/usr/share/fonts/noto-cjk/NotoSansCJK-Bold.ttc", "/usr/share/fonts/truetype/noto/NotoSansCJK-Bold.ttc", "/usr/share/fonts/truetype/droid/DroidSansFallbackFull.ttf", "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", "/System/Library/Fonts/STHeiti Medium.ttc", "/System/Library/Fonts/PingFang.ttc", ] for path in candidates: if os.path.isfile(path): return path return "/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc" def _find_real_sfx(effect_type: str) -> Optional[str]: """Find a real SFX file in assets/sfx/ matching the effect type. Mapping: swish -> basketball_swish.mp3 slam -> basketball_dunk_rim.mp3, impact_*.mp3 whistle -> whistle_short.mp3, whistle_swoosh.mp3 """ sfx_dir = os.path.join(_ASSETS_DIR, "sfx") if not os.path.isdir(sfx_dir): return None sfx_mapping = { "swish": ["basketball_swish.mp3", "basketball_bounce_01.mp3"], "slam": ["basketball_dunk_rim.mp3", "impact_01.mp3", "impact_02.mp3", "impact_03.mp3"], "whistle": ["whistle_short.mp3", "whistle_swoosh.mp3", "basketball_buzzer.mp3"], } candidates = sfx_mapping.get(effect_type, []) for candidate in candidates: path = os.path.join(sfx_dir, candidate) if os.path.isfile(path): return path # Fallback: try to find any file with the effect_type in the name pattern = os.path.join(sfx_dir, f"*{effect_type}*.mp3") matches = glob_module.glob(pattern) if matches: return random.choice(matches) return None def _find_real_bgm() -> Optional[str]: """Find a random real BGM file in assets/bgm/.""" bgm_dir = os.path.join(_ASSETS_DIR, "bgm") if not os.path.isdir(bgm_dir): return None pattern = os.path.join(bgm_dir, "*.mp3") matches = glob_module.glob(pattern) if matches: return random.choice(matches) return None def get_video_info(video_path: str) -> Dict[str, Any]: try: cmd = [ "ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", video_path, ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) if result.returncode == 0: info = json.loads(result.stdout) duration = float(info.get("format", {}).get("duration", 0)) streams = info.get("streams", []) video_stream = next((s for s in streams if s["codec_type"] == "video"), None) width = int(video_stream["width"]) if video_stream else 0 height = int(video_stream["height"]) if video_stream else 0 fps = 30 if video_stream and "r_frame_rate" in video_stream: parts = video_stream["r_frame_rate"].split("/") if len(parts) == 2 and int(parts[1]) > 0: fps = int(parts[0]) / int(parts[1]) return {"duration": duration, "width": width, "height": height, "fps": fps} except (FileNotFoundError, subprocess.TimeoutExpired, json.JSONDecodeError) as e: logger.info(f"ffprobe not available or failed: {e}, trying cv2 fallback") try: import cv2 cap = cv2.VideoCapture(video_path) if cap.isOpened(): width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) fps = cap.get(cv2.CAP_PROP_FPS) total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) duration = total_frames / fps if fps > 0 else 0 cap.release() return {"duration": duration, "width": width, "height": height, "fps": fps} except ImportError: logger.warning("cv2 not available for video info fallback") except Exception as e: logger.warning(f"cv2 video info fallback failed: {e}") logger.warning("All video info methods failed, using defaults") return {"duration": 0, "width": 1920, "height": 1080, "fps": 30} def extract_clip_with_effects( video_path: str, start: float, end: float, output_path: str, peak_time: float = 0.0, ): """合并提取+调色+暗角为一次FFmpeg调用，大幅减少重编码次数 相比分别调用 extract_clip → apply_color_grading → apply_vignette， 此函数将所有滤镜合并到一条 filter_complex 中，只需一次编码。 """ clip_duration = end - start filter_parts = [] # 1. 调色 filter_parts.append( "eq=contrast=1.35:brightness=0.05:saturation=1.5," "colorbalance=rs=0.1:bs=-0.05:rm=0.05:bm=-0.03," "unsharp=5:5:1.0:5:5:0.0" ) # 2. 暗角（静态暗角，避免条件表达式的复杂性） filter_parts.append("vignette=angle=0.4:mode=forward") filter_complex = f"[0:v]{','.join(filter_parts)}[v]" # 使用 -ss 在 -i 之前（快速 seeking），-t 指定时长（避免 -to 时间轴歧义） clip_dur = end - start cmd = [ "ffmpeg", "-y", "-ss", str(start), "-i", video_path, "-t", str(clip_dur), "-filter_complex", filter_complex, "-map", "[v]", "-map", "0:a?", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-g", "30", "-keyint_min", "30", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) if result.returncode != 0: logger.warning(f"[extract_clip_with_effects] combined filter failed: {result.stderr[:300]}") # fallback 1: simple extract without effects, -ss before -i cmd_simple = [ "ffmpeg", "-y", "-ss", str(start), "-i", video_path, "-t", str(clip_dur), "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-g", "30", "-keyint_min", "30", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] r2 = subprocess.run(cmd_simple, capture_output=True, text=True, timeout=60) if r2.returncode != 0: logger.warning(f"[extract_clip_with_effects] simple extract failed: {r2.stderr[:300]}") # fallback 2: -ss after -i（慢但更精确） cmd_slow = [ "ffmpeg", "-y", "-i", video_path, "-ss", str(start), "-to", str(end), "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-g", "30", "-keyint_min", "30", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] subprocess.run(cmd_slow, capture_output=True, text=True, timeout=60) except subprocess.TimeoutExpired: logger.warning("[extract_clip_with_effects] timed out, doing simple extract") try: cmd_simple = [ "ffmpeg", "-y", "-ss", str(start), "-i", video_path, "-t", str(clip_dur), "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-g", "30", "-keyint_min", "30", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] subprocess.run(cmd_simple, capture_output=True, text=True, timeout=60) except Exception: pass def extract_clip(video_path: str, start: float, end: float, output_path: str): cmd = [ "ffmpeg", "-y", "-i", video_path, "-ss", str(start), "-to", str(end), "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-g", "30", "-keyint_min", "30", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] subprocess.run(cmd, capture_output=True, check=True, timeout=60) def apply_slow_motion(input_path: str, output_path: str, factor: float = 2.0): cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]setpts={factor}*PTS[v];[0:a]atempo={1/factor}[a]", "-map", "[v]", "-map", "[a]", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-g", "30", "-keyint_min", "30", "-c:a", "aac", "-b:a", "128k", "-vsync", "cfr", "-movflags", "+faststart", output_path, ] subprocess.run(cmd, capture_output=True, check=True, timeout=120) def apply_peak_slow_motion(input_path: str, output_path: str, peak_time: float, clip_duration: float, slow_factor: float = 3.0, slow_duration: float = 1.5): """关键帧变速：只在 peak_time 附近慢放，其他部分正常速度 慢放区域：peak_time 前后各 slow_duration/2 秒 有音频时使用分段拼接方式：将视频切为三段，仅中间段做视频+音频慢放， 然后用 FFmpeg concat 合并，确保音视频时长同步。 无音频时使用 setpts 条件表达式实现局部慢放。 """ slow_start = max(0.0, peak_time - slow_duration / 2) slow_end = min(clip_duration, peak_time + slow_duration / 2) # 检查输入是否有音频流 probe_cmd = ["ffprobe", "-v", "error", "-select_streams", "a", "-show_entries", "stream=codec_name", "-of", "csv=p=0", input_path] probe_result = subprocess.run(probe_cmd, capture_output=True, text=True, timeout=10) has_audio = bool(probe_result.stdout.strip()) try: if has_audio: # 有音频：使用分段拼接方式，确保音视频同步 # 分三段：[0, slow_start] + [slow_start, slow_end]慢放 + [slow_end, end] import tempfile # 如果慢放区域覆盖整个片段，直接全局慢放 if slow_start < 0.3 and (clip_duration - slow_end) < 0.3: apply_slow_motion(input_path, output_path, factor=slow_factor) return tmp_dir = tempfile.mkdtemp(prefix="peak_slow_") try: part1 = os.path.join(tmp_dir, "part1.mp4") part2_slow = os.path.join(tmp_dir, "part2_slow.mp4") part3 = os.path.join(tmp_dir, "part3.mp4") # Part 1: 0 → slow_start（正常速度） if slow_start > 0.05: cmd1 = [ "ffmpeg", "-y", "-i", input_path, "-t", str(slow_start), "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-g", "30", "-keyint_min", "30", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", part1, ] r1 = subprocess.run(cmd1, capture_output=True, text=True, timeout=60) if r1.returncode != 0: logger.warning(f"[apply_peak_slow_motion] part1 failed: {r1.stderr[:200]}") else: part1 = None # Part 2: slow_start → slow_end（慢放，视频+音频同步变速） # atempo 只支持 0.5~2.0 范围，slow_factor > 2 时需要链式 atempo tempo_val = 1.0 / slow_factor if tempo_val < 0.5: tempo1 = 0.5 tempo2 = tempo_val / tempo1 audio_filter = f"[0:a]atempo={tempo1},atempo={tempo2:.4f}[a]" else: audio_filter = f"[0:a]atempo={tempo_val:.4f}[a]" cmd2 = [ "ffmpeg", "-y", "-i", input_path, "-ss", str(slow_start), "-to", str(slow_end), "-filter_complex", f"[0:v]setpts={slow_factor}*PTS[v];{audio_filter}", "-map", "[v]", "-map", "[a]", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-g", "30", "-keyint_min", "30", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", part2_slow, ] r2 = subprocess.run(cmd2, capture_output=True, text=True, timeout=60) if r2.returncode != 0: logger.warning(f"[apply_peak_slow_motion] part2_slow failed: {r2.stderr[:200]}") raise RuntimeError("part2_slow extraction failed") # Part 3: slow_end → end（正常速度） if (clip_duration - slow_end) > 0.05: cmd3 = [ "ffmpeg", "-y", "-i", input_path, "-ss", str(slow_end), "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-g", "30", "-keyint_min", "30", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", part3, ] r3 = subprocess.run(cmd3, capture_output=True, text=True, timeout=60) if r3.returncode != 0: logger.warning(f"[apply_peak_slow_motion] part3 failed: {r3.stderr[:200]}") else: part3 = None # 构建 concat 文件 concat_file = os.path.join(tmp_dir, "concat.txt") with open(concat_file, "w") as f: if part1 and os.path.exists(part1) and os.path.getsize(part1) > 0: f.write(f"file '{part1}'\n") if os.path.exists(part2_slow) and os.path.getsize(part2_slow) > 0: f.write(f"file '{part2_slow}'\n") if part3 and os.path.exists(part3) and os.path.getsize(part3) > 0: f.write(f"file '{part3}'\n") # 合并 cmd_concat = [ "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", concat_file, "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-g", "30", "-keyint_min", "30", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] result = subprocess.run(cmd_concat, capture_output=True, text=True, timeout=120) if result.returncode != 0: logger.warning(f"[apply_peak_slow_motion] concat failed: {result.stderr[:200]}") raise RuntimeError("concat failed") finally: import shutil shutil.rmtree(tmp_dir, ignore_errors=True) else: # 无音频：只处理视频（使用 setpts 条件表达式实现局部慢放） video_filter = ( f"[0:v]setpts='if(between(t,{slow_start},{slow_end})," f"{slow_factor}*PTS,PTS)'[v]" ) cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", video_filter, "-map", "[v]", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-g", "30", "-keyint_min", "30", "-movflags", "+faststart", output_path, ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) if result.returncode == 0 and os.path.exists(output_path) and os.path.getsize(output_path) > 0: return logger.warning(f"[apply_peak_slow_motion] no-audio filter failed: {result.stderr[:300]}") raise RuntimeError("no-audio filter failed") # 验证输出 if os.path.exists(output_path) and os.path.getsize(output_path) > 0: return raise RuntimeError("output file missing or empty") except (subprocess.TimeoutExpired, Exception) as e: logger.warning(f"[apply_peak_slow_motion] segment approach failed: {e}") # Fallback：全局慢放（最简单最可靠） logger.info("[apply_peak_slow_motion] falling back to global slow motion") apply_slow_motion(input_path, output_path, factor=slow_factor) def apply_ken_burns_zoom(input_path: str, output_path: str, peak_time: float, clip_duration: float, zoom_factor: float = 1.15, zoom_duration: float = 0.8): """Ken Burns zoom-in 效果：在 peak_time 附近平滑放大镜头 zoom_factor: 最大放大倍数（1.15 = 放大 15%） zoom_duration: zoom 持续时间（秒） """ info = get_video_info(input_path) width = info.get("width", 1920) height = info.get("height", 1080) if width <= 0 or height <= 0: import shutil shutil.copy2(input_path, output_path) return # 确保 width/height 为偶数 width = width + (width % 2) height = height + (height % 2) # zoom 时间范围 zoom_start = max(0.0, peak_time - zoom_duration / 2) zoom_peak = peak_time zoom_end = min(clip_duration, peak_time + zoom_duration / 2) # 使用 zoompan 滤镜实现平滑 zoom # zoompan 参数：z=zoom_level, d=duration_in_frames, x/y=pan_position fps = 30 total_frames = int(clip_duration * fps) zoom_start_frame = int(zoom_start * fps) zoom_peak_frame = int(zoom_peak * fps) zoom_end_frame = int(zoom_end * fps) # 构建 zoompan 表达式 # 在 zoom_start 之前：z=1（正常） # 从 zoom_start 到 zoom_peak：z 从 1 线性增加到 zoom_factor # 从 zoom_peak 到 zoom_end：z 从 zoom_factor 线性减少到 1 # zoom_end 之后：z=1（正常） zoom_expr = ( f"if(lte(on, {zoom_start_frame}), 1," f"if(lte(on, {zoom_peak_frame})," f" 1+({zoom_factor}-1)*(on-{zoom_start_frame})/({zoom_peak_frame}-{zoom_start_frame})," f"if(lte(on, {zoom_end_frame})," f" {zoom_factor}-({zoom_factor}-1)*(on-{zoom_peak_frame})/({zoom_end_frame}-{zoom_peak_frame})," f" 1)))" ) # x/y 居中 x_expr = f"iw/2-(iw/zoom)/2" y_expr = f"ih/2-(ih/zoom)/2" cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]zoompan=z='{zoom_expr}':x='{x_expr}':y='{y_expr}':d={total_frames}:s={width}x{height}:fps={fps}[v]", "-map", "[v]", "-map", "0:a?", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) if result.returncode != 0: logger.warning(f"[apply_ken_burns_zoom] zoompan failed: {result.stderr[:200]}") import shutil shutil.copy2(input_path, output_path) except subprocess.TimeoutExpired: logger.warning("[apply_ken_burns_zoom] zoompan timed out, copying original") import shutil shutil.copy2(input_path, output_path) except Exception as e: logger.warning(f"[apply_ken_burns_zoom] error: {e}") import shutil shutil.copy2(input_path, output_path) def apply_cpu_zoom_in(input_path: str, output_path: str, peak_time: float, clip_duration: float, zoom_factor: float = 1.2, zoom_duration: float = 1.0): """CPU 友好的推镜头效果：使用 scale+crop 动态缩放替代 zoompan 相比 zoompan（CPU 上极慢），此方案使用 crop 的 x/y 参数做动画， 从 1.0x 渐进到 zoom_factor 再回退，模拟推镜头效果。 CPU 开销小，3s 片段约 2s 处理时间。 """ info = get_video_info(input_path) width = info.get("width", 1920) height = info.get("height", 1080) if width <= 0 or height <= 0: import shutil shutil.copy2(input_path, output_path) return # 确保 width/height 为偶数 width = width + (width % 2) height = height + (height % 2) # zoom 时间范围 zoom_start = max(0.0, peak_time - zoom_duration / 2) zoom_end = min(clip_duration, peak_time + zoom_duration / 2) # 使用 crop + scale 实现动态推镜头 # crop 宽高随时间变化：从 iw/1.0 逐渐缩小到 iw/zoom_factor（即放大效果） # 然后再从 iw/zoom_factor 逐渐恢复到 iw/1.0 # crop_x/crop_y 居中 if zoom_start < peak_time and peak_time < zoom_end: # 渐进推镜头：zoom_start → peak_time 放大，peak_time → zoom_end 缩回 crop_w_expr = ( f"iw/if(between(t,{zoom_start},{peak_time})," f"1+({zoom_factor}-1)*(t-{zoom_start})/({peak_time}-{zoom_start})," f"if(between(t,{peak_time},{zoom_end})," f"{zoom_factor}-({zoom_factor}-1)*(t-{peak_time})/({zoom_end}-{peak_time})," f"1))" ) crop_h_expr = crop_w_expr.replace("iw/", "ih/") x_expr = f"(iw-ow)/2" y_expr = f"(ih-oh)/2" filter_complex = ( f"[0:v]crop={crop_w_expr}:{crop_h_expr}:{x_expr}:{y_expr}," f"scale={width}:{height}[v]" ) else: # 时间不足，做静态放大 filter_complex = ( f"[0:v]scale=iw*{zoom_factor}:ih*{zoom_factor}," f"crop={width}:{height}:(iw-{width})/2:(ih-{height})/2[v]" ) cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", filter_complex, "-map", "[v]", "-map", "0:a?", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) if result.returncode != 0: logger.warning(f"[apply_cpu_zoom_in] failed: {result.stderr[:200]}") import shutil shutil.copy2(input_path, output_path) except subprocess.TimeoutExpired: logger.warning("[apply_cpu_zoom_in] timed out, copying original") import shutil shutil.copy2(input_path, output_path) except Exception as e: logger.warning(f"[apply_cpu_zoom_in] error: {e}") import shutil shutil.copy2(input_path, output_path) def apply_flash_effect(input_path: str, output_path: str, peak_time: float, clip_duration: float, flash_duration: float = 0.2, brightness_peak: float = 1.0): """闪光效果：peak_time 时刻全屏白色闪烁 使用 eq=brightness 在 peak_time 瞬间提升亮度，flash_duration 内衰减回正常。 """ flash_start = max(0.0, peak_time - flash_duration / 2) flash_end = min(clip_duration, peak_time + flash_duration / 2) # 亮度表达式：flash_start 时亮度突增 brightness_peak，flash_end 时衰减回 0 brightness_expr = ( f"if(between(t,{flash_start},{flash_end})," f"{brightness_peak}*(1-(t-{flash_start})/{flash_duration}),0)" ) cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]eq=brightness='{brightness_expr}'[v]", "-map", "[v]", "-map", "0:a?", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) if result.returncode != 0: logger.warning(f"[apply_flash_effect] failed: {result.stderr[:200]}") import shutil shutil.copy2(input_path, output_path) except (subprocess.TimeoutExpired, Exception) as e: logger.warning(f"[apply_flash_effect] error: {e}") import shutil shutil.copy2(input_path, output_path) def apply_shake_effect(input_path: str, output_path: str, peak_time: float, clip_duration: float, shake_duration: float = 0.4, shake_amplitude: int = 25): """画面震动效果：peak_time 附近画面做正弦波位移，增强力量感 使用 overlay + 正弦波偏移实现画面抖动。 """ shake_start = max(0.0, peak_time - shake_duration / 2) shake_end = min(clip_duration, peak_time + shake_duration / 2) info = get_video_info(input_path) width = info.get("width", 1920) height = info.get("height", 1080) if width <= 0 or height <= 0: import shutil shutil.copy2(input_path, output_path) return # 确保 width/height 为偶数 width = width + (width % 2) height = height + (height % 2) # 震动频率 15Hz，振幅 shake_amplitude 像素 freq = 15 x_expr = f"{shake_amplitude}+{shake_amplitude}*sin(2*PI*{freq}*(t-{shake_start}))" y_expr = f"{shake_amplitude}+{shake_amplitude}*cos(2*PI*{freq}*(t-{shake_start}))" enable_expr = f"between(t,{shake_start},{shake_end})" # 使用 crop + overlay 实现震动 crop_w = width - 2 * shake_amplitude crop_h = height - 2 * shake_amplitude # 确保 crop 尺寸为偶数 crop_w = crop_w + (crop_w % 2) crop_h = crop_h + (crop_h % 2) filter_complex = ( f"[0:v]split[bg][fg];" f"[fg]crop={crop_w}:{crop_h}:{shake_amplitude}:{shake_amplitude}[cropped];" f"[bg][cropped]overlay=x='{x_expr}':y='{y_expr}':enable='{enable_expr}'[v]" ) cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", filter_complex, "-map", "[v]", "-map", "0:a?", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) if result.returncode != 0: logger.warning(f"[apply_shake_effect] failed: {result.stderr[:200]}") import shutil shutil.copy2(input_path, output_path) except (subprocess.TimeoutExpired, Exception) as e: logger.warning(f"[apply_shake_effect] error: {e}") import shutil shutil.copy2(input_path, output_path) def apply_glitch_effect(input_path: str, output_path: str, peak_time: float, clip_duration: float, glitch_duration: float = 0.15, shift_pixels: int = 8): """RGB 色差分离（Glitch）效果：peak_time 附近短暂 RGB 通道偏移 在进球瞬间（peak_time）短暂应用 RGB 通道水平偏移， 营造数字故障感，增强视觉冲击力。 """ glitch_start = max(0.0, peak_time - glitch_duration / 2) glitch_end = min(clip_duration, peak_time + glitch_duration / 2) info = get_video_info(input_path) width = info.get("width", 1920) height = info.get("height", 1080) if width <= 0 or height <= 0: import shutil shutil.copy2(input_path, output_path) return # 确保 width/height 为偶数 width = width + (width % 2) height = height + (height % 2) # 使用 colorchannelmixer 实现 RGB 色差偏移 # 在 glitch_start ~ glitch_end 期间，R 通道向左偏移，B 通道向右偏移 # colorchannelmixer 可以混合通道来实现偏移效果 # 但 FFmpeg 没有直接的通道偏移滤镜，使用 crop + overlay 方式 # 简化方案：使用 eq 滤镜在 glitch 期间增强对比度+饱和度 + 轻微色相偏移 # 配合闪光效果已经足够产生 Glitch 感 enable_expr = f"between(t,{glitch_start},{glitch_end})" # 在 glitch 期间：高对比度 + 高饱和 + 色相偏移 15 度 filter_complex = ( f"[0:v]eq=contrast='if({enable_expr},2.0,1.0)':" f"saturation='if({enable_expr},2.0,1.0)':" f"brightness='if({enable_expr},0.1,0.0)'," f"hue=h='if({enable_expr},15,0)'[v]" ) cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", filter_complex, "-map", "[v]", "-map", "0:a?", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) if result.returncode != 0: logger.warning(f"[apply_glitch_effect] failed: {result.stderr[:200]}") import shutil shutil.copy2(input_path, output_path) except (subprocess.TimeoutExpired, Exception) as e: logger.warning(f"[apply_glitch_effect] error: {e}") import shutil shutil.copy2(input_path, output_path) def apply_combined_effects(input_path: str, output_path: str, peak_time: float, clip_duration: float, flash_duration: float = 0.2, brightness_peak: float = 1.0, shake_duration: float = 0.4, shake_amplitude: int = 25, glitch_duration: float = 0.15): """合并闪光+震屏+Glitch为一次FFmpeg调用，大幅减少重编码次数 将三种特效合并到一条 filter_complex 中： 1. 闪光：peak_time 附近亮度突增 2. 震屏：peak_time 附近画面正弦波位移 3. Glitch：peak_time 附近对比度+饱和度+色相偏移 """ info = get_video_info(input_path) width = info.get("width", 1920) height = info.get("height", 1080) if width <= 0 or height <= 0: import shutil shutil.copy2(input_path, output_path) return width = width + (width % 2) height = height + (height % 2) # 时间范围 flash_start = max(0.0, peak_time - flash_duration / 2) flash_end = min(clip_duration, peak_time + flash_duration / 2) shake_start = max(0.0, peak_time - shake_duration / 2) shake_end = min(clip_duration, peak_time + shake_duration / 2) glitch_start = max(0.0, peak_time - glitch_duration / 2) glitch_end = min(clip_duration, peak_time + glitch_duration / 2) # 1. 闪光亮度表达式 brightness_expr = ( f"if(between(t,{flash_start},{flash_end})," f"{brightness_peak}*(1-(t-{flash_start})/{flash_duration}),0)" ) # 2. Glitch 效果：对比度+饱和度+色相偏移 glitch_enable = f"between(t,{glitch_start},{glitch_end})" # 3. 震屏效果：crop + overlay crop_w = width - 2 * shake_amplitude crop_h = height - 2 * shake_amplitude crop_w = crop_w + (crop_w % 2) crop_h = crop_h + (crop_h % 2) freq = 15 x_expr = f"{shake_amplitude}+{shake_amplitude}*sin(2*PI*{freq}*(t-{shake_start}))" y_expr = f"{shake_amplitude}+{shake_amplitude}*cos(2*PI*{freq}*(t-{shake_start}))" shake_enable = f"between(t,{shake_start},{shake_end})" # 合并滤镜链：eq(闪光+Glitch) → split → crop → overlay(震屏) # 注意：hue 滤镜参数格式是 h=角度值 hue_expr = f"if({glitch_enable},15,0)" filter_complex = ( f"[0:v]eq=" f"brightness='{brightness_expr}':" f"contrast='if({glitch_enable},2.0,1.0)':" f"saturation='if({glitch_enable},2.0,1.0)'," f"hue=h='{hue_expr}'," f"split[bg][fg];" f"[fg]crop={crop_w}:{crop_h}:{shake_amplitude}:{shake_amplitude}[cropped];" f"[bg][cropped]overlay=x='{x_expr}':y='{y_expr}':enable='{shake_enable}'[v]" ) cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", filter_complex, "-map", "[v]", "-map", "0:a?", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) if result.returncode != 0: logger.warning(f"[apply_combined_effects] failed: {result.stderr[:300]}") import shutil shutil.copy2(input_path, output_path) except (subprocess.TimeoutExpired, Exception) as e: logger.warning(f"[apply_combined_effects] error: {e}") import shutil shutil.copy2(input_path, output_path) def apply_replay_effect(input_path: str, output_path: str, peak_time: float, clip_duration: float, replay_duration: float = 1.0, replay_count: int = 2): """重播效果：peak_time 附近的片段重复播放，第二次慢放 将片段切为三段：[start, peak-replay] + [peak-replay, peak+replay] + [peak+replay, end] 中间段重复 replay_count 次（第一次原速，后续慢放 2x） 然后用 FFmpeg concat 合并。 """ import tempfile replay_start = max(0.0, peak_time - replay_duration / 2) replay_end = min(clip_duration, peak_time + replay_duration / 2) # 如果片段太短，不做重播 if replay_start < 0.3 or (clip_duration - replay_end) < 0.3: import shutil shutil.copy2(input_path, output_path) return tmp_dir = tempfile.mkdtemp(prefix="replay_") try: # 切分三段 part1 = os.path.join(tmp_dir, "part1.mp4") part2 = os.path.join(tmp_dir, "part2.mp4") part3 = os.path.join(tmp_dir, "part3.mp4") # Part 1: start → replay_start cmd1 = [ "ffmpeg", "-y", "-i", input_path, "-t", str(replay_start), "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", part1, ] subprocess.run(cmd1, capture_output=True, text=True, timeout=60) # Part 2: replay_start → replay_end (原速) cmd2 = [ "ffmpeg", "-y", "-i", input_path, "-ss", str(replay_start), "-to", str(replay_end), "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", part2, ] subprocess.run(cmd2, capture_output=True, text=True, timeout=60) # Part 3: replay_end → end cmd3 = [ "ffmpeg", "-y", "-i", input_path, "-ss", str(replay_end), "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", part3, ] subprocess.run(cmd3, capture_output=True, text=True, timeout=60) # Part 2 慢放版本 part2_slow = os.path.join(tmp_dir, "part2_slow.mp4") apply_slow_motion(part2, part2_slow, factor=2.0) # 构建 concat 文件：part1 + part2(原速) + part2_slow(慢放) + part3 concat_file = os.path.join(tmp_dir, "concat.txt") with open(concat_file, "w") as f: f.write(f"file '{part1}'\n") f.write(f"file '{part2}'\n") if os.path.exists(part2_slow) and os.path.getsize(part2_slow) > 0: f.write(f"file '{part2_slow}'\n") f.write(f"file '{part3}'\n") # 合并 cmd_concat = [ "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", concat_file, "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] result = subprocess.run(cmd_concat, capture_output=True, text=True, timeout=120) if result.returncode != 0: logger.warning(f"[apply_replay_effect] concat failed: {result.stderr[:200]}") import shutil shutil.copy2(input_path, output_path) except (subprocess.TimeoutExpired, Exception) as e: logger.warning(f"[apply_replay_effect] error: {e}") import shutil shutil.copy2(input_path, output_path) finally: import shutil shutil.rmtree(tmp_dir, ignore_errors=True) def apply_color_grading(input_path: str, output_path: str): cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", "[0:v]eq=contrast=1.2:brightness=0.03:saturation=1.3," "colorbalance=rs=0.08:bs=-0.04:rm=0.04:bm=-0.02," "unsharp=5:5:0.8:5:5:0.0[v]", "-map", "[v]", "-c:a", "copy", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-movflags", "+faststart", output_path, ] subprocess.run(cmd, capture_output=True, check=True, timeout=120) def apply_vignette(input_path: str, output_path: str, mode: str = "peak", peak_time: float = 0.0, clip_duration: float = 0.0): """暗角效果：边缘暗化强调主体 mode: 'static' 始终暗角, 'peak' 只在 peak_time 附近加强暗角 """ if mode == "peak" and peak_time > 0 and clip_duration > 0: enable_strong = f"between(t,{max(0,peak_time-0.5)},{min(clip_duration,peak_time+0.5)})" cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]vignette=angle=0.3:mode=forward:enable='1'," f"vignette=angle=0.5:mode=forward:enable='{enable_strong}'[v]", "-map", "[v]", "-c:a", "copy", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-movflags", "+faststart", output_path, ] else: cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", "[0:v]vignette=angle=0.3:mode=forward[v]", "-map", "[v]", "-c:a", "copy", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-movflags", "+faststart", output_path, ] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) if result.returncode != 0: logger.warning(f"[apply_vignette] failed: {result.stderr[:200]}") import shutil shutil.copy2(input_path, output_path) except subprocess.TimeoutExpired: logger.warning("[apply_vignette] timed out, copying original") import shutil shutil.copy2(input_path, output_path) def add_highlight_overlay(input_path: str, output_path: str, text: str, peak_time: float, clip_duration: float, fontcolor: str = "white", fontsize: int = 90): """高光文字叠加：在 peak_time 附近显示进球/2+1/神仙球等文字，带缩放弹入动画 文字在 peak_time 前 0.3s 出现，持续 1.5s，带 zoom-in 动画和淡入淡出。 正确处理音频流（-map 0:a?），避免丢失音频。 """ fontfile = _find_cjk_font() # 文字出现时间：peak 前 0.3s 到 peak 后 1.2s text_start = max(0.0, peak_time - 0.3) text_end = min(clip_duration, peak_time + 1.2) fade_in = 0.2 fade_out = 0.3 # zoom-in 动画：字号从 30 渐变到目标大小 fs_expr = f"if(lt(t,{text_start + fade_in}),30+({fontsize}-30)*(t-{text_start})/{fade_in},{fontsize})" enable = f"between(t,{text_start},{text_end})" alpha_expr = ( f"if(between(t,{text_start},{text_start + fade_in}),(t-{text_start})/{fade_in}," f"if(between(t,{text_end - fade_out},{text_end}),({text_end}-t)/{fade_out},1))" ) filter_complex = ( f"[0:v]drawtext=text='{text}':fontfile={fontfile}:" f"fontcolor={fontcolor}:fontsize='{fs_expr}':" f"borderw=5:bordercolor=black:" f"x=(w-text_w)/2:y=h*0.75:" f"alpha='{alpha_expr}':" f"enable='{enable}'[v]" ) cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", filter_complex, "-map", "[v]", "-map", "0:a?", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) if result.returncode != 0: logger.warning(f"[add_highlight_overlay] failed: {result.stderr[:200]}") import shutil shutil.copy2(input_path, output_path) except (subprocess.TimeoutExpired, Exception) as e: logger.warning(f"[add_highlight_overlay] error: {e}") import shutil shutil.copy2(input_path, output_path) def add_subtitle(input_path: str, output_path: str, text: str, start: float, end: float, fontfile: str = "/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc", fontcolor: str = "yellow", fontsize: int = 80, effect: str = "fade"): """添加字幕，支持多种动画效果 effect: 'fade' 淡入淡出, 'bounce' 弹入, 'zoom' 缩放弹入, 'slide_left' 左滑入, 'slide_right' 右滑入, 'zoom_in' 从小到大弹入, 'flash' 闪烁出现, 'shake' 震动出现, 'rotate_in' 旋转缩放进入, 'explode' 爆炸扩散进入 """ enable = f"between(t,{start},{end})" duration = end - start fade_in = min(0.3, duration * 0.2) fade_out = min(0.3, duration * 0.2) if effect == "bounce": # 弹入效果：从下方弹入 + 缩放 y_expr = f"h*0.75-if(gt(t,{start}),0,min(1,(t-{start})/{fade_in}))*40*if(lt(t,{start}+{fade_in}),1,1)" cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]drawtext=text='{text}':fontfile={fontfile}:" f"fontcolor={fontcolor}:fontsize={fontsize}:" f"borderw=4:bordercolor=black:" f"x=(w-text_w)/2:y={y_expr}:" f"alpha='if(between(t,{start},{start+fade_in}),(t-{start})/{fade_in}," f"if(between(t,{end-fade_out},{end}),({end}-t)/{fade_out},1))':" f"enable='{enable}'[v]", "-map", "[v]", "-c:a", "copy", output_path, ] elif effect == "zoom": # 缩放弹入效果 cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]drawtext=text='{text}':fontfile={fontfile}:" f"fontcolor={fontcolor}:fontsize={fontsize}:" f"borderw=4:bordercolor=black:" f"x=(w-text_w)/2:y=h*0.75:" f"alpha='if(between(t,{start},{start+fade_in}),(t-{start})/{fade_in}," f"if(between(t,{end-fade_out},{end}),({end}-t)/{fade_out},1))':" f"enable='{enable}'[v]", "-map", "[v]", "-c:a", "copy", output_path, ] elif effect == "slide_left": # 左滑入效果 x_expr = f"if(lt(t,{start+fade_in}),(w-text_w)*(1-(t-{start})/{fade_in}),(w-text_w)/2)" cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]drawtext=text='{text}':fontfile={fontfile}:" f"fontcolor={fontcolor}:fontsize={fontsize}:" f"borderw=4:bordercolor=black:" f"x={x_expr}:y=h*0.75:" f"alpha='if(between(t,{start},{start+fade_in}),(t-{start})/{fade_in}," f"if(between(t,{end-fade_out},{end}),({end}-t)/{fade_out},1))':" f"enable='{enable}'[v]", "-map", "[v]", "-c:a", "copy", output_path, ] elif effect == "slide_right": # 右滑入效果 x_expr = f"if(lt(t,{start+fade_in}),0+(w-text_w)/2*(t-{start})/{fade_in},(w-text_w)/2)" cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]drawtext=text='{text}':fontfile={fontfile}:" f"fontcolor={fontcolor}:fontsize={fontsize}:" f"borderw=4:bordercolor=black:" f"x={x_expr}:y=h*0.75:" f"alpha='if(between(t,{start},{start+fade_in}),(t-{start})/{fade_in}," f"if(between(t,{end-fade_out},{end}),({end}-t)/{fade_out},1))':" f"enable='{enable}'[v]", "-map", "[v]", "-c:a", "copy", output_path, ] elif effect == "zoom_in": # 从小到大弹入效果：fontsize 从 20 渐变到目标大小 fs_expr = f"if(lt(t,{start+fade_in}),20+({fontsize}-20)*(t-{start})/{fade_in},{fontsize})" cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]drawtext=text='{text}':fontfile={fontfile}:" f"fontcolor={fontcolor}:fontsize={fs_expr}:" f"borderw=4:bordercolor=black:" f"x=(w-text_w)/2:y=h*0.75:" f"alpha='if(between(t,{start},{start+fade_in}),(t-{start})/{fade_in}," f"if(between(t,{end-fade_out},{end}),({end}-t)/{fade_out},1))':" f"enable='{enable}'[v]", "-map", "[v]", "-c:a", "copy", output_path, ] elif effect == "flash": # 闪烁出现效果：快速闪烁 3 次后稳定显示 blink_period = fade_in / 3 alpha_expr = ( f"if(between(t,{start},{start+fade_in})," f"if(eq(mod(int((t-{start})/{blink_period}),2),0),1,0.2)," f"if(between(t,{end-fade_out},{end}),({end}-t)/{fade_out},1))" ) cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]drawtext=text='{text}':fontfile={fontfile}:" f"fontcolor={fontcolor}:fontsize={fontsize}:" f"borderw=4:bordercolor=black:" f"x=(w-text_w)/2:y=h*0.75:" f"alpha='{alpha_expr}':" f"enable='{enable}'[v]", "-map", "[v]", "-c:a", "copy", output_path, ] elif effect == "shake": # 震动出现效果：出现时 x 方向随机偏移 shake_amp = 15 x_expr = f"(w-text_w)/2+{shake_amp}*sin(2*PI*12*(t-{start}))" shake_alpha = ( f"if(between(t,{start},{start+fade_in}),(t-{start})/{fade_in}," f"if(between(t,{end-fade_out},{end}),({end}-t)/{fade_out},1))" ) # 震动在 fade_in 后衰减 x_expr_damped = ( f"(w-text_w)/2+{shake_amp}*sin(2*PI*12*(t-{start}))" f"*if(lt(t,{start+fade_in}),1,max(0,1-(t-{start+fade_in})/0.3))" ) cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]drawtext=text='{text}':fontfile={fontfile}:" f"fontcolor={fontcolor}:fontsize={fontsize}:" f"borderw=4:bordercolor=black:" f"x='{x_expr_damped}':y=h*0.75:" f"alpha='{shake_alpha}':" f"enable='{enable}'[v]", "-map", "[v]", "-c:a", "copy", output_path, ] elif effect == "rotate_in": # 旋转缩放进入：用 fontsize 变化模拟旋转效果（从小到大+左右偏移） fs_expr = f"if(lt(t,{start+fade_in}),20+({fontsize}-20)*(t-{start})/{fade_in},{fontsize})" # 旋转用 x 偏移模拟：从左侧偏移到居中 x_expr = f"if(lt(t,{start+fade_in}),(w-text_w)/2-30*(1-(t-{start})/{fade_in}),(w-text_w)/2)" cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]drawtext=text='{text}':fontfile={fontfile}:" f"fontcolor={fontcolor}:fontsize={fs_expr}:" f"borderw=4:bordercolor=black:" f"x='{x_expr}':y=h*0.75:" f"alpha='if(between(t,{start},{start+fade_in}),(t-{start})/{fade_in}," f"if(between(t,{end-fade_out},{end}),({end}-t)/{fade_out},1))':" f"enable='{enable}'[v]", "-map", "[v]", "-c:a", "copy", output_path, ] elif effect == "explode": # 爆炸扩散进入：fontsize 从极大缩小到目标 + 透明度渐变 fs_expr = f"if(lt(t,{start+fade_in}),{fontsize*3}-({fontsize*3}-{fontsize})*(t-{start})/{fade_in},{fontsize})" cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]drawtext=text='{text}':fontfile={fontfile}:" f"fontcolor={fontcolor}:fontsize={fs_expr}:" f"borderw=4:bordercolor=black:" f"x=(w-text_w)/2:y=h*0.75:" f"alpha='if(between(t,{start},{start+fade_in}),(t-{start})/{fade_in}," f"if(between(t,{end-fade_out},{end}),({end}-t)/{fade_out},1))':" f"enable='{enable}'[v]", "-map", "[v]", "-c:a", "copy", output_path, ] else: # 默认 fade 效果 cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]drawtext=text='{text}':fontfile={fontfile}:" f"fontcolor={fontcolor}:fontsize={fontsize}:" f"borderw=4:bordercolor=black:" f"x=(w-text_w)/2:y=h*0.75:" f"alpha='if(between(t,{start},{start+fade_in}),(t-{start})/{fade_in}," f"if(between(t,{end-fade_out},{end}),({end}-t)/{fade_out},1))':" f"enable='{enable}'[v]", "-map", "[v]", "-c:a", "copy", output_path, ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: # fallback 到简单字幕 logger.warning(f"[add_subtitle] animated subtitle failed, falling back: {result.stderr[:200]}") cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]drawtext=text='{text}':fontfile={fontfile}:" f"fontcolor={fontcolor}:fontsize={fontsize}:" f"borderw=4:bordercolor=black:" f"x=(w-text_w)/2:y=h*0.75:enable='{enable}'[v]", "-map", "[v]", "-c:a", "copy", output_path, ] subprocess.run(cmd, capture_output=True, check=True) def concatenate_with_transition(clips: list, output_path: str, transition: str = "fade", duration: float = 0.5, transitions: list = None): """合并多个片段，支持每个转场独立指定类型 transition: 全局默认转场类型（当 transitions 未提供时使用） transitions: 每对相邻片段的转场类型列表，长度为 len(clips)-1 音频处理：使用 acrossfade 进行音频交叉淡入淡出，与视频 xfade 同步。 对于没有音频流的片段，自动填充静音轨道，确保合并后所有片段都有声音。 修复：添加视频标准化（scale+fps+format+setsar）确保所有片段格式一致， 避免 xfade 因分辨率/像素格式/帧率不一致而失败。 """ if len(clips) == 1: import shutil shutil.copy2(clips[0], output_path) return # 构建 transitions 列表 if transitions is None: transitions = [transition] * (len(clips) - 1) # 检查每个片段是否有音频流，并获取时长 clip_has_audio = [] clip_durations = [] for clip in clips: probe_a = ["ffprobe", "-v", "error", "-select_streams", "a", "-show_entries", "stream=codec_name", "-of", "csv=p=0", clip] probe_a_result = subprocess.run(probe_a, capture_output=True, text=True, timeout=10) clip_has_audio.append(bool(probe_a_result.stdout.strip())) probe_d = ["ffprobe", "-v", "quiet", "-show_entries", "format=duration", "-of", "csv=p=0", clip] probe_d_result = subprocess.run(probe_d, capture_output=True, text=True, timeout=10) try: d = float(probe_d_result.stdout.strip()) clip_durations.append(max(0.1, d)) except (ValueError, TypeError): clip_durations.append(0.0) # 动态计算 FFmpeg 超时时间：基于片段总时长 # preset=fast 在 1080p 上约 20fps，xfade 滤镜更慢 # 每秒视频给 20 秒处理时间，最少 600 秒（10 分钟），上限 7200 秒（2 小时） total_clip_duration = sum(clip_durations) ffmpeg_timeout = min(7200, max(600, int(total_clip_duration * 20))) # 获取目标尺寸（使用第一个片段的尺寸作为标准） first_info = get_video_info(clips[0]) target_w = int(first_info.get("width", 1920)) target_h = int(first_info.get("height", 1080)) # 确保尺寸为偶数 target_w = target_w + (target_w % 2) target_h = target_h + (target_h % 2) # 优先使用 xfade + acrossfade（视频转场 + 音频交叉淡入淡出，音视频同步） try: # 大视频跳过 xfade（需同时解码所有片段，内存占用高且速度慢），直接用 concat filter if total_clip_duration > 120 or len(clips) > 8: logger.info( f"[concatenate_with_transition] 跳过 xfade（视频过长或片段过多: " f"{total_clip_duration:.1f}s, {len(clips)} clips），直接用 concat filter" ) raise ValueError("skip_xfade") inputs = [] for clip in clips: inputs.extend(["-i", clip]) filter_parts = [] # 视频标准化：统一尺寸、帧率、像素格式、SAR，避免 xfade 因格式不一致失败 for i in range(len(clips)): filter_parts.append( f"[{i}:v]scale={target_w}:{target_h},fps=30,format=yuv420p,setsar=1[v{i}_norm]" ) # 视频 xfade 链（使用标准化后的流） # 标签流转：i=0 输入 [v0_norm][v1_norm] → 输出 [v0] # i=1 输入 [v0][v2_norm] → 输出 [v1] (in1 引用上一个 xfade 的输出) # i=2 输入 [v1][v3_norm] → 输出 [v2] offset = 0 for i in range(len(clips) - 1): if i == 0: in1 = f"[v0_norm]" else: in1 = f"[v{i-1}]" in2 = f"[v{i+1}_norm]" out = f"[v{i}]" offset += clip_durations[i] - duration trans = transitions[i] if i < len(transitions) else transition filter_parts.append( f"{in1}{in2}xfade=transition={trans}:duration={duration}:offset={offset}{out}" ) # 音频处理：统一格式（含 sample_fmts=fltp）+ 静音填充 + acrossfade 交叉淡入淡出 audio_filter_parts = [] for i in range(len(clips)): if clip_has_audio[i]: audio_filter_parts.append( f"[{i}:a]aformat=sample_fmts=fltp:sample_rates=44100:channel_layouts=stereo[a{i}_src]" ) else: d = max(0.1, clip_durations[i]) audio_filter_parts.append( f"anullsrc=r=44100:cl=stereo:d={d:.3f}[a{i}_src]" ) for i in range(len(clips) - 1): if i == 0: a_in1 = "[a0_src]" else: a_in1 = f"[a{i-1}]" a_in2 = f"[a{i+1}_src]" a_out = f"[a{i}]" audio_filter_parts.append( f"{a_in1}{a_in2}acrossfade=d={duration}{a_out}" ) filter_complex = ";".join(filter_parts + audio_filter_parts) audio_map = f"[a{len(clips)-2}]" cmd = ["ffmpeg", "-y"] + inputs + [ "-filter_complex", filter_complex, "-map", f"[v{len(clips)-2}]", "-map", audio_map, "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-g", "30", "-keyint_min", "30", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=ffmpeg_timeout) if result.returncode == 0 and os.path.exists(output_path) and os.path.getsize(output_path) > 10240: # 验证输出时长是否合理（至少应该是片段总时长的一半） expected_min_duration = sum(clip_durations) - duration * (len(clips) - 1) - 1.0 out_info = get_video_info(output_path) out_dur = out_info.get("duration", 0) if out_dur >= expected_min_duration * 0.5: return logger.warning( f"[concatenate_with_transition] xfade output too short: {out_dur:.2f}s " f"(expected ~{expected_min_duration:.2f}s), trying fallback. stderr: {result.stderr[:300]}" ) else: logger.warning( f"[concatenate_with_transition] xfade+acrossfade failed. " f"returncode={result.returncode}, stderr: {result.stderr[:500]}" ) except ValueError as e: if str(e) == "skip_xfade": pass # 主动跳过 xfade，继续走 concat filter fallback else: logger.warning(f"[concatenate_with_transition] xfade+acrossfade failed: {e}") except subprocess.TimeoutExpired: logger.warning(f"[concatenate_with_transition] xfade+acrossfade timed out ({ffmpeg_timeout}s)") except Exception as e: logger.warning(f"[concatenate_with_transition] xfade+acrossfade failed: {e}") # Fallback 1: concat filter（无转场效果，但保证音视频同步且所有片段有音频） try: inputs = [] for clip in clips: inputs.extend(["-i", clip]) filter_parts = [] # 视频统一尺寸+格式后 concat v_labels = [] for i in range(len(clips)): label = f"[v{i}]" v_labels.append(label) filter_parts.append( f"[{i}:v]scale={target_w}:{target_h},fps=30,format=yuv420p,setsar=1{label}" ) filter_parts.append(f"{''.join(v_labels)}concat=n={len(clips)}:v=1:a=0[vout]") # 音频统一格式（含 sample_fmts）+ 静音填充后 concat a_labels = [] for i in range(len(clips)): label = f"[a{i}]" a_labels.append(label) if clip_has_audio[i]: filter_parts.append(f"[{i}:a]aformat=sample_fmts=fltp:sample_rates=44100:channel_layouts=stereo{label}") else: d = max(0.1, clip_durations[i]) filter_parts.append(f"anullsrc=r=44100:cl=stereo:d={d:.3f}{label}") filter_parts.append(f"{''.join(a_labels)}concat=n={len(clips)}:v=0:a=1[aout]") filter_complex = ";".join(filter_parts) cmd = ["ffmpeg", "-y"] + inputs + [ "-filter_complex", filter_complex, "-map", "[vout]", "-map", "[aout]", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-g", "30", "-keyint_min", "30", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=ffmpeg_timeout) if result.returncode == 0 and os.path.exists(output_path) and os.path.getsize(output_path) > 10240: # 验证输出时长 expected_min_duration = sum(clip_durations) - 1.0 out_info = get_video_info(output_path) out_dur = out_info.get("duration", 0) if out_dur >= expected_min_duration * 0.5: return logger.warning( f"[concatenate_with_transition] concat filter output too short: {out_dur:.2f}s, " f"trying next fallback. stderr: {result.stderr[:300]}" ) else: logger.warning(f"[concatenate_with_transition] concat filter failed. stderr: {result.stderr[:500]}") except subprocess.TimeoutExpired: logger.warning(f"[concatenate_with_transition] concat filter timed out ({ffmpeg_timeout}s)") except Exception as e: logger.warning(f"[concatenate_with_transition] concat filter failed: {e}") # Fallback 2: concat demuxer + 重编码（无转场，最可靠） try: import tempfile tmp_dir = tempfile.mkdtemp(prefix="concat_") concat_file = os.path.join(tmp_dir, "concat.txt") with open(concat_file, "w") as f: for clip in clips: f.write(f"file '{clip}'\n") cmd = [ "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", concat_file, "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-g", "30", "-keyint_min", "30", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=ffmpeg_timeout) import shutil shutil.rmtree(tmp_dir, ignore_errors=True) if result.returncode == 0 and os.path.exists(output_path) and os.path.getsize(output_path) > 10240: # 验证输出时长 expected_min_duration = sum(clip_durations) - 1.0 out_info = get_video_info(output_path) out_dur = out_info.get("duration", 0) if out_dur >= expected_min_duration * 0.5: return logger.warning( f"[concatenate_with_transition] concat demuxer output too short: {out_dur:.2f}s, " f"trying per-clip re-encode fallback. stderr: {result.stderr[:300]}" ) else: logger.warning(f"[concatenate_with_transition] concat demuxer failed. stderr: {result.stderr[:500]}") except subprocess.TimeoutExpired: logger.warning(f"[concatenate_with_transition] concat demuxer timed out ({ffmpeg_timeout}s)") except Exception as e: logger.warning(f"[concatenate_with_transition] concat demuxer failed: {e}") # Fallback 3: 逐片段重编码到统一格式后再用 concat demuxer 合并 # 这是最可靠的方案：先标准化每个片段，再合并 try: import tempfile import shutil tmp_dir = tempfile.mkdtemp(prefix="concat_norm_") normalized_clips = [] success = True for i, clip in enumerate(clips): norm_path = os.path.join(tmp_dir, f"norm_{i:03d}.mp4") cmd = [ "ffmpeg", "-y", "-i", clip, "-vf", f"scale={target_w}:{target_h},fps=30,format=yuv420p,setsar=1", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-ar", "44100", "-ac", "2", "-movflags", "+faststart", norm_path, ] r = subprocess.run(cmd, capture_output=True, text=True, timeout=max(300, int(clip_durations[i] * 20))) if r.returncode != 0 or not os.path.exists(norm_path) or os.path.getsize(norm_path) < 1024: logger.warning(f"[concatenate_with_transition] normalize clip {i} failed: {r.stderr[:200]}") success = False break normalized_clips.append(norm_path) if success and len(normalized_clips) == len(clips): concat_file = os.path.join(tmp_dir, "concat.txt") with open(concat_file, "w") as f: for clip in normalized_clips: f.write(f"file '{clip}'\n") cmd = [ "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", concat_file, "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=ffmpeg_timeout) shutil.rmtree(tmp_dir, ignore_errors=True) if result.returncode == 0 and os.path.exists(output_path) and os.path.getsize(output_path) > 10240: out_info = get_video_info(output_path) out_dur = out_info.get("duration", 0) expected_min_duration = sum(clip_durations) - 1.0 if out_dur >= expected_min_duration * 0.5: logger.info(f"[concatenate_with_transition] per-clip re-encode fallback succeeded, duration={out_dur:.2f}s") return logger.warning(f"[concatenate_with_transition] per-clip re-encode output too short: {out_dur:.2f}s") else: logger.warning(f"[concatenate_with_transition] per-clip re-encode concat failed: {result.stderr[:300]}") else: shutil.rmtree(tmp_dir, ignore_errors=True) except subprocess.TimeoutExpired: logger.warning("[concatenate_with_transition] per-clip re-encode timed out") except Exception as e: logger.warning(f"[concatenate_with_transition] per-clip re-encode failed: {e}") # 所有方案都失败，抛出异常而非静默复制第一个片段 raise RuntimeError( f"[concatenate_with_transition] ALL methods failed! " f"clips={len(clips)}, durations={clip_durations}, timeout={ffmpeg_timeout}s. " f"Unable to concatenate highlight clips." ) def generate_hls(video_path: str, output_dir: str): os.makedirs(output_dir, exist_ok=True) cmd = [ "ffmpeg", "-y", "-i", video_path, "-c:v", "libx264", "-c:a", "aac", "-hls_time", "2", "-hls_list_size", "0", "-hls_segment_filename", os.path.join(output_dir, "segment_%03d.ts"), os.path.join(output_dir, "playlist.m3u8"), ] subprocess.run(cmd, capture_output=True, check=True, timeout=120) def generate_gif(video_path: str, output_path: str, width: int = 480, fps: int = 10): cmd = [ "ffmpeg", "-y", "-i", video_path, "-vf", f"fps={fps},scale={width}:-1:flags=lanczos", output_path, ] subprocess.run(cmd, capture_output=True, check=True, timeout=60) def crop_vertical(input_path: str, output_path: str, track_data: Dict[str, Any] = None): """竖屏裁剪：优先使用 pyautoflip 智能裁剪，回退到 FFmpeg 居中裁剪""" # 尝试使用 pyautoflip 智能裁剪 try: from pyautoflip import AutoFlip autoflip = AutoFlip( input_path=input_path, output_path=output_path, target_aspect_ratio="9:16", detection_mode="person", ) autoflip.process() if os.path.exists(output_path) and os.path.getsize(output_path) > 0: logger.info("[crop_vertical] pyautoflip 智能裁剪成功") return except ImportError: logger.info("[crop_vertical] pyautoflip 不可用，使用 FFmpeg 裁剪") except Exception as e: logger.warning(f"[crop_vertical] pyautoflip 失败，回退到 FFmpeg: {e}") # 原有的 FFmpeg 裁剪逻辑 info = get_video_info(input_path) width = info["width"] height = info["height"] if width <= 0 or height <= 0: import shutil shutil.copy2(input_path, output_path) return target_w = 1080 target_h = 1920 target_ratio = target_w / target_h use_tracking = ( track_data is not None and "track_center_x" in track_data and "track_center_y" in track_data ) if use_tracking: input_ratio = width / height if input_ratio > target_ratio: crop_h = height crop_w = int(height * target_ratio) else: crop_w = width crop_h = int(width / target_ratio) crop_w = min(crop_w, width) crop_h = min(crop_h, height) cx = int(track_data["track_center_x"] * width) cy = int(track_data["track_center_y"] * height) x = max(0, min(cx - crop_w // 2, width - crop_w)) y = max(0, min(cy - crop_h // 2, height - crop_h)) out_w = min(target_w, crop_w) out_h = min(target_h, crop_h) cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]crop={crop_w}:{crop_h}:{x}:{y},scale={out_w}:{out_h}[v]", "-map", "[v]", "-map", "0:a?", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] subprocess.run(cmd, capture_output=True, check=True, timeout=60) else: input_ratio = width / height if input_ratio > target_ratio: scale_w = target_w scale_h = int(target_w / input_ratio) else: scale_h = target_h scale_w = int(target_h * input_ratio) scale_w = scale_w + (scale_w % 2) scale_h = scale_h + (scale_h % 2) pad_x = (target_w - scale_w) // 2 pad_y = (target_h - scale_h) // 2 cmd = [ "ffmpeg", "-y", "-i", input_path, "-filter_complex", f"[0:v]scale={scale_w}:{scale_h},pad={target_w}:{target_h}:{pad_x}:{pad_y}:black[v]", "-map", "[v]", "-map", "0:a?", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-maxrate", "6M", "-bufsize", "12M", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", output_path, ] subprocess.run(cmd, capture_output=True, check=True, timeout=60) def interpolate_frames(input_path: str, output_path: str, target_fps: int = 60): # Skip minterpolate on CPU - extremely slow, minimal benefit logger.info("[interpolate_frames] Skipping frame interpolation on CPU for performance") import shutil shutil.copy2(input_path, output_path) # ── 片头片尾渲染引擎（numpy + PIL 逐帧渲染，炫酷+篮球+科技感）── # 设计：深蓝黑背景 + 透视科技网格 + 飘动粒子 + 发光篮球 + 旋转 HUD 环 + # 弹性弹出文字 + 扫描线 + 闪光 + 混入 sfx 音效 _IO_BG = (5, 8, 22) # 深蓝黑 _IO_ORANGE = (255, 107, 43) # 篮球橙 _IO_RED = (204, 0, 0) # 红 _IO_GOLD = (255, 215, 0) # 金 _IO_CYAN = (0, 229, 255) # 科技青 _IO_WHITE = (255, 255, 255) _IO_BALL_DARK = (180, 70, 20) # glow 缓存，避免每帧重复高斯模糊 _glow_cache: Dict[Tuple[int, Tuple[int, int, int], float], Any] = {} def _io_clamp(v, lo=0.0, hi=1.0): return max(lo, min(hi, v)) def _io_ease_out_cubic(t): return 1 - (1 - t) ** 3 def _io_ease_out_back(t, s=1.7): return 1 + (s + 1) * (t - 1) ** 3 + s * (t - 1) ** 2 def _io_make_basketball(size): """绘制带纹路和高光的篮球（RGBA PIL Image）""" from PIL import Image, ImageDraw, ImageFilter img = Image.new("RGBA", (size, size), (0, 0, 0, 0)) d = ImageDraw.Draw(img) cx = cy = size / 2 r = size / 2 - 2 # 球体径向渐变 for i in range(int(r), 0, -1): ratio = i / r cr = int(_IO_ORANGE[0] * (0.55 + 0.45 * (1 - ratio)) + _IO_BALL_DARK[0] * (ratio * 0.3)) cg = int(_IO_ORANGE[1] * (0.55 + 0.45 * (1 - ratio)) + _IO_BALL_DARK[1] * (ratio * 0.3)) cb = int(_IO_ORANGE[2] * (0.55 + 0.45 * (1 - ratio)) + _IO_BALL_DARK[2] * (ratio * 0.3)) d.ellipse([cx - i, cy - i, cx + i, cy + i], fill=(cr, cg, cb, 255)) # 高光 hl = Image.new("RGBA", (size, size), (0, 0, 0, 0)) hd = ImageDraw.Draw(hl) hd.ellipse([cx - r * 0.5, cy - r * 0.6, cx + r * 0.1, cy - r * 0.1], fill=(255, 230, 180, 90)) hl = hl.filter(ImageFilter.GaussianBlur(size // 12)) img = Image.alpha_composite(img, hl) d = ImageDraw.Draw(img) # 经纬线 lw = max(2, size // 60) d.line([cx, cy - r, cx, cy + r], fill=(20, 10, 0, 230), width=lw) d.line([cx - r, cy, cx + r, cy], fill=(20, 10, 0, 230), width=lw) bbox = [cx - r, cy - r, cx + r, cy + r] d.arc(bbox, start=30, end=150, fill=(20, 10, 0, 230), width=lw) d.arc(bbox, start=210, end=330, fill=(20, 10, 0, 230), width=lw) d.ellipse([cx - r, cy - r, cx + r, cy + r], outline=(30, 15, 0, 255), width=lw) return img def _io_make_glow(size, color, intensity=1.0): """生成发光圆斑（带缓存）""" from PIL import Image, ImageDraw, ImageFilter key = (size, color, round(intensity, 2)) if key in _glow_cache: return _glow_cache[key] size = max(4, int(size)) img = Image.new("RGBA", (size, size), (0, 0, 0, 0)) d = ImageDraw.Draw(img) cx = cy = size / 2 r = size / 2 for i in range(int(r), 0, -2): a = int(255 * intensity * (1 - i / r) ** 2.5) d.ellipse([cx - i, cy - i, cx + i, cy + i], fill=(color[0], color[1], color[2], a)) img = img.filter(ImageFilter.GaussianBlur(size // 8)) if len(_glow_cache) < 16: _glow_cache[key] = img return img def _io_draw_hud_rings(base, cx, cy, t, scale=1.0): """绘制旋转的科技 HUD 环""" from PIL import ImageDraw import math d = ImageDraw.Draw(base, "RGBA") r1 = 180 * scale r2 = 220 * scale r3 = 260 * scale rot = t * 60 rot2 = -t * 40 # 外环刻度 for i in range(60): ang = math.radians(rot + i * 6) long = (i % 5 == 0) r_in = r3 - (18 if long else 8) d.line([cx + r_in * math.cos(ang), cy + r_in * math.sin(ang), cx + r3 * math.cos(ang), cy + r3 * math.sin(ang)], fill=(*(_IO_CYAN if long else (0, 150, 200)), 200), width=2 if long else 1) # 中环弧 d.arc([cx - r2, cy - r2, cx + r2, cy + r2], start=rot, end=rot + 120, fill=(*_IO_ORANGE, 220), width=4) d.arc([cx - r2, cy - r2, cx + r2, cy + r2], start=rot + 180, end=rot + 260, fill=(*_IO_ORANGE, 180), width=4) # 内环刻度 for i in range(12): ang = math.radians(rot2 + i * 30) d.line([cx + (r1 - 14) * math.cos(ang), cy + (r1 - 14) * math.sin(ang), cx + r1 * math.cos(ang), cy + r1 * math.sin(ang)], fill=(*_IO_CYAN, 160), width=3) d.ellipse([cx - r1, cy - r1, cx + r1, cy + r1], outline=(*_IO_CYAN, 120), width=2) # 角标 for ang_deg in [45, 135, 225, 315]: ang = math.radians(ang_deg + rot * 0.3) px = cx + r3 * math.cos(ang) py = cy + r3 * math.sin(ang) sz = 16 * scale d.line([px - sz, py, px + sz, py], fill=(*_IO_GOLD, 220), width=3) d.line([px, py - sz, px, py + sz], fill=(*_IO_GOLD, 220), width=3) def _io_draw_grid(base, t, w, h): """透视科技网格背景""" from PIL import ImageDraw d = ImageDraw.Draw(base, "RGBA") horizon = h * 0.55 scroll = (t * 80) % 60 for i in range(20): ratio = (i * 60 + scroll) / (20 * 60) y = horizon + (h - horizon) * (ratio ** 2.2) if horizon < y < h: alpha = int(80 * (1 - ratio * 0.5)) d.line([0, y, w, y], fill=(*_IO_CYAN, alpha), width=1) vanish_x = w / 2 for i in range(-12, 13): x_bottom = w / 2 + i * 160 alpha = int(60 * (1 - abs(i) / 14)) d.line([vanish_x, horizon, x_bottom, h], fill=(*_IO_CYAN, alpha), width=1) for i in range(30): a = int(60 * (1 - i / 30)) d.line([0, horizon - i, w, horizon - i], fill=(*_IO_ORANGE, a), width=1) def _io_draw_glow_text(base, text, font, pos, color, glow_color, glow_radius=8, alpha=255, border_color=None, border_w=0): """绘制带发光+描边的文字""" from PIL import Image, ImageDraw, ImageFilter glow_img = Image.new("RGBA", base.size, (0, 0, 0, 0)) gd = ImageDraw.Draw(glow_img) gd.text(pos, text, font=font, fill=(*glow_color, 200), anchor="mm") glow_img = glow_img.filter(ImageFilter.GaussianBlur(glow_radius)) base.alpha_composite(glow_img) d = ImageDraw.Draw(base, "RGBA") if border_color and border_w > 0: for dx in range(-border_w, border_w + 1): for dy in range(-border_w, border_w + 1): if dx == 0 and dy == 0: continue d.text((pos[0] + dx, pos[1] + dy), text, font=font, fill=(*border_color, alpha), anchor="mm") d.text(pos, text, font=font, fill=(*color, alpha), anchor="mm") class _IOParticles: """粒子系统""" def __init__(self, n, w, h, seed=0): import numpy as np rng = np.random.default_rng(seed) self.n = n self.x = rng.uniform(0, w, n).astype(np.float32) self.y = rng.uniform(0, h, n).astype(np.float32) self.vx = rng.uniform(-15, 15, n).astype(np.float32) self.vy = rng.uniform(-25, -5, n).astype(np.float32) self.size = rng.uniform(1, 3, n).astype(np.float32) self.hue = rng.uniform(0, 1, n) self.w = w self.h = h self._rng = np.random.default_rng() def update(self, dt, upward_boost=0.0): import numpy as np self.x += self.vx * dt self.y += self.vy * dt - upward_boost * dt self.vy += 5 * dt mask = (self.y < -10) | (self.y > self.h + 10) | (self.x < -10) | (self.x > self.w + 10) if mask.any(): self.x[mask] = self._rng.uniform(0, self.w, mask.sum()) self.y[mask] = self.h + 5 self.vy[mask] = self._rng.uniform(-40, -10, mask.sum()) def draw(self, base): from PIL import ImageDraw d = ImageDraw.Draw(base, "RGBA") for i in range(self.n): r = self.size[i] col = _IO_ORANGE if self.hue[i] < 0.5 else _IO_GOLD a = int(180 * _io_clamp(1 - (self.h - self.y[i]) / self.h)) d.ellipse([self.x[i] - r, self.y[i] - r, self.x[i] + r, self.y[i] + r], fill=(*col, a)) def _io_render_intro_frame(t, duration, w, h, ball, particles, font_path): """渲染片头单帧""" from PIL import Image, ImageDraw, ImageFont import math frame = Image.new("RGBA", (w, h), (*_IO_BG, 255)) _io_draw_grid(frame, t * 0.5, w, h) particles.draw(frame) cx, cy = w * 0.32, h * 0.5 # 篮球入场 if t < 0.7: p = _io_ease_out_cubic(t / 0.7) scale = 0.2 + 0.8 * p glow = _io_make_glow(int(600 * p), _IO_ORANGE, 0.8) frame.alpha_composite(glow, (int(cx - glow.width / 2), int(cy - glow.height / 2))) else: scale = 1.0 + 0.03 * math.sin(t * 4) ball_rot = ball.rotate(t * 120, expand=False, resample=Image.BICUBIC) bs = int(ball.width * scale) if bs > 10: ball_sized = ball_rot.resize((bs, bs), Image.LANCZOS) glow = _io_make_glow(int(bs * 1.6), _IO_ORANGE, 0.5) frame.alpha_composite(glow, (int(cx - glow.width / 2), int(cy - glow.height / 2))) frame.alpha_composite(ball_sized, (int(cx - bs / 2), int(cy - bs / 2))) # HUD 环 hud_alpha = _io_clamp((t - 0.5) / 0.5) if hud_alpha > 0: hud_layer = Image.new("RGBA", (w, h), (0, 0, 0, 0)) _io_draw_hud_rings(hud_layer, cx, cy, t, 1.0) hud_layer.putalpha(hud_layer.split()[3].point(lambda a: int(a * hud_alpha))) frame.alpha_composite(hud_layer) # 文字 if t >= 0.9 and font_path: tx, ty = w * 0.68, h * 0.42 title_p = _io_ease_out_back(_io_clamp((t - 0.9) / 0.6)) title_alpha = int(255 * _io_clamp((t - 0.9) / 0.3)) ts = int(130 * title_p) if ts > 10: ft = ImageFont.truetype(font_path, ts) _io_draw_glow_text(frame, "篮球高光工坊", ft, (tx, ty), _IO_WHITE, _IO_ORANGE, glow_radius=12, alpha=title_alpha, border_color=_IO_RED, border_w=4) sub_p = _io_clamp((t - 1.3) / 0.4) if sub_p > 0: fs = ImageFont.truetype(font_path, 56) _io_draw_glow_text(frame, "BASKETBALL HIGHLIGHTS", fs, (tx, ty + 110), _IO_GOLD, _IO_ORANGE, glow_radius=6, alpha=int(255 * sub_p), border_color=(0, 0, 0), border_w=3) tag_p = _io_clamp((t - 1.7) / 0.4) if tag_p > 0: tag_alpha = int(255 * tag_p * (0.7 + 0.3 * math.sin(t * 5))) ftg = ImageFont.truetype(font_path, 40) _io_draw_glow_text(frame, "◆ HOOPS ◆ TECH ◆", ftg, (tx, ty + 190), _IO_CYAN, _IO_CYAN, glow_radius=4, alpha=tag_alpha) # 扫描线 if 1.8 < t < 2.6: sp = (t - 1.8) / 0.8 sy = int(h * sp) scan = Image.new("RGBA", (w, 60), (0, 0, 0, 0)) sd = ImageDraw.Draw(scan) for i in range(60): a = int(120 * (1 - abs(i - 30) / 30)) sd.line([0, i, w, i], fill=(*_IO_CYAN, a), width=1) frame.alpha_composite(scan, (0, sy - 30)) # 闪光 if 2.75 < t < 2.95: flash_a = int(180 * (1 - (t - 2.75) / 0.2)) frame.alpha_composite(Image.new("RGBA", (w, h), (255, 255, 255, flash_a))) # 淡入淡出 bg = Image.new("RGBA", (w, h), (*_IO_BG, 255)) if t < 0.2: a = int(255 * (t / 0.2)) frame.putalpha(frame.split()[3].point(lambda x: int(x * a / 255))) frame = Image.alpha_composite(bg, frame) if t > duration - 0.3: a = int(255 * (duration - t) / 0.3) frame.putalpha(frame.split()[3].point(lambda x: int(x * a / 255))) frame = Image.alpha_composite(bg, frame) return frame.convert("RGB") def _io_render_outro_frame(t, duration, w, h, ball, particles, font_path): """渲染片尾单帧""" from PIL import Image, ImageDraw, ImageFont import math import numpy as np frame = Image.new("RGBA", (w, h), (*_IO_BG, 255)) _io_draw_grid(frame, t * 0.4, w, h) particles.draw(frame) cx, cy = w * 0.5, h * 0.42 # 篮球弹入 if t < 0.8: bt = t / 0.8 bounce = abs(math.sin(bt * math.pi * 2)) scale = 0.6 + 0.4 * _io_ease_out_cubic(bt) cy_b = cy + (1 - bounce) * 60 else: scale = 1.0 + 0.03 * math.sin(t * 4) cy_b = cy ball_rot = ball.rotate(t * 90, expand=False, resample=Image.BICUBIC) bs = int(ball.width * scale) if bs > 10: ball_sized = ball_rot.resize((bs, bs), Image.LANCZOS) glow = _io_make_glow(int(bs * 1.5), _IO_ORANGE, 0.5) frame.alpha_composite(glow, (int(cx - glow.width / 2), int(cy_b - glow.height / 2))) frame.alpha_composite(ball_sized, (int(cx - bs / 2), int(cy_b - bs / 2))) # HUD 环 hud_alpha = _io_clamp((t - 0.4) / 0.5) if hud_alpha > 0: hud_layer = Image.new("RGBA", (w, h), (0, 0, 0, 0)) _io_draw_hud_rings(hud_layer, cx, cy_b, t, 1.0) hud_layer.putalpha(hud_layer.split()[3].point(lambda a: int(a * hud_alpha))) frame.alpha_composite(hud_layer) # 文字 if t >= 0.7 and font_path: ty = h * 0.72 title_p = _io_ease_out_back(_io_clamp((t - 0.7) / 0.5)) title_alpha = int(255 * _io_clamp((t - 0.7) / 0.3)) ts = int(130 * title_p) if ts > 10: ft = ImageFont.truetype(font_path, ts) _io_draw_glow_text(frame, "感谢观看", ft, (w / 2, ty), _IO_WHITE, _IO_ORANGE, glow_radius=12, alpha=title_alpha, border_color=_IO_RED, border_w=4) sub_p = _io_clamp((t - 1.1) / 0.4) if sub_p > 0: fs = ImageFont.truetype(font_path, 56) _io_draw_glow_text(frame, "更多高光 敬请期待", fs, (w / 2, ty + 100), _IO_GOLD, _IO_ORANGE, glow_radius=6, alpha=int(255 * sub_p), border_color=(0, 0, 0), border_w=3) # 庆祝粒子 if t >= 1.3: d = ImageDraw.Draw(frame, "RGBA") rng = np.random.default_rng(int(t * 10)) for _ in range(8): px = rng.uniform(w * 0.2, w * 0.8) py = h - (t - 1.3) * 200 - rng.uniform(0, 100) if py > 0: r = rng.uniform(2, 5) d.ellipse([px - r, py - r, px + r, py + r], fill=(*_IO_GOLD, 200)) # 淡入淡出 bg = Image.new("RGBA", (w, h), (*_IO_BG, 255)) if t < 0.2: a = int(255 * (t / 0.2)) frame.putalpha(frame.split()[3].point(lambda x: int(x * a / 255))) frame = Image.alpha_composite(bg, frame) if t > duration - 0.4: a = int(255 * (duration - t) / 0.4) frame.putalpha(frame.split()[3].point(lambda x: int(x * a / 255))) frame = Image.alpha_composite(bg, frame) return frame.convert("RGB") def _io_encode(render_fn, output_path, duration, w, h, font_path, audio_mix): """逐帧渲染并通过 stdin pipe 给 ffmpeg 编码，混入音效""" import subprocess fps = 30 n_frames = int(fps * duration) sfx_dir = os.path.join(_ASSETS_DIR, "sfx") cmd = ["ffmpeg", "-y", "-f", "rawvideo", "-pix_fmt", "rgb24", "-s", f"{w}x{h}", "-r", str(fps), "-i", "-"] valid_audio = [] for sfx_name, start, dur in audio_mix: p = os.path.join(sfx_dir, sfx_name) if os.path.exists(p): cmd += ["-i", p] valid_audio.append((start, dur)) audio_parts = [] for i, (start, dur) in enumerate(valid_audio, 1): audio_parts.append(f"[{i}:a]atrim=0:{dur},adelay={int(start*1000)}|{int(start*1000)},asetpts=N/SR[a{i}]") if audio_parts: amix = "".join(f"[a{i}]" for i in range(1, len(valid_audio) + 1)) filter_complex = ";".join(audio_parts) + f";{amix}amix=inputs={len(valid_audio)}:duration=longest[aout]" cmd += ["-filter_complex", filter_complex, "-map", "0:v", "-map", "[aout]"] else: cmd += ["-map", "0:v"] cmd += ["-c:v", "libx264", "-crf", "20", "-preset", "fast", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "192k", "-t", f"{duration}", "-movflags", "+faststart", output_path] proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stderr=subprocess.PIPE) try: ball = _io_make_basketball(400) particles = _IOParticles(60, w, h, seed=42) for i in range(n_frames): t = i / fps particles.update(1 / fps, upward_boost=20) frame = render_fn(t, duration, w, h, ball, particles, font_path) proc.stdin.write(frame.tobytes()) proc.stdin.close() except (BrokenPipeError, Exception) as e: logger.warning(f"[intro/outro] render pipe error: {e}") _, err = proc.communicate() if proc.returncode != 0: logger.warning(f"[intro/outro] ffmpeg failed: {err.decode()[-400:]}") return False return True def _try_use_pregenerated(clip_type: str, output_path: str, width: int, height: int) -> bool: """优先使用预生成的片头/片尾视频文件（效果与预览完全一致）。 预生成文件为 1920x1080@30fps，如果目标分辨率相同则直接复制； 否则用 ffmpeg scale 到目标分辨率。 """ import shutil src = os.path.join(_ASSETS_DIR, f"{clip_type}_1920x1080.mp4") if not os.path.isfile(src): return False if width == 1920 and height == 1080: shutil.copy2(src, output_path) logger.info(f"[{clip_type}] 使用预生成文件（直接复制）") return True # 分辨率不同，scale 后复制 try: cmd = ["ffmpeg", "-y", "-i", src, "-vf", f"scale={width}:{height}", "-c:v", "libx264", "-crf", "20", "-preset", "fast", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "192k", "-movflags", "+faststart", output_path] r = subprocess.run(cmd, capture_output=True, text=True, timeout=30) if r.returncode == 0 and os.path.exists(output_path) and os.path.getsize(output_path) > 0: logger.info(f"[{clip_type}] 使用预生成文件（scale 到 {width}x{height}）") return True logger.warning(f"[{clip_type}] scale failed: {r.stderr[:200]}") except Exception as e: logger.warning(f"[{clip_type}] scale error: {e}") return False def generate_intro(output_path: str, duration: float = 3.0, width: int = 1920, height: int = 1080): """生成炫酷的篮球科技风片头（3秒） 优先使用预生成文件（效果与预览完全一致），失败则动态渲染，再失败回退到简版。 """ # 方案1：预生成文件 if _try_use_pregenerated("intro", output_path, width, height): return # 方案2：numpy+PIL 动态渲染 font_path = _find_cjk_font() audio_mix = [ ("impact_01.mp3", 0.0, 0.8), ("whistle_swoosh.mp3", 0.6, 1.0), ("basketball_buzzer.mp3", 2.7, 0.5), ] try: ok = _io_encode(_io_render_intro_frame, output_path, duration, width, height, font_path, audio_mix) if not ok: logger.warning("[generate_intro] numpy+PIL render failed, fallback to simple") _generate_intro_simple(output_path, duration, width, height) except Exception as e: logger.warning(f"[generate_intro] render failed: {e}, fallback to simple") _generate_intro_simple(output_path, duration, width, height) def _generate_intro_simple(output_path: str, duration: float, width: int, height: int): """简单版片头（回退方案）""" fontfile = _find_cjk_font() cmd = [ "ffmpeg", "-y", "-f", "lavfi", "-i", f"color=c=0x0a0014:s={width}x{height}:d={duration}:r=30", "-filter_complex", f"[0:v]drawtext=text='篮球高光工坊':fontfile={fontfile}:" f"fontcolor=white:fontsize=140:borderw=8:bordercolor=0xCC0000:" f"x=(w-text_w)/2:y=(h-text_h)/2-60," f"drawtext=text='BASKETBALL HIGHLIGHTS':fontfile={fontfile}:" f"fontcolor=0xFFD700:fontsize=70:borderw=4:bordercolor=black:" f"x=(w-text_w)/2:y=(h-text_h)/2+100," f"fade=in:0:15,fade=out:st={duration-0.5}:d=0.5[v]", "-map", "[v]", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-pix_fmt", "yuv420p", "-movflags", "+faststart", output_path, ] subprocess.run(cmd, capture_output=True, check=True, timeout=30) def generate_outro(output_path: str, duration: float = 3.0, width: int = 1920, height: int = 1080): """生成炫酷的篮球科技风片尾（3秒） 优先使用预生成文件（效果与预览完全一致），失败则动态渲染，再失败回退到简版。 """ # 方案1：预生成文件 if _try_use_pregenerated("outro", output_path, width, height): return # 方案2：numpy+PIL 动态渲染 font_path = _find_cjk_font() audio_mix = [ ("basketball_bounce_01.mp3", 0.0, 0.8), ("crowd_cheering.mp3", 0.7, 1.8), ("whistle_short.mp3", 2.5, 0.5), ] try: ok = _io_encode(_io_render_outro_frame, output_path, duration, width, height, font_path, audio_mix) if not ok: logger.warning("[generate_outro] numpy+PIL render failed, fallback to simple") _generate_outro_simple(output_path, duration, width, height) except Exception as e: logger.warning(f"[generate_outro] render failed: {e}, fallback to simple") _generate_outro_simple(output_path, duration, width, height) def _generate_outro_simple(output_path: str, duration: float, width: int, height: int): """简单版片尾（回退方案）""" fontfile = _find_cjk_font() cmd = [ "ffmpeg", "-y", "-f", "lavfi", "-i", f"color=c=0x0a0014:s={width}x{height}:d={duration}:r=30", "-filter_complex", f"[0:v]drawtext=text='感谢观看':fontfile={fontfile}:" f"fontcolor=white:fontsize=140:borderw=8:bordercolor=0xCC0000:" f"x=(w-text_w)/2:y=(h-text_h)/2-60," f"drawtext=text='更多高光敬请期待':fontfile={fontfile}:" f"fontcolor=0xFFD700:fontsize=70:borderw=4:bordercolor=black:" f"x=(w-text_w)/2:y=(h-text_h)/2+100," f"fade=in:0:10,fade=out:st={duration-0.5}:d=0.5[v]", "-map", "[v]", "-c:v", "libx264", "-crf", "24", "-preset", "fast", "-pix_fmt", "yuv420p", "-movflags", "+faststart", output_path, ] subprocess.run(cmd, capture_output=True, check=True, timeout=30) def generate_sound_effect(effect_type: str, duration: float, output_path: str): # Try to use real SFX file from assets/sfx/ real_sfx = _find_real_sfx(effect_type) if real_sfx: logger.info(f"[SFX] Using real audio file: {real_sfx}") try: # Trim the real SFX to the desired duration fade_out_start = max(0, duration - 0.1) cmd = [ "ffmpeg", "-y", "-i", real_sfx, "-t", f"{duration:.2f}", "-af", f"afade=t=in:st=0:d=0.05,afade=t=out:st={fade_out_start:.2f}:d=0.1", "-c:a", "aac", "-b:a", "64k", output_path, ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode == 0: return logger.warning(f"[SFX] Real SFX failed: {result.stderr[:200]}") except Exception as e: logger.warning(f"[SFX] Failed to use real SFX {real_sfx}: {e}, falling back to synthetic") # Fallback: generate synthetic sound effect fade_out_start = max(0, duration - 0.1) if effect_type == "swish": cmd = [ "ffmpeg", "-y", "-f", "lavfi", "-i", f"anoisesrc=d={duration}:c=white:r=44100:a=0.3", "-filter_complex", f"[0:a]highpass=f=2000,lowpass=f=8000," f"afade=t=in:st=0:d=0.05,afade=t=out:st={fade_out_start:.2f}:d=0.1," f"asetrate=44100,aformat=sample_fmts=s16[out]", "-map", "[out]", "-c:a", "aac", "-b:a", "64k", output_path, ] elif effect_type == "slam": cmd = [ "ffmpeg", "-y", "-f", "lavfi", "-i", f"anoisesrc=d={duration}:c=brown:r=44100:a=0.5", "-filter_complex", f"[0:a]lowpass=f=500," f"afade=t=in:st=0:d=0.02,afade=t=out:st={fade_out_start:.2f}:d=0.15," f"asetrate=44100,aformat=sample_fmts=s16[out]", "-map", "[out]", "-c:a", "aac", "-b:a", "64k", output_path, ] elif effect_type == "whistle": cmd = [ "ffmpeg", "-y", "-f", "lavfi", "-i", f"sine=frequency=3500:duration={duration}", "-filter_complex", f"[0:a]volume=0.2," f"afade=t=in:st=0:d=0.05,afade=t=out:st={fade_out_start:.2f}:d=0.1," f"asetrate=44100,aformat=sample_fmts=s16[out]", "-map", "[out]", "-c:a", "aac", "-b:a", "64k", output_path, ] else: cmd = [ "ffmpeg", "-y", "-f", "lavfi", "-i", f"anoisesrc=d={duration}:c=pink:r=44100:a=0.1", "-filter_complex", f"[0:a]lowpass=f=1000,afade=t=in:st=0:d=0.05,afade=t=out:st={fade_out_start:.2f}:d=0.1[out]", "-map", "[out]", "-c:a", "aac", "-b:a", "64k", output_path, ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: logger.warning(f"[SFX] Synthetic SFX generation failed: {result.stderr[:200]}") # Last resort: create a minimal silent file cmd = [ "ffmpeg", "-y", "-f", "lavfi", "-i", f"anullsrc=r=44100:cl=mono", "-t", f"{duration:.2f}", "-c:a", "aac", "-b:a", "64k", output_path, ] subprocess.run(cmd, capture_output=True, check=True) def add_sound_effect(video_path: str, effect_type: str, output_path: str, offset: float = 0.0): sfx_path = video_path + f".{effect_type}.m4a" try: generate_sound_effect(effect_type, 0.8, sfx_path) # 检查视频是否有音频流 probe_cmd = ["ffprobe", "-v", "error", "-select_streams", "a", "-show_entries", "stream=codec_name", "-of", "csv=p=0", video_path] probe_result = subprocess.run(probe_cmd, capture_output=True, text=True, timeout=10) has_audio = bool(probe_result.stdout.strip()) if has_audio: # 有音频：混音 cmd = [ "ffmpeg", "-y", "-i", video_path, "-i", sfx_path, "-filter_complex", f"[1:a]adelay={int(offset*1000)}|{int(offset*1000)},volume=0.5[sfx];" f"[0:a][sfx]amix=inputs=2:duration=first:dropout_transition=2[a]", "-map", "0:v", "-map", "[a]", "-c:v", "copy", "-c:a", "aac", "-shortest", output_path, ] else: # 无音频：直接添加音效作为音轨 cmd = [ "ffmpeg", "-y", "-i", video_path, "-i", sfx_path, "-filter_complex", f"[1:a]adelay={int(offset*1000)}|{int(offset*1000)},volume=0.5[a]", "-map", "0:v", "-map", "[a]", "-c:v", "copy", "-c:a", "aac", "-shortest", output_path, ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: import shutil shutil.copy2(video_path, output_path) except Exception: import shutil shutil.copy2(video_path, output_path) finally: if os.path.exists(sfx_path): os.remove(sfx_path) def _probe_streams(path: str) -> dict: """探测视频文件的视频和音频流参数""" info = {"has_video": False, "has_audio": False, "width": 0, "height": 0, "fps": 0, "v_codec": "", "pix_fmt": "", "a_codec": "", "sample_rate": 0, "channels": 0, "duration": 0} try: cmd = ["ffprobe", "-v", "quiet", "-print_format", "json", "-show_streams", "-show_format", path] r = subprocess.run(cmd, capture_output=True, text=True, timeout=10) if r.returncode == 0: data = json.loads(r.stdout) info["duration"] = float(data.get("format", {}).get("duration", 0)) for s in data.get("streams", []): if s["codec_type"] == "video": info["has_video"] = True info["width"] = int(s.get("width", 0)) info["height"] = int(s.get("height", 0)) info["v_codec"] = s.get("codec_name", "") info["pix_fmt"] = s.get("pix_fmt", "") fr = s.get("r_frame_rate", "30/1").split("/") if len(fr) == 2 and int(fr[1]) > 0: info["fps"] = round(int(fr[0]) / int(fr[1])) elif s["codec_type"] == "audio": info["has_audio"] = True info["a_codec"] = s.get("codec_name", "") info["sample_rate"] = int(s.get("sample_rate", 0)) info["channels"] = int(s.get("channels", 0)) except Exception: pass return info def _concat_demuxer(files: list, output_path: str, timeout: int = 60) -> bool: """使用 concat demuxer + -c copy 拼接视频（不重编码，极快）""" import tempfile list_path = os.path.join(os.path.dirname(output_path), f"_concat_{os.path.basename(output_path)}.txt") try: with open(list_path, "w") as f: for fp in files: # 转义单引号 safe = fp.replace("'", r"'\''") f.write(f"file '{safe}'\n") cmd = ["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", list_path, "-c", "copy", "-fflags", "+genpts", "-avoid_negative_ts", "make_zero", "-movflags", "+faststart", output_path] r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) if r.returncode == 0 and os.path.exists(output_path) and os.path.getsize(output_path) > 0: return True logger.warning(f"[concat_demuxer] failed: {r.stderr[:300]}") return False except Exception as e: logger.warning(f"[concat_demuxer] error: {e}") return False finally: if os.path.exists(list_path): os.remove(list_path) def _normalize_clip(clip_path: str, ref_info: dict, output_path: str, timeout: int = 60) -> bool: """将片头/片尾的参数统一到与主视频一致（只重编码短片段，很快）""" clip_info = _probe_streams(clip_path) clip_duration = clip_info.get("duration", 3.0) or 3.0 filters = [] if ref_info.get("width") and ref_info.get("height"): filters.append(f"scale={ref_info['width']}:{ref_info['height']}") filters.append("fps=30") filters.append("format=yuv420p") vf = ",".join(filters) cmd = ["ffmpeg", "-y", "-i", clip_path] if ref_info.get("has_audio"): sr = ref_info.get("sample_rate", 44100) or 44100 ch = ref_info.get("channels", 2) or 2 cmd += ["-f", "lavfi", "-t", f"{clip_duration}", "-i", f"anullsrc=r={sr}:cl={'stereo' if ch >= 2 else 'mono'}"] cmd += ["-filter_complex", f"[0:v]{vf}[v];[1:a][0:a?]amix=inputs=2:duration=first:dropout_transition=0[a]"] cmd += ["-map", "[v]", "-map", "[a]"] else: cmd += ["-vf", vf, "-map", "0:v"] cmd += ["-c:v", "libx264", "-crf", "20", "-preset", "fast", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "192k", "-r", "30", "-video_track_timescale", "15360", "-x264-params", "bframes=0", "-g", "30", "-keyint_min", "30", "-fflags", "+genpts", "-avoid_negative_ts", "make_zero", "-movflags", "+faststart", output_path] try: r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) if r.returncode == 0 and os.path.exists(output_path) and os.path.getsize(output_path) > 0: return True logger.warning(f"[normalize_clip] failed: {r.stderr[:300]}") return False except Exception as e: logger.warning(f"[normalize_clip] error: {e}") return False def prepend_intro(main_path: str, intro_path: str, output_path: str): """将片头拼接到主视频前面。使用 concat filter 重编码确保时间戳连续。""" main_info = _probe_streams(main_path) intro_info = _probe_streams(intro_path) intro_duration = intro_info.get("duration", 3.0) or 3.0 has_main_audio = main_info.get("has_audio", False) has_intro_audio = intro_info.get("has_audio", False) # 统一使用主视频的音频参数 sr = main_info.get("sample_rate", 44100) or 44100 ch = 2 if main_info.get("channels", 2) >= 2 else 1 ch_str = "stereo" if ch >= 2 else "mono" ch_layout = "stereo" if ch >= 2 else "mono" # concat filter 重编码（禁用B帧确保PTS严格递增，浏览器兼容性好） # 用 aformat 统一音频参数，避免采样率/通道不一致导致静音 try: if has_main_audio and has_intro_audio: # 两者都有音频，统一格式后 concat filter_complex = ( f"[0:a]aformat=sample_rates={sr}:channel_layouts={ch_layout}[a0];" f"[1:a]aformat=sample_rates={sr}:channel_layouts={ch_layout}[a1];" f"[0:v][1:v]concat=n=2:v=1:a=0[v];" f"[a0][a1]concat=n=2:v=0:a=1[a]" ) cmd = [ "ffmpeg", "-y", "-i", intro_path, "-i", main_path, "-filter_complex", filter_complex, "-map", "[v]", "-map", "[a]", ] elif has_main_audio: # 主视频有音频，片头无音频：用静音补片头 filter_complex = ( f"anullsrc=r={sr}:cl={ch_str}[silent];" f"[silent]atrim=0:{intro_duration},aformat=sample_rates={sr}:channel_layouts={ch_layout}[sil];" f"[1:a]aformat=sample_rates={sr}:channel_layouts={ch_layout}[a1];" f"[0:v][1:v]concat=n=2:v=1:a=0[v];" f"[sil][a1]concat=n=2:v=0:a=1[a]" ) cmd = [ "ffmpeg", "-y", "-i", intro_path, "-i", main_path, "-filter_complex", filter_complex, "-map", "[v]", "-map", "[a]", ] else: # 主视频无音频，只拼视频 filter_complex = "[0:v][1:v]concat=n=2:v=1:a=0[v]" cmd = [ "ffmpeg", "-y", "-i", intro_path, "-i", main_path, "-filter_complex", filter_complex, "-map", "[v]", ] cmd += [ "-c:v", "libx264", "-crf", "22", "-preset", "ultrafast", "-pix_fmt", "yuv420p", "-x264-params", "bframes=0", "-video_track_timescale", "15360", "-r", "30", "-g", "30", "-keyint_min", "30", "-fflags", "+genpts", "-avoid_negative_ts", "make_zero", ] if has_main_audio: cmd += ["-c:a", "aac", "-b:a", "128k", "-ar", str(sr), "-ac", str(ch)] cmd += ["-movflags", "+faststart", output_path] result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) if result.returncode == 0 and os.path.exists(output_path) and os.path.getsize(output_path) > 0: logger.info("[prepend_intro] concat filter succeeded") return logger.warning(f"[prepend_intro] concat filter failed: {result.stderr[:300]}") except Exception as e: logger.warning(f"[prepend_intro] concat filter error: {e}") # 回退：直接复制主视频 import shutil logger.warning("[prepend_intro] all methods failed, copying main video") shutil.copy2(main_path, output_path) def append_outro(main_path: str, outro_path: str, output_path: str): """将片尾拼接到主视频后面。使用 concat filter 重编码确保时间戳连续。""" main_info = _probe_streams(main_path) outro_info = _probe_streams(outro_path) outro_duration = outro_info.get("duration", 3.0) or 3.0 has_main_audio = main_info.get("has_audio", False) has_outro_audio = outro_info.get("has_audio", False) sr = main_info.get("sample_rate", 44100) or 44100 ch = 2 if main_info.get("channels", 2) >= 2 else 1 ch_str = "stereo" if ch >= 2 else "mono" ch_layout = "stereo" if ch >= 2 else "mono" try: if has_main_audio and has_outro_audio: filter_complex = ( f"[0:a]aformat=sample_rates={sr}:channel_layouts={ch_layout}[a0];" f"[1:a]aformat=sample_rates={sr}:channel_layouts={ch_layout}[a1];" f"[0:v][1:v]concat=n=2:v=1:a=0[v];" f"[a0][a1]concat=n=2:v=0:a=1[a]" ) cmd = [ "ffmpeg", "-y", "-i", main_path, "-i", outro_path, "-filter_complex", filter_complex, "-map", "[v]", "-map", "[a]", ] elif has_main_audio: filter_complex = ( f"anullsrc=r={sr}:cl={ch_str}[silent];" f"[silent]atrim=0:{outro_duration},aformat=sample_rates={sr}:channel_layouts={ch_layout}[sil];" f"[0:a]aformat=sample_rates={sr}:channel_layouts={ch_layout}[a0];" f"[0:v][1:v]concat=n=2:v=1:a=0[v];" f"[a0][sil]concat=n=2:v=0:a=1[a]" ) cmd = [ "ffmpeg", "-y", "-i", main_path, "-i", outro_path, "-filter_complex", filter_complex, "-map", "[v]", "-map", "[a]", ] else: filter_complex = "[0:v][1:v]concat=n=2:v=1:a=0[v]" cmd = [ "ffmpeg", "-y", "-i", main_path, "-i", outro_path, "-filter_complex", filter_complex, "-map", "[v]", ] cmd += [ "-c:v", "libx264", "-crf", "22", "-preset", "ultrafast", "-pix_fmt", "yuv420p", "-x264-params", "bframes=0", "-video_track_timescale", "15360", "-r", "30", "-g", "30", "-keyint_min", "30", "-fflags", "+genpts", "-avoid_negative_ts", "make_zero", ] if has_main_audio: cmd += ["-c:a", "aac", "-b:a", "128k", "-ar", str(sr), "-ac", str(ch)] cmd += ["-movflags", "+faststart", output_path] result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) if result.returncode == 0 and os.path.exists(output_path) and os.path.getsize(output_path) > 0: logger.info("[append_outro] concat filter succeeded") return logger.warning(f"[append_outro] concat filter failed: {result.stderr[:300]}") except Exception as e: logger.warning(f"[append_outro] concat filter error: {e}") # 回退：直接复制主视频 import shutil logger.warning("[append_outro] all methods failed, copying main video") shutil.copy2(main_path, output_path) def generate_ambient_bgm(duration: float, output_path: str): """Generate background music for the highlight video. Priority: 1. Use a real BGM file from assets/bgm/ (trimmed/faded to match duration) 2. Fallback: generate beat-driven synthetic BGM with FFmpeg """ # Try to use real BGM from assets real_bgm = _find_real_bgm() if real_bgm: logger.info(f"[BGM] Using real audio file: {real_bgm}") try: dur = max(1.0, duration + 4.0) fade_in = 2.0 fade_out = 3.0 cmd = [ "ffmpeg", "-y", "-i", real_bgm, "-t", str(dur), "-af", f"afade=t=in:st=0:d={fade_in},afade=t=out:st={dur - fade_out}:d={fade_out},volume=0.5", "-c:a", "aac", "-b:a", "128k", output_path, ] subprocess.run(cmd, capture_output=True, check=True) return except Exception as e: logger.warning(f"[BGM] Failed to use real BGM {real_bgm}: {e}, falling back to synthetic") # Fallback: generate beat-driven synthetic BGM _generate_synthetic_bgm(duration, output_path) def _generate_synthetic_bgm(duration: float, output_path: str): """Generate a beat-driven synthetic BGM using FFmpeg aevalsrc. Creates a hip-hop/sports style beat with: - Kick drum on every beat (exponential decay envelope) - Snare on beats 2 and 4 (offset pattern) - Hi-hat on every 8th note (fast decay) - Sustained bass line - Atmospheric pad chord """ dur = max(1.0, duration + 4.0) bpm = 90 beat = 60.0 / bpm # ~0.667s per beat hbeat = beat / 2 # ~0.333s per half beat # Kick drum: low freq burst with exponential decay, repeating every beat kick_expr = f"0.4*exp(-12*mod(t,{beat}))*sin(2*PI*60*t)" # Snare: mid-freq burst on beats 2 and 4 (offset by half beat, period = 2 beats) snare_expr = f"0.25*exp(-15*mod(t+{hbeat},{beat*2}))*sin(2*PI*200*t)" # Hi-hat: high freq burst every half beat with fast decay hihat_expr = f"0.12*exp(-25*mod(t,{hbeat}))*sin(2*PI*9000*t)" # Bass: sustained low note bass_expr = f"0.18*sin(2*PI*55*t)" # Pad: soft chord (A minor: A3, C4, E4) pad_expr = f"0.04*sin(2*PI*220*t)+0.03*sin(2*PI*277*t)+0.03*sin(2*PI*330*t)" filter_complex = ( f"aevalsrc='{kick_expr}':s=44100:d={dur}," f"afade=t=in:st=0:d=2,afade=t=out:st={dur-3}:d=3[kick];" f"aevalsrc='{snare_expr}':s=44100:d={dur}," f"afade=t=in:st=0:d=2,afade=t=out:st={dur-3}:d=3[snare];" f"aevalsrc='{hihat_expr}':s=44100:d={dur}," f"afade=t=in:st=0:d=2,afade=t=out:st={dur-3}:d=3[hihat];" f"aevalsrc='{bass_expr}':s=44100:d={dur}," f"afade=t=in:st=0:d=2,afade=t=out:st={dur-3}:d=3[bass];" f"aevalsrc='{pad_expr}':s=44100:d={dur}," f"afade=t=in:st=0:d=3,afade=t=out:st={dur-3}:d=3[pad];" f"[kick][snare][hihat][bass][pad]amix=inputs=5:duration=longest," f"lowpass=f=4000,highpass=f=30," f"afade=t=in:st=0:d=2,afade=t=out:st={dur-3}:d=3," f"aformat=sample_fmts=s16:sample_rates=44100[out]" ) cmd = [ "ffmpeg", "-y", "-filter_complex", filter_complex, "-map", "[out]", "-c:a", "aac", "-b:a", "128k", output_path, ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: logger.warning(f"[BGM] Beat-driven synthetic BGM failed: {result.stderr[:300]}") # Ultimate fallback: simple ambient drone _generate_simple_bgm(duration, output_path) def _generate_simple_bgm(duration: float, output_path: str): """Simple ambient BGM fallback (original sine wave approach).""" dur = max(1.0, duration + 4.0) cmd = [ "ffmpeg", "-y", "-f", "lavfi", "-i", f"sine=frequency=110:duration={dur}", "-f", "lavfi", "-i", f"sine=frequency=165:duration={dur}", "-f", "lavfi", "-i", f"sine=frequency=220:duration={dur}", "-f", "lavfi", "-i", f"anoisesrc=d={dur}:c=pink:r=22050:a=0.02", "-filter_complex", f"[0:a]volume=0.15[bass];" f"[1:a]volume=0.10[mid];" f"[2:a]volume=0.08[high];" f"[3:a]lowpass=f=400,volume=0.05[noise];" f"[bass][mid][high][noise]amix=inputs=4:duration=longest," f"lowpass=f=1200,afade=t=in:st=0:d=3,afade=t=out:st={dur-3}:d=3[out]", "-map", "[out]", "-c:a", "aac", "-b:a", "128k", output_path, ] subprocess.run(cmd, capture_output=True, check=True) def detect_beat_times(audio_path: str) -> List[float]: """用 librosa 检测音频节拍时间点 返回: 节拍时间列表，如 [0.23, 0.69, 1.15, ...] """ try: import librosa y, sr = librosa.load(audio_path, sr=22050) tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr) beat_times = librosa.frames_to_time(beat_frames, sr=sr).tolist() logger.info(f"[BeatSync] Detected {len(beat_times)} beats, tempo={tempo:.1f} BPM") return beat_times except ImportError: logger.warning("[BeatSync] librosa not available, skipping beat detection") return [] except Exception as e: logger.warning(f"[BeatSync] Beat detection failed: {e}") return [] def align_to_beat(time_point: float, beat_times: List[float], tolerance: float = 0.5) -> float: """将时间点对齐到最近的节拍点 tolerance: 最大对齐距离（秒），超过则不对齐 """ if not beat_times: return time_point closest = min(beat_times, key=lambda b: abs(b - time_point)) if abs(closest - time_point) <= tolerance: return closest return time_point def mix_audio(video_path: str, bgm_path: str, output_path: str, bgm_volume: float = 0.3, video_volume: float = 0.8): probe_cmd = ["ffprobe", "-v", "error", "-select_streams", "a", "-show_entries", "stream=codec_name", "-of", "csv=p=0", video_path] result = subprocess.run(probe_cmd, capture_output=True, text=True) if not result.stdout.strip(): cmd = [ "ffmpeg", "-y", "-i", video_path, "-i", bgm_path, "-filter_complex", f"[1:a]volume={bgm_volume},afade=t=in:st=0:d=2,afade=t=out:st=30:d=2[a2];" f"[0:v]copy[v]", "-map", "[v]", "-map", "[a2]", "-c:v", "copy", "-c:a", "aac", "-shortest", output_path, ] subprocess.run(cmd, capture_output=True, check=True) return cmd = [ "ffmpeg", "-y", "-i", video_path, "-i", bgm_path, "-filter_complex", f"[0:a]volume={video_volume}[a1];" f"[1:a]volume={bgm_volume},afade=t=in:st=0:d=2,afade=t=out:st=30:d=2[a2];" f"[a2]sidechaincompress=threshold=0.15:ratio=6:attack=10:release=200[bgm_ducked];" f"[a1][bgm_ducked]amix=inputs=2:duration=first:dropout_transition=3[a]", "-map", "0:v", "-map", "[a]", "-c:v", "copy", "-c:a", "aac", "-shortest", output_path, ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: logger.warning(f"[mix_audio] sidechaincompress failed, falling back to simple mix: {result.stderr[:200]}") cmd = [ "ffmpeg", "-y", "-i", video_path, "-i", bgm_path, "-filter_complex", f"[0:a]volume={video_volume}[a1];" f"[1:a]volume={bgm_volume},afade=t=in:st=0:d=2,afade=t=out:st=30:d=2[a2];" f"[a1][a2]amix=inputs=2:duration=first[a]", "-map", "0:v", "-map", "[a]", "-c:v", "copy", "-c:a", "aac", "-shortest", output_path, ] subprocess.run(cmd, capture_output=True, check=True)