| 1 | # Copyright (C) 2025 AIDC-AI |
| 2 | # |
| 3 | # Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | # you may not use this file except in compliance with the License. |
| 5 | # You may obtain a copy of the License at |
| 6 | # http://www.apache.org/licenses/LICENSE-2.0 |
| 7 | # Unless required by applicable law or agreed to in writing, software |
| 8 | # distributed under the License is distributed on an "AS IS" BASIS, |
| 9 | # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 10 | # See the License for the specific language governing permissions and |
| 11 | # limitations under the License. |
| 12 | |
| 13 | """ |
| 14 | Video Processing Service |
| 15 | |
| 16 | High-performance video composition service built on ffmpeg-python. |
| 17 | |
| 18 | Features: |
| 19 | - Video concatenation |
| 20 | - Audio/video merging |
| 21 | - Background music addition |
| 22 | - Image to video conversion |
| 23 | |
| 24 | Note: Requires FFmpeg to be installed on the system. |
| 25 | """ |
| 26 | |
| 27 | import os |
| 28 | import shutil |
| 29 | import tempfile |
| 30 | import uuid |
| 31 | from pathlib import Path |
| 32 | from typing import List, Literal, Optional |
| 33 | |
| 34 | import ffmpeg |
| 35 | from loguru import logger |
| 36 | |
| 37 | from pixelle_video.utils.os_util import ( |
| 38 | get_resource_path, |
| 39 | list_resource_files, |
| 40 | resource_exists |
| 41 | ) |
| 42 | |
| 43 | |
| 44 | def check_ffmpeg() -> None: |
| 45 | """ |
| 46 | Check if FFmpeg is installed on the system |
| 47 | |
| 48 | Raises: |
| 49 | RuntimeError: If FFmpeg is not found |
| 50 | """ |
| 51 | if not shutil.which("ffmpeg"): |
| 52 | raise RuntimeError( |
| 53 | "FFmpeg not found. Please install it:\n" |
| 54 | " macOS: brew install ffmpeg\n" |
| 55 | " Ubuntu/Debian: apt-get install ffmpeg\n" |
| 56 | " Windows: https://ffmpeg.org/download.html" |
| 57 | ) |
| 58 | |
| 59 | |
| 60 | class VideoService: |
| 61 | """ |
| 62 | Video compositor for common video processing tasks |
| 63 | |
| 64 | Uses ffmpeg-python for high-performance video processing. |
| 65 | All operations preserve video quality when possible (stream copy). |
| 66 | |
| 67 | Examples: |
| 68 | >>> compositor = VideoCompositor() |
| 69 | >>> |
| 70 | >>> # Concatenate videos |
| 71 | >>> compositor.concat_videos( |
| 72 | ... ["intro.mp4", "main.mp4", "outro.mp4"], |
| 73 | ... "final.mp4" |
| 74 | ... ) |
| 75 | >>> |
| 76 | >>> # Add voiceover |
| 77 | >>> compositor.merge_audio_video( |
| 78 | ... "visual.mp4", |
| 79 | ... "voiceover.mp3", |
| 80 | ... "final.mp4" |
| 81 | ... ) |
| 82 | >>> |
| 83 | >>> # Add background music |
| 84 | >>> compositor.add_bgm( |
| 85 | ... "video.mp4", |
| 86 | ... "music.mp3", |
| 87 | ... "final.mp4", |
| 88 | ... bgm_volume=0.3 |
| 89 | ... ) |
| 90 | >>> |
| 91 | >>> # Create video from image + audio |
| 92 | >>> compositor.create_video_from_image( |
| 93 | ... "frame.png", |
| 94 | ... "narration.mp3", |
| 95 | ... "segment.mp4" |
| 96 | ... ) |
| 97 | """ |
| 98 | |
| 99 | def __init__(self): |
| 100 | self._ffmpeg_checked = False |
| 101 | |
| 102 | def _ensure_ffmpeg(self): |
| 103 | """Lazily check FFmpeg availability on first use, not at import time""" |
| 104 | if not self._ffmpeg_checked: |
| 105 | check_ffmpeg() |
| 106 | self._ffmpeg_checked = True |
| 107 | |
| 108 | def concat_videos( |
| 109 | self, |
| 110 | videos: List[str], |
| 111 | output: str, |
| 112 | method: Literal["demuxer", "filter"] = "demuxer", |
| 113 | bgm_path: Optional[str] = None, |
| 114 | bgm_volume: float = 0.2, |
| 115 | bgm_mode: Literal["once", "loop"] = "loop" |
| 116 | ) -> str: |
| 117 | """ |
| 118 | Concatenate multiple videos into one |
| 119 | |
| 120 | Args: |
| 121 | videos: List of video file paths to concatenate |
| 122 | output: Output video file path |
| 123 | method: Concatenation method |
| 124 | - "demuxer": Fast, no re-encoding (requires identical formats) |
| 125 | - "filter": Slower but handles different formats |
| 126 | bgm_path: Background music file path (optional) |
| 127 | - None: No BGM |
| 128 | """ |
| 129 | self._ensure_ffmpeg() |
| 130 | |
| 131 | if not videos: |
| 132 | raise ValueError("Videos list cannot be empty") |
| 133 | |
| 134 | if len(videos) == 1: |
| 135 | logger.info(f"Only one video provided, copying to {output}") |
| 136 | shutil.copy(videos[0], output) |
| 137 | return output |
| 138 | |
| 139 | logger.info(f"Concatenating {len(videos)} videos using {method} method") |
| 140 | |
| 141 | # Step 1: Concatenate videos |
| 142 | if bgm_path: |
| 143 | # If BGM needed, concatenate to temp file first |
| 144 | temp_output = output.replace('.mp4', '_no_bgm.mp4') |
| 145 | concat_result = self._concat_demuxer(videos, temp_output) if method == "demuxer" else self._concat_filter(videos, temp_output) |
| 146 | |
| 147 | # Step 2: Add BGM |
| 148 | logger.info(f"Adding BGM: {bgm_path} (volume={bgm_volume}, mode={bgm_mode})") |
| 149 | final_result = self._add_bgm_to_video( |
| 150 | video=concat_result, |
| 151 | bgm_path=bgm_path, |
| 152 | output=output, |
| 153 | volume=bgm_volume, |
| 154 | mode=bgm_mode |
| 155 | ) |
| 156 | |
| 157 | # Clean up temp file |
| 158 | if os.path.exists(temp_output): |
| 159 | os.unlink(temp_output) |
| 160 | |
| 161 | return final_result |
| 162 | else: |
| 163 | # No BGM, direct concatenation |
| 164 | if method == "demuxer": |
| 165 | return self._concat_demuxer(videos, output) |
| 166 | else: |
| 167 | return self._concat_filter(videos, output) |
| 168 | |
| 169 | def _concat_demuxer(self, videos: List[str], output: str) -> str: |
| 170 | """ |
| 171 | Concatenate using concat demuxer (fast, no re-encoding) |
| 172 | |
| 173 | FFmpeg equivalent: |
| 174 | ffmpeg -f concat -safe 0 -i filelist.txt -c copy output.mp4 |
| 175 | """ |
| 176 | # Create temporary file list |
| 177 | with tempfile.NamedTemporaryFile( |
| 178 | mode='w', |
| 179 | delete=False, |
| 180 | suffix='.txt', |
| 181 | encoding='utf-8' |
| 182 | ) as f: |
| 183 | for video in videos: |
| 184 | abs_path = Path(video).absolute() |
| 185 | escaped_path = str(abs_path).replace("'", "'\\''") |
| 186 | f.write(f"file '{escaped_path}'\n") |
| 187 | filelist = f.name |
| 188 | |
| 189 | try: |
| 190 | logger.debug(f"Created filelist: {filelist}") |
| 191 | ( |
| 192 | ffmpeg |
| 193 | .input(filelist, format='concat', safe=0) |
| 194 | .output(output, c='copy') |
| 195 | .overwrite_output() |
| 196 | .run(capture_stdout=True, capture_stderr=True) |
| 197 | ) |
| 198 | logger.success(f"Videos concatenated successfully: {output}") |
| 199 | return output |
| 200 | except ffmpeg.Error as e: |
| 201 | error_msg = e.stderr.decode() if e.stderr else str(e) |
| 202 | logger.error(f"FFmpeg concat error: {error_msg}") |
| 203 | raise RuntimeError(f"Failed to concatenate videos: {error_msg}") |
| 204 | finally: |
| 205 | if os.path.exists(filelist): |
| 206 | os.unlink(filelist) |
| 207 | |
| 208 | def _concat_filter(self, videos: List[str], output: str) -> str: |
| 209 | """ |
| 210 | Concatenate using concat filter (slower but handles different formats) |
| 211 | |
| 212 | FFmpeg equivalent: |
| 213 | ffmpeg -i v1.mp4 -i v2.mp4 -filter_complex "[0:v][0:a][1:v][1:a]concat=n=2:v=1:a=1[v][a]" |
| 214 | -map "[v]" -map "[a]" output.mp4 |
| 215 | """ |
| 216 | try: |
| 217 | # Build filter_complex string manually |
| 218 | n = len(videos) |
| 219 | |
| 220 | # Build input stream labels: [0:v][0:a][1:v][1:a]... |
| 221 | stream_spec = "".join([f"[{i}:v][{i}:a]" for i in range(n)]) |
| 222 | filter_complex = f"{stream_spec}concat=n={n}:v=1:a=1[v][a]" |
| 223 | |
| 224 | # Build ffmpeg command |
| 225 | cmd = ['ffmpeg'] |
| 226 | for video in videos: |
| 227 | cmd.extend(['-i', video]) |
| 228 | cmd.extend([ |
| 229 | '-filter_complex', filter_complex, |
| 230 | '-map', '[v]', |
| 231 | '-map', '[a]', |
| 232 | '-y', # Overwrite output |
| 233 | output |
| 234 | ]) |
| 235 | |
| 236 | # Run command |
| 237 | import subprocess |
| 238 | result = subprocess.run( |
| 239 | cmd, |
| 240 | capture_output=True, |
| 241 | text=True, |
| 242 | check=True |
| 243 | ) |
| 244 | |
| 245 | logger.success(f"Videos concatenated successfully: {output}") |
| 246 | return output |
| 247 | except subprocess.CalledProcessError as e: |
| 248 | error_msg = e.stderr if e.stderr else str(e) |
| 249 | logger.error(f"FFmpeg concat filter error: {error_msg}") |
| 250 | raise RuntimeError(f"Failed to concatenate videos: {error_msg}") |
| 251 | except Exception as e: |
| 252 | logger.error(f"Concatenation error: {e}") |
| 253 | raise RuntimeError(f"Failed to concatenate videos: {e}") |
| 254 | |
| 255 | def _get_video_duration(self, video: str) -> float: |
| 256 | """Get video duration in seconds""" |
| 257 | try: |
| 258 | probe = ffmpeg.probe(video) |
| 259 | duration = float(probe['format']['duration']) |
| 260 | return duration |
| 261 | except Exception as e: |
| 262 | logger.warning(f"Failed to get video duration: {e}") |
| 263 | return 0.0 |
| 264 | |
| 265 | def _get_audio_duration(self, audio: str) -> float: |
| 266 | """Get audio duration in seconds""" |
| 267 | try: |
| 268 | probe = ffmpeg.probe(audio) |
| 269 | duration = float(probe['format']['duration']) |
| 270 | return duration |
| 271 | except Exception as e: |
| 272 | logger.warning(f"Failed to get audio duration: {e}, using estimate") |
| 273 | # Fallback: estimate based on file size (very rough) |
| 274 | import os |
| 275 | file_size = os.path.getsize(audio) |
| 276 | # Assume ~16kbps for MP3, so 2KB per second |
| 277 | estimated_duration = file_size / 2000 |
| 278 | return max(1.0, estimated_duration) # At least 1 second |
| 279 | |
| 280 | def has_audio_stream(self, video: str) -> bool: |
| 281 | """ |
| 282 | Check if video has audio stream |
| 283 | |
| 284 | Args: |
| 285 | video: Video file path |
| 286 | |
| 287 | Returns: |
| 288 | True if video has audio stream, False otherwise |
| 289 | """ |
| 290 | try: |
| 291 | probe = ffmpeg.probe(video) |
| 292 | audio_streams = [s for s in probe.get('streams', []) if s['codec_type'] == 'audio'] |
| 293 | has_audio = len(audio_streams) > 0 |
| 294 | logger.debug(f"Video {video} has_audio={has_audio}") |
| 295 | return has_audio |
| 296 | except Exception as e: |
| 297 | logger.warning(f"Failed to probe video audio streams: {e}, assuming no audio") |
| 298 | return False |
| 299 | |
| 300 | def merge_audio_video( |
| 301 | self, |
| 302 | video: str, |
| 303 | audio: str, |
| 304 | output: str, |
| 305 | replace_audio: bool = True, |
| 306 | audio_volume: float = 1.0, |
| 307 | video_volume: float = 0.0, |
| 308 | pad_strategy: str = "freeze", # "freeze" (freeze last frame) or "black" (black screen) |
| 309 | auto_adjust_duration: bool = True, # Automatically adjust video duration to match audio |
| 310 | duration_tolerance: float = 0.3, # Tolerance for video being longer than audio (seconds) |
| 311 | ) -> str: |
| 312 | """ |
| 313 | Merge audio with video with intelligent duration adjustment |
| 314 | |
| 315 | Automatically handles duration mismatches between video and audio: |
| 316 | - If video < audio: Pad video to match audio (avoid black screen) |
| 317 | - If video > audio (within tolerance): Keep as-is (acceptable) |
| 318 | - If video > audio (exceeds tolerance): Trim video to match audio |
| 319 | |
| 320 | Automatically handles videos with or without audio streams. |
| 321 | - If video has no audio: adds the audio track |
| 322 | - If video has audio and replace_audio=True: replaces with new audio |
| 323 | - If video has audio and replace_audio=False: mixes both audio tracks |
| 324 | |
| 325 | Args: |
| 326 | video: Video file path |
| 327 | audio: Audio file path |
| 328 | output: Output video file path |
| 329 | replace_audio: If True, replace video's audio; if False, mix with original |
| 330 | audio_volume: Volume of the new audio (0.0 to 1.0+) |
| 331 | video_volume: Volume of original video audio (0.0 to 1.0+) |
| 332 | Only used when replace_audio=False |
| 333 | pad_strategy: Strategy to pad video if audio is longer |
| 334 | - "freeze": Freeze last frame (default) |
| 335 | - "black": Fill with black screen |
| 336 | auto_adjust_duration: Enable intelligent duration adjustment (default: True) |
| 337 | duration_tolerance: Tolerance for video being longer than audio in seconds (default: 0.3) |
| 338 | Videos within this tolerance won't be trimmed |
| 339 | |
| 340 | Returns: |
| 341 | Path to the output video file |
| 342 | |
| 343 | Raises: |
| 344 | RuntimeError: If FFmpeg execution fails |
| 345 | |
| 346 | Note: |
| 347 | - Uses the longer duration between video and audio |
| 348 | - When audio is longer, video is padded using pad_strategy |
| 349 | - When video is longer, audio is looped or extended |
| 350 | - Automatically detects if video has audio |
| 351 | - When video is silent, audio is added regardless of replace_audio |
| 352 | - When replace_audio=True and video has audio, original audio is removed |
| 353 | - When replace_audio=False and video has audio, original and new audio are mixed |
| 354 | """ |
| 355 | self._ensure_ffmpeg() |
| 356 | |
| 357 | # Get durations of video and audio |
| 358 | video_duration = self._get_video_duration(video) |
| 359 | audio_duration = self._get_audio_duration(audio) |
| 360 | |
| 361 | logger.info(f"Video duration: {video_duration:.2f}s, Audio duration: {audio_duration:.2f}s") |
| 362 | |
| 363 | # Intelligent duration adjustment (if enabled) |
| 364 | if auto_adjust_duration: |
| 365 | diff = video_duration - audio_duration |
| 366 | |
| 367 | if diff < 0: |
| 368 | # Video shorter than audio → Must pad to avoid black screen |
| 369 | logger.warning(f"⚠️ Video shorter than audio by {abs(diff):.2f}s, padding required") |
| 370 | video = self._pad_video_to_duration(video, audio_duration, pad_strategy) |
| 371 | video_duration = audio_duration # Update duration after padding |
| 372 | logger.info(f"📌 Padded video to {audio_duration:.2f}s") |
| 373 | |
| 374 | elif diff > duration_tolerance: |
| 375 | # Video significantly longer than audio → Trim |
| 376 | logger.info(f"⚠️ Video longer than audio by {diff:.2f}s (tolerance: {duration_tolerance}s)") |
| 377 | video = self._trim_video_to_duration(video, audio_duration) |
| 378 | video_duration = audio_duration # Update duration after trimming |
| 379 | logger.info(f"✂️ Trimmed video to {audio_duration:.2f}s") |
| 380 | |
| 381 | else: # 0 <= diff <= duration_tolerance |
| 382 | # Video slightly longer but within tolerance → Keep as-is |
| 383 | logger.info(f"✅ Duration acceptable: video={video_duration:.2f}s, audio={audio_duration:.2f}s (diff={diff:.2f}s)") |
| 384 | |
| 385 | # Determine target duration (max of both) |
| 386 | target_duration = max(video_duration, audio_duration) |
| 387 | logger.info(f"Target output duration: {target_duration:.2f}s") |
| 388 | |
| 389 | # Check if video has audio stream |
| 390 | video_has_audio = self.has_audio_stream(video) |
| 391 | |
| 392 | # Prepare video stream (potentially with padding) |
| 393 | input_video = ffmpeg.input(video) |
| 394 | video_stream = input_video.video |
| 395 | |
| 396 | # Pad video if audio is longer |
| 397 | if audio_duration > video_duration: |
| 398 | pad_duration = audio_duration - video_duration |
| 399 | logger.info(f"Audio is longer, padding video by {pad_duration:.2f}s using '{pad_strategy}' strategy") |
| 400 | |
| 401 | if pad_strategy == "freeze": |
| 402 | # Freeze last frame: tpad filter |
| 403 | video_stream = video_stream.filter('tpad', stop_mode='clone', stop_duration=pad_duration) |
| 404 | else: # black |
| 405 | # Generate black frames for padding duration |
| 406 | # Get video properties |
| 407 | probe = ffmpeg.probe(video) |
| 408 | video_info = next(s for s in probe['streams'] if s['codec_type'] == 'video') |
| 409 | width = int(video_info['width']) |
| 410 | height = int(video_info['height']) |
| 411 | fps_str = video_info['r_frame_rate'] |
| 412 | fps_num, fps_den = map(int, fps_str.split('/')) |
| 413 | fps = fps_num / fps_den if fps_den != 0 else 30 |
| 414 | |
| 415 | # Create black video for padding |
| 416 | black_video_path = self._get_unique_temp_path("black_pad", os.path.basename(output)) |
| 417 | black_input = ffmpeg.input( |
| 418 | f'color=c=black:s={width}x{height}:r={fps}', |
| 419 | f='lavfi', |
| 420 | t=pad_duration |
| 421 | ) |
| 422 | |
| 423 | # Concatenate original video with black padding |
| 424 | video_stream = ffmpeg.concat(video_stream, black_input.video, v=1, a=0) |
| 425 | |
| 426 | # Prepare audio stream (pad if needed to match target duration) |
| 427 | input_audio = ffmpeg.input(audio) |
| 428 | audio_stream = input_audio.audio.filter('volume', audio_volume) |
| 429 | |
| 430 | # Pad audio with silence if video is longer |
| 431 | if video_duration > audio_duration: |
| 432 | pad_duration = video_duration - audio_duration |
| 433 | logger.info(f"Video is longer, padding audio with {pad_duration:.2f}s silence") |
| 434 | # Use apad to add silence at the end |
| 435 | audio_stream = audio_stream.filter('apad', whole_dur=target_duration) |
| 436 | |
| 437 | if not video_has_audio: |
| 438 | logger.info(f"Video has no audio stream, adding audio track") |
| 439 | # Video is silent, just add the audio |
| 440 | try: |
| 441 | ( |
| 442 | ffmpeg |
| 443 | .output( |
| 444 | video_stream, |
| 445 | audio_stream, |
| 446 | output, |
| 447 | vcodec='libx264', # Re-encode video if padded |
| 448 | acodec='aac', |
| 449 | audio_bitrate='192k' |
| 450 | ) |
| 451 | .overwrite_output() |
| 452 | .run(capture_stdout=True, capture_stderr=True) |
| 453 | ) |
| 454 | |
| 455 | logger.success(f"Audio added to silent video: {output}") |
| 456 | return output |
| 457 | except ffmpeg.Error as e: |
| 458 | error_msg = e.stderr.decode() if e.stderr else str(e) |
| 459 | logger.error(f"FFmpeg error adding audio to silent video: {error_msg}") |
| 460 | raise RuntimeError(f"Failed to add audio to video: {error_msg}") |
| 461 | |
| 462 | # Video has audio, proceed with merging |
| 463 | logger.info(f"Merging audio with video (replace={replace_audio})") |
| 464 | |
| 465 | try: |
| 466 | if replace_audio: |
| 467 | # Replace audio: use only new audio, ignore original |
| 468 | ( |
| 469 | ffmpeg |
| 470 | .output( |
| 471 | video_stream, |
| 472 | audio_stream, |
| 473 | output, |
| 474 | vcodec='libx264', # Re-encode video if padded |
| 475 | acodec='aac', |
| 476 | audio_bitrate='192k' |
| 477 | ) |
| 478 | .overwrite_output() |
| 479 | .run(capture_stdout=True, capture_stderr=True) |
| 480 | ) |
| 481 | else: |
| 482 | # Mix audio: combine original and new audio |
| 483 | mixed_audio = ffmpeg.filter( |
| 484 | [ |
| 485 | input_video.audio.filter('volume', video_volume), |
| 486 | audio_stream |
| 487 | ], |
| 488 | 'amix', |
| 489 | inputs=2, |
| 490 | duration='longest' # Use longest audio |
| 491 | ) |
| 492 | |
| 493 | ( |
| 494 | ffmpeg |
| 495 | .output( |
| 496 | video_stream, |
| 497 | mixed_audio, |
| 498 | output, |
| 499 | vcodec='libx264', # Re-encode video if padded |
| 500 | acodec='aac', |
| 501 | audio_bitrate='192k' |
| 502 | ) |
| 503 | .overwrite_output() |
| 504 | .run(capture_stdout=True, capture_stderr=True) |
| 505 | ) |
| 506 | |
| 507 | logger.success(f"Audio merged successfully: {output}") |
| 508 | return output |
| 509 | except ffmpeg.Error as e: |
| 510 | error_msg = e.stderr.decode() if e.stderr else str(e) |
| 511 | logger.error(f"FFmpeg merge error: {error_msg}") |
| 512 | raise RuntimeError(f"Failed to merge audio and video: {error_msg}") |
| 513 | |
| 514 | def overlay_image_on_video( |
| 515 | self, |
| 516 | video: str, |
| 517 | overlay_image: str, |
| 518 | output: str, |
| 519 | scale_mode: str = "contain" |
| 520 | ) -> str: |
| 521 | """ |
| 522 | Overlay a transparent image on top of video |
| 523 | |
| 524 | Args: |
| 525 | video: Base video file path |
| 526 | overlay_image: Transparent overlay image path (e.g., rendered HTML with transparent background) |
| 527 | output: Output video file path |
| 528 | scale_mode: How to scale the base video to fit the overlay size |
| 529 | - "contain": Scale video to fit within overlay dimensions (letterbox/pillarbox) |
| 530 | - "cover": Scale video to cover overlay dimensions (may crop) |
| 531 | - "stretch": Stretch video to exact overlay dimensions |
| 532 | |
| 533 | Returns: |
| 534 | Path to the output video file |
| 535 | |
| 536 | Raises: |
| 537 | RuntimeError: If FFmpeg execution fails |
| 538 | |
| 539 | Note: |
| 540 | - Overlay image should have transparent background |
| 541 | - Video is scaled to match overlay dimensions based on scale_mode |
| 542 | - Final video size matches overlay image size |
| 543 | - Video codec is re-encoded to support overlay |
| 544 | """ |
| 545 | self._ensure_ffmpeg() |
| 546 | logger.info(f"Overlaying image on video (scale_mode={scale_mode})") |
| 547 | |
| 548 | try: |
| 549 | # Get overlay image dimensions |
| 550 | overlay_probe = ffmpeg.probe(overlay_image) |
| 551 | overlay_stream = next(s for s in overlay_probe['streams'] if s['codec_type'] == 'video') |
| 552 | overlay_width = int(overlay_stream['width']) |
| 553 | overlay_height = int(overlay_stream['height']) |
| 554 | |
| 555 | logger.debug(f"Overlay dimensions: {overlay_width}x{overlay_height}") |
| 556 | |
| 557 | input_video = ffmpeg.input(video) |
| 558 | input_overlay = ffmpeg.input(overlay_image) |
| 559 | |
| 560 | # Scale video to fit overlay size using scale_mode |
| 561 | if scale_mode == "contain": |
| 562 | # Scale to fit (letterbox/pillarbox if aspect ratio differs) |
| 563 | # Use scale filter with force_original_aspect_ratio=decrease and pad to center |
| 564 | scaled_video = ( |
| 565 | input_video |
| 566 | .filter('scale', overlay_width, overlay_height, force_original_aspect_ratio='decrease') |
| 567 | .filter('pad', overlay_width, overlay_height, '(ow-iw)/2', '(oh-ih)/2', color='black') |
| 568 | ) |
| 569 | elif scale_mode == "cover": |
| 570 | # Scale to cover (crop if aspect ratio differs) |
| 571 | scaled_video = ( |
| 572 | input_video |
| 573 | .filter('scale', overlay_width, overlay_height, force_original_aspect_ratio='increase') |
| 574 | .filter('crop', overlay_width, overlay_height) |
| 575 | ) |
| 576 | else: # stretch |
| 577 | # Stretch to exact dimensions |
| 578 | scaled_video = input_video.filter('scale', overlay_width, overlay_height) |
| 579 | |
| 580 | # Overlay the transparent image on top of the scaled video |
| 581 | output_stream = ffmpeg.overlay(scaled_video, input_overlay) |
| 582 | |
| 583 | ( |
| 584 | ffmpeg |
| 585 | .output(output_stream, output, |
| 586 | vcodec='libx264', |
| 587 | pix_fmt='yuv420p', |
| 588 | preset='medium', |
| 589 | crf=23) |
| 590 | .overwrite_output() |
| 591 | .run(capture_stdout=True, capture_stderr=True) |
| 592 | ) |
| 593 | |
| 594 | logger.success(f"Image overlaid on video: {output}") |
| 595 | return output |
| 596 | except ffmpeg.Error as e: |
| 597 | error_msg = e.stderr.decode() if e.stderr else str(e) |
| 598 | logger.error(f"FFmpeg overlay error: {error_msg}") |
| 599 | raise RuntimeError(f"Failed to overlay image on video: {error_msg}") |
| 600 | |
| 601 | def create_video_from_image( |
| 602 | self, |
| 603 | image: str, |
| 604 | audio: str, |
| 605 | output: str, |
| 606 | fps: int = 30, |
| 607 | ) -> str: |
| 608 | """ |
| 609 | Create video from static image and audio |
| 610 | |
| 611 | Args: |
| 612 | image: Image file path |
| 613 | audio: Audio file path |
| 614 | output: Output video path |
| 615 | fps: Frames per second |
| 616 | |
| 617 | Returns: |
| 618 | Path to the output video |
| 619 | |
| 620 | Raises: |
| 621 | RuntimeError: If FFmpeg execution fails |
| 622 | |
| 623 | Note: |
| 624 | - Image is displayed as static frame for the duration of audio |
| 625 | - Video duration matches audio duration |
| 626 | - Useful for creating video segments from storyboard frames |
| 627 | |
| 628 | Example: |
| 629 | >>> compositor.create_video_from_image( |
| 630 | ... "frame.png", |
| 631 | ... "narration.mp3", |
| 632 | ... "segment.mp4" |
| 633 | ... ) |
| 634 | """ |
| 635 | self._ensure_ffmpeg() |
| 636 | logger.info("Creating video from image and audio") |
| 637 | |
| 638 | try: |
| 639 | # Get audio duration to ensure exact video duration match |
| 640 | probe = ffmpeg.probe(audio) |
| 641 | audio_duration = float(probe['format']['duration']) |
| 642 | logger.debug(f"Audio duration: {audio_duration:.3f}s") |
| 643 | |
| 644 | # Input image with loop (loop=1 means loop indefinitely) |
| 645 | # Use framerate to set input framerate |
| 646 | input_image = ffmpeg.input(image, loop=1, framerate=fps) |
| 647 | input_audio = ffmpeg.input(audio) |
| 648 | |
| 649 | # Combine image and audio |
| 650 | # Use -t to explicitly set video duration = audio duration |
| 651 | ( |
| 652 | ffmpeg |
| 653 | .output( |
| 654 | input_image, |
| 655 | input_audio, |
| 656 | output, |
| 657 | t=audio_duration, # Force video duration to match audio exactly |
| 658 | vcodec='libx264', |
| 659 | acodec='aac', |
| 660 | pix_fmt='yuv420p', |
| 661 | audio_bitrate='192k', |
| 662 | preset='medium', |
| 663 | crf=23, |
| 664 | **{'b:v': '2M'} # Video bitrate |
| 665 | ) |
| 666 | .overwrite_output() |
| 667 | .run(capture_stdout=True, capture_stderr=True) |
| 668 | ) |
| 669 | |
| 670 | logger.success(f"Video created from image: {output} (duration: {audio_duration:.3f}s)") |
| 671 | return output |
| 672 | except ffmpeg.Error as e: |
| 673 | error_msg = e.stderr.decode() if e.stderr else str(e) |
| 674 | logger.error(f"FFmpeg error creating video from image: {error_msg}") |
| 675 | raise RuntimeError(f"Failed to create video from image: {error_msg}") |
| 676 | |
| 677 | def add_bgm( |
| 678 | self, |
| 679 | video: str, |
| 680 | bgm: str, |
| 681 | output: str, |
| 682 | bgm_volume: float = 0.3, |
| 683 | loop: bool = True, |
| 684 | fade_in: float = 0.0, |
| 685 | fade_out: float = 0.0, |
| 686 | ) -> str: |
| 687 | """ |
| 688 | Add background music to video |
| 689 | |
| 690 | Args: |
| 691 | video: Video file path |
| 692 | bgm: Background music file path |
| 693 | output: Output video file path |
| 694 | bgm_volume: BGM volume relative to original (0.0 to 1.0+) |
| 695 | loop: If True, loop BGM to match video duration |
| 696 | fade_in: BGM fade-in duration in seconds |
| 697 | fade_out: BGM fade-out duration in seconds (not yet implemented) |
| 698 | |
| 699 | Returns: |
| 700 | Path to the output video file |
| 701 | |
| 702 | Raises: |
| 703 | RuntimeError: If FFmpeg execution fails |
| 704 | |
| 705 | Note: |
| 706 | - BGM is mixed with original video audio |
| 707 | - If loop=True, BGM repeats until video ends |
| 708 | - Fade effects are applied to BGM only |
| 709 | """ |
| 710 | self._ensure_ffmpeg() |
| 711 | logger.info(f"Adding BGM to video (volume={bgm_volume}, loop={loop})") |
| 712 | |
| 713 | try: |
| 714 | input_video = ffmpeg.input(video) |
| 715 | |
| 716 | # Configure BGM input with looping if needed |
| 717 | bgm_input = ffmpeg.input( |
| 718 | bgm, |
| 719 | stream_loop=-1 if loop else 0 # -1 = infinite loop |
| 720 | ) |
| 721 | |
| 722 | # Apply volume adjustment to BGM |
| 723 | bgm_audio = bgm_input.audio.filter('volume', bgm_volume) |
| 724 | |
| 725 | # Apply fade effects if specified |
| 726 | if fade_in > 0: |
| 727 | bgm_audio = bgm_audio.filter('afade', type='in', duration=fade_in) |
| 728 | # Note: fade_out at the end requires knowing the duration, which is complex |
| 729 | # For now, we skip fade_out in this implementation |
| 730 | # A more advanced implementation would need to: |
| 731 | # 1. Get video duration |
| 732 | # 2. Calculate fade_out start time |
| 733 | # 3. Apply fade filter with specific start_time |
| 734 | |
| 735 | # Mix original audio with BGM |
| 736 | mixed_audio = ffmpeg.filter( |
| 737 | [input_video.audio, bgm_audio], |
| 738 | 'amix', |
| 739 | inputs=2, |
| 740 | duration='first' # Use video's duration |
| 741 | ) |
| 742 | |
| 743 | ( |
| 744 | ffmpeg |
| 745 | .output( |
| 746 | input_video.video, |
| 747 | mixed_audio, |
| 748 | output, |
| 749 | vcodec='copy', |
| 750 | acodec='aac', |
| 751 | audio_bitrate='192k' |
| 752 | ) |
| 753 | .overwrite_output() |
| 754 | .run(capture_stdout=True, capture_stderr=True) |
| 755 | ) |
| 756 | |
| 757 | logger.success(f"BGM added successfully: {output}") |
| 758 | return output |
| 759 | except ffmpeg.Error as e: |
| 760 | error_msg = e.stderr.decode() if e.stderr else str(e) |
| 761 | logger.error(f"FFmpeg BGM error: {error_msg}") |
| 762 | raise RuntimeError(f"Failed to add BGM: {error_msg}") |
| 763 | |
| 764 | def _add_bgm_to_video( |
| 765 | self, |
| 766 | video: str, |
| 767 | bgm_path: str, |
| 768 | output: str, |
| 769 | volume: float = 0.2, |
| 770 | mode: Literal["once", "loop"] = "loop" |
| 771 | ) -> str: |
| 772 | """ |
| 773 | Internal helper to add BGM to video with path resolution |
| 774 | |
| 775 | Args: |
| 776 | video: Video file path |
| 777 | bgm_path: BGM path (can be preset name or custom path) |
| 778 | output: Output file path |
| 779 | volume: BGM volume (0.0-1.0) |
| 780 | mode: "once" or "loop" |
| 781 | |
| 782 | Returns: |
| 783 | Path to output video |
| 784 | |
| 785 | Raises: |
| 786 | FileNotFoundError: If BGM file not found |
| 787 | """ |
| 788 | # Resolve BGM path (raises FileNotFoundError if not found) |
| 789 | resolved_bgm = self._resolve_bgm_path(bgm_path) |
| 790 | |
| 791 | # Add BGM using existing method |
| 792 | loop = (mode == "loop") |
| 793 | return self.add_bgm( |
| 794 | video=video, |
| 795 | bgm=resolved_bgm, |
| 796 | output=output, |
| 797 | bgm_volume=volume, |
| 798 | loop=loop, |
| 799 | fade_in=0.0 |
| 800 | ) |
| 801 | |
| 802 | def _get_unique_temp_path(self, prefix: str, original_filename: str) -> str: |
| 803 | """ |
| 804 | Generate unique temporary file path to avoid concurrent conflicts |
| 805 | |
| 806 | Args: |
| 807 | prefix: Prefix for the temp file (e.g., "trimmed", "padded", "black_pad") |
| 808 | original_filename: Original filename to preserve in temp path |
| 809 | |
| 810 | Returns: |
| 811 | Unique temporary file path with format: temp/{prefix}_{uuid}_{original_filename} |
| 812 | |
| 813 | Example: |
| 814 | >>> self._get_unique_temp_path("trimmed", "video.mp4") |
| 815 | >>> # Returns: "temp/trimmed_a3f2d8c1_video.mp4" |
| 816 | """ |
| 817 | from pixelle_video.utils.os_util import get_temp_path |
| 818 | |
| 819 | unique_id = uuid.uuid4().hex[:8] |
| 820 | return get_temp_path(f"{prefix}_{unique_id}_{original_filename}") |
| 821 | |
| 822 | def _resolve_bgm_path(self, bgm_path: str) -> str: |
| 823 | """ |
| 824 | Resolve BGM path (filename or custom path) with custom override support |
| 825 | |
| 826 | Search priority: |
| 827 | 1. Direct path (absolute or relative) |
| 828 | 2. data/bgm/{filename} (custom) |
| 829 | 3. bgm/{filename} (default) |
| 830 | |
| 831 | Args: |
| 832 | bgm_path: Can be: |
| 833 | - Filename with extension (e.g., "default.mp3", "happy.mp3"): auto-resolved from bgm/ or data/bgm/ |
| 834 | - Custom file path (absolute or relative) |
| 835 | |
| 836 | Returns: |
| 837 | Resolved absolute path |
| 838 | |
| 839 | Raises: |
| 840 | FileNotFoundError: If BGM file not found |
| 841 | """ |
| 842 | # Try direct path first (absolute or relative) |
| 843 | if os.path.exists(bgm_path): |
| 844 | return os.path.abspath(bgm_path) |
| 845 | |
| 846 | # Try as filename in resource directories (custom > default) |
| 847 | if resource_exists("bgm", bgm_path): |
| 848 | return get_resource_path("bgm", bgm_path) |
| 849 | |
| 850 | # Not found - provide helpful error message |
| 851 | tried_paths = [ |
| 852 | os.path.abspath(bgm_path), |
| 853 | f"data/bgm/{bgm_path} or bgm/{bgm_path}" |
| 854 | ] |
| 855 | |
| 856 | # List available BGM files |
| 857 | available_bgm = self._list_available_bgm() |
| 858 | available_msg = f"\n Available BGM files: {', '.join(available_bgm)}" if available_bgm else "" |
| 859 | |
| 860 | raise FileNotFoundError( |
| 861 | f"BGM file not found: '{bgm_path}'\n" |
| 862 | f" Tried paths:\n" |
| 863 | f" 1. {tried_paths[0]}\n" |
| 864 | f" 2. {tried_paths[1]}" |
| 865 | f"{available_msg}" |
| 866 | ) |
| 867 | |
| 868 | def _list_available_bgm(self) -> list[str]: |
| 869 | """ |
| 870 | List available BGM files (merged from bgm/ and data/bgm/) |
| 871 | |
| 872 | Returns: |
| 873 | List of filenames (with extensions), sorted |
| 874 | """ |
| 875 | try: |
| 876 | # Use resource API to get merged list |
| 877 | all_files = list_resource_files("bgm") |
| 878 | |
| 879 | # Filter to audio files only |
| 880 | audio_extensions = ('.mp3', '.wav', '.ogg', '.flac', '.m4a', '.aac') |
| 881 | return sorted([f for f in all_files if f.lower().endswith(audio_extensions)]) |
| 882 | except Exception as e: |
| 883 | logger.warning(f"Failed to list BGM files: {e}") |
| 884 | return [] |
| 885 | |
| 886 | def _trim_video_to_duration(self, video: str, target_duration: float) -> str: |
| 887 | """ |
| 888 | Trim video to specified duration |
| 889 | |
| 890 | Args: |
| 891 | video: Input video file path |
| 892 | target_duration: Target duration in seconds |
| 893 | |
| 894 | Returns: |
| 895 | Path to trimmed video (temp file) |
| 896 | |
| 897 | Raises: |
| 898 | RuntimeError: If FFmpeg execution fails |
| 899 | """ |
| 900 | output = self._get_unique_temp_path("trimmed", os.path.basename(video)) |
| 901 | |
| 902 | try: |
| 903 | # Use stream copy when possible for fast trimming |
| 904 | input_stream = ffmpeg.input(video, t=target_duration) |
| 905 | output_kwargs = {"vcodec": "copy"} |
| 906 | if self.has_audio_stream(video): |
| 907 | output_kwargs["acodec"] = "copy" |
| 908 | ( |
| 909 | input_stream |
| 910 | .output(output, **output_kwargs) |
| 911 | .overwrite_output() |
| 912 | .run(capture_stdout=True, capture_stderr=True, quiet=True) |
| 913 | ) |
| 914 | return output |
| 915 | except ffmpeg.Error as e: |
| 916 | error_msg = e.stderr.decode() if e.stderr else str(e) |
| 917 | logger.error(f"FFmpeg error trimming video: {error_msg}") |
| 918 | raise RuntimeError(f"Failed to trim video: {error_msg}") |
| 919 | |
| 920 | def _pad_video_to_duration(self, video: str, target_duration: float, pad_strategy: str = "freeze") -> str: |
| 921 | """ |
| 922 | Pad video to specified duration by extending the last frame or adding black frames |
| 923 | |
| 924 | Args: |
| 925 | video: Input video file path |
| 926 | target_duration: Target duration in seconds |
| 927 | pad_strategy: Padding strategy - "freeze" (freeze last frame) or "black" (black screen) |
| 928 | |
| 929 | Returns: |
| 930 | Path to padded video (temp file) |
| 931 | |
| 932 | Raises: |
| 933 | RuntimeError: If FFmpeg execution fails |
| 934 | """ |
| 935 | output = self._get_unique_temp_path("padded", os.path.basename(video)) |
| 936 | |
| 937 | video_duration = self._get_video_duration(video) |
| 938 | pad_duration = target_duration - video_duration |
| 939 | |
| 940 | if pad_duration <= 0: |
| 941 | # No padding needed, return original |
| 942 | return video |
| 943 | |
| 944 | try: |
| 945 | input_video = ffmpeg.input(video) |
| 946 | video_stream = input_video.video |
| 947 | |
| 948 | if pad_strategy == "freeze": |
| 949 | # Freeze last frame using tpad filter |
| 950 | video_stream = video_stream.filter('tpad', stop_mode='clone', stop_duration=pad_duration) |
| 951 | |
| 952 | # Output with re-encoding (tpad requires it) |
| 953 | ( |
| 954 | ffmpeg |
| 955 | .output( |
| 956 | video_stream, |
| 957 | output, |
| 958 | vcodec='libx264', |
| 959 | preset='fast', |
| 960 | crf=23 |
| 961 | ) |
| 962 | .overwrite_output() |
| 963 | .run(capture_stdout=True, capture_stderr=True, quiet=True) |
| 964 | ) |
| 965 | else: # black |
| 966 | # Generate black frames for padding duration |
| 967 | # Get video properties |
| 968 | probe = ffmpeg.probe(video) |
| 969 | video_info = next(s for s in probe['streams'] if s['codec_type'] == 'video') |
| 970 | width = int(video_info['width']) |
| 971 | height = int(video_info['height']) |
| 972 | fps_str = video_info['r_frame_rate'] |
| 973 | fps_num, fps_den = map(int, fps_str.split('/')) |
| 974 | fps = fps_num / fps_den if fps_den != 0 else 30 |
| 975 | |
| 976 | # Create black video for padding |
| 977 | black_input = ffmpeg.input( |
| 978 | f'color=c=black:s={width}x{height}:r={fps}', |
| 979 | f='lavfi', |
| 980 | t=pad_duration |
| 981 | ) |
| 982 | |
| 983 | # Concatenate original video with black padding |
| 984 | video_stream = ffmpeg.concat(video_stream, black_input.video, v=1, a=0) |
| 985 | |
| 986 | ( |
| 987 | ffmpeg |
| 988 | .output( |
| 989 | video_stream, |
| 990 | output, |
| 991 | vcodec='libx264', |
| 992 | preset='fast', |
| 993 | crf=23 |
| 994 | ) |
| 995 | .overwrite_output() |
| 996 | .run(capture_stdout=True, capture_stderr=True, quiet=True) |
| 997 | ) |
| 998 | |
| 999 | return output |
| 1000 | except ffmpeg.Error as e: |
| 1001 | error_msg = e.stderr.decode() if e.stderr else str(e) |
| 1002 | logger.error(f"FFmpeg error padding video: {error_msg}") |
| 1003 | raise RuntimeError(f"Failed to pad video: {error_msg}") |
| 1004 | |
| 1005 |