| 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 | Asset-Based Video Pipeline |
| 15 | |
| 16 | Generates marketing videos from user-provided assets (images/videos) rather than |
| 17 | AI-generated media. Ideal for small businesses with existing media libraries. |
| 18 | |
| 19 | Workflow: |
| 20 | 1. Analyze uploaded assets (images/videos) |
| 21 | 2. Generate script based on user intent and available assets |
| 22 | 3. Match assets to script scenes |
| 23 | 4. Compose final video with narrations |
| 24 | |
| 25 | Example: |
| 26 | pipeline = AssetBasedPipeline(pixelle_video) |
| 27 | result = await pipeline( |
| 28 | assets=["/path/img1.jpg", "/path/img2.jpg"], |
| 29 | video_title="Pet Store Year-End Sale", |
| 30 | intent="Promote our pet store's year-end sale with a warm and friendly tone", |
| 31 | duration=30 |
| 32 | ) |
| 33 | """ |
| 34 | |
| 35 | from typing import List, Dict, Any, Optional, Callable |
| 36 | from pathlib import Path |
| 37 | import math |
| 38 | from datetime import datetime |
| 39 | |
| 40 | from loguru import logger |
| 41 | from pydantic import BaseModel, Field |
| 42 | |
| 43 | from pixelle_video.pipelines.linear import LinearVideoPipeline, PipelineContext |
| 44 | from pixelle_video.models.progress import ProgressEvent |
| 45 | from pixelle_video.utils.os_util import ( |
| 46 | create_task_output_dir, |
| 47 | get_task_final_video_path, |
| 48 | get_task_frame_path, |
| 49 | ) |
| 50 | |
| 51 | # Type alias for progress callback |
| 52 | ProgressCallback = Optional[Callable[[ProgressEvent], None]] |
| 53 | |
| 54 | |
| 55 | # ==================== Structured Output Models ==================== |
| 56 | |
| 57 | class SceneScript(BaseModel): |
| 58 | """Single scene in the video script""" |
| 59 | scene_number: int = Field(description="Scene number starting from 1") |
| 60 | asset_path: str = Field(description="Path to the asset file for this scene") |
| 61 | narrations: List[str] = Field(description="List of narration sentences for this scene (1-5 sentences)") |
| 62 | duration: int = Field(description="Estimated duration in seconds for this scene") |
| 63 | |
| 64 | |
| 65 | class VideoScript(BaseModel): |
| 66 | """Complete video script with scenes""" |
| 67 | scenes: List[SceneScript] = Field(description="List of scenes in the video") |
| 68 | |
| 69 | |
| 70 | class AssetBasedPipeline(LinearVideoPipeline): |
| 71 | """ |
| 72 | Asset-Based Video Pipeline |
| 73 | |
| 74 | Generates videos from user-provided assets instead of AI-generated media. |
| 75 | """ |
| 76 | |
| 77 | def __init__(self, core): |
| 78 | """ |
| 79 | Initialize pipeline |
| 80 | |
| 81 | Args: |
| 82 | core: PixelleVideoCore instance |
| 83 | """ |
| 84 | super().__init__(core) |
| 85 | self.asset_index: Dict[str, Any] = {} # In-memory asset metadata |
| 86 | |
| 87 | async def __call__( |
| 88 | self, |
| 89 | assets: List[str], |
| 90 | video_title: str = "", |
| 91 | intent: Optional[str] = None, |
| 92 | duration: int = 30, |
| 93 | source: str = "runninghub", |
| 94 | bgm_path: Optional[str] = None, |
| 95 | bgm_volume: float = 0.2, |
| 96 | bgm_mode: str = "loop", |
| 97 | progress_callback: ProgressCallback = None, |
| 98 | **kwargs |
| 99 | ) -> PipelineContext: |
| 100 | """ |
| 101 | Execute pipeline with user-provided assets |
| 102 | |
| 103 | Args: |
| 104 | assets: List of asset file paths |
| 105 | video_title: Video title |
| 106 | intent: Video intent/purpose (defaults to video_title) |
| 107 | duration: Target duration in seconds |
| 108 | source: Workflow source ("runninghub" or "selfhost") |
| 109 | bgm_path: Path to background music file (optional) |
| 110 | bgm_volume: BGM volume (0.0-1.0, default 0.2) |
| 111 | bgm_mode: BGM mode ("loop" or "once", default "loop") |
| 112 | progress_callback: Optional callback for progress updates |
| 113 | **kwargs: Additional parameters |
| 114 | |
| 115 | Returns: |
| 116 | Pipeline context with generated video |
| 117 | """ |
| 118 | from pixelle_video.pipelines.linear import PipelineContext |
| 119 | |
| 120 | # Store progress callback |
| 121 | self._progress_callback = progress_callback |
| 122 | |
| 123 | # Create custom context with asset-specific parameters |
| 124 | ctx = PipelineContext( |
| 125 | input_text=intent or video_title, # Use intent or title as input_text |
| 126 | params={ |
| 127 | "assets": assets, |
| 128 | "video_title": video_title, |
| 129 | "intent": intent or video_title, |
| 130 | "duration": duration, |
| 131 | "source": source, |
| 132 | "bgm_path": bgm_path, |
| 133 | "bgm_volume": bgm_volume, |
| 134 | "bgm_mode": bgm_mode, |
| 135 | **kwargs |
| 136 | } |
| 137 | ) |
| 138 | |
| 139 | # Store request parameters in context for easy access |
| 140 | ctx.request = ctx.params |
| 141 | |
| 142 | try: |
| 143 | # Execute pipeline lifecycle |
| 144 | await self.setup_environment(ctx) |
| 145 | await self.determine_title(ctx) |
| 146 | await self.generate_content(ctx) |
| 147 | await self.plan_visuals(ctx) |
| 148 | await self.initialize_storyboard(ctx) |
| 149 | await self.produce_assets(ctx) |
| 150 | await self.post_production(ctx) |
| 151 | await self.finalize(ctx) |
| 152 | |
| 153 | return ctx |
| 154 | |
| 155 | except Exception as e: |
| 156 | await self.handle_exception(ctx, e) |
| 157 | raise |
| 158 | |
| 159 | def _emit_progress(self, event: ProgressEvent): |
| 160 | """Emit progress event to callback if available""" |
| 161 | if self._progress_callback: |
| 162 | self._progress_callback(event) |
| 163 | |
| 164 | async def setup_environment(self, context: PipelineContext) -> PipelineContext: |
| 165 | """ |
| 166 | Analyze uploaded assets and build asset index |
| 167 | |
| 168 | Args: |
| 169 | context: Pipeline context with assets list |
| 170 | |
| 171 | Returns: |
| 172 | Updated context with asset_index |
| 173 | """ |
| 174 | # Create isolated task directory |
| 175 | task_dir, task_id = create_task_output_dir() |
| 176 | context.task_id = task_id |
| 177 | context.task_dir = Path(task_dir) # Convert to Path for easier usage |
| 178 | |
| 179 | # Determine final video path |
| 180 | context.final_video_path = get_task_final_video_path(task_id) |
| 181 | |
| 182 | logger.info(f"📁 Task directory created: {task_dir}") |
| 183 | logger.info("🔍 Analyzing uploaded assets...") |
| 184 | |
| 185 | assets: List[str] = context.request.get("assets", []) |
| 186 | if not assets: |
| 187 | raise ValueError("No assets provided. Please upload at least one image or video.") |
| 188 | |
| 189 | total_assets = len(assets) |
| 190 | logger.info(f"Found {total_assets} assets to analyze") |
| 191 | |
| 192 | # Emit initial progress (0-15% for asset analysis) |
| 193 | self._emit_progress(ProgressEvent( |
| 194 | event_type="analyzing_assets", |
| 195 | progress=0.01, |
| 196 | frame_current=0, |
| 197 | frame_total=total_assets, |
| 198 | extra_info="start" |
| 199 | )) |
| 200 | |
| 201 | self.asset_index = {} |
| 202 | |
| 203 | for i, asset_path in enumerate(assets, 1): |
| 204 | asset_path_obj = Path(asset_path) |
| 205 | |
| 206 | if not asset_path_obj.exists(): |
| 207 | logger.warning(f"Asset not found: {asset_path}") |
| 208 | continue |
| 209 | |
| 210 | logger.info(f"Analyzing asset {i}/{total_assets}: {asset_path_obj.name}") |
| 211 | |
| 212 | # Emit progress for this asset |
| 213 | progress = 0.01 + (i - 1) / total_assets * 0.14 # 1% - 15% |
| 214 | self._emit_progress(ProgressEvent( |
| 215 | event_type="analyzing_asset", |
| 216 | progress=progress, |
| 217 | frame_current=i, |
| 218 | frame_total=total_assets, |
| 219 | extra_info=asset_path_obj.name |
| 220 | )) |
| 221 | |
| 222 | # Determine asset type |
| 223 | asset_type = self._get_asset_type(asset_path_obj) |
| 224 | |
| 225 | if asset_type == "image": |
| 226 | analysis_source = context.request.get("source", "runninghub") |
| 227 | if analysis_source == "api": |
| 228 | description = await self.core.api_asset_analysis.analyze_image( |
| 229 | asset_path, |
| 230 | model=context.request.get("analysis_vlm_model"), |
| 231 | ) |
| 232 | else: |
| 233 | # Analyze image using ImageAnalysisService |
| 234 | description = await self.core.image_analysis( |
| 235 | asset_path, |
| 236 | source=analysis_source, |
| 237 | workflow=context.request.get("analysis_image_workflow"), |
| 238 | ) |
| 239 | |
| 240 | self.asset_index[asset_path] = { |
| 241 | "path": asset_path, |
| 242 | "type": "image", |
| 243 | "name": asset_path_obj.name, |
| 244 | "description": description |
| 245 | } |
| 246 | |
| 247 | logger.info(f"✅ Image analyzed: {description[:50]}...") |
| 248 | |
| 249 | elif asset_type == "video": |
| 250 | analysis_source = context.request.get("source", "runninghub") |
| 251 | try: |
| 252 | if analysis_source == "api": |
| 253 | description = await self.core.api_asset_analysis.analyze_video( |
| 254 | asset_path, |
| 255 | model=context.request.get("analysis_vlm_model"), |
| 256 | ) |
| 257 | else: |
| 258 | # Analyze video using VideoAnalysisService |
| 259 | description = await self.core.video_analysis( |
| 260 | asset_path, |
| 261 | source=analysis_source, |
| 262 | workflow=context.request.get("analysis_video_workflow"), |
| 263 | ) |
| 264 | |
| 265 | self.asset_index[asset_path] = { |
| 266 | "path": asset_path, |
| 267 | "type": "video", |
| 268 | "name": asset_path_obj.name, |
| 269 | "description": description |
| 270 | } |
| 271 | |
| 272 | logger.info(f"✅ Video analyzed: {description[:50]}...") |
| 273 | except Exception as e: |
| 274 | logger.warning(f"Video analysis failed for {asset_path_obj.name}: {e}, using fallback") |
| 275 | self.asset_index[asset_path] = { |
| 276 | "path": asset_path, |
| 277 | "type": "video", |
| 278 | "name": asset_path_obj.name, |
| 279 | "description": "Video asset (analysis failed)" |
| 280 | } |
| 281 | |
| 282 | else: |
| 283 | logger.warning(f"Unknown asset type: {asset_path}") |
| 284 | |
| 285 | logger.success(f"✅ Asset analysis complete: {len(self.asset_index)} assets indexed") |
| 286 | |
| 287 | # Store asset index in context |
| 288 | context.asset_index = self.asset_index |
| 289 | |
| 290 | # Emit completion of asset analysis |
| 291 | self._emit_progress(ProgressEvent( |
| 292 | event_type="analyzing_assets", |
| 293 | progress=0.15, |
| 294 | frame_current=total_assets, |
| 295 | frame_total=total_assets, |
| 296 | extra_info="complete" |
| 297 | )) |
| 298 | |
| 299 | return context |
| 300 | |
| 301 | async def determine_title(self, context: PipelineContext) -> PipelineContext: |
| 302 | """ |
| 303 | Use user-provided title if available, otherwise leave empty |
| 304 | |
| 305 | Args: |
| 306 | context: Pipeline context |
| 307 | |
| 308 | Returns: |
| 309 | Updated context with title (may be empty) |
| 310 | """ |
| 311 | title = context.request.get("video_title") |
| 312 | |
| 313 | if title: |
| 314 | context.title = title |
| 315 | logger.info(f"📝 Video title: {title} (user-specified)") |
| 316 | else: |
| 317 | context.title = "" |
| 318 | logger.info(f"📝 No video title specified (will be hidden in template)") |
| 319 | |
| 320 | return context |
| 321 | |
| 322 | async def generate_content(self, context: PipelineContext) -> PipelineContext: |
| 323 | """ |
| 324 | Generate video script using LLM with structured output |
| 325 | |
| 326 | LLM directly assigns assets to scenes - no complex matching logic needed. |
| 327 | |
| 328 | Args: |
| 329 | context: Pipeline context |
| 330 | |
| 331 | Returns: |
| 332 | Updated context with generated script (scenes already have asset_path assigned) |
| 333 | """ |
| 334 | from pixelle_video.prompts.asset_script_generation import build_asset_script_prompt |
| 335 | |
| 336 | logger.info("🤖 Generating video script with LLM...") |
| 337 | |
| 338 | # Emit progress for script generation (15% - 25%) |
| 339 | self._emit_progress(ProgressEvent( |
| 340 | event_type="generating_script", |
| 341 | progress=0.16 |
| 342 | )) |
| 343 | |
| 344 | # Build prompt for LLM |
| 345 | intent = context.request.get("intent", context.input_text) |
| 346 | duration = context.request.get("duration", 30) |
| 347 | title = context.title # May be empty if user didn't provide one |
| 348 | |
| 349 | # Prepare asset descriptions with full paths for LLM to reference |
| 350 | asset_info = [] |
| 351 | for asset_path, metadata in self.asset_index.items(): |
| 352 | asset_info.append(f"- Path: {asset_path}\n Description: {metadata['description']}") |
| 353 | |
| 354 | assets_text = "\n".join(asset_info) |
| 355 | |
| 356 | # Build prompt using the centralized prompt function |
| 357 | prompt = build_asset_script_prompt( |
| 358 | intent=intent, |
| 359 | duration=duration, |
| 360 | assets_text=assets_text, |
| 361 | title=title |
| 362 | ) |
| 363 | |
| 364 | # Call LLM with structured output |
| 365 | script: VideoScript = await self.core.llm( |
| 366 | prompt=prompt, |
| 367 | response_type=VideoScript, |
| 368 | temperature=0.8, |
| 369 | max_tokens=4000 |
| 370 | ) |
| 371 | |
| 372 | # Convert to dict format for compatibility with downstream code |
| 373 | context.script = [scene.model_dump() for scene in script.scenes] |
| 374 | |
| 375 | # Validate asset paths exist |
| 376 | for scene in context.script: |
| 377 | asset_path = scene.get("asset_path") |
| 378 | if asset_path not in self.asset_index: |
| 379 | # Find closest match (in case LLM slightly modified the path) |
| 380 | matched = False |
| 381 | for known_path in self.asset_index.keys(): |
| 382 | if Path(known_path).name == Path(asset_path).name: |
| 383 | scene["asset_path"] = known_path |
| 384 | matched = True |
| 385 | logger.warning(f"Corrected asset path: {asset_path} -> {known_path}") |
| 386 | break |
| 387 | |
| 388 | if not matched: |
| 389 | # Fallback to first available asset |
| 390 | fallback_path = list(self.asset_index.keys())[0] |
| 391 | logger.warning(f"Unknown asset path '{asset_path}', using fallback: {fallback_path}") |
| 392 | scene["asset_path"] = fallback_path |
| 393 | |
| 394 | logger.success(f"✅ Generated script with {len(context.script)} scenes") |
| 395 | |
| 396 | # Emit progress after script generation |
| 397 | self._emit_progress(ProgressEvent( |
| 398 | event_type="generating_script", |
| 399 | progress=0.25, |
| 400 | extra_info="complete" |
| 401 | )) |
| 402 | |
| 403 | # Log script preview |
| 404 | for scene in context.script: |
| 405 | narrations = scene.get("narrations", []) |
| 406 | if isinstance(narrations, str): |
| 407 | narrations = [narrations] |
| 408 | narration_preview = " | ".join([n[:30] + "..." if len(n) > 30 else n for n in narrations[:2]]) |
| 409 | asset_name = Path(scene.get("asset_path", "unknown")).name |
| 410 | logger.info(f"Scene {scene['scene_number']} [{asset_name}]: {narration_preview}") |
| 411 | |
| 412 | return context |
| 413 | |
| 414 | async def plan_visuals(self, context: PipelineContext) -> PipelineContext: |
| 415 | """ |
| 416 | Prepare matched scenes from LLM-generated script |
| 417 | |
| 418 | Since LLM already assigned asset_path in generate_content, this method |
| 419 | simply converts the script format to matched_scenes format. |
| 420 | |
| 421 | Args: |
| 422 | context: Pipeline context |
| 423 | |
| 424 | Returns: |
| 425 | Updated context with matched_scenes |
| 426 | """ |
| 427 | logger.info("🎯 Preparing scene-asset mapping...") |
| 428 | |
| 429 | # LLM already assigned asset_path to each scene in generate_content |
| 430 | # Just convert to matched_scenes format for downstream compatibility |
| 431 | context.matched_scenes = [ |
| 432 | { |
| 433 | **scene, |
| 434 | "matched_asset": scene["asset_path"] # Alias for compatibility |
| 435 | } |
| 436 | for scene in context.script |
| 437 | ] |
| 438 | |
| 439 | # Log asset usage summary |
| 440 | asset_usage = {} |
| 441 | for scene in context.matched_scenes: |
| 442 | asset = scene["matched_asset"] |
| 443 | asset_usage[asset] = asset_usage.get(asset, 0) + 1 |
| 444 | |
| 445 | logger.info(f"📊 Asset usage summary:") |
| 446 | for asset_path, count in asset_usage.items(): |
| 447 | logger.info(f" {Path(asset_path).name}: {count} scene(s)") |
| 448 | |
| 449 | return context |
| 450 | |
| 451 | async def initialize_storyboard(self, context: PipelineContext) -> PipelineContext: |
| 452 | """ |
| 453 | Initialize storyboard from matched scenes |
| 454 | |
| 455 | Args: |
| 456 | context: Pipeline context |
| 457 | |
| 458 | Returns: |
| 459 | Updated context with storyboard |
| 460 | """ |
| 461 | from pixelle_video.models.storyboard import ( |
| 462 | Storyboard, |
| 463 | StoryboardFrame, |
| 464 | StoryboardConfig |
| 465 | ) |
| 466 | from datetime import datetime |
| 467 | |
| 468 | # Extract all narrations in order for compatibility |
| 469 | all_narrations = [] |
| 470 | for scene in context.matched_scenes: |
| 471 | narrations = scene.get("narrations", [scene.get("narration", "")]) |
| 472 | if isinstance(narrations, str): |
| 473 | narrations = [narrations] |
| 474 | all_narrations.extend(narrations) |
| 475 | |
| 476 | context.narrations = all_narrations |
| 477 | |
| 478 | # Get template dimensions |
| 479 | # Use asset_default.html template which supports both image and video assets |
| 480 | # (conditionally shows background image or provides transparent overlay) |
| 481 | template_name = "1080x1920/asset_default.html" |
| 482 | # Extract dimensions from template name (e.g., "1080x1920") |
| 483 | try: |
| 484 | dims = template_name.split("/")[0].split("x") |
| 485 | media_width = int(dims[0]) |
| 486 | media_height = int(dims[1]) |
| 487 | except: |
| 488 | # Default to 1080x1920 |
| 489 | media_width = 1080 |
| 490 | media_height = 1920 |
| 491 | |
| 492 | # Create StoryboardConfig |
| 493 | context.config = StoryboardConfig( |
| 494 | task_id=context.task_id, |
| 495 | n_storyboard=len(context.matched_scenes), # Number of scenes |
| 496 | min_narration_words=5, |
| 497 | max_narration_words=50, |
| 498 | video_fps=30, |
| 499 | tts_inference_mode="local", |
| 500 | voice_id=context.params.get("voice_id", "zh-CN-YunjianNeural"), |
| 501 | tts_speed=context.params.get("tts_speed", 1.2), |
| 502 | media_width=media_width, |
| 503 | media_height=media_height, |
| 504 | frame_template=template_name, |
| 505 | template_params=context.params.get("template_params") |
| 506 | ) |
| 507 | |
| 508 | # Create Storyboard |
| 509 | context.storyboard = Storyboard( |
| 510 | title=context.title, |
| 511 | config=context.config, |
| 512 | created_at=datetime.now() |
| 513 | ) |
| 514 | |
| 515 | # Create StoryboardFrames - one per scene |
| 516 | for i, scene in enumerate(context.matched_scenes): |
| 517 | # Get first narration for the frame (we'll combine audios later) |
| 518 | narrations = scene.get("narrations", [scene.get("narration", "")]) |
| 519 | if isinstance(narrations, str): |
| 520 | narrations = [narrations] |
| 521 | |
| 522 | # Use first narration as the main text (for subtitle) |
| 523 | # We'll combine all narrations in the audio |
| 524 | main_narration = " ".join(narrations) # Combine for subtitle display |
| 525 | |
| 526 | frame = StoryboardFrame( |
| 527 | index=i, |
| 528 | narration=main_narration, |
| 529 | image_prompt=None, # We're using user assets, not generating images |
| 530 | created_at=datetime.now() |
| 531 | ) |
| 532 | |
| 533 | # Get asset path and determine actual media type from asset_index |
| 534 | asset_path = scene["matched_asset"] |
| 535 | asset_metadata = self.asset_index.get(asset_path, {}) |
| 536 | asset_type = asset_metadata.get("type", "image") # Default to image if not found |
| 537 | |
| 538 | # Set media type and path based on actual asset type |
| 539 | if asset_type == "video": |
| 540 | frame.media_type = "video" |
| 541 | frame.video_path = asset_path |
| 542 | logger.debug(f"Scene {i}: Using video asset: {Path(asset_path).name}") |
| 543 | else: |
| 544 | frame.media_type = "image" |
| 545 | frame.image_path = asset_path |
| 546 | logger.debug(f"Scene {i}: Using image asset: {Path(asset_path).name}") |
| 547 | |
| 548 | # Store scene info for later audio generation |
| 549 | frame._scene_data = scene # Temporary storage for multi-narration |
| 550 | |
| 551 | context.storyboard.frames.append(frame) |
| 552 | |
| 553 | logger.info(f"✅ Created storyboard with {len(context.storyboard.frames)} scenes") |
| 554 | |
| 555 | return context |
| 556 | |
| 557 | async def produce_assets(self, context: PipelineContext) -> PipelineContext: |
| 558 | """ |
| 559 | Generate scene videos using FrameProcessor (asset + multiple narrations + template) |
| 560 | |
| 561 | Args: |
| 562 | context: Pipeline context |
| 563 | |
| 564 | Returns: |
| 565 | Updated context with processed frames |
| 566 | """ |
| 567 | logger.info("🎬 Producing scene videos...") |
| 568 | |
| 569 | storyboard = context.storyboard |
| 570 | config = context.config |
| 571 | total_frames = len(storyboard.frames) |
| 572 | |
| 573 | # Progress range: 30% - 85% for frame production |
| 574 | base_progress = 0.30 |
| 575 | progress_range = 0.55 # 85% - 30% |
| 576 | |
| 577 | for i, frame in enumerate(storyboard.frames, 1): |
| 578 | logger.info(f"Producing scene {i}/{total_frames}...") |
| 579 | |
| 580 | # Emit progress for this frame (each frame has 4 steps: audio, combine, duration, compose) |
| 581 | frame_progress = base_progress + (i - 1) / total_frames * progress_range |
| 582 | self._emit_progress(ProgressEvent( |
| 583 | event_type="frame_step", |
| 584 | progress=frame_progress, |
| 585 | frame_current=i, |
| 586 | frame_total=total_frames, |
| 587 | step=1, |
| 588 | action="audio" |
| 589 | )) |
| 590 | |
| 591 | # Get scene data with narrations |
| 592 | scene = frame._scene_data |
| 593 | narrations = scene.get("narrations", [scene.get("narration", "")]) |
| 594 | if isinstance(narrations, str): |
| 595 | narrations = [narrations] |
| 596 | |
| 597 | logger.info(f"Scene {i} has {len(narrations)} narration(s)") |
| 598 | |
| 599 | # Step 1: Generate audio for each narration and combine |
| 600 | narration_audios = [] |
| 601 | for j, narration_text in enumerate(narrations, 1): |
| 602 | audio_path = Path(context.task_dir) / "frames" / f"{i:02d}_narration_{j}.mp3" |
| 603 | audio_path.parent.mkdir(parents=True, exist_ok=True) |
| 604 | |
| 605 | await self.core.tts( |
| 606 | text=narration_text, |
| 607 | output_path=str(audio_path), |
| 608 | voice=config.voice_id, |
| 609 | speed=config.tts_speed |
| 610 | ) |
| 611 | |
| 612 | narration_audios.append(str(audio_path)) |
| 613 | logger.debug(f" Narration {j}/{len(narrations)}: {narration_text[:30]}...") |
| 614 | |
| 615 | # Concatenate all narration audios for this scene |
| 616 | if len(narration_audios) > 1: |
| 617 | from pixelle_video.utils.os_util import get_task_frame_path |
| 618 | |
| 619 | # Emit progress for combining audio |
| 620 | frame_progress = base_progress + ((i - 1) + 0.25) / total_frames * progress_range |
| 621 | self._emit_progress(ProgressEvent( |
| 622 | event_type="frame_step", |
| 623 | progress=frame_progress, |
| 624 | frame_current=i, |
| 625 | frame_total=total_frames, |
| 626 | step=2, |
| 627 | action="audio" |
| 628 | )) |
| 629 | |
| 630 | combined_audio_path = Path(context.task_dir) / "frames" / f"{i:02d}_audio.mp3" |
| 631 | |
| 632 | # Use FFmpeg to concatenate audio files |
| 633 | import subprocess |
| 634 | |
| 635 | # Create a file list for FFmpeg concat |
| 636 | filelist_path = Path(context.task_dir) / "frames" / f"{i:02d}_audiolist.txt" |
| 637 | with open(filelist_path, 'w') as f: |
| 638 | for audio_file in narration_audios: |
| 639 | escaped_path = str(Path(audio_file).absolute()).replace("'", "'\\''") |
| 640 | f.write(f"file '{escaped_path}'\n") |
| 641 | |
| 642 | # Concatenate audio files |
| 643 | concat_cmd = [ |
| 644 | 'ffmpeg', |
| 645 | '-f', 'concat', |
| 646 | '-safe', '0', |
| 647 | '-i', str(filelist_path), |
| 648 | '-c', 'copy', |
| 649 | '-y', |
| 650 | str(combined_audio_path) |
| 651 | ] |
| 652 | |
| 653 | subprocess.run(concat_cmd, check=True, capture_output=True) |
| 654 | frame.audio_path = str(combined_audio_path) |
| 655 | |
| 656 | logger.info(f"✅ Combined {len(narration_audios)} narrations into one audio") |
| 657 | else: |
| 658 | frame.audio_path = narration_audios[0] |
| 659 | |
| 660 | # Step 2: Use FrameProcessor to generate composed frame and video |
| 661 | # FrameProcessor will handle: |
| 662 | # - Template rendering (with proper dimensions) |
| 663 | # - Subtitle composition |
| 664 | # - Video segment creation |
| 665 | # - Proper file naming in frames/ |
| 666 | |
| 667 | # Since we already have the audio and image, we bypass some steps |
| 668 | # by manually calling the composition steps |
| 669 | |
| 670 | # Emit progress for duration calculation |
| 671 | frame_progress = base_progress + ((i - 1) + 0.5) / total_frames * progress_range |
| 672 | self._emit_progress(ProgressEvent( |
| 673 | event_type="frame_step", |
| 674 | progress=frame_progress, |
| 675 | frame_current=i, |
| 676 | frame_total=total_frames, |
| 677 | step=3, |
| 678 | action="compose" |
| 679 | )) |
| 680 | |
| 681 | # Get audio duration for frame duration |
| 682 | import subprocess |
| 683 | duration_cmd = [ |
| 684 | 'ffprobe', |
| 685 | '-v', 'error', |
| 686 | '-show_entries', 'format=duration', |
| 687 | '-of', 'default=noprint_wrappers=1:nokey=1', |
| 688 | frame.audio_path |
| 689 | ] |
| 690 | duration_result = subprocess.run(duration_cmd, capture_output=True, text=True, check=True) |
| 691 | frame.duration = float(duration_result.stdout.strip()) |
| 692 | |
| 693 | api_video_workflow = context.request.get("api_video_workflow") |
| 694 | api_video_generated_for_frame = False |
| 695 | if api_video_workflow and frame.media_type == "image" and frame.image_path: |
| 696 | logger.info(f"Animating scene {i} image via API workflow: {api_video_workflow}") |
| 697 | api_video_path = get_task_frame_path(context.task_id, frame.index, "video") |
| 698 | api_video_params = dict(context.request.get("api_video_params") or {}) |
| 699 | api_video_params.pop("use_narration_audio_as_driving_audio", None) |
| 700 | workflow_info = self._get_api_workflow_info(api_video_workflow) |
| 701 | adapter_abilities = set((workflow_info or {}).get("adapter_ability_types") or []) |
| 702 | |
| 703 | if "audio_driven_i2v" in adapter_abilities: |
| 704 | api_video_params["audio_path"] = frame.audio_path |
| 705 | |
| 706 | # Asset-based scenes should follow narration duration, not the UI default duration. |
| 707 | api_video_params.pop("duration", None) |
| 708 | api_duration = max(1, int(math.ceil(frame.duration or 5))) |
| 709 | |
| 710 | reference_image_path = frame.image_path |
| 711 | if getattr(context, "_last_api_video_tail_frame", None): |
| 712 | reference_image_path = context._last_api_video_tail_frame |
| 713 | logger.info( |
| 714 | f"Scene {i}: using previous video tail frame as API first-frame reference: " |
| 715 | f"{reference_image_path}" |
| 716 | ) |
| 717 | |
| 718 | media_result = await self.core.media( |
| 719 | prompt=frame.narration or context.input_text or "", |
| 720 | workflow=api_video_workflow, |
| 721 | media_type="video", |
| 722 | image_path=reference_image_path, |
| 723 | output_path=api_video_path, |
| 724 | duration=api_duration, |
| 725 | width=config.media_width, |
| 726 | height=config.media_height, |
| 727 | **api_video_params, |
| 728 | ) |
| 729 | frame.media_type = "video" |
| 730 | frame.video_path = media_result.url |
| 731 | api_video_generated_for_frame = True |
| 732 | tail_frame_path = Path(context.task_dir) / "frames" / f"{i:02d}_api_tail_reference.png" |
| 733 | extracted_tail = self._extract_video_tail_frame( |
| 734 | frame.video_path, |
| 735 | str(tail_frame_path), |
| 736 | ) |
| 737 | if extracted_tail: |
| 738 | context._last_api_video_tail_frame = extracted_tail |
| 739 | logger.success(f"✅ API video generated for scene {i}: {frame.video_path}") |
| 740 | |
| 741 | # Emit progress for video composition |
| 742 | frame_progress = base_progress + ((i - 1) + 0.75) / total_frames * progress_range |
| 743 | self._emit_progress(ProgressEvent( |
| 744 | event_type="frame_step", |
| 745 | progress=frame_progress, |
| 746 | frame_current=i, |
| 747 | frame_total=total_frames, |
| 748 | step=4, |
| 749 | action="video" |
| 750 | )) |
| 751 | |
| 752 | # Use FrameProcessor for proper composition |
| 753 | processed_frame = await self.core.frame_processor( |
| 754 | frame=frame, |
| 755 | storyboard=storyboard, |
| 756 | config=config, |
| 757 | total_frames=total_frames |
| 758 | ) |
| 759 | |
| 760 | if api_video_workflow and not api_video_generated_for_frame and processed_frame.video_segment_path: |
| 761 | tail_frame_path = Path(context.task_dir) / "frames" / f"{i:02d}_tail_reference.png" |
| 762 | extracted_tail = self._extract_video_tail_frame( |
| 763 | processed_frame.video_segment_path, |
| 764 | str(tail_frame_path), |
| 765 | ) |
| 766 | if extracted_tail: |
| 767 | context._last_api_video_tail_frame = extracted_tail |
| 768 | |
| 769 | storyboard.total_duration += processed_frame.duration or frame.duration or 0 |
| 770 | |
| 771 | logger.success(f"✅ Scene {i} complete") |
| 772 | |
| 773 | # Emit completion of frame production |
| 774 | self._emit_progress(ProgressEvent( |
| 775 | event_type="processing_frame", |
| 776 | progress=0.85, |
| 777 | frame_current=total_frames, |
| 778 | frame_total=total_frames |
| 779 | )) |
| 780 | |
| 781 | return context |
| 782 | |
| 783 | async def post_production(self, context: PipelineContext) -> PipelineContext: |
| 784 | """ |
| 785 | Concatenate scene videos and add BGM |
| 786 | |
| 787 | Args: |
| 788 | context: Pipeline context |
| 789 | |
| 790 | Returns: |
| 791 | Updated context with final video path |
| 792 | """ |
| 793 | logger.info("🎞️ Concatenating scenes...") |
| 794 | |
| 795 | # Emit progress for concatenation (85% - 95%) |
| 796 | self._emit_progress(ProgressEvent( |
| 797 | event_type="concatenating", |
| 798 | progress=0.86 |
| 799 | )) |
| 800 | |
| 801 | # Collect video segments from storyboard frames |
| 802 | scene_videos = [frame.video_segment_path for frame in context.storyboard.frames] |
| 803 | |
| 804 | # Generate filename: use title if provided, otherwise use task_id or default name |
| 805 | if context.title: |
| 806 | filename = f"{context.title}.mp4" |
| 807 | else: |
| 808 | filename = f"{context.task_id}.mp4" # Use task_id as filename when title is empty |
| 809 | |
| 810 | final_video_path = Path(context.task_dir) / filename |
| 811 | |
| 812 | # Get BGM parameters |
| 813 | bgm_path = context.request.get("bgm_path") |
| 814 | bgm_volume = context.request.get("bgm_volume", 0.2) |
| 815 | bgm_mode = context.request.get("bgm_mode", "loop") |
| 816 | |
| 817 | if bgm_path: |
| 818 | logger.info(f"🎵 Adding BGM: {bgm_path} (volume={bgm_volume}, mode={bgm_mode})") |
| 819 | |
| 820 | self.core.video.concat_videos( |
| 821 | videos=scene_videos, |
| 822 | output=str(final_video_path), |
| 823 | bgm_path=bgm_path, |
| 824 | bgm_volume=bgm_volume, |
| 825 | bgm_mode=bgm_mode |
| 826 | ) |
| 827 | |
| 828 | context.final_video_path = str(final_video_path) |
| 829 | context.storyboard.final_video_path = str(final_video_path) |
| 830 | context.storyboard.completed_at = datetime.now() |
| 831 | |
| 832 | if not context.storyboard.total_duration: |
| 833 | context.storyboard.total_duration = self._probe_video_duration(str(final_video_path)) |
| 834 | |
| 835 | logger.success(f"✅ Final video: {final_video_path}") |
| 836 | |
| 837 | # Emit completion of concatenation |
| 838 | self._emit_progress(ProgressEvent( |
| 839 | event_type="concatenating", |
| 840 | progress=0.95, |
| 841 | extra_info="complete" |
| 842 | )) |
| 843 | |
| 844 | return context |
| 845 | |
| 846 | async def finalize(self, context: PipelineContext) -> PipelineContext: |
| 847 | """ |
| 848 | Finalize and return result |
| 849 | |
| 850 | Args: |
| 851 | context: Pipeline context |
| 852 | |
| 853 | Returns: |
| 854 | Final context |
| 855 | """ |
| 856 | logger.success(f"🎉 Asset-based video generation complete!") |
| 857 | logger.info(f"Video: {context.final_video_path}") |
| 858 | |
| 859 | # Emit completion |
| 860 | self._emit_progress(ProgressEvent( |
| 861 | event_type="completed", |
| 862 | progress=1.0 |
| 863 | )) |
| 864 | |
| 865 | # Persist metadata for history tracking |
| 866 | await self._persist_task_data(context) |
| 867 | |
| 868 | return context |
| 869 | |
| 870 | async def _persist_task_data(self, ctx: PipelineContext): |
| 871 | """ |
| 872 | Persist task metadata and storyboard to filesystem for history tracking |
| 873 | """ |
| 874 | from pathlib import Path |
| 875 | |
| 876 | try: |
| 877 | storyboard = ctx.storyboard |
| 878 | task_id = ctx.task_id |
| 879 | |
| 880 | if not task_id: |
| 881 | logger.warning("No task_id in context, skipping persistence") |
| 882 | return |
| 883 | |
| 884 | # Get file size |
| 885 | video_path_obj = Path(ctx.final_video_path) |
| 886 | file_size = video_path_obj.stat().st_size if video_path_obj.exists() else 0 |
| 887 | |
| 888 | # Build metadata |
| 889 | input_params = { |
| 890 | "text": ctx.input_text, |
| 891 | "mode": "asset_based", |
| 892 | "title": ctx.title or "", |
| 893 | "n_scenes": len(storyboard.frames) if storyboard else 0, |
| 894 | "assets": ctx.request.get("assets", []), |
| 895 | "intent": ctx.request.get("intent"), |
| 896 | "duration": ctx.request.get("duration"), |
| 897 | "source": ctx.request.get("source"), |
| 898 | "voice_id": ctx.request.get("voice_id"), |
| 899 | "tts_speed": ctx.request.get("tts_speed"), |
| 900 | } |
| 901 | |
| 902 | metadata = { |
| 903 | "task_id": task_id, |
| 904 | "created_at": storyboard.created_at.isoformat() if storyboard and storyboard.created_at else None, |
| 905 | "completed_at": storyboard.completed_at.isoformat() if storyboard and storyboard.completed_at else None, |
| 906 | "status": "completed", |
| 907 | |
| 908 | "input": input_params, |
| 909 | |
| 910 | "result": { |
| 911 | "video_path": ctx.final_video_path, |
| 912 | "duration": storyboard.total_duration if storyboard else 0, |
| 913 | "file_size": file_size, |
| 914 | "n_frames": len(storyboard.frames) if storyboard else 0 |
| 915 | }, |
| 916 | |
| 917 | "config": { |
| 918 | "llm_model": self.core.config.get("llm", {}).get("model", "unknown"), |
| 919 | "llm_base_url": self.core.config.get("llm", {}).get("base_url", "unknown"), |
| 920 | "source": ctx.request.get("source", "runninghub"), |
| 921 | } |
| 922 | } |
| 923 | |
| 924 | # Save metadata |
| 925 | await self.core.persistence.save_task_metadata(task_id, metadata) |
| 926 | logger.info(f"💾 Saved task metadata: {task_id}") |
| 927 | |
| 928 | # Save storyboard |
| 929 | if storyboard: |
| 930 | await self.core.persistence.save_storyboard(task_id, storyboard) |
| 931 | logger.info(f"💾 Saved storyboard: {task_id}") |
| 932 | |
| 933 | except Exception as e: |
| 934 | logger.error(f"Failed to persist task data: {e}") |
| 935 | # Don't raise - persistence failure shouldn't break video generation |
| 936 | |
| 937 | # Helper methods |
| 938 | |
| 939 | def _get_asset_type(self, path: Path) -> str: |
| 940 | """Determine asset type from file extension""" |
| 941 | image_exts = {".jpg", ".jpeg", ".png", ".gif", ".webp"} |
| 942 | video_exts = {".mp4", ".mov", ".avi", ".mkv", ".webm"} |
| 943 | |
| 944 | ext = path.suffix.lower() |
| 945 | |
| 946 | if ext in image_exts: |
| 947 | return "image" |
| 948 | elif ext in video_exts: |
| 949 | return "video" |
| 950 | else: |
| 951 | return "unknown" |
| 952 | |
| 953 | def _get_api_workflow_info(self, workflow_key: str) -> dict: |
| 954 | """Find API workflow metadata for capability-aware adapter behavior.""" |
| 955 | try: |
| 956 | for workflow in self.core.api_media.list_workflows(): |
| 957 | if workflow.get("key") == workflow_key: |
| 958 | return workflow |
| 959 | except Exception as exc: |
| 960 | logger.warning(f"Failed to read API workflow metadata for {workflow_key}: {exc}") |
| 961 | return {} |
| 962 | |
| 963 | def _extract_video_tail_frame(self, video_path: str, output_path: str) -> Optional[str]: |
| 964 | """Extract the last visible frame from a generated scene video.""" |
| 965 | try: |
| 966 | import subprocess |
| 967 | |
| 968 | Path(output_path).parent.mkdir(parents=True, exist_ok=True) |
| 969 | extract_cmd = [ |
| 970 | "ffmpeg", |
| 971 | "-hide_banner", |
| 972 | "-loglevel", |
| 973 | "error", |
| 974 | "-sseof", |
| 975 | "-0.15", |
| 976 | "-i", |
| 977 | video_path, |
| 978 | "-frames:v", |
| 979 | "1", |
| 980 | "-q:v", |
| 981 | "2", |
| 982 | "-y", |
| 983 | output_path, |
| 984 | ] |
| 985 | subprocess.run(extract_cmd, capture_output=True, text=True, check=True) |
| 986 | if Path(output_path).exists(): |
| 987 | logger.info(f"Extracted tail reference frame: {output_path}") |
| 988 | return output_path |
| 989 | except Exception as exc: |
| 990 | logger.debug(f"Primary tail-frame extraction failed for {video_path}: {exc}") |
| 991 | |
| 992 | try: |
| 993 | import subprocess |
| 994 | |
| 995 | probe_cmd = [ |
| 996 | "ffprobe", |
| 997 | "-v", |
| 998 | "error", |
| 999 | "-show_entries", |
| 1000 | "format=duration", |
| 1001 | "-of", |
| 1002 | "default=noprint_wrappers=1:nokey=1", |
| 1003 | video_path, |
| 1004 | ] |
| 1005 | probe = subprocess.run(probe_cmd, capture_output=True, text=True, check=True) |
| 1006 | duration = float(probe.stdout.strip() or 0) |
| 1007 | seek_time = max(duration - 0.5, 0) |
| 1008 | fallback_cmd = [ |
| 1009 | "ffmpeg", |
| 1010 | "-hide_banner", |
| 1011 | "-loglevel", |
| 1012 | "error", |
| 1013 | "-i", |
| 1014 | video_path, |
| 1015 | "-ss", |
| 1016 | f"{seek_time:.3f}", |
| 1017 | "-frames:v", |
| 1018 | "1", |
| 1019 | "-q:v", |
| 1020 | "2", |
| 1021 | "-y", |
| 1022 | output_path, |
| 1023 | ] |
| 1024 | subprocess.run(fallback_cmd, capture_output=True, text=True, check=True) |
| 1025 | if Path(output_path).exists(): |
| 1026 | logger.info(f"Extracted tail reference frame with fallback: {output_path}") |
| 1027 | return output_path |
| 1028 | except Exception as exc: |
| 1029 | logger.warning(f"Failed to extract tail frame from {video_path}: {exc}") |
| 1030 | return None |
| 1031 | |
| 1032 | def _probe_video_duration(self, video_path: str) -> float: |
| 1033 | """Return video duration using ffprobe, or 0 on failure.""" |
| 1034 | try: |
| 1035 | import subprocess |
| 1036 | |
| 1037 | result = subprocess.run( |
| 1038 | [ |
| 1039 | "ffprobe", |
| 1040 | "-v", |
| 1041 | "error", |
| 1042 | "-show_entries", |
| 1043 | "format=duration", |
| 1044 | "-of", |
| 1045 | "default=noprint_wrappers=1:nokey=1", |
| 1046 | video_path, |
| 1047 | ], |
| 1048 | capture_output=True, |
| 1049 | text=True, |
| 1050 | check=True, |
| 1051 | ) |
| 1052 | return float(result.stdout.strip() or 0) |
| 1053 | except Exception as exc: |
| 1054 | logger.warning(f"Failed to probe video duration for {video_path}: {exc}") |
| 1055 | return 0.0 |
| 1056 | |
| 1057 |