V 3.1.7
1. 移除软件包自带的本地 whisper(需单独安装) 2. 重构版本底层依赖,移除外部依赖 3. 修复 首页 暗黑模式不兼容的问题 4. 修复 SD 合并提示词报错
This commit is contained in:
@@ -0,0 +1,2 @@
|
||||
@echo off
|
||||
pyinstaller --upx-dir="C:\\Users\\27698\\Desktop\\upx-4.2.4-win64\upx.exe" local_whisper.py
|
||||
@@ -0,0 +1,170 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
import public_tools
|
||||
from pathlib import Path
|
||||
from huggingface_hub import hf_hub_download
|
||||
from faster_whisper import WhisperModel
|
||||
|
||||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
|
||||
|
||||
# 判断sys.argv 的长度,如果小于2,说明没有传入参数,设置初始参数
|
||||
# "C:\\Users\\27698\\Desktop\\LAITool\\resources\\scripts\\Lai.exe" -c "D:/来推项目集/7.4/娱乐:江湖大哥退休,去拍电影/scripts/output_crop_00001.json" "NVIDIA"
|
||||
# if len(sys.argv) < 2:
|
||||
# sys.argv = [
|
||||
# "C:\\Users\\27698\\Desktop\\LAITool\\resources\\scripts\\Lai.exe",
|
||||
# "-w",
|
||||
# "C:\\Users\\27698\\Desktop\\测试\\test\\mjTestoutput_crop_00001.mp4",
|
||||
# "C:\\Users\\27698\\Desktop\\测试\\test\data\\frame",
|
||||
# "C:\\Users\\27698\\Desktop\\测试\\test\\tmp\\input_crop",
|
||||
# 30,
|
||||
# "NVIDIA",
|
||||
# ]
|
||||
|
||||
print(sys.argv)
|
||||
|
||||
if len(sys.argv) < 2:
|
||||
print("Params: <runtime-config.json>")
|
||||
exit(0)
|
||||
|
||||
if getattr(sys, "frozen", False):
|
||||
cript_directory = os.path.dirname(sys.executable)
|
||||
elif __file__:
|
||||
cript_directory = os.path.dirname(__file__)
|
||||
|
||||
|
||||
def GetText(out_folder, mp3_folder):
|
||||
text = []
|
||||
# 先获取模型
|
||||
print("正在下载或加载模型")
|
||||
sys.stdout.flush()
|
||||
model_path = Path(
|
||||
hf_hub_download(repo_id="Systran/faster-whisper-large-v3", filename="model.bin")
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="config.json",
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="preprocessor_config.json",
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="tokenizer.json",
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="vocabulary.json",
|
||||
)
|
||||
model = WhisperModel(
|
||||
model_size_or_path=os.path.dirname(model_path),
|
||||
device="auto",
|
||||
local_files_only=True,
|
||||
)
|
||||
print("模型加载成功,开始识别")
|
||||
sys.stdout.flush()
|
||||
# 拿到指定文件夹里面的所有的MP3文件
|
||||
mp3_list = []
|
||||
for root, dirs, files in os.walk(mp3_folder):
|
||||
for file in files:
|
||||
if file.endswith(".mp3"):
|
||||
mp3_list.append(os.path.join(root, file))
|
||||
|
||||
for mp in mp3_list:
|
||||
segments, info = model.transcribe(
|
||||
mp,
|
||||
beam_size=5,
|
||||
language="zh",
|
||||
vad_filter=True,
|
||||
vad_parameters=dict(min_silence_duration_ms=1000),
|
||||
)
|
||||
tmp_text = ""
|
||||
for segment in segments:
|
||||
tmp_text += segment.text + "。"
|
||||
print(mp + "识别完成")
|
||||
sys.stdout.flush()
|
||||
text.append(tmp_text)
|
||||
|
||||
# 数据写出
|
||||
print("文本全部识别成功,正在写出")
|
||||
sys.stdout.flush()
|
||||
tools = public_tools.PublicTools()
|
||||
tools.write_to_file(text, os.path.join(out_folder, "文案.txt"))
|
||||
print("写出完成")
|
||||
sys.stdout.flush()
|
||||
|
||||
|
||||
def GetTextTask(out_folder, mp, name):
|
||||
text = []
|
||||
# 先获取模型
|
||||
print("正在下载或加载模型")
|
||||
sys.stdout.flush()
|
||||
model_path = Path(
|
||||
hf_hub_download(repo_id="Systran/faster-whisper-large-v3", filename="model.bin")
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="config.json",
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="preprocessor_config.json",
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="tokenizer.json",
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="vocabulary.json",
|
||||
)
|
||||
model = WhisperModel(
|
||||
model_size_or_path=os.path.dirname(model_path),
|
||||
device="auto",
|
||||
local_files_only=True,
|
||||
)
|
||||
print("模型加载成功,开始识别")
|
||||
sys.stdout.flush()
|
||||
segments, info = model.transcribe(
|
||||
mp,
|
||||
beam_size=5,
|
||||
language="zh",
|
||||
vad_filter=True,
|
||||
vad_parameters=dict(min_silence_duration_ms=1000),
|
||||
)
|
||||
tmp_text = ""
|
||||
for segment in segments:
|
||||
tmp_text += segment.text + "。"
|
||||
print(mp + "识别完成")
|
||||
sys.stdout.flush()
|
||||
text.append(tmp_text)
|
||||
|
||||
# 数据写出
|
||||
sys.stdout.flush()
|
||||
tools = public_tools.PublicTools()
|
||||
tools.write_to_file(text, os.path.join(out_folder, name + ".txt"))
|
||||
sys.stdout.flush()
|
||||
|
||||
|
||||
# GetTextTask(
|
||||
# "C:\\Users\\27698\\Desktop\\测试\\mjTest",
|
||||
# "C:\\Users\\27698\\Desktop\\测试\\mjTest\\data\\frame\\00001.mp4",
|
||||
# "00001",
|
||||
# )
|
||||
|
||||
if sys.argv[1] == "-ts":
|
||||
GetText(
|
||||
sys.argv[2],
|
||||
sys.argv[3],
|
||||
)
|
||||
elif sys.argv[1] == "-t":
|
||||
GetTextTask(
|
||||
sys.argv[2],
|
||||
sys.argv[3],
|
||||
sys.argv[4],
|
||||
)
|
||||
else:
|
||||
print("Params: <runtime-config.json>")
|
||||
exit(0)
|
||||
@@ -0,0 +1,50 @@
|
||||
# -*- mode: python ; coding: utf-8 -*-
|
||||
|
||||
from PyInstaller.building.datastruct import Tree
|
||||
from PyInstaller.utils.hooks import get_package_paths
|
||||
|
||||
PACKAGE_DIRECTORY = get_package_paths('faster_whisper')[1]
|
||||
datas = [(PACKAGE_DIRECTORY, 'faster_whisper')]
|
||||
|
||||
|
||||
a = Analysis(
|
||||
['local_whisper.py'],
|
||||
pathex=[],
|
||||
binaries=[],
|
||||
datas=[],
|
||||
hiddenimports=[],
|
||||
hookspath=[],
|
||||
hooksconfig={},
|
||||
runtime_hooks=[],
|
||||
excludes=[],
|
||||
noarchive=False,
|
||||
optimize=0,
|
||||
)
|
||||
pyz = PYZ(a.pure)
|
||||
|
||||
exe = EXE(
|
||||
pyz,
|
||||
a.scripts,
|
||||
[],
|
||||
exclude_binaries=True,
|
||||
name='local_whisper',
|
||||
debug=False,
|
||||
bootloader_ignore_signals=False,
|
||||
strip=False,
|
||||
upx=True,
|
||||
console=True,
|
||||
disable_windowed_traceback=False,
|
||||
argv_emulation=False,
|
||||
target_arch=None,
|
||||
codesign_identity=None,
|
||||
entitlements_file=None,
|
||||
)
|
||||
coll = COLLECT(
|
||||
exe,
|
||||
a.binaries,
|
||||
a.datas,
|
||||
strip=False,
|
||||
upx=True,
|
||||
upx_exclude=[],
|
||||
name='local_whisper',
|
||||
)
|
||||
@@ -0,0 +1,351 @@
|
||||
# 读取文件的方法
|
||||
import json
|
||||
import os
|
||||
import win32api
|
||||
import win32con
|
||||
import pywintypes
|
||||
import shutil
|
||||
import re
|
||||
|
||||
|
||||
class PublicTools:
|
||||
"""
|
||||
一些公用的基础方法
|
||||
"""
|
||||
|
||||
def delete_path(self, path):
|
||||
"""
|
||||
删除指定路径的文件或者是文件夹
|
||||
"""
|
||||
# 检查路径是否存在
|
||||
if not os.path.exists(path):
|
||||
return
|
||||
|
||||
# 检查路径是文件还是文件夹
|
||||
if os.path.isfile(path):
|
||||
# 是文件,执行删除
|
||||
try:
|
||||
os.remove(path)
|
||||
except Exception as e:
|
||||
raise e
|
||||
elif os.path.isdir(path):
|
||||
# 是文件夹,执行删除
|
||||
try:
|
||||
shutil.rmtree(path)
|
||||
except Exception as e:
|
||||
raise e
|
||||
else:
|
||||
raise
|
||||
|
||||
def list_files_by_extension(self, folder_path, extension):
|
||||
"""
|
||||
读取指定文件夹下面的所有的指定拓展文件命的文件列表
|
||||
"""
|
||||
file_list = []
|
||||
for root, dirs, files in os.walk(folder_path):
|
||||
for file in files:
|
||||
if file.endswith(extension):
|
||||
file_list.append(os.path.join(root, file))
|
||||
elif file.endswith(extension.upper()):
|
||||
file_list.append(os.path.join(root, file))
|
||||
return file_list
|
||||
|
||||
def get_fonts_from_registry(self, key_path):
|
||||
"""
|
||||
获取注册表中安装的字体文件
|
||||
"""
|
||||
font_names = []
|
||||
try:
|
||||
key = win32api.RegOpenKeyEx(
|
||||
(
|
||||
win32con.HKEY_LOCAL_MACHINE
|
||||
if "HKEY_LOCAL_MACHINE" in key_path
|
||||
else win32con.HKEY_CURRENT_USER
|
||||
),
|
||||
key_path.split("\\", 1)[1],
|
||||
0,
|
||||
win32con.KEY_READ,
|
||||
)
|
||||
i = 0
|
||||
while True:
|
||||
try:
|
||||
value = win32api.RegEnumValue(key, i)
|
||||
font_name = value[0]
|
||||
# 使用正则表达式移除括号及其内容
|
||||
font_name = re.sub(r"\s*\([^)]*\)$", "", font_name)
|
||||
font_names.append(font_name)
|
||||
i += 1
|
||||
except pywintypes.error as e:
|
||||
if e.winerror == 259: # 没有更多的数据
|
||||
break
|
||||
else:
|
||||
raise
|
||||
finally:
|
||||
try:
|
||||
win32api.RegCloseKey(key)
|
||||
except:
|
||||
pass
|
||||
return font_names
|
||||
|
||||
def get_installed_fonts(self):
|
||||
"""
|
||||
获取字体文件名称并返回
|
||||
"""
|
||||
system_fonts = self.get_fonts_from_registry(
|
||||
"HKEY_LOCAL_MACHINE\\SOFTWARE\\Microsoft\\Windows NT\\CurrentVersion\\Fonts"
|
||||
)
|
||||
user_fonts = self.get_fonts_from_registry(
|
||||
"HKEY_CURRENT_USER\\Software\\Microsoft\\Windows NT\\CurrentVersion\\Fonts"
|
||||
)
|
||||
all_fonts = list(set(system_fonts + user_fonts)) # 合并并去重
|
||||
return all_fonts
|
||||
|
||||
# 将RRGGBB转换为BBGGRR
|
||||
def convert_rrggbb_to_bbggrr(self, rrggbb):
|
||||
"""
|
||||
将RRGGBB转换为BBGGRR
|
||||
"""
|
||||
if len(rrggbb) == 7:
|
||||
rr = rrggbb[1:3]
|
||||
gg = rrggbb[3:5]
|
||||
bb = rrggbb[5:7]
|
||||
return bb + gg + rr
|
||||
else:
|
||||
return "Invalid input"
|
||||
|
||||
def write_to_file(self, arr, filename):
|
||||
with open(filename, "w",encoding='utf-8') as f:
|
||||
for item in arr:
|
||||
f.write("%s\n" % item)
|
||||
|
||||
|
||||
# 读取文件
|
||||
def read_file(fileType):
|
||||
txt_path = input(f"输入{fileType}文件路径:")
|
||||
txt_path = remove_prefix_and_suffix(txt_path, '"', '"')
|
||||
while txt_path.strip() == "":
|
||||
txt_path = input(f"输入{fileType}文件路径:")
|
||||
|
||||
while os.path.exists(txt_path) == False:
|
||||
print("文件路径不存在错误:")
|
||||
txt_path = input(f"输入{fileType}文件路径:")
|
||||
txt_path = remove_prefix_and_suffix(txt_path, '"', '"')
|
||||
return txt_path
|
||||
|
||||
|
||||
def format_time_ms(milliseconds):
|
||||
"""
|
||||
时间转换将ms->小时:分钟:秒.毫秒格式
|
||||
"""
|
||||
seconds = milliseconds / 1000
|
||||
# 计算小时、分钟和秒
|
||||
hours = int(seconds // 3600)
|
||||
minutes = int((seconds % 3600) // 60)
|
||||
seconds = seconds % 60
|
||||
# 格式化字符串
|
||||
# 使用`%02d`确保小时和分钟总是显示为两位数,`%.2f`确保秒数显示两位小数
|
||||
formatted_time = f"{hours}:{minutes:02d}:{seconds:05.2f}"
|
||||
return formatted_time
|
||||
|
||||
|
||||
# 删除满足条件的开头和结尾
|
||||
def remove_prefix_and_suffix(input_str, prefix_to_remove, suffix_to_remove):
|
||||
if input_str.startswith(prefix_to_remove):
|
||||
# 删除开头
|
||||
input_str = input_str[len(prefix_to_remove) :]
|
||||
|
||||
if input_str.endswith(suffix_to_remove):
|
||||
# 删除结尾
|
||||
input_str = input_str[: -len(suffix_to_remove)]
|
||||
|
||||
return input_str
|
||||
|
||||
|
||||
# 判断文件夹下面是不是有特定的文件夹
|
||||
def check_if_folder_exists(parent_folder, target_folder_name):
|
||||
# 获取文件夹列表
|
||||
subfolders = [f.name for f in os.scandir(parent_folder) if f.is_dir()]
|
||||
|
||||
# 检查特定文件夹是否存在
|
||||
if target_folder_name in subfolders:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
# 检查指定文件夹中是否存在特定文件。
|
||||
def file_exists_in_folder(folder_path: str, file_name: str) -> bool:
|
||||
# 构建完整的文件路径
|
||||
file_path = os.path.join(folder_path, file_name)
|
||||
|
||||
# 返回文件是否存在
|
||||
return os.path.isfile(file_path)
|
||||
|
||||
|
||||
# 秒数转换,保留一位小数
|
||||
def convert_to_seconds(number, count):
|
||||
seconds = number / 1000000
|
||||
rounded_number = round(seconds, count)
|
||||
return rounded_number
|
||||
|
||||
|
||||
def is_empty(obj):
|
||||
if obj is None:
|
||||
return True
|
||||
elif isinstance(obj, str):
|
||||
return len(obj) == 0
|
||||
elif isinstance(obj, list):
|
||||
return len(obj) == 0
|
||||
elif isinstance(obj, dict):
|
||||
return len(obj) == 0
|
||||
return False
|
||||
|
||||
|
||||
def opt_dict(obj, key, default=None):
|
||||
if obj is None:
|
||||
return default
|
||||
if key in obj:
|
||||
v = obj[key]
|
||||
if not is_empty(v):
|
||||
return v
|
||||
return default
|
||||
|
||||
|
||||
def read_config(path, webui=True):
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
runtime_config = json.load(f)
|
||||
|
||||
if "config" not in runtime_config:
|
||||
print("no filed 'config' in json")
|
||||
return None
|
||||
|
||||
config = runtime_config["config"]
|
||||
if "webui" not in config:
|
||||
print("no filed 'webui' in 'config'")
|
||||
return None
|
||||
|
||||
setting_config_path = config["setting"]
|
||||
if not os.path.exists(setting_config_path):
|
||||
setting_config_path = "config/" + setting_config_path
|
||||
if not os.path.exists(setting_config_path):
|
||||
setting_config_path = "../" + setting_config_path
|
||||
|
||||
# read config
|
||||
with open(setting_config_path, "r", encoding="utf-8") as f:
|
||||
setting_config = json.load(f)
|
||||
|
||||
# set workspace parent:根目录
|
||||
if "workspace" in setting_config:
|
||||
setting_config["workspace"]["parent"] = runtime_config["workspace"]
|
||||
else:
|
||||
setting_config["workspace"] = {"parent": runtime_config["workspace"]}
|
||||
setting_config["video"] = opt_dict(runtime_config, "video")
|
||||
|
||||
# merge setting config
|
||||
if "setting" in config:
|
||||
setting_config.update(runtime_config["setting"])
|
||||
|
||||
# webui config
|
||||
if webui:
|
||||
webui_config_path = config["webui"]
|
||||
if not os.path.exists(webui_config_path):
|
||||
webui_config_path = "config/webui/" + webui_config_path
|
||||
if not os.path.exists(webui_config_path):
|
||||
webui_config_path = "../" + webui_config_path
|
||||
|
||||
with open(webui_config_path, "r", encoding="utf-8") as f:
|
||||
webui_config = json.load(f)
|
||||
|
||||
# merge webui config
|
||||
if "webui" in runtime_config:
|
||||
webui_config.update(runtime_config["webui"])
|
||||
|
||||
return webui_config, setting_config
|
||||
return setting_config
|
||||
|
||||
|
||||
TAG_MODE_NONE = ""
|
||||
|
||||
|
||||
# 工作路径
|
||||
class Workspace:
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
root: str,
|
||||
input: str,
|
||||
output: str,
|
||||
input_crop: str,
|
||||
output_crop: str,
|
||||
input_tag: str,
|
||||
input_mask: str,
|
||||
input_crop_mask: str,
|
||||
crop_info: str,
|
||||
):
|
||||
self.root = root
|
||||
self.input = input
|
||||
self.output = output
|
||||
self.input_crop = input_crop
|
||||
self.output_crop = output_crop
|
||||
self.input_tag = input_tag
|
||||
self.input_mask = input_mask
|
||||
self.input_crop_mask = input_crop_mask
|
||||
self.crop_info = crop_info
|
||||
|
||||
|
||||
# 定义一个倍数函数
|
||||
def round_up(num, mul):
|
||||
return (num // mul + 1) * mul
|
||||
|
||||
|
||||
class SettingConfig:
|
||||
|
||||
def __init__(self, config: dict, workParent):
|
||||
self.config = config
|
||||
self.webui_work_api = None
|
||||
self.workParent = workParent
|
||||
|
||||
def to_dict(self):
|
||||
return self.__dict__
|
||||
|
||||
def get_tag_mode(self):
|
||||
tag_cfg = opt_dict(self.config, "tag")
|
||||
return opt_dict(tag_cfg, "mode", TAG_MODE_NONE)
|
||||
|
||||
def get_tag_actions(self):
|
||||
tag_cfg = opt_dict(self.config, "tag")
|
||||
return opt_dict(tag_cfg, "actions", [])
|
||||
|
||||
def get_workspace_config(self) -> Workspace:
|
||||
workspace_config = opt_dict(self.config, "workspace")
|
||||
tmp_config = opt_dict(workspace_config, "tmp")
|
||||
|
||||
input = opt_dict(workspace_config, "input", "input")
|
||||
output = opt_dict(workspace_config, "output", "output")
|
||||
workspace_parent = self.workParent
|
||||
|
||||
tmp_parent = opt_dict(tmp_config, "parent", "tmp")
|
||||
input_crop = opt_dict(tmp_config, "input_crop", "input_crop")
|
||||
output_crop = opt_dict(tmp_config, "output_crop", "output_crop")
|
||||
input_tag = opt_dict(tmp_config, "input_tag", "input_crop")
|
||||
input_mask = opt_dict(tmp_config, "input_mask", "input_mask")
|
||||
input_crop_mask = opt_dict(tmp_config, "input_crop_mask", "input_crop_mask")
|
||||
crop_info = opt_dict(tmp_config, "crop_info", "crop_info.txt")
|
||||
|
||||
tmp_path = os.path.join(workspace_parent, tmp_parent)
|
||||
|
||||
return Workspace(
|
||||
workspace_parent,
|
||||
os.path.join(workspace_parent, input),
|
||||
os.path.join(workspace_parent, output),
|
||||
os.path.join(tmp_path, input_crop),
|
||||
os.path.join(tmp_path, output_crop),
|
||||
os.path.join(tmp_path, input_tag),
|
||||
os.path.join(tmp_path, input_mask),
|
||||
os.path.join(tmp_path, input_crop_mask),
|
||||
os.path.join(tmp_path, crop_info),
|
||||
)
|
||||
|
||||
def enable_tag(self):
|
||||
tag_cfg = opt_dict(self.config, "tag")
|
||||
return opt_dict(tag_cfg, "enable", True)
|
||||
@@ -0,0 +1,307 @@
|
||||
# pip install scenedetect opencv-python -i https://pypi.tuna.tsinghua.edu.cn/simple
|
||||
|
||||
from scenedetect.video_manager import VideoManager
|
||||
from scenedetect.scene_manager import SceneManager
|
||||
from scenedetect.stats_manager import StatsManager
|
||||
from scenedetect.detectors.content_detector import ContentDetector
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import subprocess
|
||||
from huggingface_hub import hf_hub_download
|
||||
from faster_whisper import WhisperModel
|
||||
from pathlib import Path
|
||||
import public_tools
|
||||
|
||||
# 获取智能画面分割的时间或者秒数
|
||||
def find_scenes(video_path, sensitivity):
|
||||
print(
|
||||
"正在计算分镜数据" + "sensitivity:" + str(sensitivity) + "path : " + video_path
|
||||
)
|
||||
sys.stdout.flush()
|
||||
video_manager = VideoManager([video_path])
|
||||
stats_manager = StatsManager()
|
||||
scene_manager = SceneManager(stats_manager)
|
||||
|
||||
# 使用contect-detector
|
||||
scene_manager.add_detector(ContentDetector(threshold=float(sensitivity)))
|
||||
|
||||
shijian_list = []
|
||||
|
||||
try:
|
||||
video_manager.set_downscale_factor()
|
||||
video_manager.start()
|
||||
scene_manager.detect_scenes(frame_source=video_manager)
|
||||
scene_list = scene_manager.get_scene_list()
|
||||
print("分镜数据列表:")
|
||||
sys.stdout.flush()
|
||||
for i, scene in enumerate(scene_list):
|
||||
shijian_list.append([scene[0].get_timecode(), scene[1].get_timecode()])
|
||||
print(
|
||||
"Scene %2d: Start %s / Frame %d, End %s / Frame %d"
|
||||
% (
|
||||
i + 1,
|
||||
scene[0].get_timecode(),
|
||||
scene[0].get_frames(),
|
||||
scene[1].get_timecode(),
|
||||
scene[1].get_frames(),
|
||||
)
|
||||
)
|
||||
sys.stdout.flush()
|
||||
finally:
|
||||
video_manager.release()
|
||||
|
||||
return shijian_list
|
||||
|
||||
|
||||
# 如果不存在就创建
|
||||
def createDir(file_dir):
|
||||
# 如果不存在文件夹,就创建
|
||||
if not os.path.isdir(file_dir):
|
||||
os.mkdir(file_dir)
|
||||
|
||||
|
||||
# 切分一个视频
|
||||
def ClipVideo(video_path, out_folder, image_out_folder, sensitivity, gpu_type):
|
||||
shijian_list = find_scenes(video_path, sensitivity) # 多组时间列表
|
||||
shijian_list_len = len(shijian_list)
|
||||
|
||||
print("总共有%s个场景" % str(shijian_list_len))
|
||||
sys.stdout.flush()
|
||||
video_list = []
|
||||
for i in range(0, shijian_list_len):
|
||||
start_time_str = shijian_list[i][0]
|
||||
end_time_str = shijian_list[i][1]
|
||||
|
||||
print("开始输出第" + str(i + 1) + "个分镜")
|
||||
video_name = "{:05d}".format(i + 1)
|
||||
out_video_file = os.path.join(out_folder, video_name + ".mp4")
|
||||
sys.stdout.flush()
|
||||
video_list.append(
|
||||
{
|
||||
"start_time_str": start_time_str,
|
||||
"end_time_str": end_time_str,
|
||||
"out_video_file": out_video_file,
|
||||
"video_name": video_name,
|
||||
}
|
||||
)
|
||||
|
||||
# 使用 ffmpeg 裁剪视频
|
||||
command = []
|
||||
command.append("ffmpeg")
|
||||
command.append("-i")
|
||||
command.append(video_path)
|
||||
command.append("-ss")
|
||||
command.append(start_time_str)
|
||||
command.append("-to")
|
||||
command.append(end_time_str)
|
||||
command.append("-c:v")
|
||||
|
||||
if gpu_type == "NVIDIA":
|
||||
command.append("h264_nvenc")
|
||||
elif gpu_type == "AMD":
|
||||
command.append("h264_amf")
|
||||
else:
|
||||
command.append("libx264")
|
||||
|
||||
command.append("-preset")
|
||||
command.append("fast")
|
||||
command.append("-c:a")
|
||||
command.append("copy")
|
||||
command.append(out_video_file)
|
||||
command.append("-loglevel")
|
||||
command.append("error")
|
||||
|
||||
subprocess.run(
|
||||
command,
|
||||
check=True,
|
||||
stderr=subprocess.PIPE,
|
||||
)
|
||||
|
||||
print("分镜输出完成。开始抽帧")
|
||||
sys.stdout.flush()
|
||||
for vi in video_list:
|
||||
h, m, s = vi["start_time_str"].split(":")
|
||||
start_seconds = int(h) * 3600 + int(m) * 60 + float(s)
|
||||
|
||||
h, m, s = vi["end_time_str"].split(":")
|
||||
end_seconds = int(h) * 3600 + int(m) * 60 + float(s)
|
||||
print("正在抽帧:" + vi["video_name"])
|
||||
sys.stdout.flush()
|
||||
subprocess.run(
|
||||
[
|
||||
"ffmpeg",
|
||||
"-ss",
|
||||
str((end_seconds - start_seconds) / 2),
|
||||
"-i",
|
||||
vi["out_video_file"],
|
||||
"-frames:v",
|
||||
"1",
|
||||
os.path.join(image_out_folder, vi["video_name"] + ".png"),
|
||||
"-loglevel",
|
||||
"error",
|
||||
]
|
||||
)
|
||||
|
||||
print("抽帧完成,开始识别文案")
|
||||
sys.stdout.flush()
|
||||
return video_list
|
||||
|
||||
|
||||
def SplitAudio(video_out_folder, video_list):
|
||||
# ffmpeg -i input_file.mp4 -vn -ab 128k output_file.mp3
|
||||
print("正在分离音频!!")
|
||||
mp3_list = []
|
||||
sys.stdout.flush()
|
||||
for v in video_list:
|
||||
mp3_path = os.path.join(video_out_folder, v["video_name"] + ".mp3")
|
||||
mp3_list.append(mp3_path)
|
||||
subprocess.run(
|
||||
[
|
||||
"ffmpeg",
|
||||
"-i",
|
||||
v["out_video_file"],
|
||||
"-vn",
|
||||
"-ab",
|
||||
"128k",
|
||||
mp3_path,
|
||||
"-loglevel",
|
||||
"error",
|
||||
],
|
||||
check=True,
|
||||
)
|
||||
return mp3_list
|
||||
|
||||
|
||||
def GetText(out_folder, mp3_list):
|
||||
text = []
|
||||
# 先获取模型
|
||||
print("正在下载或加载模型")
|
||||
sys.stdout.flush()
|
||||
model_path = Path(
|
||||
hf_hub_download(repo_id="Systran/faster-whisper-large-v3", filename="model.bin")
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="config.json",
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="preprocessor_config.json",
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="tokenizer.json",
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="vocabulary.json",
|
||||
)
|
||||
model = WhisperModel(
|
||||
model_size_or_path=os.path.dirname(model_path),
|
||||
device="auto",
|
||||
local_files_only=True,
|
||||
)
|
||||
print("模型加载成功,开始识别")
|
||||
sys.stdout.flush()
|
||||
for mp in mp3_list:
|
||||
segments, info = model.transcribe(
|
||||
mp,
|
||||
beam_size=5,
|
||||
language="zh",
|
||||
vad_filter=True,
|
||||
vad_parameters=dict(min_silence_duration_ms=1000),
|
||||
)
|
||||
tmp_text = ""
|
||||
for segment in segments:
|
||||
tmp_text += segment.text + "。"
|
||||
print(mp + "识别完成")
|
||||
sys.stdout.flush()
|
||||
text.append(tmp_text)
|
||||
|
||||
# 数据写出
|
||||
print("文本全部识别成功,正在写出")
|
||||
sys.stdout.flush()
|
||||
tools = public_tools.PublicTools()
|
||||
tools.write_to_file(text, os.path.join(out_folder, "文案.txt"))
|
||||
print("写出完成")
|
||||
sys.stdout.flush()
|
||||
|
||||
|
||||
def GetTextTask(out_folder, mp, name):
|
||||
text = []
|
||||
# 先获取模型
|
||||
print("正在下载或加载模型")
|
||||
sys.stdout.flush()
|
||||
model_path = Path(
|
||||
hf_hub_download(repo_id="Systran/faster-whisper-large-v3", filename="model.bin")
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="config.json",
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="preprocessor_config.json",
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="tokenizer.json",
|
||||
)
|
||||
hf_hub_download(
|
||||
repo_id="Systran/faster-whisper-large-v3",
|
||||
filename="vocabulary.json",
|
||||
)
|
||||
model = WhisperModel(
|
||||
model_size_or_path=os.path.dirname(model_path),
|
||||
device="auto",
|
||||
local_files_only=True,
|
||||
)
|
||||
print("模型加载成功,开始识别")
|
||||
sys.stdout.flush()
|
||||
segments, info = model.transcribe(
|
||||
mp,
|
||||
beam_size=5,
|
||||
language="zh",
|
||||
vad_filter=True,
|
||||
vad_parameters=dict(min_silence_duration_ms=1000),
|
||||
)
|
||||
tmp_text = ""
|
||||
for segment in segments:
|
||||
tmp_text += segment.text + "。"
|
||||
print(mp + "识别完成")
|
||||
sys.stdout.flush()
|
||||
text.append(tmp_text)
|
||||
|
||||
# 数据写出
|
||||
sys.stdout.flush()
|
||||
tools = public_tools.PublicTools()
|
||||
tools.write_to_file(text, os.path.join(out_folder, name + ".txt"))
|
||||
sys.stdout.flush()
|
||||
|
||||
|
||||
def get_fram(video_path, out_path, sensitivity):
|
||||
try:
|
||||
shijian_list = find_scenes(video_path, sensitivity) # 多组时间列表
|
||||
print("总共有%s个场景" % str(len(shijian_list)))
|
||||
print("开始输出json")
|
||||
print(shijian_list)
|
||||
# 将数组中的消息写道json文件中
|
||||
with open(out_path, "w") as file:
|
||||
# 将数组写入到指定的json文件
|
||||
json.dump(shijian_list, file)
|
||||
print("输出完成")
|
||||
except Exception as e:
|
||||
print("出现错误" + str(e))
|
||||
exit(0)
|
||||
|
||||
|
||||
def init(video_path, video_out_folder, image_out_folder, sensitivity, gpu_type):
|
||||
v_l = ClipVideo(
|
||||
video_path, video_out_folder, image_out_folder, sensitivity, gpu_type
|
||||
)
|
||||
|
||||
# 开始分离音频
|
||||
m_l = SplitAudio(video_out_folder, v_l)
|
||||
# 开始识别字幕
|
||||
GetText(os.path.dirname(video_out_folder), m_l)
|
||||
Reference in New Issue
Block a user