| 1 | import os |
| 2 | import re |
| 3 | import logging |
| 4 | |
| 5 | from models.llm_client import LLM |
| 6 | |
| 7 | from .api_media import generate_image_api, generate_video_api |
| 8 | from .api_media import parse_api_workflow |
| 9 | from .storage import append_artifact, task_output_dir, update_task |
| 10 | from .tts import generate_edge_tts |
| 11 | from .utils import ( |
| 12 | artifact, |
| 13 | concat_audios, |
| 14 | concat_videos, |
| 15 | copy_input_file, |
| 16 | extract_last_frame, |
| 17 | media_duration_seconds, |
| 18 | replace_video_audio, |
| 19 | run_blocking, |
| 20 | speed_audio_to_duration, |
| 21 | write_json, |
| 22 | write_text, |
| 23 | ) |
| 24 | from models.config_model import video_capabilities |
| 25 | |
| 26 | logger = logging.getLogger(__name__) |
| 27 | |
| 28 | |
| 29 | def required_param(params: dict, key: str) -> str: |
| 30 | value = params.get(key) |
| 31 | if not value: |
| 32 | raise ValueError(f"digital_human pipeline requires {key}") |
| 33 | return str(value) |
| 34 | |
| 35 | |
| 36 | def split_by_periods(text: str) -> list[str]: |
| 37 | parts = re.findall(r"[^。．.]+[。．.]?|[^。．.]+$", text.strip()) |
| 38 | return [part.strip() for part in parts if part.strip()] |
| 39 | |
| 40 | |
| 41 | async def run(task_id: str, params: dict) -> tuple[dict, list[dict]]: |
| 42 | output_dir = task_output_dir(task_id) |
| 43 | os.makedirs(output_dir, exist_ok=True) |
| 44 | |
| 45 | mode = params.get("mode") or "customize" |
| 46 | logger.info("Digital human pipeline started: task_id=%s mode=%s", task_id, mode) |
| 47 | character_image = params.get("character_image_path") or params.get("character_asset") |
| 48 | goods_image = params.get("goods_image_path") or params.get("goods_asset") |
| 49 | goods_title = params.get("goods_title") or "" |
| 50 | goods_text = params.get("goods_text") or params.get("text") or "" |
| 51 | if not character_image: |
| 52 | raise ValueError("digital_human requires character_image_path") |
| 53 | if mode == "digital" and not goods_image: |
| 54 | raise ValueError("digital mode requires goods_image_path") |
| 55 | if mode == "customize" and not goods_text.strip(): |
| 56 | raise ValueError("customize mode requires goods_text") |
| 57 | if mode == "digital" and not (goods_text.strip() or goods_title.strip()): |
| 58 | raise ValueError("digital mode requires goods_text or goods_title") |
| 59 | |
| 60 | character_image = copy_input_file(character_image, output_dir, "character") |
| 61 | if goods_image: |
| 62 | goods_image = copy_input_file(goods_image, output_dir, "goods") |
| 63 | |
| 64 | llm = LLM() |
| 65 | llm_model = required_param(params, "llm_model") |
| 66 | if not goods_text.strip(): |
| 67 | update_task(task_id, progress=15, message="Generating digital-human script") |
| 68 | logger.info("Generating digital-human script from title: %s", goods_title) |
| 69 | goods_text = await run_blocking( |
| 70 | llm.query, |
| 71 | f"请为商品“{goods_title}”写一段适合数字人口播短视频的中文推广文案。要求自然、有吸引力，控制在80字以内，只输出文案正文。", |
| 72 | model=llm_model, |
| 73 | ) |
| 74 | goods_text = goods_text.strip() |
| 75 | if not goods_title.strip(): |
| 76 | update_task(task_id, progress=18, message="Generating digital-human title") |
| 77 | logger.info("Generating digital-human title from script: task_id=%s", task_id) |
| 78 | goods_title = await run_blocking( |
| 79 | llm.query, |
| 80 | f"为下面的数字人口播文案生成一个简短中文标题，只输出标题：\n{goods_text}", |
| 81 | model=llm_model, |
| 82 | ) |
| 83 | goods_title = goods_title.strip().splitlines()[0] |
| 84 | logger.info("Digital-human script ready: task_id=%s chars=%d", task_id, len(goods_text)) |
| 85 | script_path = write_text(os.path.join(output_dir, "script.txt"), goods_text) |
| 86 | append_artifact(task_id, artifact(script_path, "text", "script")) |
| 87 | |
| 88 | generated_image = None |
| 89 | reference_images = [character_image] |
| 90 | if mode == "digital" and goods_image: |
| 91 | reference_images.append(goods_image) |
| 92 | image_model = required_param(params, "image_model") |
| 93 | update_task(task_id, progress=30, message="Generating digital-human reference image") |
| 94 | logger.info("Generating digital-human reference image: model=%s refs=%d", image_model, len(reference_images)) |
| 95 | image_prompt = ( |
| 96 | f"Create a polished vertical digital-human product promotion image. " |
| 97 | f"Use reference image 1 as the presenter and reference image 2 as the product. " |
| 98 | f"Make the scene commercial, clean, and suitable for a spoken short video. " |
| 99 | f"Script: {goods_text}" |
| 100 | ) |
| 101 | generated_image = await run_blocking( |
| 102 | generate_image_api, |
| 103 | prompt=image_prompt, |
| 104 | model=image_model, |
| 105 | output_dir=output_dir, |
| 106 | task_id=task_id, |
| 107 | image_paths=reference_images, |
| 108 | video_ratio=params.get("video_ratio") or "9:16", |
| 109 | resolution=params.get("image_resolution") or "1080P", |
| 110 | ) |
| 111 | reference_images = [generated_image] |
| 112 | append_artifact(task_id, artifact(generated_image, "image", "generated_reference")) |
| 113 | |
| 114 | update_task(task_id, progress=62, message="Preparing digital-human video segments") |
| 115 | video_model = required_param(params, "video_model") |
| 116 | provider, resolved_video_model = parse_api_workflow(video_model, "video") |
| 117 | duration_contract = video_capabilities(provider, resolved_video_model).get("duration") or {} |
| 118 | max_duration = int(duration_contract.get("max") or 10) |
| 119 | min_duration = int(duration_contract.get("min") or 2) |
| 120 | segment_seconds = max(min_duration, max_duration) |
| 121 | narration_sentences = split_by_periods(goods_text) or [goods_text] |
| 122 | logger.info( |
| 123 | "Digital-human narration split: task_id=%s sentences=%d model=%s max_segment=%ss", |
| 124 | task_id, |
| 125 | len(narration_sentences), |
| 126 | resolved_video_model, |
| 127 | segment_seconds, |
| 128 | ) |
| 129 | |
| 130 | audio_segments = [] |
| 131 | audio_segment_texts = [] |
| 132 | audio_artifacts = [] |
| 133 | sentence_audio_paths = [] |
| 134 | for idx, sentence in enumerate(narration_sentences, 1): |
| 135 | update_task( |
| 136 | task_id, |
| 137 | progress=45 + int(15 * idx / max(len(narration_sentences), 1)), |
| 138 | message=f"Generating narration audio {idx}/{len(narration_sentences)}", |
| 139 | ) |
| 140 | logger.info( |
| 141 | "Generating TTS sentence %d/%d: chars=%d", |
| 142 | idx, |
| 143 | len(narration_sentences), |
| 144 | len(sentence), |
| 145 | ) |
| 146 | sentence_audio_path = os.path.join(output_dir, f"narration_sentence_{idx:02d}.mp3") |
| 147 | await generate_edge_tts( |
| 148 | sentence, |
| 149 | output_path=sentence_audio_path, |
| 150 | voice=params.get("tts_voice", "zh-CN-YunjianNeural"), |
| 151 | speed=float(params.get("tts_speed", 1.0)), |
| 152 | ) |
| 153 | sentence_audio_paths.append(sentence_audio_path) |
| 154 | sentence_duration = media_duration_seconds(sentence_audio_path) |
| 155 | if sentence_duration and sentence_duration > segment_seconds: |
| 156 | logger.info( |
| 157 | "Sentence audio exceeds model duration; speeding up: sentence=%d duration=%.2fs max=%ss", |
| 158 | idx, |
| 159 | sentence_duration, |
| 160 | segment_seconds, |
| 161 | ) |
| 162 | sped_audio_path = os.path.join(output_dir, f"narration_sentence_{idx:02d}_speed.mp3") |
| 163 | sped_audio_path = await run_blocking( |
| 164 | speed_audio_to_duration, |
| 165 | sentence_audio_path, |
| 166 | sped_audio_path, |
| 167 | segment_seconds, |
| 168 | ) |
| 169 | audio_segments.append(sped_audio_path) |
| 170 | audio_segment_texts.append(sentence) |
| 171 | audio_artifact = artifact(sped_audio_path, "audio", f"narration_sentence_{idx:02d}_speed") |
| 172 | audio_artifacts.append(audio_artifact) |
| 173 | append_artifact(task_id, audio_artifact) |
| 174 | else: |
| 175 | audio_segments.append(sentence_audio_path) |
| 176 | audio_segment_texts.append(sentence) |
| 177 | audio_artifact = artifact(sentence_audio_path, "audio", f"narration_sentence_{idx:02d}") |
| 178 | audio_artifacts.append(audio_artifact) |
| 179 | append_artifact(task_id, audio_artifact) |
| 180 | |
| 181 | audio_duration = sum(media_duration_seconds(path) or 0 for path in audio_segments) or None |
| 182 | logger.info( |
| 183 | "Digital-human audio prepared: task_id=%s segments=%d total_duration=%s", |
| 184 | task_id, |
| 185 | len(audio_segments), |
| 186 | f"{audio_duration:.2f}s" if audio_duration else "unknown", |
| 187 | ) |
| 188 | |
| 189 | subject_prompt = "参考图中的人物面对镜头自然口播。" |
| 190 | if mode == "digital": |
| 191 | subject_prompt += "结合商品信息，生成竖屏商业口播视频。" |
| 192 | segment_videos = [] |
| 193 | tail_frame = None |
| 194 | for idx, segment_audio_path in enumerate(audio_segments, 1): |
| 195 | segment_text = audio_segment_texts[idx - 1] if idx - 1 < len(audio_segment_texts) else goods_text |
| 196 | prompt = f"{subject_prompt} 口播文案：{segment_text}" |
| 197 | progress = 65 + int(25 * idx / max(len(audio_segments), 1)) |
| 198 | update_task( |
| 199 | task_id, |
| 200 | progress=progress, |
| 201 | message=f"Calling digital-human video API {idx}/{len(audio_segments)}", |
| 202 | ) |
| 203 | segment_duration = media_duration_seconds(segment_audio_path) or audio_duration or max_duration |
| 204 | safe_segment_duration = max(min_duration, min(max_duration, int(round(segment_duration)))) |
| 205 | logger.info( |
| 206 | "Generating digital-human video segment %d/%d: audio=%s duration=%ss tail_frame=%s", |
| 207 | idx, |
| 208 | len(audio_segments), |
| 209 | segment_audio_path, |
| 210 | safe_segment_duration, |
| 211 | bool(tail_frame), |
| 212 | ) |
| 213 | segment_video_path = os.path.join(output_dir, f"video_part_{idx:02d}.mp4") |
| 214 | segment_reference_images = [tail_frame] if tail_frame else reference_images |
| 215 | await run_blocking( |
| 216 | generate_video_api, |
| 217 | prompt=prompt, |
| 218 | model=video_model, |
| 219 | output_path=segment_video_path, |
| 220 | image_path=tail_frame, |
| 221 | duration=safe_segment_duration, |
| 222 | video_ratio=params.get("video_ratio") or "9:16", |
| 223 | reference_image_paths=segment_reference_images, |
| 224 | reference_audio_path=segment_audio_path, |
| 225 | audio=True, |
| 226 | negative_prompt=params.get("negative_prompt"), |
| 227 | video_resolution=params.get("video_resolution") or params.get("resolution"), |
| 228 | watermark=params.get("watermark"), |
| 229 | prompt_extend=params.get("prompt_extend"), |
| 230 | ) |
| 231 | segment_videos.append(segment_video_path) |
| 232 | append_artifact(task_id, artifact(segment_video_path, "video", f"video_part_{idx:02d}")) |
| 233 | |
| 234 | if idx < len(audio_segments): |
| 235 | tail_frame = os.path.join(output_dir, f"tail_frame_{idx:02d}.jpg") |
| 236 | await run_blocking(extract_last_frame, segment_video_path, tail_frame) |
| 237 | append_artifact(task_id, artifact(tail_frame, "image", f"tail_frame_{idx:02d}")) |
| 238 | |
| 239 | silent_video_path = segment_videos[0] if len(segment_videos) == 1 else concat_videos(segment_videos, os.path.join(output_dir, "final_video_only.mp4")) |
| 240 | if not silent_video_path: |
| 241 | raise RuntimeError("Digital-human video generation did not produce a final video.") |
| 242 | narration_audio_path = concat_audios(audio_segments, os.path.join(output_dir, "final_narration.mp3")) |
| 243 | if not narration_audio_path: |
| 244 | raise RuntimeError("Digital-human narration audio was not produced.") |
| 245 | with_audio_path = replace_video_audio( |
| 246 | silent_video_path, |
| 247 | narration_audio_path, |
| 248 | os.path.join(output_dir, "final.mp4"), |
| 249 | ) |
| 250 | final_video_path = with_audio_path |
| 251 | append_artifact(task_id, artifact(narration_audio_path, "audio", "final_narration")) |
| 252 | append_artifact(task_id, artifact(final_video_path, "video", "final")) |
| 253 | logger.info("Digital human pipeline completed: task_id=%s final_video=%s", task_id, final_video_path) |
| 254 | |
| 255 | request_path = write_json(os.path.join(output_dir, "request.json"), { |
| 256 | "mode": mode, |
| 257 | "goods_title": goods_title, |
| 258 | "goods_text": goods_text, |
| 259 | "character_image": character_image, |
| 260 | "goods_image": goods_image, |
| 261 | "generated_image": generated_image, |
| 262 | "video_model": video_model, |
| 263 | "audio_duration": audio_duration, |
| 264 | "narration_sentences": narration_sentences, |
| 265 | "sentence_audio_paths": sentence_audio_paths, |
| 266 | "audio_segments": audio_segments, |
| 267 | "audio_segment_texts": audio_segment_texts, |
| 268 | "final_narration_audio": narration_audio_path, |
| 269 | "segment_videos": segment_videos, |
| 270 | "silent_video_path": silent_video_path, |
| 271 | "with_audio_path": with_audio_path, |
| 272 | }) |
| 273 | |
| 274 | artifacts = [ |
| 275 | artifact(request_path, "text", "request"), |
| 276 | artifact(script_path, "text", "script"), |
| 277 | artifact(character_image, "image", "character"), |
| 278 | artifact(narration_audio_path, "audio", "final_narration"), |
| 279 | artifact(final_video_path, "video", "final"), |
| 280 | ] |
| 281 | artifacts.extend(audio_artifacts) |
| 282 | if goods_image: |
| 283 | artifacts.append(artifact(goods_image, "image", "goods")) |
| 284 | if generated_image: |
| 285 | artifacts.append(artifact(generated_image, "image", "generated_reference")) |
| 286 | for item in artifacts: |
| 287 | append_artifact(task_id, item) |
| 288 | |
| 289 | output = { |
| 290 | "script": goods_text, |
| 291 | "script_path": script_path, |
| 292 | "audio_path": audio_segments[0] if audio_segments else None, |
| 293 | "audio_paths": audio_segments, |
| 294 | "final_audio_path": narration_audio_path, |
| 295 | "audio_segment_texts": audio_segment_texts, |
| 296 | "video_path": final_video_path, |
| 297 | "video_parts": segment_videos, |
| 298 | "generated_image": generated_image, |
| 299 | } |
| 300 | return output, artifacts |
| 301 |