返回 Pixelle-Video
standard.py
根目录 / pixelle_video / pipelines / standard.py
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 Standard Video Generation Pipeline
15
16 Standard workflow for generating short videos from topic or fixed script.
17 This is the default pipeline for general-purpose video generation.
18 Refactored to use LinearVideoPipeline (Template Method Pattern).
19 """
20
21 from datetime import datetime
22 from pathlib import Path
23 from typing import Optional, Callable, Literal, List
24 import asyncio
25 import shutil
26
27 from loguru import logger
28
29 from pixelle_video.pipelines.linear import LinearVideoPipeline, PipelineContext
30 from pixelle_video.models.progress import ProgressEvent
31 from pixelle_video.models.storyboard import (
32 Storyboard,
33 StoryboardFrame,
34 StoryboardConfig,
35 ContentMetadata,
36 VideoGenerationResult
37 )
38 from pixelle_video.utils.content_generators import (
39 generate_title,
40 generate_narrations_from_topic,
41 split_narration_script,
42 generate_image_prompts,
43 )
44 from pixelle_video.utils.os_util import (
45 create_task_output_dir,
46 get_task_final_video_path
47 )
48 from pixelle_video.utils.template_util import get_template_type
49 from pixelle_video.utils.prompt_helper import build_image_prompt
50 from pixelle_video.services.video import VideoService
51
52
53
54
55 class StandardPipeline(LinearVideoPipeline):
56 """
57 Standard video generation pipeline
58
59 Workflow:
60 1. Generate/determine title
61 2. Generate narrations (from topic or split fixed script)
62 3. Generate image prompts for each narration
63 4. For each frame:
64 - Generate audio (TTS)
65 - Generate image
66 - Compose frame with template
67 - Create video segment
68 5. Concatenate all segments
69 6. Add BGM (optional)
70
71 Supports two modes:
72 - "generate": LLM generates narrations from topic
73 - "fixed": Use provided script as-is (each line = one narration)
74 """
75
76 # ==================== Lifecycle Methods ====================
77
78 async def setup_environment(self, ctx: PipelineContext):
79 """Step 1: Setup task directory and environment."""
80 text = ctx.input_text
81 mode = ctx.params.get("mode", "generate")
82
83 logger.info(f"🚀 Starting StandardPipeline in '{mode}' mode")
84 logger.info(f" Text length: {len(text)} chars")
85
86 # Create isolated task directory
87 task_dir, task_id = create_task_output_dir()
88 ctx.task_id = task_id
89 ctx.task_dir = task_dir
90
91 logger.info(f"📁 Task directory created: {task_dir}")
92 logger.info(f" Task ID: {task_id}")
93
94 # Determine final video path
95 output_path = ctx.params.get("output_path")
96 if output_path is None:
97 ctx.final_video_path = get_task_final_video_path(task_id)
98 else:
99 # We will copy to this path in finalize/post_production
100 # For internal processing, we still use the task dir path?
101 # Actually StandardPipeline logic used get_task_final_video_path as the target for concat
102 # and then copied. Let's stick to that.
103 ctx.final_video_path = get_task_final_video_path(task_id)
104 logger.info(f" Will copy final video to: {output_path}")
105
106 async def generate_content(self, ctx: PipelineContext):
107 """Step 2: Generate or process script/narrations."""
108 mode = ctx.params.get("mode", "generate")
109 text = ctx.input_text
110 n_scenes = ctx.params.get("n_scenes", 5)
111 min_words = ctx.params.get("min_narration_words", 5)
112 max_words = ctx.params.get("max_narration_words", 20)
113
114 if mode == "generate":
115 self._report_progress(ctx.progress_callback, "generating_narrations", 0.05)
116 ctx.narrations = await generate_narrations_from_topic(
117 self.llm,
118 topic=text,
119 n_scenes=n_scenes,
120 min_words=min_words,
121 max_words=max_words
122 )
123 logger.info(f"✅ Generated {len(ctx.narrations)} narrations")
124 else: # fixed
125 self._report_progress(ctx.progress_callback, "splitting_script", 0.05)
126 split_mode = ctx.params.get("split_mode", "paragraph")
127 ctx.narrations = await split_narration_script(text, split_mode=split_mode)
128 logger.info(f"✅ Split script into {len(ctx.narrations)} segments (mode={split_mode})")
129 logger.info(f" Note: n_scenes={n_scenes} is ignored in fixed mode")
130
131 async def determine_title(self, ctx: PipelineContext):
132 """Step 3: Determine or generate video title."""
133 # Note: Swapped order with generate_content in base class call,
134 # but in StandardPipeline original code, title was determined BEFORE narrations.
135 # However, LinearVideoPipeline defines generate_content BEFORE determine_title.
136 # This is fine as they are independent in StandardPipeline logic.
137
138 title = ctx.params.get("title")
139 mode = ctx.params.get("mode", "generate")
140 text = ctx.input_text
141
142 if title:
143 ctx.title = title
144 logger.info(f" Title: '{title}' (user-specified)")
145 else:
146 self._report_progress(ctx.progress_callback, "generating_title", 0.01)
147 if mode == "generate":
148 ctx.title = await generate_title(self.llm, text, strategy="auto")
149 logger.info(f" Title: '{ctx.title}' (auto-generated)")
150 else: # fixed
151 ctx.title = await generate_title(self.llm, text, strategy="llm")
152 logger.info(f" Title: '{ctx.title}' (LLM-generated)")
153
154 async def plan_visuals(self, ctx: PipelineContext):
155 """Step 4: Generate image prompts or visual descriptions."""
156 # Detect template type to determine if media generation is needed
157 frame_template = ctx.params.get("frame_template") or "1080x1920/default.html"
158
159 template_name = Path(frame_template).name
160 template_type = get_template_type(template_name)
161 template_requires_media = (template_type in ["image", "video"])
162
163 if template_type == "image":
164 logger.info(f"📸 Template requires image generation")
165 elif template_type == "video":
166 logger.info(f"🎬 Template requires video generation")
167 else: # static
168 logger.info(f"⚡ Static template - skipping media generation pipeline")
169 logger.info(f" 💡 Benefits: Faster generation + Lower cost + No ComfyUI dependency")
170
171 # Only generate image prompts if template requires media
172 if template_requires_media:
173 self._report_progress(ctx.progress_callback, "generating_image_prompts", 0.15)
174
175 prompt_prefix = ctx.params.get("prompt_prefix")
176 min_words = ctx.params.get("min_image_prompt_words", 30)
177 max_words = ctx.params.get("max_image_prompt_words", 60)
178
179 # Override prompt_prefix if provided
180 original_prefix = None
181 if prompt_prefix is not None:
182 image_config = self.core.config.get("comfyui", {}).get("image", {})
183 original_prefix = image_config.get("prompt_prefix")
184 image_config["prompt_prefix"] = prompt_prefix
185 logger.info(f"Using custom prompt_prefix: '{prompt_prefix}'")
186
187 try:
188 # Create progress callback wrapper for image prompt generation
189 def image_prompt_progress(completed: int, total: int, message: str):
190 batch_progress = completed / total if total > 0 else 0
191 overall_progress = 0.15 + (batch_progress * 0.15)
192 self._report_progress(
193 ctx.progress_callback,
194 "generating_image_prompts",
195 overall_progress,
196 extra_info=message
197 )
198
199 # Generate base image prompts
200 base_image_prompts = await generate_image_prompts(
201 self.llm,
202 narrations=ctx.narrations,
203 min_words=min_words,
204 max_words=max_words,
205 progress_callback=image_prompt_progress
206 )
207
208 # Apply prompt prefix
209 image_config = self.core.config.get("comfyui", {}).get("image", {})
210 prompt_prefix_to_use = prompt_prefix if prompt_prefix is not None else image_config.get("prompt_prefix", "")
211
212 ctx.image_prompts = []
213 for base_prompt in base_image_prompts:
214 final_prompt = build_image_prompt(base_prompt, prompt_prefix_to_use)
215 ctx.image_prompts.append(final_prompt)
216
217 finally:
218 # Restore original prompt_prefix
219 if original_prefix is not None:
220 image_config["prompt_prefix"] = original_prefix
221
222 logger.info(f"✅ Generated {len(ctx.image_prompts)} image prompts")
223 else:
224 # Static template - skip image prompt generation entirely
225 ctx.image_prompts = [None] * len(ctx.narrations)
226 logger.info(f"⚡ Skipped image prompt generation (static template)")
227 logger.info(f" 💡 Savings: {len(ctx.narrations)} LLM calls + {len(ctx.narrations)} media generations")
228
229 async def initialize_storyboard(self, ctx: PipelineContext):
230 """Step 5: Create Storyboard object and frames."""
231 # === Handle TTS parameter compatibility ===
232 tts_inference_mode = ctx.params.get("tts_inference_mode")
233 tts_voice = ctx.params.get("tts_voice")
234 voice_id = ctx.params.get("voice_id")
235 tts_workflow = ctx.params.get("tts_workflow")
236
237 final_voice_id = None
238 final_tts_workflow = tts_workflow
239
240 if tts_inference_mode:
241 # New API from web UI
242 if tts_inference_mode == "local":
243 final_voice_id = tts_voice or "zh-CN-YunjianNeural"
244 final_tts_workflow = None
245 logger.debug(f"TTS Mode: local (voice={final_voice_id})")
246 elif tts_inference_mode == "comfyui":
247 final_voice_id = None
248 logger.debug(f"TTS Mode: comfyui (workflow={final_tts_workflow})")
249 else:
250 # Old API
251 final_voice_id = voice_id or tts_voice or "zh-CN-YunjianNeural"
252 logger.debug(f"TTS Mode: legacy (voice_id={final_voice_id}, workflow={final_tts_workflow})")
253
254 # Create config
255 ctx.config = StoryboardConfig(
256 task_id=ctx.task_id,
257 n_storyboard=len(ctx.narrations), # Use actual length
258 min_narration_words=ctx.params.get("min_narration_words", 5),
259 max_narration_words=ctx.params.get("max_narration_words", 20),
260 min_image_prompt_words=ctx.params.get("min_image_prompt_words", 30),
261 max_image_prompt_words=ctx.params.get("max_image_prompt_words", 60),
262 video_fps=ctx.params.get("video_fps", 30),
263 tts_inference_mode=tts_inference_mode or "local",
264 voice_id=final_voice_id,
265 tts_workflow=final_tts_workflow,
266 tts_speed=ctx.params.get("tts_speed", 1.2),
267 ref_audio=ctx.params.get("ref_audio"),
268 media_width=ctx.params.get("media_width"),
269 media_height=ctx.params.get("media_height"),
270 media_workflow=ctx.params.get("media_workflow"),
271 api_video_params=ctx.params.get("api_video_params"),
272 frame_template=ctx.params.get("frame_template") or "1080x1920/default.html",
273 template_params=ctx.params.get("template_params")
274 )
275
276 # Create storyboard
277 ctx.storyboard = Storyboard(
278 title=ctx.title,
279 config=ctx.config,
280 content_metadata=ctx.params.get("content_metadata"),
281 created_at=datetime.now()
282 )
283
284 # Create frames
285 for i, (narration, image_prompt) in enumerate(zip(ctx.narrations, ctx.image_prompts)):
286 frame = StoryboardFrame(
287 index=i,
288 narration=narration,
289 image_prompt=image_prompt,
290 created_at=datetime.now()
291 )
292 ctx.storyboard.frames.append(frame)
293
294 async def produce_assets(self, ctx: PipelineContext):
295 """Step 6: Generate audio, images, and render frames (Core processing)."""
296 storyboard = ctx.storyboard
297 config = ctx.config
298
299 # Check if using RunningHub workflows for parallel processing
300 is_runninghub = (
301 (config.tts_workflow and config.tts_workflow.startswith("runninghub/")) or
302 (config.media_workflow and config.media_workflow.startswith("runninghub/"))
303 )
304
305 # Get concurrent limit from config_manager (supports hot reload without restart)
306 from pixelle_video.config import config_manager
307 runninghub_concurrent_limit = config_manager.config.comfyui.runninghub_concurrent_limit or 1
308
309 if is_runninghub and runninghub_concurrent_limit > 1:
310 logger.info(f"🚀 Using parallel processing for RunningHub workflows (max {runninghub_concurrent_limit} concurrent)")
311
312 semaphore = asyncio.Semaphore(runninghub_concurrent_limit)
313 completed_count = 0
314
315 async def process_frame_with_semaphore(i: int, frame: StoryboardFrame):
316 nonlocal completed_count
317 async with semaphore:
318 base_progress = 0.2
319 frame_range = 0.6
320 per_frame_progress = frame_range / len(storyboard.frames)
321
322 # Create frame-specific progress callback
323 def frame_progress_callback(event: ProgressEvent):
324 overall_progress = base_progress + (per_frame_progress * completed_count) + (per_frame_progress * event.progress)
325 if ctx.progress_callback:
326 adjusted_event = ProgressEvent(
327 event_type=event.event_type,
328 progress=overall_progress,
329 frame_current=i+1,
330 frame_total=len(storyboard.frames),
331 step=event.step,
332 action=event.action
333 )
334 ctx.progress_callback(adjusted_event)
335
336 # Report frame start
337 self._report_progress(
338 ctx.progress_callback,
339 "processing_frame",
340 base_progress + (per_frame_progress * completed_count),
341 frame_current=i+1,
342 frame_total=len(storyboard.frames)
343 )
344
345 processed_frame = await self.core.frame_processor(
346 frame=frame,
347 storyboard=storyboard,
348 config=config,
349 total_frames=len(storyboard.frames),
350 progress_callback=frame_progress_callback
351 )
352
353 completed_count += 1
354 logger.info(f"✅ Frame {i+1} completed ({processed_frame.duration:.2f}s) [{completed_count}/{len(storyboard.frames)}]")
355 return i, processed_frame
356
357 # Create all tasks and execute in parallel
358 tasks = [process_frame_with_semaphore(i, frame) for i, frame in enumerate(storyboard.frames)]
359 results = await asyncio.gather(*tasks)
360
361 # Update frames in order and calculate total duration
362 for idx, processed_frame in sorted(results, key=lambda x: x[0]):
363 storyboard.frames[idx] = processed_frame
364 storyboard.total_duration += processed_frame.duration
365
366 logger.info(f"✅ All frames processed in parallel (total duration: {storyboard.total_duration:.2f}s)")
367 else:
368 # Serial processing for non-RunningHub workflows
369 logger.info("⚙️ Using serial processing (non-RunningHub workflow)")
370
371 for i, frame in enumerate(storyboard.frames):
372 base_progress = 0.2
373 frame_range = 0.6
374 per_frame_progress = frame_range / len(storyboard.frames)
375
376 # Create frame-specific progress callback
377 def frame_progress_callback(event: ProgressEvent):
378 overall_progress = base_progress + (per_frame_progress * i) + (per_frame_progress * event.progress)
379 if ctx.progress_callback:
380 adjusted_event = ProgressEvent(
381 event_type=event.event_type,
382 progress=overall_progress,
383 frame_current=event.frame_current,
384 frame_total=event.frame_total,
385 step=event.step,
386 action=event.action
387 )
388 ctx.progress_callback(adjusted_event)
389
390 # Report frame start
391 self._report_progress(
392 ctx.progress_callback,
393 "processing_frame",
394 base_progress + (per_frame_progress * i),
395 frame_current=i+1,
396 frame_total=len(storyboard.frames)
397 )
398
399 processed_frame = await self.core.frame_processor(
400 frame=frame,
401 storyboard=storyboard,
402 config=config,
403 total_frames=len(storyboard.frames),
404 progress_callback=frame_progress_callback
405 )
406 storyboard.total_duration += processed_frame.duration
407 logger.info(f"✅ Frame {i+1} completed ({processed_frame.duration:.2f}s)")
408
409 async def post_production(self, ctx: PipelineContext):
410 """Step 7: Concatenate videos and add BGM."""
411 self._report_progress(ctx.progress_callback, "concatenating", 0.85)
412
413 storyboard = ctx.storyboard
414 segment_paths = [frame.video_segment_path for frame in storyboard.frames]
415
416 video_service = VideoService()
417
418 final_video_path = video_service.concat_videos(
419 videos=segment_paths,
420 output=ctx.final_video_path,
421 bgm_path=ctx.params.get("bgm_path"),
422 bgm_volume=ctx.params.get("bgm_volume", 0.2),
423 bgm_mode=ctx.params.get("bgm_mode", "loop")
424 )
425
426 storyboard.final_video_path = final_video_path
427 storyboard.completed_at = datetime.now()
428
429 # Copy to user-specified path if provided
430 user_specified_output = ctx.params.get("output_path")
431 if user_specified_output:
432 Path(user_specified_output).parent.mkdir(parents=True, exist_ok=True)
433 shutil.copy2(final_video_path, user_specified_output)
434 logger.info(f"📹 Final video copied to: {user_specified_output}")
435 ctx.final_video_path = user_specified_output
436 storyboard.final_video_path = user_specified_output
437
438 logger.success(f"🎬 Video generation completed: {ctx.final_video_path}")
439
440 async def finalize(self, ctx: PipelineContext) -> VideoGenerationResult:
441 """Step 8: Create result object and persist metadata."""
442 self._report_progress(ctx.progress_callback, "completed", 1.0)
443
444 video_path_obj = Path(ctx.final_video_path)
445 file_size = video_path_obj.stat().st_size
446
447 result = VideoGenerationResult(
448 video_path=ctx.final_video_path,
449 storyboard=ctx.storyboard,
450 duration=ctx.storyboard.total_duration,
451 file_size=file_size
452 )
453
454 ctx.result = result
455
456 logger.info(f"✅ Generated video: {ctx.final_video_path}")
457 logger.info(f" Duration: {ctx.storyboard.total_duration:.2f}s")
458 logger.info(f" Size: {file_size / (1024*1024):.2f} MB")
459 logger.info(f" Frames: {len(ctx.storyboard.frames)}")
460
461 # Persist metadata
462 await self._persist_task_data(ctx)
463
464 return result
465
466 async def _persist_task_data(self, ctx: PipelineContext):
467 """
468 Persist task metadata and storyboard to filesystem
469 """
470 try:
471 storyboard = ctx.storyboard
472 result = ctx.result
473 task_id = storyboard.config.task_id
474
475 if not task_id:
476 logger.warning("No task_id in storyboard, skipping persistence")
477 return
478
479 # Build metadata
480 input_with_title = ctx.params.copy()
481 input_with_title["text"] = ctx.input_text # Ensure text is included
482 if not input_with_title.get("title"):
483 input_with_title["title"] = storyboard.title
484
485 metadata = {
486 "task_id": task_id,
487 "created_at": storyboard.created_at.isoformat() if storyboard.created_at else None,
488 "completed_at": storyboard.completed_at.isoformat() if storyboard.completed_at else None,
489 "status": "completed",
490
491 "input": input_with_title,
492
493 "result": {
494 "video_path": result.video_path,
495 "duration": result.duration,
496 "file_size": result.file_size,
497 "n_frames": len(storyboard.frames)
498 },
499
500 "config": {
501 "llm_model": self.core.config.get("llm", {}).get("model", "unknown"),
502 "llm_base_url": self.core.config.get("llm", {}).get("base_url", "unknown"),
503 "comfyui_url": self.core.config.get("comfyui", {}).get("comfyui_url", "unknown"),
504 "runninghub_enabled": bool(self.core.config.get("comfyui", {}).get("runninghub_api_key")),
505 }
506 }
507
508 # Save metadata
509 await self.core.persistence.save_task_metadata(task_id, metadata)
510 logger.info(f"💾 Saved task metadata: {task_id}")
511
512 # Save storyboard
513 await self.core.persistence.save_storyboard(task_id, storyboard)
514 logger.info(f"💾 Saved storyboard: {task_id}")
515
516 except Exception as e:
517 logger.error(f"Failed to persist task data: {e}")
518 # Don't raise - persistence failure shouldn't break video generation
519
519 lines PYTHON