| 1 | import json |
| 2 | import locale |
| 3 | import os |
| 4 | import re |
| 5 | import shutil |
| 6 | from functools import lru_cache |
| 7 | from pathlib import Path |
| 8 | import threading |
| 9 | from typing import Any |
| 10 | from uuid import uuid4 |
| 11 | |
| 12 | from loguru import logger |
| 13 | |
| 14 | from app.models import const |
| 15 | |
| 16 | |
| 17 | def get_response(status: int, data: Any = None, message: str = ""): |
| 18 | obj = { |
| 19 | "status": status, |
| 20 | } |
| 21 | if data: |
| 22 | obj["data"] = data |
| 23 | if message: |
| 24 | obj["message"] = message |
| 25 | return obj |
| 26 | |
| 27 | |
| 28 | def to_json(obj): |
| 29 | try: |
| 30 | # Define a helper function to handle different types of objects |
| 31 | def serialize(o): |
| 32 | # If the object is a serializable type, return it directly |
| 33 | if isinstance(o, (int, float, bool, str)) or o is None: |
| 34 | return o |
| 35 | # If the object is binary data, convert it to a base64-encoded string |
| 36 | elif isinstance(o, bytes): |
| 37 | return "*** binary data ***" |
| 38 | # If the object is a dictionary, recursively process each key-value pair |
| 39 | elif isinstance(o, dict): |
| 40 | return {k: serialize(v) for k, v in o.items()} |
| 41 | # If the object is a list or tuple, recursively process each element |
| 42 | elif isinstance(o, (list, tuple)): |
| 43 | return [serialize(item) for item in o] |
| 44 | # If the object is a custom type, attempt to return its __dict__ attribute |
| 45 | elif hasattr(o, "__dict__"): |
| 46 | return serialize(o.__dict__) |
| 47 | # Return None for other cases (or choose to raise an exception) |
| 48 | else: |
| 49 | return None |
| 50 | |
| 51 | # Use the serialize function to process the input object |
| 52 | serialized_obj = serialize(obj) |
| 53 | |
| 54 | # Serialize the processed object into a JSON string |
| 55 | return json.dumps(serialized_obj, ensure_ascii=False, indent=4) |
| 56 | except Exception as e: |
| 57 | logger.error(f"failed to serialize object to json: {str(e)}") |
| 58 | return None |
| 59 | |
| 60 | |
| 61 | def get_uuid(remove_hyphen: bool = False): |
| 62 | u = str(uuid4()) |
| 63 | if remove_hyphen: |
| 64 | u = u.replace("-", "") |
| 65 | return u |
| 66 | |
| 67 | |
| 68 | def root_dir(): |
| 69 | return os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))) |
| 70 | |
| 71 | |
| 72 | def storage_dir(sub_dir: str = "", create: bool = False): |
| 73 | d = os.path.join(root_dir(), "storage") |
| 74 | if sub_dir: |
| 75 | d = os.path.join(d, sub_dir) |
| 76 | if create and not os.path.exists(d): |
| 77 | os.makedirs(d) |
| 78 | |
| 79 | return d |
| 80 | |
| 81 | |
| 82 | def resource_dir(sub_dir: str = ""): |
| 83 | d = os.path.join(root_dir(), "resource") |
| 84 | if sub_dir: |
| 85 | d = os.path.join(d, sub_dir) |
| 86 | return d |
| 87 | |
| 88 | |
| 89 | def task_dir(sub_dir: str = ""): |
| 90 | d = os.path.join(storage_dir(), "tasks") |
| 91 | if sub_dir: |
| 92 | d = os.path.join(d, sub_dir) |
| 93 | if not os.path.exists(d): |
| 94 | os.makedirs(d) |
| 95 | return d |
| 96 | |
| 97 | |
| 98 | def font_dir(sub_dir: str = ""): |
| 99 | d = resource_dir("fonts") |
| 100 | if sub_dir: |
| 101 | d = os.path.join(d, sub_dir) |
| 102 | if not os.path.exists(d): |
| 103 | os.makedirs(d) |
| 104 | return d |
| 105 | |
| 106 | |
| 107 | def song_dir(sub_dir: str = ""): |
| 108 | d = resource_dir("songs") |
| 109 | if sub_dir: |
| 110 | d = os.path.join(d, sub_dir) |
| 111 | if not os.path.exists(d): |
| 112 | os.makedirs(d) |
| 113 | return d |
| 114 | |
| 115 | |
| 116 | def public_dir(sub_dir: str = ""): |
| 117 | d = resource_dir("public") |
| 118 | if sub_dir: |
| 119 | d = os.path.join(d, sub_dir) |
| 120 | if not os.path.exists(d): |
| 121 | os.makedirs(d) |
| 122 | return d |
| 123 | |
| 124 | |
| 125 | def get_ffmpeg_binary() -> str: |
| 126 | """ |
| 127 | 解析当前进程应该使用的 FFmpeg 可执行文件。 |
| 128 | |
| 129 | 增加原因： |
| 130 | 1. 视频编码、静音音频生成、pydub 音频转码都依赖 FFmpeg； |
| 131 | 2. Windows 便携包、Docker 和用户自定义安装目录经常出现 PATH 不一致； |
| 132 | 3. 集中解析可以让所有调用方使用同一套优先级，减少某条链路能跑、 |
| 133 | 另一条链路找不到 FFmpeg 的现场问题。 |
| 134 | |
| 135 | 优先级： |
| 136 | 1. IMAGEIO_FFMPEG_EXE：MoviePy/imageio 约定的显式配置； |
| 137 | 2. 系统 PATH 中的 ffmpeg； |
| 138 | 3. imageio-ffmpeg 依赖提供的内置二进制； |
| 139 | 4. 字符串 "ffmpeg" 兜底，交给 subprocess 在运行时暴露更具体错误。 |
| 140 | """ |
| 141 | configured_ffmpeg = os.environ.get("IMAGEIO_FFMPEG_EXE") |
| 142 | if configured_ffmpeg: |
| 143 | return configured_ffmpeg |
| 144 | |
| 145 | system_ffmpeg = shutil.which("ffmpeg") |
| 146 | if system_ffmpeg: |
| 147 | return system_ffmpeg |
| 148 | |
| 149 | try: |
| 150 | import imageio_ffmpeg |
| 151 | |
| 152 | bundled_ffmpeg = imageio_ffmpeg.get_ffmpeg_exe() |
| 153 | if bundled_ffmpeg: |
| 154 | return bundled_ffmpeg |
| 155 | except Exception as exc: |
| 156 | logger.warning(f"failed to resolve bundled ffmpeg binary: {str(exc)}") |
| 157 | |
| 158 | return "ffmpeg" |
| 159 | |
| 160 | |
| 161 | def run_in_background(func, *args, **kwargs): |
| 162 | def run(): |
| 163 | try: |
| 164 | func(*args, **kwargs) |
| 165 | except Exception as e: |
| 166 | logger.error(f"run_in_background error: {e}", exc_info=True) |
| 167 | |
| 168 | thread = threading.Thread(target=run, daemon=False) |
| 169 | thread.start() |
| 170 | return thread |
| 171 | |
| 172 | |
| 173 | def time_convert_seconds_to_hmsm(seconds) -> str: |
| 174 | hours = int(seconds // 3600) |
| 175 | seconds = seconds % 3600 |
| 176 | minutes = int(seconds // 60) |
| 177 | milliseconds = int(seconds * 1000) % 1000 |
| 178 | seconds = int(seconds % 60) |
| 179 | return "{:02d}:{:02d}:{:02d},{:03d}".format(hours, minutes, seconds, milliseconds) |
| 180 | |
| 181 | |
| 182 | def text_to_srt(idx: int, msg: str, start_time: float, end_time: float) -> str: |
| 183 | start_time = time_convert_seconds_to_hmsm(start_time) |
| 184 | end_time = time_convert_seconds_to_hmsm(end_time) |
| 185 | srt = """%d |
| 186 | %s --> %s |
| 187 | %s |
| 188 | """ % ( |
| 189 | idx, |
| 190 | start_time, |
| 191 | end_time, |
| 192 | msg, |
| 193 | ) |
| 194 | return srt |
| 195 | |
| 196 | |
| 197 | def str_contains_punctuation(word): |
| 198 | for p in const.PUNCTUATIONS: |
| 199 | if p in word: |
| 200 | return True |
| 201 | return False |
| 202 | |
| 203 | |
| 204 | def split_string_by_punctuations(s): |
| 205 | result = [] |
| 206 | txt = "" |
| 207 | |
| 208 | previous_char = "" |
| 209 | next_char = "" |
| 210 | for i in range(len(s)): |
| 211 | char = s[i] |
| 212 | if char == "\n": |
| 213 | result.append(txt.strip()) |
| 214 | txt = "" |
| 215 | continue |
| 216 | |
| 217 | if i > 0: |
| 218 | previous_char = s[i - 1] |
| 219 | if i < len(s) - 1: |
| 220 | next_char = s[i + 1] |
| 221 | |
| 222 | if char == "." and previous_char.isdigit() and next_char.isdigit(): |
| 223 | # # In the case of "withdraw 10,000, charged at 2.5% fee", the dot in "2.5" should not be treated as a line break marker |
| 224 | txt += char |
| 225 | continue |
| 226 | |
| 227 | if char == "," and previous_char.isdigit() and next_char.isdigit(): |
| 228 | # 英文数字里的千分位逗号不是断句符，例如 "1,000 years"。 |
| 229 | # Edge TTS 的 word boundary 通常会把这种数字整体作为连续内容返回； |
| 230 | # 如果这里拆成 "1" 和 "000 years"，后续字幕聚合会无法匹配脚本原文， |
| 231 | # 进而错误回退到 Whisper。 |
| 232 | txt += char |
| 233 | continue |
| 234 | |
| 235 | if char not in const.PUNCTUATIONS: |
| 236 | txt += char |
| 237 | else: |
| 238 | result.append(txt.strip()) |
| 239 | txt = "" |
| 240 | result.append(txt.strip()) |
| 241 | # filter empty string |
| 242 | result = list(filter(None, result)) |
| 243 | return result |
| 244 | |
| 245 | |
| 246 | def normalize_script_for_subtitle_matching(video_script: str) -> str: |
| 247 | """ |
| 248 | 清理字幕匹配前的脚本文本。 |
| 249 | |
| 250 | 用户可能手动输入 Markdown 分隔符、标题强调或 `_` 这类格式符号。 |
| 251 | 这些字符通常不会出现在 TTS/Whisper 的识别结果里；如果继续参与 |
| 252 | 字幕逐行匹配，脚本行数量会大于真实字幕行数量，最终可能补出 |
| 253 | `00:00:00,000 --> 00:00:00,000`，导致剪辑软件无法导入 SRT。 |
| 254 | """ |
| 255 | video_script = video_script or "" |
| 256 | underscore_count = video_script.count("_") |
| 257 | video_script = video_script.replace("_", "") |
| 258 | cleaned_lines = [] |
| 259 | removed_separator_lines = 0 |
| 260 | for line in video_script.splitlines(): |
| 261 | line = line.strip() |
| 262 | # Markdown 分隔符或强调符号单独成行时不会被 TTS 朗读，必须从 |
| 263 | # 脚本行里移除，避免字幕聚合卡在这类“不可发声”的目标行上。 |
| 264 | if re.fullmatch(r"[-*_]{3,}", line): |
| 265 | removed_separator_lines += 1 |
| 266 | continue |
| 267 | cleaned_lines.append(line) |
| 268 | |
| 269 | normalized_script = "\n".join(cleaned_lines).strip() |
| 270 | if underscore_count or removed_separator_lines: |
| 271 | logger.debug( |
| 272 | "normalized script for subtitle matching, " |
| 273 | f"removed underscores: {underscore_count}, " |
| 274 | f"removed markdown separator lines: {removed_separator_lines}" |
| 275 | ) |
| 276 | return normalized_script |
| 277 | |
| 278 | |
| 279 | def md5(text): |
| 280 | import hashlib |
| 281 | |
| 282 | return hashlib.md5(text.encode("utf-8")).hexdigest() |
| 283 | |
| 284 | |
| 285 | def get_system_locale(): |
| 286 | try: |
| 287 | loc = locale.getdefaultlocale() |
| 288 | # zh_CN, zh_TW return zh |
| 289 | # en_US, en_GB return en |
| 290 | language_code = loc[0].split("_")[0] |
| 291 | return language_code |
| 292 | except Exception: |
| 293 | return "en" |
| 294 | |
| 295 | |
| 296 | @lru_cache(maxsize=None) |
| 297 | def load_locales(i18n_dir): |
| 298 | # WebUI 每次交互都会触发 Streamlit 重新执行脚本，语言文件运行期不会变化， |
| 299 | # 因此缓存解析结果，避免反复读取和解析所有 i18n JSON 文件。 |
| 300 | _locales = {} |
| 301 | for root, dirs, files in os.walk(i18n_dir): |
| 302 | for file in files: |
| 303 | if file.endswith(".json"): |
| 304 | lang = file.split(".")[0] |
| 305 | with open(os.path.join(root, file), "r", encoding="utf-8") as f: |
| 306 | _locales[lang] = json.loads(f.read()) |
| 307 | return _locales |
| 308 | |
| 309 | |
| 310 | def parse_extension(filename): |
| 311 | return Path(filename).suffix.lower().lstrip('.') |
| 312 |