# -*- coding: utf-8 -*- """ 剪映草稿管理器模块 """ # 列举导入模块 import hashlib import json from pathlib import Path from random import choice 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()) WORKFLOWS = { "存量_纯图_竖版": [[("底纹", { "clip_settings": [ { "scale_x": 1.0, "scale_y": 1.0, }, ], # 剪辑设置 },)], [("底纹", { "clip_settings": [ { "scale_x": 1.0, "scale_y": 1.0, }, ], # 剪辑设置 },)], [("箭头", { "clip_settings": [ { "scale_x": 0.35, "scale_y": 0.35, "transform_x": 0, "transform_y": -0.92, }, ], # 剪辑设置 },)], [("提示语", { "volume": [1.0], # 播放音量 "clip_settings": [ { "scale_x": 1.0, "scale_y": 1.0, }, ], # 剪辑设置 },)],] } class DraftsManager: """ 剪映草稿管理器 """ def __init__( self, project_name: str, drafts_folder_path: str = r"E:\JianYingPro Drafts", ): """ 初始化剪映草稿管理器 :param project_name: 项目名称 :param drafts_folder_path: 剪映草稿文件夹路径，默认为 E:\\JianYingPro Drafts """ # 项目名称 self.project_name = project_name # 构建素材文件夹路径 self.materials_folder_path = Path(rf"E:\jianying\materials\{project_name}") # 检查素材文件夹是否存在，若不存在则抛出异常 if not self.materials_folder_path.exists(): raise RuntimeError("素材文件夹不存在") # 初始化素材 self.materials = self.init_materials() # 构建成品文件夹路径 self.products_folder_path = Path(rf"E:\jianying\products\{project_name}") # 若成品文件夹已存在则先删除在创建 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.drafts_folder = DraftFolder(folder_path=drafts_folder_path) # 实例化缓存 self.caches = Caches() def init_materials(self) -> Dict[str, List[Any]]: """ 初始化素材 :return: 所有素材 """ materials = {} # 获取工作流 workflow = WORKFLOWS.get(self.project_name) if not workflow: raise RuntimeError(f"未就 {self.project_name} 配置工作流") # 根据项目工作流初始化素材 for track in workflow: for segment in track: material_folder_name, _ = segment # 构建素材文件夹路径 material_folder_path = self.materials_folder_path / material_folder_name # 若素材文件夹不存在则抛出异常 if not material_folder_path.exists(): raise RuntimeError(f"素材文件夹 {material_folder_name} 不存在") # 遍历素材文件夹内所有素材路径 materials[material_folder_name] = [ material_path for material_path in material_folder_path.rglob( "*" ) ] return materials def batch_create( self, draft_counts: int, video_width: int = 1080, video_height: int = 1920, video_fps: int = 30, capture_duration: Optional[int] = None, ) -> None: """ 批量创建草稿 :param draft_counts: 草稿数 :param video_width: 视频宽度（单位为像素），默认为 1080 :param video_height: 视频高度（单位为像素），默认为 1920 :param video_fps: 视频帧率（单位为帧/秒），默认为 30 :param capture_duration: 截取背景视频时长（单位为秒），默认为 None :return: 无 """ # 获取工作流 workflow = WORKFLOWS[self.project_name] index = 1 # 草稿索引 while True: scheme = [] # 草稿生成方案 for track in workflow: for segment in track: material_folder_name, configuration = segment # 随机选择素材路径 material_path = choice(self.materials[material_folder_name]) # 素材后缀名 suffix = material_path.suffix.lower() # 根据素材类型匹配节点名称 match suffix: case ".mp4" | ".mov" | ".png": node_name = "add_video" case _: raise RuntimeError(f"未就 {suffix} 配置匹配节点") # 根据节点名称获取节点配置 scheme.append( { node_name: { } } ) # 若非生成字幕则在工作流配置添加轨道名称 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_subtitle_video"]["material_path"].as_posix() ).duration if "add_subtitle_video" in configurations else VideoMaterial( path=configurations["add_background_video"]["material_path"].as_posix() ).duration ) # 若截取背景视频时长非空则将背景视频时长设置为截取背景视频时长 if capture_duration: background_video_duration = capture_duration * 1_000_000 # 设置背景视频在轨道上的范围为 (0, 背景视频时长) configurations["add_background_video"]["target_timerange"] = ( 0, background_video_duration, ) if "add_subtitle_video" in configurations: # 若包含添加字幕视频则在添加背景视频时其播放音量设置为 0 configurations["add_background_video"]["volume"] = 0.0 # 设置字幕视频在轨道上的范围为 (0, 背景视频时长) configurations["add_subtitle_video"]["target_timerange"] = ( 0, background_video_duration, ) # 中贴视频特殊处理：背景视频、背景音频和字幕视频在轨道上的范围为 (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 if "add_mid_roll_video" in configurations else 0 ) ), VideoMaterial( path=configurations["add_post_roll_video"][ "material_path" ].as_posix() ).duration, ), } } ) # 添加保存节点 configurations.update( { "save": {}, } ) return configurations # 定义视频时时长 duration = VideoMaterial( path=workflow["add_background_video"][ "material_path" ].as_posix() ).duration # 转换为微秒单位 if capture_duration: video_duration = capture_duration * 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}({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_image": print("-> 正在添加声明图片...", end="") draft.add_video_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() index += 1 if index > draft_counts: break 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哈希值的大写十六进制作为草稿名称 )