import json import os import re import shutil import subprocess import time import uuid from typing import Any, Dict, List, Optional, Set, Tuple from loguru import logger from app.models.schema import VideoClipParams MICROSECONDS = 1_000_000 DRAFT_PATH_PLACEHOLDER = "##_draftpath_placeholder_0E685133-18CE-45ED-8CB8-2904A212EC80_##" DRAFT_PATH_PLACEHOLDER_PATTERN = re.compile(r"^##_draftpath_placeholder_[^#]+_##/") MATERIAL_COLLECTION_KEYS = [ "ai_translates", "audio_balances", "audio_effects", "audio_fades", "audio_pannings", "audio_pitch_shifts", "audio_track_indexes", "audios", "beats", "canvases", "chromas", "color_curves", "common_mask", "digital_human_model_dressing", "digital_humans", "drafts", "effects", "flowers", "green_screens", "handwrites", "hsl", "hsl_curves", "images", "log_color_wheels", "loudnesses", "manual_beautys", "manual_deformations", "material_animations", "material_colors", "multi_language_refs", "placeholder_infos", "placeholders", "plugin_effects", "primary_color_wheels", "realtime_denoises", "shapes", "smart_crops", "smart_relights", "sound_channel_mappings", "speeds", "stickers", "tail_leaders", "text_templates", "texts", "time_marks", "transitions", "video_effects", "video_radius", "video_shadows", "video_strokes", "video_trackings", "videos", "vocal_beautifys", "vocal_separations", ] DRAFT_PACKAGE_DIRECTORIES = [ "qr_upload", "matting", "common_attachment", "Resources/audioAlg", "Resources/digitalHuman", "Resources/restore_lut", "Resources/videoAlg", "subdraft", "adjust_mask", "assets/audio", "assets/video", "smart_crop", ] DEFAULT_DRAFT_COVER_BYTES = bytes([ 0xFF, 0xD8, 0xFF, 0xDB, 0x00, 0x43, 0x00, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xC0, 0x00, 0x0B, 0x08, 0x00, 0x01, 0x00, 0x01, 0x01, 0x01, 0x11, 0x00, 0xFF, 0xC4, 0x00, 0x14, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0xFF, 0xC4, 0x00, 0x14, 0x10, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0xFF, 0xDA, 0x00, 0x08, 0x01, 0x01, 0x00, 0x00, 0x3F, 0x00, 0x7F, 0xFF, 0xD9, ]) def _write_json_file(file_path: str, data: Dict[str, Any]) -> None: os.makedirs(os.path.dirname(file_path), exist_ok=True) with open(file_path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, separators=(",", ":")) def _floor_duration_to_milliseconds(duration: float) -> float: return int(max(duration, 0.0) * 1000) / 1000.0 def _seconds_to_microseconds(seconds: float) -> int: return int(round(max(seconds, 0.0) * MICROSECONDS)) def _get_media_duration_ffprobe(media_file: str) -> float: cmd = [ "ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", media_file, ] result = subprocess.run(cmd, capture_output=True, text=True, check=True) return float(result.stdout.strip()) def _get_cached_media_duration(media_file: str, duration_cache: Dict[str, float]) -> float: if media_file not in duration_cache: duration_cache[media_file] = _floor_duration_to_milliseconds( _get_media_duration_ffprobe(media_file) ) return duration_cache[media_file] def _clamp_duration_to_media( requested_duration: float, media_file: str, duration_cache: Dict[str, float], media_label: str, source_start_time: float = 0.0, ) -> float: requested_duration = _floor_duration_to_milliseconds(requested_duration) actual_duration = _get_cached_media_duration(media_file, duration_cache) available_duration = _floor_duration_to_milliseconds( max(actual_duration - max(source_start_time, 0.0), 0.0) ) safe_duration = min(requested_duration, available_duration) logger.info( f"{media_label}实际时长: {actual_duration:.6f}秒, " f"可用时长: {available_duration:.6f}秒, 请求时长: {requested_duration:.3f}秒" ) if safe_duration < requested_duration: logger.warning( f"{media_label}短于脚本时长，已将剪映片段时长从 " f"{requested_duration:.3f}秒 调整为 {safe_duration:.3f}秒" ) return safe_duration def _get_video_metadata_ffprobe( media_file: str, metadata_cache: Dict[str, Tuple[int, int, int]], ) -> Tuple[int, int, int]: if media_file in metadata_cache: return metadata_cache[media_file] try: cmd = [ "ffprobe", "-v", "error", "-show_entries", "stream=width,height:format=duration", "-of", "json", media_file, ] result = subprocess.run(cmd, capture_output=True, text=True, check=True) info = json.loads(result.stdout or "{}") stream = next( ( item for item in info.get("streams", []) if item.get("width") and item.get("height") ), {}, ) duration = _floor_duration_to_milliseconds( float(info.get("format", {}).get("duration") or 0.0) ) width = int(stream.get("width") or 1920) height = int(stream.get("height") or 1080) metadata_cache[media_file] = (_seconds_to_microseconds(duration), width, height) except Exception as e: logger.warning(f"读取视频元信息失败，将使用默认分辨率: {media_file}, {e}") duration = _floor_duration_to_milliseconds(_get_media_duration_ffprobe(media_file)) metadata_cache[media_file] = (_seconds_to_microseconds(duration), 1920, 1080) return metadata_cache[media_file] def _format_draft_uuid(draft_id: str) -> str: compact = draft_id.replace("-", "") if not re.fullmatch(r"[a-fA-F0-9]{32}", compact): return draft_id return "-".join([ compact[0:8], compact[8:12], compact[12:16], compact[16:20], compact[20:32], ]).upper() def _detect_platform(draft_root_path: str) -> str: return "windows" if re.match(r"^(?:[a-zA-Z]:[\\/]|\\\\)", draft_root_path) else "mac" def _create_platform_info(draft_root_path: str) -> Dict[str, Any]: return { "app_id": 3704, "app_source": "lv", "app_version": "10.6.0", "device_id": "", "hard_disk_id": "", "mac_address": "", "os": _detect_platform(draft_root_path), "os_version": "", } def _default_function_assistant_info() -> Dict[str, Any]: return { "audio_noise_segid_list": [], "auto_adjust": False, "auto_adjust_fixed": False, "auto_adjust_fixed_value": 50.0, "auto_adjust_segid_list": [], "auto_caption": False, "auto_caption_segid_list": [], "auto_caption_template_id": "", "caption_opt": False, "caption_opt_segid_list": [], "color_correction": False, "color_correction_fixed": False, "color_correction_fixed_value": 50.0, "color_correction_segid_list": [], "deflicker_segid_list": [], "enhance_quality": False, "enhance_quality_fixed": False, "enhance_quality_segid_list": [], "enhance_voice_segid_list": [], "enhande_voice": False, "enhande_voice_fixed": False, "eye_correction": False, "eye_correction_segid_list": [], "fixed_rec_applied": False, "fps": {"den": 1, "num": 0}, "normalize_loudness": False, "normalize_loudness_audio_denoise_segid_list": [], "normalize_loudness_fixed": False, "normalize_loudness_segid_list": [], "retouch": False, "retouch_fixed": False, "retouch_segid_list": [], "smart_rec_applied": False, "smart_segid_list": [], "smooth_slow_motion": False, "smooth_slow_motion_fixed": False, "video_noise_segid_list": [], } def _safe_file_name(file_path: str, fallback: str) -> str: name = os.path.basename(file_path) or fallback name = re.sub(r'[<>:"|?*\x00-\x1f/\\]+', "_", name).strip(" ._") return name or fallback def _normalize_asset_path(file_path: Optional[str], fallback: str) -> str: normalized = (file_path or "").replace("\\", "/").lstrip("./") without_draft_placeholder = DRAFT_PATH_PLACEHOLDER_PATTERN.sub("", normalized) if without_draft_placeholder.startswith("assets/"): return without_draft_placeholder assets_index = without_draft_placeholder.rfind("/assets/") if assets_index >= 0: return without_draft_placeholder[assets_index + 1:] return fallback def _to_draft_material_path(relative_path: str) -> str: return f"{DRAFT_PATH_PLACEHOLDER}/{relative_path}" def _unique_relative_asset_path( directory: str, file_name: str, used_paths: Set[str], ) -> str: base_name, ext = os.path.splitext(file_name) candidate_name = file_name counter = 2 while True: relative_path = f"{directory}/{candidate_name}" if relative_path not in used_paths: used_paths.add(relative_path) return relative_path candidate_name = f"{base_name}_{counter}{ext}" counter += 1 def _copy_asset_into_draft(source_file: str, draft_path: str, relative_path: str) -> None: destination = os.path.join(draft_path, *relative_path.split("/")) os.makedirs(os.path.dirname(destination), exist_ok=True) if os.path.abspath(source_file) != os.path.abspath(destination): shutil.copy2(source_file, destination) def _register_asset( source_file: str, draft_path: str, asset_dir: str, fallback_name: str, used_paths: Set[str], asset_path_cache: Dict[str, str], ) -> str: source_key = os.path.abspath(source_file) if source_key in asset_path_cache: return asset_path_cache[source_key] file_name = _safe_file_name(source_file, fallback_name) relative_path = _unique_relative_asset_path(asset_dir, file_name, used_paths) _copy_asset_into_draft(source_file, draft_path, relative_path) asset_path_cache[source_key] = relative_path return relative_path def _create_unique_draft_path(drafts_root: str, draft_name: str) -> Tuple[str, str]: folder_base = _safe_file_name(draft_name, f"NarratoAI_{int(time.time())}") folder_name = folder_base counter = 2 while os.path.exists(os.path.join(drafts_root, folder_name)): folder_name = f"{folder_base}_{counter}" counter += 1 return folder_name, os.path.join(drafts_root, folder_name) def _create_material_collections() -> Dict[str, List[Any]]: return {key: [] for key in MATERIAL_COLLECTION_KEYS} def _create_draft_template( draft_id: str, draft_name: str, draft_root_path: str, width: int = 1920, height: int = 1080, ) -> Dict[str, Any]: now_us = int(time.time() * MICROSECONDS) platform_info = _create_platform_info(draft_root_path) return { "canvas_config": {"height": height, "ratio": "original", "width": width}, "color_space": 0, "config": { "adjust_max_index": 1, "attachment_info": [], "combination_max_index": 1, "export_range": None, "extract_audio_last_index": 1, "lyrics_recognition_id": "", "lyrics_sync": True, "lyrics_taskinfo": [], "maintrack_adsorb": True, "material_save_mode": 0, "multi_language_current": "none", "multi_language_list": [], "multi_language_main": "none", "multi_language_mode": "none", "original_sound_last_index": 1, "record_audio_last_index": 1, "sticker_max_index": 1, "subtitle_keywords_config": None, "subtitle_recognition_id": "", "subtitle_sync": True, "subtitle_taskinfo": [], "system_font_list": [], "video_mute": False, "zoom_info_params": None, }, "cover": None, "create_time": now_us, "duration": 0, "extra_info": None, "fps": 30.0, "free_render_index_mode_on": False, "group_container": None, "id": draft_id, "keyframe_graph_list": [], "keyframes": { "adjusts": [], "audios": [], "effects": [], "filters": [], "handwrites": [], "stickers": [], "texts": [], "videos": [], }, "last_modified_platform": platform_info, "materials": _create_material_collections(), "mutable_config": None, "name": draft_name, "new_version": "169.0.0", "relationships": [], "render_index_track_mode_on": True, "retouch_cover": None, "source": "default", "static_cover_image_path": "", "time_marks": None, "tracks": [], "update_time": now_us, "version": 360000, } def _create_track(track_type: str, name: str) -> Dict[str, Any]: return { "attribute": 0, "flag": 0, "id": uuid.uuid4().hex, "is_default_name": True, "name": name, "segments": [], "type": track_type, } def _create_video_material( relative_path: str, duration_us: int, width: int, height: int, ) -> Dict[str, Any]: return { "id": uuid.uuid4().hex, "path": relative_path, "type": "video", "duration": duration_us, "width": width, "height": height, "material_name": os.path.basename(relative_path), "create_time": int(time.time() * MICROSECONDS), "crop": { "lower_left_x": 0.0, "lower_left_y": 1.0, "lower_right_x": 1.0, "lower_right_y": 1.0, "upper_left_x": 0.0, "upper_left_y": 0.0, "upper_right_x": 1.0, "upper_right_y": 0.0, }, "extra_type_option": 0, "source_platform": 0, } def _create_audio_material(relative_path: str, duration_us: int) -> Dict[str, Any]: material_id = uuid.uuid4().hex return { "app_id": 0, "category_id": "", "category_name": "local", "check_flag": 1, "copyright_limit_type": "none", "duration": duration_us, "effect_id": "", "formula_id": "", "id": material_id, "intensifies_path": "", "is_ai_clone_tone": False, "is_text_edit_overdub": False, "is_ugc": False, "local_material_id": material_id, "music_id": material_id, "name": os.path.basename(relative_path), "path": relative_path, "remote_url": "", "query": "", "request_id": "", "resource_id": "", "search_id": "", "source_from": "", "source_platform": 0, "team_id": "", "text_id": "", "tone_category_id": "", "tone_category_name": "", "tone_effect_id": "", "tone_effect_name": "", "tone_platform": "", "tone_second_category_id": "", "tone_second_category_name": "", "tone_speaker": "", "tone_type": "", "type": "extract_music", "video_id": "", "wave_points": [], } def _create_video_segment( material_id: str, source_start_us: int, duration_us: int, target_start_us: int, volume: float, ) -> Dict[str, Any]: return { "id": uuid.uuid4().hex, "material_id": material_id, "target_timerange": {"start": target_start_us, "duration": duration_us}, "source_timerange": {"start": source_start_us, "duration": duration_us}, "speed": 1.0, "volume": volume, "enable_adjust": True, "enable_color_curves": True, "enable_color_match_adjust": False, "enable_color_wheels": True, "enable_lut": True, "enable_smart_color_adjust": False, "extra_material_refs": [], "hdr_settings": {"intensity": 1.0, "mode": 1, "nits": 1000}, "uniform_scale": {"on": True, "value": 1.0}, "clip": { "alpha": 1.0, "flip": {"horizontal": False, "vertical": False}, "rotation": 0.0, "scale": {"x": 1.0, "y": 1.0}, "transform": {"x": 0.0, "y": 0.0}, }, "common_keyframes": [], } def _create_audio_segment( material_id: str, duration_us: int, target_start_us: int, volume: float, ) -> Dict[str, Any]: return { "id": uuid.uuid4().hex, "material_id": material_id, "target_timerange": {"start": target_start_us, "duration": duration_us}, "source_timerange": {"start": 0, "duration": duration_us}, "speed": 1.0, "volume": volume, "extra_material_refs": [], "clip": None, "hdr_settings": None, "uniform_scale": None, "common_keyframes": [], } def _normalize_video_material(material: Dict[str, Any]) -> Dict[str, Any]: fallback_path = f"assets/video/{material.get('material_name') or 'source.mp4'}" result = { "aigc_history_id": "", "aigc_item_id": "", "aigc_type": "none", "audio_fade": None, "beauty_body_auto_preset": None, "beauty_body_preset_id": "", "beauty_face_auto_preset": None, "beauty_face_auto_preset_infos": [], "beauty_face_preset_infos": [], "cartoon_path": "", "category_id": "", "category_name": "local", "check_flag": 65535, "content_feature_info": None, "corner_pin": None, "crop_ratio": "free", "crop_scale": 1.0, "formula_id": "", "freeze": None, "has_audio": True, "has_sound_separated": False, "intensifies_audio_path": "", "intensifies_path": "", "is_ai_generate_content": False, "is_copyright": False, "is_set_beauty_mode": False, "is_text_edit_overdub": False, "is_unified_beauty_mode": False, "live_photo_cover_path": "", "live_photo_timestamp": 0, "local_id": "", "local_material_from": 0, "local_material_id": "", "material_id": "", "material_url": "", "matting": None, "media_path": "", "multi_camera_info": None, "object_locked": None, "origin_material_id": "", "picture_from": "none", "picture_set_category_id": "", "picture_set_category_name": "", "request_id": "", "reverse_intensifies_path": "", "reverse_path": "", "smart_match_info": None, "smart_motion": None, "source": 0, "stable": None, "surface_trackings": None, "team_id": "", "unique_id": "", "video_algorithm": None, "video_mask_shadow": None, "video_mask_stroke": None, } result.update(material) result["path"] = _to_draft_material_path( _normalize_asset_path(material.get("path"), fallback_path) ) result["type"] = "video" return result def _normalize_audio_material(material: Dict[str, Any]) -> Dict[str, Any]: fallback_path = f"assets/audio/{material.get('name') or 'audio.mp3'}" result = { "ai_music_enter_from": "", "ai_music_generate_scene": "", "ai_music_type": 0, "aigc_history_id": "", "aigc_item_id": "", "app_id": 0, "category_id": "", "category_name": "local", "check_flag": 1, "cloned_model_type": "", "copyright_limit_type": "none", "effect_id": "", "formula_id": "", "intensifies_path": "", "is_ai_clone_tone": False, "is_ai_clone_tone_post": False, "is_text_edit_overdub": False, "is_ugc": False, "lyric_type": 0, "mock_tone_speaker": "", "moyin_emotion": "", "music_source": "", "pgc_id": "", "pgc_name": "", "query": "", "request_id": "", "resource_id": "", "search_id": "", "similiar_music_info": None, "sound_separate_type": 0, "source_from": "", "source_platform": 0, "team_id": "", "text_id": "", "third_resource_id": "", "tone_category_id": "", "tone_category_name": "", "tone_effect_id": "", "tone_effect_name": "", "tone_emotion_name_key": "", "tone_emotion_role": "", "tone_emotion_scale": 0, "tone_emotion_selection": "", "tone_emotion_style": "", "tone_platform": "", "tone_second_category_id": "", "tone_second_category_name": "", "tone_speaker": "", "tone_type": "", "tts_benefit_info": None, "tts_generate_scene": 0, "tts_task_id": "", "unique_id": "", "video_id": "", "wave_points": [], } result.update(material) result["path"] = _to_draft_material_path( _normalize_asset_path(material.get("path"), fallback_path) ) result["type"] = "extract_music" return result def _normalize_materials(draft: Dict[str, Any]) -> Dict[str, List[Any]]: source = draft.get("materials", {}) materials = { key: source.get(key, []) if isinstance(source.get(key, []), list) else [] for key in MATERIAL_COLLECTION_KEYS } materials["videos"] = [_normalize_video_material(item) for item in source.get("videos", [])] materials["audios"] = [_normalize_audio_material(item) for item in source.get("audios", [])] return materials def _create_responsive_layout() -> Dict[str, Any]: return { "enable": False, "horizontal_pos_layout": 0, "size_layout": 0, "target_follow": "", "vertical_pos_layout": 0, } def _normalize_segment( segment: Dict[str, Any], track_type: str, track_index: int, track_attribute: int, ) -> Dict[str, Any]: is_video = track_type == "video" result = { "caption_info": None, "cartoon": False, "color_correct_alg_result": "", "common_keyframes": [], "desc": "", "digital_human_template_group_id": "", "enable_adjust": is_video, "enable_adjust_mask": is_video, "enable_color_adjust_pro": False, "enable_color_correct_adjust": False, "enable_color_curves": True, "enable_color_match_adjust": False, "enable_color_wheels": True, "enable_hsl": is_video, "enable_hsl_curves": True, "enable_lut": is_video, "enable_mask_shadow": False, "enable_mask_stroke": False, "enable_smart_color_adjust": False, "enable_video_mask": True, "extra_material_refs": [], "group_id": "", "hdr_settings": segment.get("hdr_settings"), "intensifies_audio": False, "is_loop": False, "is_placeholder": False, "is_tone_modify": False, "keyframe_refs": [], "last_nonzero_volume": 1.0, "lyric_keyframes": None, "raw_segment_id": "", "render_index": 0, "render_timerange": {"duration": 0, "start": 0}, "responsive_layout": _create_responsive_layout(), "reverse": False, "source": "segmentsourcenormal", "state": 0, "template_id": "", "template_scene": "default", "uniform_scale": {"on": True, "value": 1.0}, "visible": True, } result.update(segment) result["track_attribute"] = track_attribute result["track_render_index"] = track_index return result def _normalize_tracks(draft: Dict[str, Any]) -> List[Dict[str, Any]]: tracks = [] for index, track in enumerate(draft.get("tracks", [])): track_copy = dict(track) track_attribute = int(track_copy.get("attribute", 0) or 0) track_copy["segments"] = [ _normalize_segment(segment, track_copy.get("type", ""), index, track_attribute) for segment in track.get("segments", []) ] tracks.append(track_copy) return tracks def _create_draft_info( draft: Dict[str, Any], draft_name: str, draft_root_path: str, new_version: str = "169.0.0", ) -> Dict[str, Any]: info = json.loads(json.dumps(draft, ensure_ascii=False)) canvas_config = info.get("canvas_config", {}) platform_info = _create_platform_info(draft_root_path) info.update({ "canvas_config": { "background": canvas_config.get("background"), "height": canvas_config.get("height", 1080), "ratio": canvas_config.get("ratio", "original"), "width": canvas_config.get("width", 1920), }, "draft_type": "video", "function_assistant_info": _default_function_assistant_info(), "is_drop_frame_timecode": False, "last_modified_platform": platform_info, "lyrics_effects": [], "materials": _normalize_materials(info), "name": draft_name, "new_version": new_version, "path": "", "platform": platform_info, "render_index_track_mode_on": True, "smart_ads_info": {"draft_url": "", "page_from": "", "routine": ""}, "tracks": _normalize_tracks(info), "uneven_animation_template_info": { "composition": "", "content": "", "order": "", "sub_template_info_list": [], }, }) return info def _create_empty_template(draft: Dict[str, Any], draft_root_path: str) -> Dict[str, Any]: empty_draft = json.loads(json.dumps(draft, ensure_ascii=False)) empty_draft["canvas_config"] = { "background": None, "height": 0, "ratio": "original", "width": 0, } empty_draft["color_space"] = -1 empty_draft["duration"] = 0 empty_draft["keyframes"] = { "adjusts": [], "audios": [], "effects": [], "filters": [], "handwrites": [], "stickers": [], "texts": [], "videos": [], } empty_draft["materials"] = _create_material_collections() empty_draft["tracks"] = [] return _create_draft_info(empty_draft, "", draft_root_path, "75.0.0") def _create_draft_material_index_item( material: Dict[str, Any], file_name: str, metetype: str, width: int, height: int, ) -> Dict[str, Any]: duration = int(material.get("duration", 0) or 0) return { "ai_group_type": "", "create_time": -1, "duration": duration, "enter_from": 0, "extra_info": file_name, "file_Path": material.get("path", ""), "height": height, "id": material.get("id", ""), "import_time": -1, "import_time_ms": -1, "item_source": 1, "md5": "", "metetype": metetype, "roughcut_time_range": {"duration": duration, "start": 0}, "sub_time_range": {"duration": -1, "start": -1}, "type": 0, "width": width, } def _create_draft_material_index(draft: Dict[str, Any]) -> List[Dict[str, Any]]: items: List[Dict[str, Any]] = [] for video in draft.get("materials", {}).get("videos", []): relative_path = _normalize_asset_path( video.get("path"), f"assets/video/{video.get('material_name') or 'source.mp4'}", ) items.append(_create_draft_material_index_item( {**video, "path": _to_draft_material_path(relative_path)}, os.path.basename(relative_path), "video", int(video.get("width", 0) or 0), int(video.get("height", 0) or 0), )) for audio in draft.get("materials", {}).get("audios", []): relative_path = _normalize_asset_path( audio.get("path"), f"assets/audio/{audio.get('name') or 'audio.mp3'}", ) items.append(_create_draft_material_index_item( {**audio, "path": _to_draft_material_path(relative_path)}, os.path.basename(relative_path), "music", 0, 0, )) return items def _create_meta_info( draft: Dict[str, Any], draft_name: str, draft_id: str, draft_root_path: str, draft_path: str, asset_size: int, ) -> Dict[str, Any]: return { "cloud_draft_cover": False, "cloud_draft_sync": False, "cloud_package_completed_time": "", "draft_cloud_capcut_purchase_info": "", "draft_cloud_last_action_download": False, "draft_cloud_package_type": "", "draft_cloud_purchase_info": "", "draft_cloud_template_id": "", "draft_cloud_tutorial_info": "", "draft_cloud_videocut_purchase_info": "", "draft_cover": "draft_cover.jpg", "draft_deeplink_url": "", "draft_enterprise_info": { "draft_enterprise_extra": "", "draft_enterprise_id": "", "draft_enterprise_name": "", "enterprise_material": [], }, "draft_fold_path": draft_path, "draft_id": _format_draft_uuid(draft_id), "draft_is_ae_produce": False, "draft_is_ai_packaging_used": False, "draft_is_ai_shorts": False, "draft_is_ai_translate": False, "draft_is_article_video_draft": False, "draft_is_cloud_temp_draft": False, "draft_is_from_deeplink": "false", "draft_is_invisible": False, "draft_is_pippit_draft": False, "draft_is_web_article_video": False, "draft_materials": [ {"type": 0, "value": _create_draft_material_index(draft)}, {"type": 1, "value": []}, {"type": 2, "value": []}, {"type": 3, "value": []}, {"type": 6, "value": []}, {"type": 7, "value": []}, {"type": 8, "value": []}, ], "draft_materials_copied_info": [], "draft_name": draft_name, "draft_need_rename_folder": False, "draft_new_version": "", "draft_removable_storage_device": "", "draft_root_path": draft_root_path, "draft_segment_extra_info": [], "draft_timeline_materials_size_": asset_size, "draft_type": "", "draft_web_article_video_enter_from": "", "tm_draft_cloud_completed": "", "tm_draft_cloud_entry_id": -1, "tm_draft_cloud_modified": 0, "tm_draft_cloud_parent_entry_id": -1, "tm_draft_cloud_space_id": -1, "tm_draft_cloud_user_id": -1, "tm_draft_create": draft.get("create_time", 0), "tm_draft_modified": draft.get("update_time", 0), "tm_draft_removed": 0, "tm_duration": draft.get("duration", 0), } def _create_root_meta_entry( draft: Dict[str, Any], draft_name: str, draft_id: str, draft_root_path: str, draft_path: str, asset_size: int, ) -> Dict[str, Any]: return { "cloud_draft_cover": False, "cloud_draft_sync": False, "draft_cloud_last_action_download": False, "draft_cloud_purchase_info": "", "draft_cloud_template_id": "", "draft_cloud_tutorial_info": "", "draft_cloud_videocut_purchase_info": "", "draft_cover": os.path.join(draft_path, "draft_cover.jpg"), "draft_fold_path": draft_path, "draft_id": _format_draft_uuid(draft_id), "draft_is_ai_shorts": False, "draft_is_cloud_temp_draft": False, "draft_is_invisible": False, "draft_is_web_article_video": False, "draft_json_file": os.path.join(draft_path, "draft_info.json"), "draft_name": draft_name, "draft_new_version": "", "draft_root_path": draft_root_path, "draft_timeline_materials_size": asset_size, "draft_type": "", "draft_web_article_video_enter_from": "", "streaming_edit_draft_ready": True, "tm_draft_cloud_completed": "", "tm_draft_cloud_entry_id": -1, "tm_draft_cloud_modified": 0, "tm_draft_cloud_parent_entry_id": -1, "tm_draft_cloud_space_id": -1, "tm_draft_cloud_user_id": -1, "tm_draft_create": draft.get("create_time", 0), "tm_draft_modified": draft.get("update_time", 0), "tm_draft_removed": 0, "tm_duration": draft.get("duration", 0), } def _merge_root_meta_info( existing_value: Any, next_entry: Dict[str, Any], root_path: str, ) -> Dict[str, Any]: existing = existing_value if isinstance(existing_value, dict) else {} existing_store = existing.get("all_draft_store") if not isinstance(existing_store, list): existing_store = [] all_draft_store = [ next_entry, *[ entry for entry in existing_store if ( isinstance(entry, dict) and entry.get("draft_id") != next_entry.get("draft_id") and entry.get("draft_fold_path") != next_entry.get("draft_fold_path") and entry.get("draft_name") != next_entry.get("draft_name") ) ], ] return { "all_draft_store": all_draft_store, "draft_ids": existing.get("draft_ids") if isinstance(existing.get("draft_ids"), int) else 1, "root_path": existing.get("root_path") or root_path, } def _create_draft_settings(draft: Dict[str, Any], draft_root_path: str) -> str: created_at = round(int(draft.get("create_time", 0) or 0) / MICROSECONDS) updated_at = round(int(draft.get("update_time", 0) or 0) / MICROSECONDS) return "\n".join([ "[General]", f"cloud_last_modify_platform={_detect_platform(draft_root_path)}", f"draft_create_time={created_at}", f"draft_last_edit_time={updated_at}", "real_edit_keys=1", "real_edit_seconds=0", "", ]) def _create_reference_line_attachment() -> Dict[str, Any]: return { "reference_lines_config": { "horizontal_lines": [], "is_lock": False, "is_visible": False, "vertical_lines": [], }, "safe_area_type": 0, } def _create_editing_attachment() -> Dict[str, Any]: return { "editing_draft": { "ai_remove_filter_words": { "enter_source": "", "right_id": "", }, "ai_shorts_info": { "report_params": "", "type": 0, }, "cover_extra_info": { "draft_id": "", "position": 0, "select_segment_id": "", "select_segment_source_start": 0, "select_segment_target_start": 0, "type": 1, }, "crop_info_extra": { "crop_mirror_type": 0, "crop_rotate": 0, "crop_rotate_total": 0, }, "digital_human_template_to_video_info": { "has_upload_material": False, "template_type": 0, }, "draft_used_recommend_function": "", "edit_type": 0, "eye_correct_enabled_multi_face_time": 0, "has_adjusted_render_layer": False, "image_ai_chat_info": { "before_chat_edit": False, "draft_modify_time": 0, "generate_type": "", "keyword_content": "", "keyword_type": "", "message_id": "", "model_name": "", "need_restore": False, "picture_id": "", "prompt_content": "", "prompt_from": "", "sugs_info": [], }, "is_open_expand_player": False, "is_template_text_ai_generate": False, "is_use_adjust": False, "is_use_ai_expand": False, "is_use_ai_remove": False, "is_use_ai_video": False, "is_use_audio_separation": False, "is_use_chroma_key": False, "is_use_curve_speed": False, "is_use_digital_human": False, "is_use_edit_multi_camera": False, "is_use_lip_sync": False, "is_use_lock_object": False, "is_use_loudness_unify": False, "is_use_noise_reduction": False, "is_use_one_click_beauty": False, "is_use_one_click_ultra_hd": False, "is_use_retouch_face": False, "is_use_smart_adjust_color": False, "is_use_smart_body_beautify": False, "is_use_smart_motion": False, "is_use_subtitle_recognition": False, "is_use_text_to_audio": False, "material_edit_session": { "material_edit_info": [], "session_id": "", "session_time": 0, }, "paste_segment_list": [], "profile_entrance_type": "", "publish_enter_from": "", "publish_type": "", "single_function_type": 0, "text_convert_case_types": [], "version": "1.0.0", "video_recording_create_draft": "", } } def _create_draft_virtual_store(draft: Dict[str, Any]) -> Dict[str, Any]: materials = [ *draft.get("materials", {}).get("videos", []), *draft.get("materials", {}).get("audios", []), ] return { "draft_materials": [], "draft_virtual_store": [ { "type": 0, "value": [ { "creation_time": 0, "display_name": "", "filter_type": 0, "id": "", "import_time": 0, "import_time_us": 0, "sort_sub_type": 0, "sort_type": 0, "subdraft_filter_type": 0, } ], }, { "type": 1, "value": [ {"child_id": material.get("id", ""), "parent_id": ""} for material in materials ], }, {"type": 2, "value": []}, ], } def _write_root_meta_info(draft_root_path: str, root_meta_entry: Dict[str, Any]) -> None: root_meta_path = os.path.join(draft_root_path, "root_meta_info.json") existing_value: Any = {} if os.path.exists(root_meta_path): try: with open(root_meta_path, "r", encoding="utf-8") as f: existing_value = json.load(f) except Exception as e: logger.warning(f"读取 root_meta_info.json 失败，将重建索引: {e}") _write_json_file( root_meta_path, _merge_root_meta_info(existing_value, root_meta_entry, draft_root_path), ) def _write_plaintext_draft_files( draft_root_path: str, draft_path: str, draft_name: str, draft_id: str, draft: Dict[str, Any], asset_size: int, ) -> None: draft_info = _create_draft_info(draft, draft_name, draft_root_path) empty_template = _create_empty_template(draft, draft_root_path) _write_json_file( os.path.join(draft_path, "draft_meta_info.json"), _create_meta_info(draft, draft_name, draft_id, draft_root_path, draft_path, asset_size), ) with open(os.path.join(draft_path, "draft_settings"), "w", encoding="utf-8") as f: f.write(_create_draft_settings(draft, draft_root_path)) with open(os.path.join(draft_path, "draft_cover.jpg"), "wb") as f: f.write(DEFAULT_DRAFT_COVER_BYTES) _write_json_file(os.path.join(draft_path, "draft_info.json"), draft_info) _write_json_file(os.path.join(draft_path, "template-2.tmp"), draft_info) _write_json_file(os.path.join(draft_path, "template.tmp"), empty_template) _write_json_file( os.path.join(draft_path, "common_attachment", "attachment_pc_timeline.json"), _create_reference_line_attachment(), ) _write_json_file(os.path.join(draft_path, "common_attachment", "attachment_action_scene.json"), {}) _write_json_file(os.path.join(draft_path, "common_attachment", "attachment_script_video.json"), {}) _write_json_file(os.path.join(draft_path, "common_attachment", "attachment_gen_ai_info.json"), {}) _write_json_file(os.path.join(draft_path, "attachment_editing.json"), _create_editing_attachment()) _write_json_file( os.path.join(draft_path, "draft_agency_config.json"), { "is_auto_agency_enabled": False, "is_auto_agency_popup": False, "is_single_agency_mode": False, "marterials": None, "use_converter": False, "video_resolution": draft.get("canvas_config", {}).get("height", 1080), }, ) with open(os.path.join(draft_path, "draft_biz_config.json"), "w", encoding="utf-8"): pass _write_json_file( os.path.join(draft_path, "draft_virtual_store.json"), _create_draft_virtual_store(draft), ) _write_json_file( os.path.join(draft_path, "performance_opt_info.json"), {"manual_cancle_precombine_segs": None, "need_auto_precombine_segs": None}, ) _write_json_file( os.path.join(draft_path, "timeline_layout.json"), { "activeTimeline": draft_id, "dockItems": [ { "dockIndex": 0, "ratio": 1, "timelineIds": [draft_id], "timelineNames": ["时间线01"], } ], "layoutOrientation": 1, }, ) _write_root_meta_info( draft_root_path, _create_root_meta_entry(draft, draft_name, draft_id, draft_root_path, draft_path, asset_size), ) def _resolve_item_audio_file(item: Dict[str, Any], output_dir: str) -> str: audio_file = "" timestamp = item.get("timestamp", "") if timestamp: audio_file = os.path.join(output_dir, f"audio_{timestamp.replace(':', '_')}.mp3") item_audio_file = item.get("audio", "") if item_audio_file and os.path.exists(item_audio_file): audio_file = item_audio_file return audio_file def write_plaintext_jianying_draft( jianying_draft_path: str, draft_name: str, new_script_list: List[Dict[str, Any]], params: VideoClipParams, output_dir: str, ) -> Tuple[str, str]: os.makedirs(jianying_draft_path, exist_ok=True) display_name = draft_name or f"NarratoAI_{int(time.time())}" folder_name, draft_path = _create_unique_draft_path(jianying_draft_path, display_name) os.makedirs(draft_path, exist_ok=False) for rel_dir in DRAFT_PACKAGE_DIRECTORIES: os.makedirs(os.path.join(draft_path, *rel_dir.split("/")), exist_ok=True) draft_id = uuid.uuid4().hex draft = _create_draft_template(draft_id, display_name, jianying_draft_path) video_track = _create_track("video", "视频轨道") audio_track = _create_track("audio", "音频轨道") draft["tracks"] = [video_track, audio_track] duration_cache: Dict[str, float] = {} metadata_cache: Dict[str, Tuple[int, int, int]] = {} used_asset_paths: Set[str] = set() asset_path_cache: Dict[str, str] = {} current_time_us = 0 for item in new_script_list: start_time = float(item.get("start_time", 0.0) or 0.0) requested_duration = float(item.get("duration", 0.0) or 0.0) timestamp = item.get("timestamp", "") ost = int(item.get("OST", 0) or 0) logger.info( f"处理片段: OST={ost}, start_time={start_time}, " f"duration={requested_duration}, timestamp={timestamp}" ) video_file = item.get("video", "") use_clipped_video = bool(video_file and os.path.exists(video_file)) if not use_clipped_video: video_file = params.video_origin_path if not video_file or not os.path.exists(video_file): logger.warning(f"视频素材不存在，跳过片段: {video_file or timestamp}") continue source_start_time = 0.0 if use_clipped_video else start_time video_duration = _clamp_duration_to_media( requested_duration, video_file, duration_cache, "视频素材" if use_clipped_video else "原始视频素材", source_start_time=source_start_time, ) audio_file = _resolve_item_audio_file(item, output_dir) audio_duration = None if ost in [0, 2] and audio_file and os.path.exists(audio_file): audio_duration = _get_cached_media_duration(audio_file, duration_cache) logger.info( f"音频文件实际时长: {audio_duration:.6f}秒, 视频片段时长: {video_duration:.3f}秒" ) segment_duration = min( video_duration, audio_duration if audio_duration is not None else video_duration, ) segment_duration = _floor_duration_to_milliseconds(segment_duration) if segment_duration <= 0: logger.warning(f"片段时长无效，跳过: {timestamp}") continue segment_duration_us = _seconds_to_microseconds(segment_duration) video_material_duration_us, width, height = _get_video_metadata_ffprobe(video_file, metadata_cache) video_relative_path = _register_asset( video_file, draft_path, "assets/video", f"video_{len(video_track['segments']) + 1}.mp4", used_asset_paths, asset_path_cache, ) video_material = _create_video_material(video_relative_path, video_material_duration_us, width, height) draft["materials"]["videos"].append(video_material) video_track["segments"].append(_create_video_segment( video_material["id"], _seconds_to_microseconds(_floor_duration_to_milliseconds(source_start_time)), segment_duration_us, current_time_us, float(getattr(params, "original_volume", 1.0) or 1.0), )) if ost in [0, 2] and audio_file and os.path.exists(audio_file): audio_material_duration_us = _seconds_to_microseconds( _get_cached_media_duration(audio_file, duration_cache) ) audio_relative_path = _register_asset( audio_file, draft_path, "assets/audio", f"audio_{len(audio_track['segments']) + 1}.mp3", used_asset_paths, asset_path_cache, ) audio_material = _create_audio_material(audio_relative_path, audio_material_duration_us) draft["materials"]["audios"].append(audio_material) audio_track["segments"].append(_create_audio_segment( audio_material["id"], segment_duration_us, current_time_us, float(getattr(params, "tts_volume", 1.0) or 1.0), )) elif ost in [0, 2]: logger.warning(f"音频文件不存在: {audio_file}") current_time_us += segment_duration_us if not video_track["segments"]: raise ValueError("没有可写入剪映草稿的视频片段") first_video = draft["materials"]["videos"][0] draft["canvas_config"]["width"] = int(first_video.get("width", 1920) or 1920) draft["canvas_config"]["height"] = int(first_video.get("height", 1080) or 1080) draft["duration"] = current_time_us draft["update_time"] = int(time.time() * MICROSECONDS) asset_size = sum( os.path.getsize(source_file) for source_file in asset_path_cache.keys() if os.path.exists(source_file) ) _write_plaintext_draft_files( jianying_draft_path, draft_path, display_name, draft_id, draft, asset_size, ) logger.info(f"剪映明文草稿包已写入: {draft_path} (folder={folder_name})") return draft_path, display_name