# -*- coding: utf-8 -*- """ 剪映草稿管理器模块 """ # 列举导入模块 import hashlib import json from pathlib import Path import random import re import shutil import sys from typing import Any, Dict, List, Optional from pyJianYingDraft import DraftFolder, VideoMaterial from caches import Caches from drafts import Drafts sys.path.append(Path(__file__).parent.parent.as_posix()) class JianYingManager: """ 剪映草稿管理器 """ def __init__( self, materials_folder_path: str, drafts_folder_path: str = r"E:\JianYingPro Drafts", ): """ 初始化 :param materials_folder_path: 素材文件夹路径。其中，文件夹名称默认为工作流名称 :param drafts_folder_path: 剪映草稿文件夹路径，默认为 E:\\JianYingPro Drafts """ try: # 初始化素材文件夹路径 self.materials_folder_path = Path(materials_folder_path) if not self.materials_folder_path.exists(): raise RuntimeError("素材文件夹路径不存在") # 初始化所有素材 self.materials = self._init_materials() # 初始化成品文件夹路径 self.products_folder_path = Path( materials_folder_path.replace("materials", "products") ) # 若成品文件夹存路径已存在则先删除 if self.products_folder_path.exists(): shutil.rmtree(self.products_folder_path) self.products_folder_path.mkdir(parents=True) # 初始化剪映草稿文件夹路径 self.drafts_folder_path = Path(drafts_folder_path) if not self.drafts_folder_path.exists(): raise RuntimeError("剪映草稿文件夹路径不存在") # 初始化节点配置 self.configurations = self._init_configurations() # 初始化剪映草稿文件夹管理器 self.drafts_folder = DraftFolder(folder_path=drafts_folder_path) # 实例化缓存 self.caches = Caches() except Exception as exception: raise RuntimeError(f"发生异常：{str(exception)}") from exception def _init_materials(self) -> Dict[str, List[Any]]: """ 初始化所有素材 :return: 所有素材 """ materials = {} # 构建字幕文本路径 subtitle_text_path = self.materials_folder_path / "字幕文本.txt" if subtitle_text_path.exists() and subtitle_text_path.is_file(): with open(subtitle_text_path, "r", encoding="utf-8") as file: # 字幕文本列表 subtitle_texts = file.readlines() if not subtitle_texts: raise RuntimeError("字幕文本为空") materials["subtitle_texts"] = subtitle_texts else: materials["subtitle_texts"] = [] # 构建字幕视频文件夹路径 subtitle_video_folder_path = self.materials_folder_path / "字幕视频" if subtitle_video_folder_path.exists() and subtitle_video_folder_path.is_dir(): materials["subtitle_video_path"] = [ subtitle_video_path for subtitle_video_path in subtitle_video_folder_path.rglob("*.mov") ] else: materials["subtitle_video_path"] = [] # 构建背景视频文件夹路径 background_video_folder_path = self.materials_folder_path / "背景视频" if ( background_video_folder_path.exists() and background_video_folder_path.is_dir() ): materials["background_video_path"] = [ background_video_path for background_video_path in background_video_folder_path.rglob("*.mp4") ] else: materials["background_video_path"] = [] # 构建中贴视频文件夹路径 mid_roll_video_folder_path = self.materials_folder_path / "中贴视频" if mid_roll_video_folder_path.exists() and mid_roll_video_folder_path.is_dir(): materials["mid_roll_video_path"] = [ mid_roll_video_path for mid_roll_video_path in mid_roll_video_folder_path.rglob("*.mp4") ] else: materials["mid_roll_video_path"] = [] # 构建后贴视频文件夹路径 post_roll_video_folder_path = self.materials_folder_path / "后贴视频" if ( post_roll_video_folder_path.exists() and post_roll_video_folder_path.is_dir() ): materials["post_roll_video_path"] = [ post_video_path for post_video_path in post_roll_video_folder_path.rglob("*.mp4") ] else: materials["post_roll_video_path"] = [] # 构建背景音频文件夹路径 background_audio_folder_path = self.materials_folder_path / "背景音频" if ( background_audio_folder_path.exists() and background_audio_folder_path.is_dir() ): materials["background_audio_path"] = [ background_audio_path for background_audio_path in background_audio_folder_path.rglob("*.mp3") ] else: materials["background_audio_path"] = [] # 构建标识图片路径 logo_image_path = self.materials_folder_path / "标识图片.png" if logo_image_path.exists() and logo_image_path.is_file(): materials["logo_image_path"] = [logo_image_path] # 有且只有一张标识 else: materials["logo_image_path"] = [] # 构建标识视频文件夹路径 logo_video_folder_path = self.materials_folder_path / "标识视频" if logo_video_folder_path.exists() and logo_video_folder_path.is_dir(): materials["logo_video_path"] = [ file_path for file_path in logo_video_folder_path.rglob("*.mov") ] else: materials["logo_video_path"] = [] # 构建声明文本路径 statement_text_path = self.materials_folder_path / "声明文本.txt" if statement_text_path.exists() and statement_text_path.is_file(): with open(statement_text_path, "r", encoding="utf-8") as file: # 声明文本列表 statement_texts = file.readlines() if not statement_texts: raise RuntimeError("声明文本为空") materials["statement_texts"] = statement_texts else: materials["statement_texts"] = [] # 构建声明视频文件夹路径 statement_video_folder_path = self.materials_folder_path / "声明视频" if ( statement_video_folder_path.exists() and statement_video_folder_path.is_dir() ): materials["statement_video_path"] = [ statement_video_path for statement_video_path in statement_video_folder_path.rglob("*.mov") ] else: materials["statement_video_path"] = [] return materials def _init_configurations( self, ) -> Dict[str, Any]: """ 初始化节点配置 :return: 节点配置 """ # 已配置工作流 workflows = { "默认工作流": [ "generate_subtitle", "add_background_video", "add_statement", "add_sticker", "add_sticker_arrow", ], # 默认工作流，先根据字幕文本合成字幕音频并生成字幕，再叠加背景视频、声明视频、非箭头贴纸视频和箭头贴纸视频 "淘宝闪购": [ "add_background_video", "add_background_audio", "add_subtitle_video", "add_statement_video", ], # 淘宝闪购，先根据字幕视频获取其持续时长并作为成品持续时长，再叠加背景视频、背景音频和生成视频 "淘宝闪购_达人": [ "add_background_video", "add_background_audio", "add_mid_roll_video", "add_post_roll_video", "add_logo_video", "add_statement_video", ], # 淘宝闪购_达人，先就背景视频、背景音频、字幕视频截取前 5 秒，再添加中贴视频、后贴视频、标识视频、声明视频和生成视频 } # 默认以素材文件夹名称为工作流名称 workflow_name = self.materials_folder_path.stem # 工作流配置 workflow = workflows.get(workflow_name, None) if not workflow: raise RuntimeError(f"该工作流 {workflow_name} 未配置") # 节点配置模板 configurations = { "generate_subtitle": { "texts": self.materials["subtitle_texts"], "style": [ {"size": 10.0}, ], # 字体样式 "effect": [ {"effect_id": "7127561998556089631"}, {"effect_id": "7166467215410187552"}, {"effect_id": "6896138122774498567"}, {"effect_id": "7166469374507765031"}, {"effect_id": "6896137924853763336"}, ], # 花字设置 }, # 生成字幕工作配置 "add_subtitle_video": { "material_path": self.materials["subtitle_video_path"], "volume": [1.0], # 播放音量 "clip_settings": [ None, ], # 图像调节设置 }, # 添加字幕视频工作配置 "add_background_video": { "material_path": self.materials["background_video_path"], "volume": [1.0], # 播放音量 "clip_settings": [ { "scale_x": 1.0, "scale_y": 1.0, }, ], # 图像调节设置 }, # 添加背景视频工作配置 "add_mid_roll_video": { "material_path": self.materials["mid_roll_video_path"], "volume": [1.0], # 播放音量 "clip_settings": [ { "scale_x": 1.0, "scale_y": 1.0, }, ], # 图像调节设置 }, # 添加中贴视频工作配置 "add_post_roll_video": { "material_path": self.materials["post_roll_video_path"], "volume": [1.0], # 播放音量 "clip_settings": [ { "scale_x": 1.0, "scale_y": 1.0, }, ], # 图像调节设置 }, # 添加后贴视频工作配置 "add_background_audio": { "material_path": self.materials["background_audio_path"], "volume": [0.6], # 播放音量 }, # 添加背景音频工作配置 "add_logo_image": { "material_path": self.materials["logo_image_path"], "clip_settings": [ { "scale_x": 0.2, "scale_y": 0.2, "transform_x": -0.78, "transform_y": 0.82, }, ], }, # 添加标识工作配置 "add_logo_video": { "material_path": self.materials["logo_video_path"], "volume": [1.0], # 播放音量 "clip_settings": [ None, ], # 图像调节设置 }, # 添加标识视频工作配置 "add_statement": { "text": self.materials["statement_texts"], "style": [ {"size": 6.0, "align": 1, "vertical": True}, ], # 文本样式 "border": [ {"width": 35.0}, {"width": 40.0}, {"width": 45.0}, ], # 描边宽度 "clip_settings": [ { "transform_x": -0.82, }, ], # 图像调节设置 }, # 添加声明工作配置 "add_statement_video": { "material_path": self.materials["statement_video_path"], "volume": [1.0], # 播放音量 "clip_settings": [ None, ], # 图像调节设置 }, # 添加声明视频工作配置 "add_sticker": { "resource_id": [ "7110124379568098568", "7019687632804334861", "6895933678262750478", "7010558788675652900", "7026858083393588487", ], "clip_settings": [ { "scale_x": 0.75, "scale_y": 0.75, "transform_x": -0.75, "transform_y": 0.75, }, { "scale_x": 0.75, "scale_y": 0.75, "transform_y": 0.75, }, { "scale_x": 0.75, "scale_y": 0.75, "transform_x": 0.75, "transform_y": 0.75, }, ], # 图像调节设置 }, # 添加非箭头贴纸工作配置 "add_sticker_arrow": { "resource_id": [ "7143078914989018379", "7142870400358255905", "7185568038027103544", "7024342011440319781", "7205042602184363322", ], "clip_settings": [ { "scale_x": 0.75, "scale_y": 0.75, "transform_x": -0.8, "transform_y": -0.62, }, ], # 图像调节设置 }, # 添加箭头贴纸工作配置 } return {node_name: configurations[node_name] for node_name in workflow} def batch_create( self, draft_counts: int, video_width: int = 1080, video_height: int = 1920, video_fps: int = 30, duration_capture: Optional[int] = None, ) -> None: """ 批量创建草稿 :param draft_counts: 草稿数 :param video_width: 视频宽度（单位为像素），默认为 1080 :param video_height: 视频高度（单位为像素），默认为 1920 :param video_fps: 视频帧率（单位为帧/秒），默认为 30 :param duration_capture: 截取背景视频时长（单位为秒），默认为 None :return: 无 """ draft_index = 1 # 草稿索引 while True: # 获取节点配置 configurations = self._get_configurations(duration_capture=duration_capture) # 若包含添加字幕视频则使用字幕视频时长作为视频时长，否则使用背景视频时长作为视频时长 if "add_subtitle_video" in configurations: video_duration = VideoMaterial( path=configurations["add_subtitle_video"][ "material_path" ].as_posix() ).duration else: video_duration = video_duration = VideoMaterial( path=configurations["add_background_video"][ "material_path" ].as_posix() ).duration if duration_capture: video_duration = duration_capture * 1_000_000 # 转换为微秒单位 # 添加中贴视频持续时长 if "add_mid_roll_video" in configurations: video_duration += VideoMaterial( path=configurations["add_mid_roll_video"][ "material_path" ].as_posix() ).duration # 添加后贴视频持续时长 if "add_post_roll_video" in configurations: video_duration += VideoMaterial( path=configurations["add_post_roll_video"][ "material_path" ].as_posix() ).duration # 生成剪映草稿名称 draft_name = self._generate_draft_name( configurations=configurations, ) # 若草稿名称已缓存则跳过 if self.caches.query(draft_name=draft_name): continue print(f"正在创建草稿 {draft_name}({draft_index}/{draft_counts})...") # 创建剪映草稿 draft = Drafts( materials_folder_path=self.materials_folder_path, drafts_folder=self.drafts_folder, draft_name=draft_name, video_width=video_width, video_height=video_height, video_fps=video_fps, video_duration=video_duration, ) for node_name in configurations: match node_name: # 添加字幕 case "generate_subtitle": print("-> 正在根据字幕文本合成字幕音频并生成字幕...", end="") draft.generate_subtitle(**configurations[node_name]) print("已完成") # 添加背景视频 case "add_background_video": print("-> 正在添加背景视频...", end="") draft.add_video_segment(**configurations[node_name]) print("已完成") # 添加背景音频 case "add_background_audio": print("-> 正在添加背景音频...", end="") draft.add_audio_segment(**configurations[node_name]) print("已完成") # 添加中贴视频 case "add_mid_roll_video": print("-> 正在添加中贴视频...", end="") draft.add_video_segment(**configurations[node_name]) print("已完成") # 添加后贴视频 case "add_post_roll_video": print("-> 正在添加后贴视频...", end="") draft.add_video_segment(**configurations[node_name]) print("已完成") # 添加字幕视频 case "add_subtitle_video": print("-> 正在添加字幕视频...", end="") draft.add_video_segment(**configurations[node_name]) print("已完成") # 添加标识 case "add_logo_image": print("-> 正在添加标识图片...", end="") draft.add_video_segment(**configurations[node_name]) print("已完成") # 添加标识视频 case "add_logo_video": print("-> 正在添加标识视频...", end="") draft.add_video_segment(**configurations[node_name]) print("已完成") # 添加声明文本 case "add_statement": print("-> 正在添加声明文本...", end="") draft.add_text_segment(**configurations[node_name]) print("已完成") # 添加声明视频 case "add_statement_video": print("-> 正在添加声明视频...", end="") draft.add_video_segment(**configurations[node_name]) print("已完成") # 添加贴纸 case _ if node_name.startswith("add_sticker"): print("-> 正在添加贴纸...", end="") draft.add_sticker(**configurations[node_name]) print("已完成") # 保存草稿 case "save": print("-> 正在保存草稿...", end="") draft.save() print("已完成") # 缓存草稿名称和所有节点配置 self.caches.update( draft_name=draft_name, configurations=configurations, ) print("已完成") print() draft_index += 1 if draft_index > draft_counts: break def _get_configurations( self, duration_capture: Optional[int] = None, ) -> Dict[str, Any]: """ 获取节点配置 :param duration_capture: 截取背景视频时长（单位为秒），默认为 None :return: 节点配置 """ configurations = {} for node_name in self.configurations: # 根据节点名称获取节点配置 configurations.update( { node_name: { key: random.choice(value) for key, value in self.configurations[node_name].items() } } ) # 若非生成字幕则在工作流配置添加轨道名称 if node_name != "generate_subtitle": configurations[node_name]["track_name"] = ( matched.group("track_name") if ( matched := re.match( pattern=r"^.+?_(?P.+)$", string=node_name, ) ) else node_name ) background_video_duration = VideoMaterial( path=configurations["add_background_video"]["material_path"].as_posix() ).duration if duration_capture: background_video_duration = duration_capture * 1_000_000 # 若包含添加字幕音频则在添加背景视频时其播放音量设置为 0 if "add_subtitle_audio" in configurations: configurations["add_background_video"]["volume"] = 0.0 # 中贴视频特殊处理：背景视频、背景音频和字幕视频在轨道上的范围为 (0, 背景视频时长)，中贴视频在轨道上的范围为 (背景视频时长, 中贴视频时长) if "add_mid_roll_video" in configurations: configurations.update( { node: { **configurations[node], "target_timerange": ( 0, background_video_duration, ), } for node in [ "add_background_video", "add_background_audio", "add_subtitle_video", ] if node in configurations } ) configurations.update( { "add_mid_roll_video": { **configurations["add_mid_roll_video"], "target_timerange": ( background_video_duration, VideoMaterial( path=configurations["add_mid_roll_video"][ "material_path" ].as_posix() ).duration, ), } } ) # 后贴视频特殊处理：后贴视频在轨道上的范围为 (背景视频时长 + 中贴视频时长, 后贴视频时长) if "add_post_roll_video" in configurations: configurations.update( { "add_post_roll_video": { **configurations["add_post_roll_video"], "target_timerange": ( ( background_video_duration + VideoMaterial( path=configurations["add_mid_roll_video"][ "material_path" ].as_posix() ).duration ), VideoMaterial( path=configurations["add_post_roll_video"][ "material_path" ].as_posix() ).duration, ), } } ) # 添加保存节点 configurations.update( { "save": {}, } ) return configurations def _generate_draft_name( self, configurations: Dict[str, Any], ) -> str: """ 生成剪映草稿名称 :param configurations: 指定工作流所有节点配置 :return: 草稿名称 """ return ( hashlib.md5( json.dumps( obj=configurations, cls=self.caches.JSONEncoder, sort_keys=True, ensure_ascii=False, ).encode("utf-8") ) # 将工作流配置序列化 .hexdigest() .upper() # MD5哈希值的大写十六进制作为草稿名称 )