694 lines
26 KiB
Python
694 lines
26 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
导出草稿模块
|
||
"""
|
||
|
||
import hashlib
|
||
import json
|
||
from pathlib import Path
|
||
from pathlib import WindowsPath
|
||
import random
|
||
import re
|
||
import shutil
|
||
import sys
|
||
import time
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
import pyJianYingDraft
|
||
|
||
from draft import JianYingDraft
|
||
from utils.sqlite import SQLite
|
||
|
||
sys.path.append(Path(__file__).parent.parent.as_posix())
|
||
|
||
|
||
# 自定义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:
|
||
"""
|
||
封装 pyJianYingDraft.JianyingController库，支持：
|
||
1、初始化素材文件夹内所有素材
|
||
2、初始化工作流和工作配置
|
||
3、导出草稿
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
materials_folder_path: str,
|
||
drafts_folder_path: str = r"E:\JianYingPro Drafts",
|
||
video_width: int = 1080,
|
||
video_height: int = 1920,
|
||
video_fps: int = 30,
|
||
):
|
||
"""
|
||
初始化
|
||
:param drafts_folder_path: 剪映草稿文件夹路径
|
||
:param materials_folder_path: 素材文件夹路径
|
||
:param video_width: 视频宽度，默认为 1080像素
|
||
:param video_height: 视频高度，默认为 1920像素
|
||
:param video_fps: 视频帧率（单位为帧/秒），默认为 30
|
||
"""
|
||
try:
|
||
# 初始化剪映草稿文件夹路径
|
||
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("素材文件夹路径不存在")
|
||
|
||
# 初始化导出文件夹路径
|
||
self.exports_folder_path = Path(
|
||
self.materials_folder_path.as_posix().replace("materials", "exports")
|
||
)
|
||
# 若导出文件夹存在则删除，再创建导出文件夹
|
||
if self.exports_folder_path.exists():
|
||
shutil.rmtree(self.exports_folder_path)
|
||
self.exports_folder_path.mkdir()
|
||
|
||
self.materials = {}
|
||
# 初始化素材文件夹内所有素材
|
||
self._init_materials()
|
||
|
||
# 初始化所有工作流
|
||
self.workflows = {
|
||
"默认": [
|
||
"add_subtitles",
|
||
"add_background_video",
|
||
"add_statement",
|
||
"add_sticker1",
|
||
"add_sticker2",
|
||
], # 默认工作流，先根据脚本合成音频，再叠加背景视频、声明视频、贴纸1视频和贴纸2视频
|
||
"淘宝闪购": [
|
||
"add_subtitles_video", # 以此作为草稿持续时长
|
||
"add_background_video",
|
||
"add_background_audio",
|
||
"add_statement_video",
|
||
], # 适用于淘宝闪购、存量抽手机
|
||
"视频号": [
|
||
"add_subtitles_video", # 以此作为草稿持续时长
|
||
"add_background_video",
|
||
"add_background_audio",
|
||
"add_logo_video",
|
||
"add_statement_video",
|
||
], # 适用于视频号
|
||
}
|
||
|
||
# 初始化所有节点配置
|
||
self.configurations = {
|
||
"add_subtitles": {
|
||
"text": self.materials["subtitles_text"],
|
||
"timbre": [
|
||
"zh-CN-XiaoxiaoNeural",
|
||
"zh-CN-XiaoyiNeural",
|
||
"zh-CN-YunjianNeural",
|
||
"zh-CN-YunxiNeural",
|
||
"zh-CN-YunxiaNeural",
|
||
"zh-CN-YunyangNeural",
|
||
], # 音色
|
||
"style": [
|
||
{"size": 9.0},
|
||
{"size": 10.0},
|
||
{"size": 11.0},
|
||
], # 字体样式
|
||
"keywords": [
|
||
"瑞幸",
|
||
], # 关键词
|
||
"effect": [
|
||
{"effect_id": "7127561998556089631"},
|
||
{"effect_id": "7166467215410187552"},
|
||
{"effect_id": "6896138122774498567"},
|
||
{"effect_id": "7166469374507765031"},
|
||
{"effect_id": "6896137924853763336"},
|
||
{"effect_id": "6896137990788091143"},
|
||
{"effect_id": "7127614731187211551"},
|
||
{"effect_id": "7127823362356743461"},
|
||
{"effect_id": "7127653467555990821"},
|
||
{"effect_id": "7127828216647011592"},
|
||
], # 花字设置
|
||
}, # 添加字幕工作配置
|
||
"add_subtitles_video": {
|
||
"material_path": self.materials["subtitles_video_material_path"],
|
||
"volume": [1.0], # 播放音量
|
||
"clip_settings": [
|
||
None,
|
||
], # 图像调节设置
|
||
}, # 添加字幕视频工作配置
|
||
"add_background_video": {
|
||
"material_path": self.materials["background_video_material_path"],
|
||
"volume": [1.0], # 播放音量
|
||
"clip_settings": [
|
||
{
|
||
"scale_x": 1.5,
|
||
"scale_y": 1.5,
|
||
},
|
||
], # 图像调节设置
|
||
}, # 添加背景视频工作配置
|
||
"add_background_audio": {
|
||
"material_path": self.materials["background_audio_material_path"],
|
||
"volume": [0.6], # 播放音量
|
||
}, # 添加背景音频工作配置
|
||
"add_logo": {
|
||
"material_path": self.materials["logo_material_path"],
|
||
"clip_settings": [
|
||
{
|
||
"scale_x": 0.2,
|
||
"scale_y": 0.2,
|
||
"transform_x": -0.78,
|
||
"transform_y": 0.82,
|
||
},
|
||
{
|
||
"scale_x": 0.2,
|
||
"scale_y": 0.2,
|
||
"transform_x": -0.68,
|
||
"transform_y": 0.82,
|
||
},
|
||
{
|
||
"scale_x": 0.2,
|
||
"scale_y": 0.2,
|
||
"transform_x": 0,
|
||
"transform_y": 0.82,
|
||
},
|
||
{
|
||
"scale_x": 0.2,
|
||
"scale_y": 0.2,
|
||
"transform_x": 0.68,
|
||
"transform_y": 0.82,
|
||
},
|
||
{
|
||
"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_material_path"],
|
||
"volume": [1.0], # 播放音量
|
||
"clip_settings": [
|
||
None,
|
||
], # 图像调节设置
|
||
}, # 添加标识视频工作配置
|
||
"add_statement": {
|
||
"text": self.materials["statement_text"],
|
||
"style": [
|
||
{"size": 5.0, "align": 1, "vertical": True},
|
||
{"size": 6.0, "align": 1, "vertical": True},
|
||
{"size": 7.0, "align": 1, "vertical": True},
|
||
], # 文本样式
|
||
"border": [
|
||
{"width": 35.0},
|
||
{"width": 40.0},
|
||
{"width": 45.0},
|
||
{"width": 50.0},
|
||
{"width": 55.0},
|
||
], # 描边宽度
|
||
"clip_settings": [
|
||
{
|
||
"transform_x": -0.80,
|
||
},
|
||
{
|
||
"transform_x": -0.82,
|
||
},
|
||
{
|
||
"transform_x": -0.84,
|
||
},
|
||
], # 图像调节设置
|
||
}, # 添加声明工作配置
|
||
"add_statement_video": {
|
||
"material_path": self.materials["statement_video_material_path"],
|
||
"volume": [1.0], # 播放音量
|
||
"clip_settings": [
|
||
None,
|
||
], # 图像调节设置
|
||
}, # 添加声明视频工作配置
|
||
"add_sticker1": {
|
||
"resource_id": [
|
||
"7110124379568098568",
|
||
"7019687632804334861",
|
||
"6895933678262750478",
|
||
"7010558788675652900",
|
||
"7026858083393588487",
|
||
"7222940306558209336",
|
||
"7120543009489341727",
|
||
"6939830545673227557",
|
||
"6939826722451754271",
|
||
"7210221631132749093",
|
||
"7138432572488453408",
|
||
"7137700067338620192",
|
||
"6895924436822674696",
|
||
"7134644683506044163",
|
||
"7062539853430279437",
|
||
],
|
||
"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,
|
||
},
|
||
], # 图像调节设置
|
||
}, # 添加贴纸1工作配置（不包含箭头类）
|
||
"add_sticker2": {
|
||
"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,
|
||
},
|
||
], # 图像调节设置
|
||
}, # 添加贴纸2工作配置
|
||
}
|
||
|
||
self.video_width, self.video_height = video_width, video_height
|
||
self.video_fps = video_fps
|
||
|
||
# 实例化缓存
|
||
self.caches = Caches()
|
||
|
||
except Exception as exception:
|
||
raise RuntimeError(f"发生异常：{str(exception)}") from exception
|
||
|
||
def _init_materials(self) -> None:
|
||
"""
|
||
初始化素材文件夹内所有素材
|
||
:return: 无
|
||
"""
|
||
# 字幕(文本)
|
||
subtitles_path = self.materials_folder_path / "字幕.txt"
|
||
if subtitles_path.exists() and subtitles_path.is_file():
|
||
with open(subtitles_path, "r", encoding="utf-8") as file:
|
||
subtitles_text = file.readlines()
|
||
if not subtitles_text:
|
||
raise RuntimeError("字幕文本为空")
|
||
self.materials["subtitles_text"] = subtitles_text
|
||
else:
|
||
self.materials["subtitles_text"] = []
|
||
|
||
# 字幕（视频）
|
||
subtitles_video_folder_path = self.materials_folder_path / "字幕视频"
|
||
if (
|
||
subtitles_video_folder_path.exists()
|
||
and subtitles_video_folder_path.is_dir()
|
||
):
|
||
self.materials["subtitles_video_material_path"] = [
|
||
file_path
|
||
for file_path in subtitles_video_folder_path.rglob("*.mov")
|
||
if file_path.is_file()
|
||
]
|
||
else:
|
||
self.materials["subtitles_video_material_path"] = []
|
||
|
||
# 背景视频
|
||
background_video_folder_path = self.materials_folder_path / "背景视频"
|
||
if (
|
||
background_video_folder_path.exists()
|
||
and background_video_folder_path.is_dir()
|
||
):
|
||
self.materials["background_video_material_path"] = [
|
||
file_path
|
||
for file_path in background_video_folder_path.rglob("*.mp4")
|
||
if file_path.is_file()
|
||
]
|
||
else:
|
||
self.materials["background_video_material_path"] = []
|
||
|
||
# 背景音频
|
||
background_audio_folder_path = self.materials_folder_path / "背景音频"
|
||
if (
|
||
background_audio_folder_path.exists()
|
||
and background_audio_folder_path.is_dir()
|
||
):
|
||
self.materials["background_audio_material_path"] = [
|
||
file_path
|
||
for file_path in background_audio_folder_path.rglob("*.mp3")
|
||
if file_path.is_file()
|
||
]
|
||
else:
|
||
self.materials["background_audio_material_path"] = []
|
||
|
||
# 标识
|
||
logo_path = self.materials_folder_path / "标识.png"
|
||
if logo_path.exists() and logo_path.is_file():
|
||
self.materials["logo_material_path"] = [logo_path] # 有且只有一张标识
|
||
else:
|
||
self.materials["logo_material_path"] = []
|
||
|
||
# 标识视频
|
||
logo_video_folder_path = self.materials_folder_path / "标识视频"
|
||
if logo_video_folder_path.exists() and logo_video_folder_path.is_dir():
|
||
self.materials["logo_video_material_path"] = [
|
||
file_path
|
||
for file_path in logo_video_folder_path.rglob("*.mov")
|
||
if file_path.is_file()
|
||
]
|
||
else:
|
||
self.materials["logo_video_material_path"] = []
|
||
|
||
# 声明文本
|
||
statement_path = self.materials_folder_path / "声明.txt"
|
||
if statement_path.exists() and statement_path.is_file():
|
||
with open(statement_path, "r", encoding="utf-8") as file:
|
||
statement_text = file.readlines()
|
||
if not statement_text:
|
||
raise RuntimeError("声明文本为空")
|
||
self.materials["statement_text"] = statement_text
|
||
else:
|
||
self.materials["statement_text"] = []
|
||
|
||
# 声明视频
|
||
statement_video_folder_path = self.materials_folder_path / "声明视频"
|
||
if (
|
||
statement_video_folder_path.exists()
|
||
and statement_video_folder_path.is_dir()
|
||
):
|
||
self.materials["statement_video_material_path"] = [
|
||
file_path
|
||
for file_path in statement_video_folder_path.rglob("*.mov")
|
||
if file_path.is_file()
|
||
]
|
||
else:
|
||
self.materials["statement_video_material_path"] = []
|
||
|
||
def export_videos(self, workflow_name: str, draft_counts: int):
|
||
"""
|
||
导出视频
|
||
:param workflow_name: 工作流名称
|
||
:param draft_counts: 每批次导出草稿数
|
||
"""
|
||
if workflow_name not in self.workflows:
|
||
raise RuntimeError(f"该工作流 {workflow_name} 未配置")
|
||
workflow = self.workflows[workflow_name]
|
||
|
||
# 若工作流包含添加背景音频，则在添加背景视频节点配置的播放音量设置为0
|
||
if "add_background_audio" in workflow:
|
||
self.configurations["add_background_video"]["volume"] = [0.0]
|
||
|
||
# 批量生成草稿
|
||
self._batch_generate_drafts(
|
||
workflow_name=workflow_name,
|
||
draft_counts=draft_counts,
|
||
)
|
||
|
||
def _batch_generate_drafts(
|
||
self,
|
||
workflow_name: str,
|
||
draft_counts: int,
|
||
) -> None:
|
||
"""
|
||
批量生成草稿
|
||
:param workflow_name: 工作流名称
|
||
:param draft_counts: 草稿数
|
||
:return: 无
|
||
"""
|
||
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}...")
|
||
|
||
# 实例化 JianYingDraft
|
||
draft = JianYingDraft(
|
||
drafts_folder=self.drafts_folder,
|
||
draft_name=draft_name,
|
||
video_width=self.video_width,
|
||
video_height=self.video_height,
|
||
video_fps=self.video_fps,
|
||
materials_folder_path=self.materials_folder_path,
|
||
)
|
||
|
||
for node in workflow_configurations:
|
||
match node["node_name"]:
|
||
# 添加字幕
|
||
case "add_subtitles":
|
||
print("-> 正在添加字幕...", end="")
|
||
draft.add_subtitles(**node["configurations"])
|
||
print("已完成")
|
||
# 添加字幕视频
|
||
case "add_subtitles_video":
|
||
print("-> 正在添加字幕视频...", end="")
|
||
draft.add_video_segment(**node["configurations"])
|
||
print("已完成")
|
||
# 添加背景视频
|
||
case "add_background_video":
|
||
print("-> 正在添加背景视频...", end="")
|
||
draft.add_video_segment(**node["configurations"])
|
||
print("已完成")
|
||
# 添加背景音频
|
||
case "add_background_audio":
|
||
print("-> 正在添加背景音频...", end="")
|
||
draft.add_audio_segment(**node["configurations"])
|
||
print("已完成")
|
||
# 添加标识
|
||
case "add_logo":
|
||
print("-> 正在添加标识...", end="")
|
||
draft.add_video_segment(**node["configurations"])
|
||
print("已完成")
|
||
# 添加标识视频
|
||
case "add_logo_video":
|
||
print("-> 正在添加标识视频...", end="")
|
||
draft.add_video_segment(**node["configurations"])
|
||
print("已完成")
|
||
# 添加声明文本
|
||
case "add_statement":
|
||
print("-> 正在添加声明...", end="")
|
||
draft.add_text_segment(**node["configurations"])
|
||
print("已完成")
|
||
# 添加声明视频
|
||
case "add_statement_video":
|
||
print("-> 正在添加声明视频...", end="")
|
||
draft.add_video_segment(**node["configurations"])
|
||
print("已完成")
|
||
# 添加贴纸
|
||
case _ if node["node_name"].startswith("add_sticker"):
|
||
print("-> 正在添加贴纸...", end="")
|
||
draft.add_sticker(**node["configurations"])
|
||
print("已完成")
|
||
# 将草稿保存至剪映草稿文件夹内
|
||
case "save":
|
||
print("-> 正在将草稿保存至剪映草稿文件夹内...", end="")
|
||
draft.save()
|
||
print("已完成")
|
||
|
||
# 缓存
|
||
self.caches.update(
|
||
draft_name=draft_name,
|
||
workflow_configurations=workflow_configurations,
|
||
)
|
||
|
||
print("已完成")
|
||
print()
|
||
|
||
draft_index += 1
|
||
if draft_index > draft_counts:
|
||
break
|
||
|
||
def _get_workflow_configurations(
|
||
self,
|
||
workflow_name: str,
|
||
) -> List[Dict[str, Any]]:
|
||
"""
|
||
获取工作流配置
|
||
:param workflow_name: 工作流名称
|
||
: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"]:
|
||
configurations["track_name"] = (
|
||
matched.group("track_name")
|
||
if (
|
||
matched := re.match(
|
||
pattern=r"^.*?_(?P<track_name>.*)$",
|
||
string=node_name,
|
||
)
|
||
)
|
||
else node_name
|
||
)
|
||
|
||
workflow_configurations.append(
|
||
{
|
||
"node_name": node_name,
|
||
"configurations": configurations,
|
||
}
|
||
)
|
||
|
||
# 添加保存节点
|
||
workflow_configurations.append(
|
||
{
|
||
"node_name": "save",
|
||
"configurations": {},
|
||
}
|
||
)
|
||
return workflow_configurations
|
||
|
||
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")
|
||
) # 将工作流配置序列化
|
||
.hexdigest()
|
||
.upper() # MD5哈希值的大写十六进制作为草稿名称
|
||
)
|