571 lines
22 KiB
Python
571 lines
22 KiB
Python
# -*- 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
|
||
|
||
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_paths"] = [
|
||
subtitle_video_path
|
||
for subtitle_video_path in subtitle_video_folder_path.rglob("*.mov")
|
||
]
|
||
else:
|
||
materials["subtitle_video_paths"] = []
|
||
|
||
# 构建背景视频文件夹路径
|
||
background_video_folder_path = self.materials_folder_path / "背景视频"
|
||
if (
|
||
background_video_folder_path.exists()
|
||
and background_video_folder_path.is_dir()
|
||
):
|
||
materials["background_video_paths"] = [
|
||
background_video_path
|
||
for background_video_path in background_video_folder_path.rglob("*.mp4")
|
||
]
|
||
else:
|
||
materials["background_video_paths"] = []
|
||
|
||
# 构建达人视频文件夹路径
|
||
kol_video_folder_path = self.materials_folder_path / "达人视频"
|
||
if kol_video_folder_path.exists() and kol_video_folder_path.is_dir():
|
||
materials["kol_video_paths"] = [
|
||
kol_video_path
|
||
for kol_video_path in kol_video_folder_path.rglob("*.mp4")
|
||
]
|
||
else:
|
||
materials["kol_video_paths"] = []
|
||
|
||
# 构建背景音频文件夹路径
|
||
background_audio_folder_path = self.materials_folder_path / "背景音频"
|
||
if (
|
||
background_audio_folder_path.exists()
|
||
and background_audio_folder_path.is_dir()
|
||
):
|
||
materials["background_audio_paths"] = [
|
||
background_audio_path
|
||
for background_audio_path in background_audio_folder_path.rglob("*.mp3")
|
||
]
|
||
else:
|
||
materials["background_audio_paths"] = []
|
||
|
||
# 构建标识图片路径
|
||
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_subtitle_video",
|
||
"add_background_video",
|
||
"add_background_audio",
|
||
"add_statement_video",
|
||
], # 淘宝闪购，先根据字幕视频获取其持续时长并作为成品持续时长，再叠加背景视频、背景音频和生成视频
|
||
"淘宝闪购_达人": [
|
||
"add_background_video",
|
||
"add_background_audio",
|
||
"add_subtitle_video",
|
||
"add_kol_video",
|
||
"add_statement_video",
|
||
], # 淘宝闪购_达人，第一段：先根据字幕视频获取前5秒作为持续时长，再叠加背景视频、背景音频；第二段：背景视频（达人）；拼接第一段和第二段再叠加声明视频
|
||
}
|
||
# 默认以素材文件夹名称为工作流名称
|
||
workflow_name = self.materials_folder_path.stem
|
||
# 工作流
|
||
workflow = workflows.get(workflow_name)
|
||
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_paths"],
|
||
"volume": [1.0], # 播放音量
|
||
"clip_settings": [
|
||
None,
|
||
], # 图像调节设置
|
||
}, # 添加字幕视频工作配置
|
||
"add_background_video": {
|
||
"material_path": self.materials["background_video_paths"],
|
||
"volume": [1.0], # 播放音量
|
||
"clip_settings": [
|
||
{
|
||
"scale_x": 1.0,
|
||
"scale_y": 1.0,
|
||
},
|
||
], # 图像调节设置
|
||
}, # 添加背景视频工作配置
|
||
"add_kol_video": {
|
||
"material_path": self.materials["kol_video_paths"],
|
||
"volume": [1.0], # 播放音量
|
||
"clip_settings": [
|
||
{
|
||
"scale_x": 1.0,
|
||
"scale_y": 1.0,
|
||
},
|
||
], # 图像调节设置
|
||
}, # 添加达人视频工作配置
|
||
"add_background_audio": {
|
||
"material_path": self.materials["background_audio_paths"],
|
||
"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,
|
||
},
|
||
], # 图像调节设置
|
||
}, # 添加箭头贴纸工作配置
|
||
}
|
||
# 达人视频特殊处理：字幕视频、背景视频和背景音频素材持续时长为 5秒，叠加达人视频
|
||
if workflow_name == "淘宝闪购_达人":
|
||
configurations.update(
|
||
{
|
||
node: {
|
||
**configurations[node],
|
||
"target_timerange": [
|
||
(0, 5_000_000),
|
||
],
|
||
}
|
||
for node in [
|
||
"add_subtitle_video",
|
||
"add_background_video",
|
||
"add_background_audio",
|
||
]
|
||
}
|
||
)
|
||
configurations.update(
|
||
{
|
||
node: {
|
||
**configurations[node],
|
||
"target_timerange": [
|
||
(5_000_000, None),
|
||
],
|
||
}
|
||
for node in [
|
||
"add_kol_video",
|
||
]
|
||
}
|
||
)
|
||
|
||
# 若包含添加背景音频节点则在添加背景视频时其播放音量设置为 0
|
||
if "add_background_audio" in workflow:
|
||
configurations["add_background_video"]["volume"] = [0.0]
|
||
|
||
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,
|
||
) -> None:
|
||
"""
|
||
批量创建草稿
|
||
:param draft_counts: 草稿数
|
||
:param video_width: 视频宽度（单位为像素），默认为 1080
|
||
:param video_height: 视频高度（单位为像素），默认为 1920
|
||
:param video_fps: 视频帧率（单位为帧/秒），默认为 30
|
||
:return: 无
|
||
"""
|
||
draft_index = 1 # 草稿索引
|
||
while True:
|
||
# 获取节点配置
|
||
configurations = self._get_configurations()
|
||
|
||
video_duration = VideoMaterial(
|
||
path=configurations["add_background_video"]["material_path"].as_posix()
|
||
).duration # 默认将背景视频素材持续时长作为视频持续时长（单位为微秒）
|
||
# 达人视频特殊处理：固定 5秒加上达人视频素材持续时长
|
||
if "add_kol_video" in configurations:
|
||
video_duration = (
|
||
5_000_000
|
||
+ VideoMaterial(
|
||
path=configurations["add_kol_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_subtitle_video":
|
||
print("-> 正在添加字幕视频...", end="")
|
||
draft.add_video_segment(**configurations[node_name])
|
||
print("已完成")
|
||
# 添加背景视频
|
||
case "add_background_video":
|
||
print("-> 正在添加背景视频...", end="")
|
||
draft.add_video_segment(**configurations[node_name])
|
||
print("已完成")
|
||
# 添加达人视频
|
||
case "add_kol_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_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,
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
获取节点配置
|
||
: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<track_name>.+_.+)$",
|
||
string=node_name,
|
||
)
|
||
)
|
||
else node_name
|
||
)
|
||
|
||
# 添加保存节点
|
||
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哈希值的大写十六进制作为草稿名称
|
||
)
|