1

短视频合成自动化/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]
+        workflow = self.workflows[workflow_name]
-        # 若工作流包含添加背景音频，则将添加背景视频工作配置中播放音量设置为0
+        # 若工作流包含添加背景音频，则在添加背景视频节点配置的播放音量设置为0
-        if "add_background_audio" in self.workflow:
+        if "add_background_audio" in workflow:
-            self.configuration["add_background_video"]["volume"] = [0.0]
+            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 draft_name in batch_draft_names:
-        for batch_start in range(0, self.draft_counts, batch_draft_counts):
+            print(f"正在导出 {draft_name}...")
-            # 当前批次所有草稿名称
+            if (self.exports_folder_path / f"{draft_name}.mp4").is_file():
-            batch_draft_names = self.draft_names[
+                print("存在相同名称的草稿，跳过")
-                batch_start : batch_start + batch_draft_counts
+                continue
-            ]
 
-            # 启动剪映专业版进程
+            jianying_controller.export_draft(
-            self._start_process()
+                draft_name=draft_name,
-            time.sleep(2)
+                output_path=self.exports_folder_path.as_posix(),
-
+            )
-            # 初始化剪映控制器
+            print("已完成")
-            jianying_controller = pyJianYingDraft.JianyingController()
+            print()
-
-            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()
 
             # 关闭剪映专业版进程
             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_index = 1  # 草稿索引
-            # 构建草稿名称
+        while True:
-            draft_name = self.project_name + f"{idx:03d}"
+            # 获取工作流配置
-            print(f"正在合成短视频 {draft_name}， {idx}/{draft_counts}...")
+            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,
             )
 
-            # 初始化当前草稿工作配置
+            for node in workflow_configurations:
-            configuration = {
+                match node["node_name"]:
-                "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:
                     # 添加字幕
                     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,
+        workflow_name: str,
-    ) -> Dict[str, Any]:
+    ) -> List[Dict[str, Any]]:
         """
-        获取工作配置
+        获取工作流配置
-        :param work: 工作，包括添加字幕、添加背景视频、添加标识、添加声明和添加贴纸
+        :param workflow_name: 工作流名称
-        :return: 工作配置
+        :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<track_name>.*)$",
+                            string=node_name,
+                        )
+                    )
+                    else node_name
+                )
 
-        parameters = {
+            workflow_configurations.append(
-            key: random.choice(value) for key, value in self.configuration[work].items()
+                {
-        }  # TODO: 考虑融合贝叶斯优化
+                    "node_name": node_name,
-        if work == "add_subtitles":
+                    "configurations": configurations,
-            parameters.pop("keywords")
+                }
+            )
 
-        # 就除添加字幕其它工作添加轨道名称
+        return workflow_configurations
-        if work != "add_subtitles" and (
-            match := re.search(r"_(?P<track_name>.+)", work)
-        ):
-            parameters["track_name"] = match.group("track_name")
 
-        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,

短视频合成自动化/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,
+    )