diff --git a/短视频合成自动化/caches.db b/短视频合成自动化/caches.db new file mode 100644 index 0000000..34265e1 Binary files /dev/null and b/短视频合成自动化/caches.db differ diff --git a/短视频合成自动化/export.py b/短视频合成自动化/export.py index bf32e5e..80973ff 100644 --- a/短视频合成自动化/export.py +++ b/短视频合成自动化/export.py @@ -8,17 +8,128 @@ import json from pathlib import Path import random import re +import shutil import subprocess import time -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional from uuid import uuid4 - +from pathlib import WindowsPath import pyJianYingDraft import win32con import win32gui - +import hashlib from draft import JianYingDraft +import sys + +sys.path.append(Path(__file__).parent.parent.as_posix()) +from utils.sqlite import SQLite + + +# 自定义JSON编码器 +class JSONEncoder(json.JSONEncoder): + def default(self, o): + # 若为WindowsPath对象则转为字符串路径 + if isinstance(o, WindowsPath): + return o.as_posix() + return super().default(o) + + +class Caches(SQLite): + """ + 缓存客户端，支持： + query：查询并返回单条缓存 + update：新增或更新单条缓存 + """ + + def __init__(self, cache_ttl: int = 30 * 86400): + """ + 初始化 + :param cache_ttl: 缓存生存时间，单位为秒，默认为30天 + """ + # 初始化 + super().__init__(database=Path(__file__).parent.resolve() / "caches.db") + self.cache_ttl = cache_ttl + + # 初始化缓存表（不清理过期缓存） + try: + with self: + self.execute( + sql=""" + CREATE TABLE IF NOT EXISTS caches + ( + --草稿名称 + draft_name TEXT PRIMARY KEY, + --工作流配置 + workflow_configurations TEXT NOT NULL, + --创建时间戳 + timestamp REAL NOT NULL + ) + """ + ) + self.execute( + sql=""" + CREATE INDEX IF NOT EXISTS idx_timestamp ON caches(timestamp) + """ + ) + except Exception as exception: + raise RuntimeError(f"初始化缓存表发生异常：{str(exception)}") from exception + + def query(self, draft_name: str) -> Optional[Dict[str, Any]]: + """ + 查询并返回单条缓存 + :param draft_name: 草稿名称 + :return: 缓存 + """ + try: + with self: + result = self.query_one( + sql=""" + SELECT workflow_configurations + FROM caches + WHERE draft_name = ? AND timestamp >= ? + """, + parameters=(draft_name, time.time() - self.cache_ttl), + ) + return ( + None + if result is None + else json.loads(result["workflow_configurations"]) + ) + except Exception as exception: + raise RuntimeError( + f"查询并获取单条缓存发生异常：{str(exception)}" + ) from exception + + def update( + self, draft_name: str, workflow_configurations: List[Dict[str, Any]] + ) -> Optional[bool]: + """ + 新增或更新单条缓存 + :param draft_name: 草稿名称 + :param workflow_configurations: 工作流配置 + :return: 成功返回True，失败返回False + """ + try: + with self: + return self.execute( + sql=""" + INSERT OR REPLACE INTO caches (draft_name, workflow_configurations, timestamp) VALUES (?, ?, ?) + """, + parameters=( + draft_name, + json.dumps( + obj=workflow_configurations, + cls=JSONEncoder, + sort_keys=True, + ensure_ascii=False, + ), + time.time(), + ), + ) + except Exception as exception: + raise RuntimeError("新增或更新缓存发生异常") from exception + class JianYingExport: """ @@ -31,7 +142,6 @@ class JianYingExport: def __init__( self, materials_folder_path: str, - program_path: str = r"E:\JianYingPro\5.9.0.11632\JianYingPro.exe", # 仅可在windows运行该脚本 drafts_folder_path: str = r"E:\JianYingPro Drafts", video_width: int = 1080, video_height: int = 1920, @@ -39,7 +149,6 @@ class JianYingExport: ): """ 初始化 - :param program_path: 剪映程序路径 :param drafts_folder_path: 剪映草稿文件夹路径 :param materials_folder_path: 素材文件夹路径 :param video_width: 视频宽度，默认为 1080像素 @@ -47,22 +156,17 @@ class JianYingExport: :param video_fps: 视频帧率（单位为帧/秒），默认为 30 """ try: - self.program_path = Path(program_path) - if not self.program_path.exists(): - raise RuntimeError("剪映程序路径不存在") - - # 初始化剪映专业版进程 - self.jianying_process = None - + # 初始化剪映草稿文件夹路径 self.drafts_folder_path = Path(drafts_folder_path) if not self.drafts_folder_path.exists(): raise RuntimeError("剪映草稿文件夹路径不存在") - # 初始化草稿文件夹管理器 + # 初始化剪映草稿文件夹管理器 self.drafts_folder = pyJianYingDraft.DraftFolder( folder_path=self.drafts_folder_path.as_posix() ) + # 初始化素材文件夹路径 self.materials_folder_path = Path(materials_folder_path) if not self.materials_folder_path.exists(): raise RuntimeError("素材文件夹路径不存在") @@ -71,15 +175,15 @@ class JianYingExport: self.exports_folder_path = Path( self.materials_folder_path.as_posix().replace("materials", "exports") ) - self.exports_folder_path.mkdir() # 若导出文件夹存在则抛出异常，需手动处理 + # 若导出文件夹存在则删除，再创建导出文件夹 + if self.exports_folder_path.exists(): + shutil.rmtree(self.exports_folder_path) + self.exports_folder_path.mkdir() self.materials = {} # 初始化素材文件夹内所有素材 self._init_materials() - # 构建项目名称 - self.project_name = self.materials_folder_path.stem - # 初始化所有工作流 self.workflows = { "0000": [ @@ -107,8 +211,8 @@ class JianYingExport: ], # 适用于视频号 } - # 初始化工作配置 - self.configuration = { + # 初始化所有节点配置 + self.configurations = { "add_subtitles": { "text": self.materials["subtitles_text"], "timbre": [ @@ -291,16 +395,14 @@ class JianYingExport: }, ], # 图像调节设置 }, # 添加贴纸2工作配置 + "save": {}, # 保存 } - # 初始化工作流 - self.workflow = [] - self.video_width, self.video_height = video_width, video_height self.video_fps = video_fps - # 初始化所有草稿名称 - self.draft_names = [] + # 实例化缓存 + self.caches = Caches() except Exception as exception: raise RuntimeError(f"发生异常：{str(exception)}") from exception @@ -406,48 +508,37 @@ class JianYingExport: else: self.materials["statement_video_material_path"] = [] - def export_videos(self, workflow: str, draft_counts: int): + def export_videos(self, workflow_name: str, draft_counts: int): """ 导出视频 - :param workflow: 工作流名称 + :param workflow_name: 工作流名称 :param draft_counts: 每批次导出草稿数 """ - if workflow not in self.workflows: + if workflow_name not in self.workflows: raise RuntimeError(f"未配置该工作流") - self.workflow = self.workflows[workflow] - # 若工作流包含添加背景音频，则将添加背景视频工作配置中播放音量设置为0 - if "add_background_audio" in self.workflow: - self.configuration["add_background_video"]["volume"] = [0.0] + workflow = self.workflows[workflow_name] + # 若工作流包含添加背景音频，则在添加背景视频节点配置的播放音量设置为0 + if "add_background_audio" in workflow: + self.configurations["add_background_video"]["volume"] = [0.0] # 按照工作流和工作配置拼接素材，批量生成草稿 - self._generate_drafts(draft_counts=draft_counts) + batch_draft_names = self._batch_generate_drafts( + workflow_name=workflow_name, + draft_counts=draft_counts, + ) - # 批次导出 - for batch_start in range(0, self.draft_counts, batch_draft_counts): - # 当前批次所有草稿名称 - batch_draft_names = self.draft_names[ - batch_start : batch_start + batch_draft_counts - ] + for draft_name in batch_draft_names: + print(f"正在导出 {draft_name}...") + if (self.exports_folder_path / f"{draft_name}.mp4").is_file(): + print("存在相同名称的草稿，跳过") + continue - # 启动剪映专业版进程 - self._start_process() - time.sleep(2) - - # 初始化剪映控制器 - jianying_controller = pyJianYingDraft.JianyingController() - - for draft_name in batch_draft_names: - print(f"正在导出 {draft_name}...") - if (self.exports_folder_path / f"{draft_name}.mp4").is_file(): - print("存在相同名称的草稿，跳过") - continue - - jianying_controller.export_draft( - draft_name=draft_name, - output_path=self.exports_folder_path.as_posix(), - ) - print("已完成") - print() + jianying_controller.export_draft( + draft_name=draft_name, + output_path=self.exports_folder_path.as_posix(), + ) + print("已完成") + print() # 关闭剪映专业版进程 self._close_process() @@ -458,19 +549,36 @@ class JianYingExport: self.drafts_folder.remove(draft_name=draft_name) time.sleep(2) - def _generate_drafts( + def _batch_generate_drafts( self, + workflow_name: str, draft_counts: int, ) -> None: """ - 按照工作流和工作配置拼接素材，批量生成草稿 + 批量生成草稿 + :param workflow_name: 工作流名称 :param draft_counts: 草稿数 :return: 无 """ - for idx in range(1, draft_counts + 1): - # 构建草稿名称 - draft_name = self.project_name + f"{idx:03d}" - print(f"正在合成短视频 {draft_name}， {idx}/{draft_counts}...") + draft_index = 1 # 草稿索引 + while True: + # 获取工作流配置 + workflow_configurations = self._get_workflow_configurations( + workflow_name=workflow_name + ) + + # 根据工作流配置生成草稿名称 + draft_name = self._generate_draft_name( + workflow_configurations=workflow_configurations, + ) + + # 若已缓存则跳过 + if self.caches.query(draft_name=draft_name): + continue + + print(f"正在生成草稿 {draft_name}...") + + exit() # 实例化 JianYingDraft draft = JianYingDraft( @@ -482,67 +590,52 @@ class JianYingExport: materials_folder_path=self.materials_folder_path, ) - # 初始化当前草稿工作配置 - configuration = { - "materials_folder_path": self.materials_folder_path.stem, - "draft_name": draft_name, - "workflows": [], - } - for work in self.workflow: - # 获取工作配置 - parameters = self._get_parameters(work=work) - configuration["workflows"].append( - { - "work": work, - "parameters": parameters, - } - ) - - match work: + for node in workflow_configurations: + match node["node_name"]: # 添加字幕 case "add_subtitles": print("-> 正在添加字幕...", end="") - draft.add_subtitles(**parameters) + draft.add_subtitles(**node["configurations"]) print("已完成") - # 添加字幕 + # 添加字幕视频 case "add_subtitles_video": print("-> 正在添加字幕视频...", end="") - draft.add_video_segment(**parameters) + draft.add_video_segment(**node["configurations"]) print("已完成") # 添加背景视频 case "add_background_video": print("-> 正在添加背景视频...", end="") - draft.add_video_segment(**parameters) + draft.add_video_segment(**node["configurations"]) print("已完成") # 添加背景音频 case "add_background_audio": print("-> 正在添加背景音频...", end="") - draft.add_audio_segment(**parameters) + draft.add_audio_segment(**node["configurations"]) print("已完成") # 添加标识 case "add_logo": print("-> 正在添加标识...", end="") - draft.add_video_segment(**parameters) + draft.add_video_segment(**node["configurations"]) print("已完成") # 添加标识视频 case "add_logo_video": print("-> 正在添加标识视频...", end="") - draft.add_video_segment(**parameters) + draft.add_video_segment(**node["configurations"]) print("已完成") # 添加声明文本 case "add_statement": print("-> 正在添加声明...", end="") - draft.add_text_segment(**parameters) + draft.add_text_segment(**node["configurations"]) print("已完成") # 添加声明视频 case "add_statement_video": print("-> 正在添加声明视频...", end="") - draft.add_video_segment(**parameters) + draft.add_video_segment(**node["configurations"]) print("已完成") # 添加贴纸 - case _ if work.startswith("add_sticker"): + case _ if node["node_name"].startswith("add_sticker"): print("-> 正在添加贴纸...", end="") - draft.add_sticker(**parameters) + draft.add_sticker(**node["configurations"]) print("已完成") # 将草稿保存至剪映草稿文件夹内 case "save": @@ -568,31 +661,66 @@ class JianYingExport: # 就所有草稿名称倒叙排序排序 self.draft_names.sort(reverse=True) - def _get_parameters( + def _get_workflow_configurations( self, - work: str, - ) -> Dict[str, Any]: + workflow_name: str, + ) -> List[Dict[str, Any]]: """ - 获取工作配置 - :param work: 工作，包括添加字幕、添加背景视频、添加标识、添加声明和添加贴纸 - :return: 工作配置 + 获取工作流配置 + :param workflow_name: 工作流名称 + :return: 工作流配置 """ - if work == "save": - return {} + # 初始化工作流配置 + workflow_configurations = [] + for node_name in self.workflows[workflow_name]: + # 根据节点名称获取节点配置 + configurations = { + key: random.choice(value) + for key, value in self.configurations[node_name].items() + } + # 若非添加字幕或保存则在工作流配置添加轨道名称 + if node_name not in ["add_subtitles", "save"]: + configurations["track_name"] = ( + matched.group("track_name") + if ( + matched := re.match( + pattern=r"^.*?_(?P.*)$", + string=node_name, + ) + ) + else node_name + ) - parameters = { - key: random.choice(value) for key, value in self.configuration[work].items() - } # TODO: 考虑融合贝叶斯优化 - if work == "add_subtitles": - parameters.pop("keywords") + workflow_configurations.append( + { + "node_name": node_name, + "configurations": configurations, + } + ) - # 就除添加字幕其它工作添加轨道名称 - if work != "add_subtitles" and ( - match := re.search(r"_(?P.+)", work) - ): - parameters["track_name"] = match.group("track_name") + return workflow_configurations - return parameters + def _generate_draft_name( + self, + workflow_configurations: List[Dict[str, Any]], + ) -> str: + """ + 根据工作流配置生成草稿名称 + :param workflow_configurations: 工作流配置 + :return: 草稿名称 + """ + return ( + hashlib.md5( + json.dumps( + obj=workflow_configurations, + cls=JSONEncoder, + sort_keys=True, + ensure_ascii=False, + ).encode("utf-8") + ) # 先将工作流配置序列化，再以MD5哈希值作为草稿名称 + .hexdigest() + .upper() + ) def _highlight_keywords( self, diff --git a/短视频合成自动化/main.py b/短视频合成自动化/main.py index 16f2632..d2946b0 100644 --- a/短视频合成自动化/main.py +++ b/短视频合成自动化/main.py @@ -9,8 +9,10 @@ if __name__ == "__main__": # 实例化 JianYingExport jianying_export = JianYingExport( materials_folder_path=r"E:\jianying\materials\淘宝闪购模版001", - draft_counts=10, ) # 导出视频 - jianying_export.export_videos(workflow_name="0001") + jianying_export.export_videos( + workflow_name="0001", + draft_counts=10, + )