#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ CXKJ 识别字幕 - 顶部三按钮：添加视频/音频/subtitle 轨道 - 渲染音频 → Qwen3ASRModel 语音识别 → 生成 SRT - 导入 SRT 字幕到时间线 使用位置：Workspace > Scripts > Utility > CXKJ_字幕识别 --worker 模式：conda python 调自身做语音识别 """ from __future__ import print_function import base64 import datetime import json import os import subprocess import sys import time import traceback # ============================================================ # 全局配置 — JSON 模板系统（对标 字幕配音 样板） # ============================================================ try: _SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) _SCRIPT_NAME = os.path.splitext(os.path.basename(os.path.abspath(__file__)))[0] except (NameError, TypeError): try: _SCRIPT_DIR = os.path.dirname(os.path.abspath(sys.argv[0])) _SCRIPT_NAME = os.path.splitext(os.path.basename(os.path.abspath(sys.argv[0])))[0] except (IndexError, TypeError, NameError): _SCRIPT_DIR = os.path.join(os.environ.get("APPDATA", os.path.expanduser("~")), "Blackmagic Design", "DaVinci Resolve", "Support", "Fusion", "Scripts", "Utility") _SCRIPT_NAME = "CXKJ_字幕识别_Win" _CONFIG_PATH = os.path.join(_SCRIPT_DIR, f"{_SCRIPT_NAME}_config.json") _CONFIG_TEMPLATE = """{ "_comment": "你是 Qwen3-ASR 配置助手。1.先问我 Qwen3 项目根目录路径 2.填入 ai_root（如 /Users/用户名/ai_studio）3.推导：python={ai_root}/.venv/bin/python3，model_path={ai_root}/Qwen3-ForcedAligner/model，asr_model_path={ai_root}/Qwen3-ASR-0.6B 4.temp_dir留空默认桌面 字幕识别缓存 5.所有路径用绝对路径，禁止~ 6.输出完整JSON让我直接粘贴回去", "ai_root": "", "python": "", "model_path": "", "asr_model_path": "", "temp_dir": "" }""" def _load_config(): cfg = {} try: with open(_CONFIG_PATH, "r", encoding="utf-8") as f: data = json.load(f) cfg = {k: v for k, v in data.items() if k != "_comment" and v} except (FileNotFoundError, json.JSONDecodeError): pass return cfg def _resolve(key, cfg, hardcoded, derive_func=None): val = cfg.get(key, "") if val: return val if hardcoded: return hardcoded root = cfg.get("ai_root", "") if root and derive_func: return derive_func(os.path.expanduser(root)) return hardcoded _CFG = _load_config() CONDA_PYTHON = _resolve("python", _CFG, "", lambda r: os.path.join(r, ".venv/bin/python3")) MODEL_PATH = _resolve("model_path", _CFG, "", lambda r: os.path.join(r, "Qwen3-ForcedAligner/model")) ASR_MODEL_PATH = _resolve("asr_model_path", _CFG, "", lambda r: os.path.join(r, "Qwen3-ASR-0.6B")) TEMP_DIR = os.path.expanduser(_resolve("temp_dir", _CFG, os.path.expanduser("~/Desktop/字幕识别缓存"), lambda r: os.path.join(r, "Desktop/字幕识别缓存"))) LOG_FILE = os.path.join(TEMP_DIR, "indextts_xml_resolve_generate.log") os.makedirs(TEMP_DIR, exist_ok=True) _script_file_path = None # ── Windows 控制台编码修复（GBK 环境输出 emoji/中文可能崩溃）── try: for _stream in (sys.stdout, sys.stderr): try: _stream.reconfigure(encoding="utf-8", errors="replace") except Exception: pass os.environ.setdefault("PYTHONIOENCODING", "utf-8") except Exception: pass def to_text(value): if isinstance(value, bytes): return value.decode("utf-8", "replace") return str(value) def append_log(text): with open(LOG_FILE, "ab") as f: f.write(to_text(text).encode("utf-8", "replace")) def ts(): return datetime.datetime.now().strftime("%Y%m%d_%H%M%S") def format_time_srt(seconds): h = int(seconds // 3600) m = int((seconds % 3600) // 60) s = int(seconds % 60) ms = int((seconds - int(seconds)) * 1000) return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}" # ════════════════════════════════════════════════════════ class LogBuffer: def __init__(self): self._lines = [] self._callback = None def set_callback(self, cb): self._callback = cb def write(self, text, tag=None): t = datetime.datetime.now().strftime("%H:%M:%S") line = f"[{t}] {text}" self._lines.append(line) append_log(line + "\n") if self._callback: self._callback(line, tag) def get_all(self): return "\n".join(self._lines) log = LogBuffer() # ════════════════════════════════════════════════════════ # ——— WORKER MODE ——— 对齐核心逻辑 # ════════════════════════════════════════════════════════ def worker_main(): """在外部 conda 环境中执行语音识别""" if len(sys.argv) < 4: print("[WORKER] 用法: ... --worker <音频> <输出SRT> [语言] [最大字数]", flush=True) sys.exit(1) audio_path = sys.argv[2] output_srt = sys.argv[3] language = sys.argv[4] if len(sys.argv) > 4 and sys.argv[4] else None max_chars = int(sys.argv[5]) if len(sys.argv) > 5 and sys.argv[5].isdigit() else 30 print(f"[WORKER] 音频: {os.path.basename(audio_path)}", flush=True) print(f"[WORKER] 语言: {language}, 每句最多 {max_chars} 字", flush=True) print("[WORKER] 正在加载 Qwen3ASRModel...", flush=True) # SSL 证书修复（系统 Python 可能缺少证书） try: import certifi os.environ["SSL_CERT_FILE"] = certifi.where() except Exception: pass import torch from qwen_asr import Qwen3ASRModel if torch.backends.mps.is_available(): device = "mps" dtype = torch.float32 elif torch.cuda.is_available(): device = "cuda" dtype = torch.bfloat16 # bfloat16：显存减半 + 更快，质量几乎不变 else: device = "cpu" dtype = torch.float32 print(f"[WORKER] 设备: {device}", flush=True) try: model = Qwen3ASRModel.from_pretrained( ASR_MODEL_PATH, forced_aligner=MODEL_PATH, forced_aligner_kwargs={"dtype": dtype, "device_map": device}, dtype=dtype, device_map=device, max_new_tokens=512, ) except Exception as _e: if "out of memory" in str(_e).lower() or "OOM" in str(_e): print("[WORKER] 显存不足(OOM)：达芬奇正占用着显卡显存。请先在达芬奇里停止占用显存的操作（播放/渲染），或重启达芬奇后再试。", flush=True) raise print("[WORKER] 模型加载完成，开始识别...", flush=True) asr_start = time.time() results = model.transcribe( audio=audio_path, language=language, return_time_stamps=True, ) result = results[0] full_text = result.text ts_result = result.time_stamps print(f"[WORKER] 识别文本: {full_text[:100]}...", flush=True) print(f"[WORKER] 时间戳单元数: {len(ts_result.items) if ts_result else 0}", flush=True) # ════════════════════════════════════════════════════════ # 断句 v3：标点优先 + 模糊对齐逐字时间戳（全程真实时间，绝不按比例从 0 估算） # ════════════════════════════════════════════════════════ _PUNCT = set("，。！？、；：,.!?;:…—~·") _SENT_END = set("。！？；!?;") def _strip(s): return "".join(c for c in s if not c.isspace() and c not in _PUNCT) items = list(ts_result.items) if (ts_result and getattr(ts_result, "items", None)) else [] # 把每个时间戳单元展开成「逐字时间」列表 char_times[i] = (start, end) char_times = [] for u in items: t = (u.text or "").strip() if not t: continue for _ in t: char_times.append((u.start_time, u.end_time)) srt_entries = [] if full_text.strip() and char_times: # 1. 用 difflib 把「全文(带标点)」模糊对齐到「时间戳字符序列」， # 容忍 ASR 重复/漏字等不一致，得到 全文内容字符 -> char_times 下标 的映射 import difflib full_content = _strip(full_text) joined_content = _strip("".join((u.text or "").strip() for u in items)) sm = difflib.SequenceMatcher(None, full_content, joined_content, autojunk=False) full_to_joined = [None] * len(full_content) for tag, i1, i2, j1, j2 in sm.get_opcodes(): if tag == "equal": for k in range(i2 - i1): full_to_joined[i1 + k] = j1 + k elif tag == "replace": n = i2 - i1 m = j2 - j1 for k in range(n): full_to_joined[i1 + k] = (j1 + min(int(k * m / n), m - 1)) if m > 0 else None elif tag == "delete": # 全文里多出来的字符（典型：ASR 重复），就近映射到前一个下标 for k in range(i2 - i1): full_to_joined[i1 + k] = (j1 - 1) if j1 > 0 else None # 修补 None：用最近邻填充 Lf = len(full_to_joined) for i in range(Lf): if full_to_joined[i] is None: for d in range(1, Lf): if i - d >= 0 and full_to_joined[i - d] is not None: full_to_joined[i] = full_to_joined[i - d] break if i + d < Lf and full_to_joined[i + d] is not None: full_to_joined[i] = full_to_joined[i + d] break # 2. 生成 token 流：每个内容字符带真实 (start, end)，标点不带时间 tokens = [] # [(char, start, end)] ci = 0 for ch in full_text: if ch.isspace(): continue if ch in _PUNCT: tokens.append((ch, None, None)) continue if ci < len(full_to_joined): j = full_to_joined[ci] if j is not None and 0 <= j < len(char_times): tokens.append((ch, char_times[j][0], char_times[j][1])) else: tokens.append((ch, None, None)) else: tokens.append((ch, None, None)) ci += 1 # 3. 工具函数：token 列表的内容字符、文本、时间范围 def tl_content(tl): return [t for t in tl if t[0] not in _PUNCT] def tl_text(tl): return "".join(t[0] for t in tl).strip() def tl_range(tl): ct = tl_content(tl) starts = [t[1] for t in ct if t[1] is not None] ends = [t[2] for t in ct if t[2] is not None] if not starts or not ends: return None, None return starts[0], ends[-1] # 4. 按句末标点切分成句子级 token 列表 sentences = [] cur = [] for t in tokens: cur.append(t) if t[0] in _SENT_END: if tl_text(cur): sentences.append(cur) cur = [] if cur and tl_text(cur): sentences.append(cur) # 5. 超长句先按逗号二次切分 def split_comma(sent, max_c): if len(tl_content(sent)) <= max_c: return [sent] out = [] cur = [] for t in sent: cur.append(t) if t[0] in "，、；：,": if tl_text(cur): out.append(cur) cur = [] if cur and tl_text(cur): out.append(cur) return out or [sent] # 6. 按字数均分一个 token 列表（内容字符均分，保留真实时间） def split_by_char(tl, max_c): ct = tl_content(tl) L = len(ct) if L <= max_c: return [tl] out = [] n = (L + max_c - 1) // max_c per = (L + n - 1) // n pos = 0 while pos < L: out.append(ct[pos:pos + per]) pos += per return out # 7. 组装：句子 → 逗号切 → 过短合并 → 字数/时长约束 → 真实时间 flat = [] for sent in sentences: for piece in split_comma(sent, max_chars): flat.append(piece) merged = [] # [(start, end, token_list)] for tl in flat: s, e = tl_range(tl) if (len(tl_content(tl)) < 4 and merged and s is not None and merged[-1][1] is not None and s - merged[-1][1] < 1.0): prev = merged[-1] merged[-1] = (prev[0], e, prev[2] + tl) else: merged.append((s, e, tl)) for s, e, tl in merged: if s is None: continue ct = tl_content(tl) dur = e - s if (dur > 7.0 and len(ct) > max_chars // 2) or len(ct) > max_chars: for chunk in split_by_char(tl, max_chars): cs, ce = tl_range(chunk) if cs is not None: srt_entries.append((round(cs, 3), round(ce, 3), tl_text(chunk))) else: srt_entries.append((round(s, 3), round(e, 3), tl_text(tl))) # 兜底：完全没有时间戳时才用（按字数比例 + 真实音频时长） if not srt_entries and full_text.strip(): print("[WORKER] 警告: 时间戳完全不可用，按字数比例分配时间", flush=True) total_dur = (items[-1].end_time if items else 30.0) text = full_text.strip() parts = [] cur = [] for ch in text: cur.append(ch) if ch in _SENT_END: parts.append("".join(cur)) cur = [] if cur: parts.append("".join(cur)) char_total = sum(len(_strip(p)) for p in parts) or 1 offset = 0.0 for p in parts: if not p.strip(): continue c = len(_strip(p)) dur = total_dur * c / char_total srt_entries.append((round(offset, 3), round(offset + dur, 3), p.strip())) offset += dur # ── 最后一步：去掉每条字幕末尾的标点符号（纯收尾，不影响前面断句逻辑）── _trail_punct = "，。！？、；：,.!?;:…—~·" srt_entries = [ (s, e, t.rstrip(_trail_punct).strip()) for (s, e, t) in srt_entries if t.rstrip(_trail_punct).strip() ] srt_lines = [] for i, (start, end, text) in enumerate(srt_entries, 1): srt_lines.append(str(i)) srt_lines.append(f"{format_time_srt(start)} --> {format_time_srt(end)}") srt_lines.append(text) srt_lines.append("") srt_content = "\n".join(srt_lines) with open(output_srt, "w", encoding="utf-8") as f: f.write(srt_content) elapsed = time.time() - asr_start print(f"[WORKER] 完成！{len(srt_entries)} 条字幕, {elapsed:.1f}秒", flush=True) print(f"[WORKER] SRT: {output_srt}", flush=True) # ════════════════════════════════════════════════════════ # Resolve 操作函数 # ════════════════════════════════════════════════════════ def get_resolve(): resolve = None fusion = globals().get("fusion") bmd = globals().get("bmd") try: if fusion: resolve = fusion.GetResolve() except Exception: pass if not resolve: try: import DaVinciResolveScript as drs resolve = drs.scriptapp("Resolve") except Exception: pass if not resolve: try: if bmd: resolve = bmd.scriptapp("Resolve") except Exception: pass return resolve def scan_audio_tracks(timeline): try: track_count = timeline.GetTrackCount("audio") except Exception: return [("1", "音频轨道 1")] tracks = [] for i in range(1, track_count + 1): try: items = timeline.GetItemListInTrack("audio", i) name = f"音频轨道 {i}" if items: first = items[0] try: name += f" ({first.GetName()})" except Exception: pass except Exception: name = f"音频轨道 {i}" tracks.append((str(i), name)) return tracks def get_project_info(resolve): info = {"project": "—", "timeline": "—", "fps": "—"} try: project = resolve.GetProjectManager().GetCurrentProject() if project: info["project"] = project.GetName() timeline = project.GetCurrentTimeline() if timeline: info["timeline"] = timeline.GetName() try: info["fps"] = f"{float(project.GetSetting('timelineFrameRate')):.3f}" except Exception: info["fps"] = "?" except Exception: pass return info def isolate_track(timeline, track_idx): """禁掉除目标轨外的所有音频轨，返回被禁的索引列表""" total = timeline.GetTrackCount("audio") disabled = [] for i in range(1, total + 1): if i != track_idx: try: if timeline.SetTrackEnable("audio", i, False): disabled.append(i) except Exception: pass return disabled def restore_tracks(timeline, disabled): """恢复被禁用的音频轨""" for i in disabled: try: timeline.SetTrackEnable("audio", i, True) except Exception: pass def start_render_job(project, output_dir, audio_name): """加载 Audio Only 预设、设置输出路径、启动渲染，返回 job_id""" preset_names = ["Audio Only", "纯音频", "仅音频", "Wave Audio", "AO"] for pn in preset_names: try: project.LoadRenderPreset(pn) log.write(f"📋 加载预设: {pn}") break except Exception: pass else: log.write("⚠️ 未找到音频预设") return None project.SetRenderSettings({ "TargetDir": output_dir, "CustomName": audio_name, }) try: project.DeleteAllRenderJobs() except Exception: pass job_id = project.AddRenderJob() if not job_id: log.write("❌ 创建任务失败") return None log.write(f"📋 任务ID: {job_id}") ok = project.StartRendering([job_id], False) log.write(f"📊 后台渲染: {ok}") return job_id if ok else None def run_worker(audio_path, script_text, output_srt, language="Chinese"): """subprocess 调 conda python + 自身 --worker 模式做对齐""" script_b64 = base64.b64encode(script_text.encode("utf-8")).decode("utf-8") script_path = getattr(sys, "argv", [None])[0] if not script_path or not os.path.exists(script_path): script_path = __file__ if not script_path or not os.path.exists(script_path): script_path = os.path.join(TEMP_DIR, "..", "CXKJ_字幕识别_Win.py") script_path = os.path.abspath(script_path) cmd = [CONDA_PYTHON, script_path, "--worker", audio_path, script_b64, output_srt, language] log.write("🚀 启动对齐...") log.write(f" Python: {CONDA_PYTHON}") try: proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env={**os.environ, "PYTHONIOENCODING": "utf-8", "PYTHONUTF8": "1", "PYTHONNOUSERSITE": "1", "PYTORCH_ENABLE_MPS_FALLBACK": "1", "PYTORCH_MPS_HIGH_WATERMARK_RATIO": "0.0"}, ) for raw in iter(proc.stdout.readline, b""): text = raw.decode("utf-8", "replace").rstrip() if text: log.write(f" {text}") proc.wait() if proc.returncode != 0: log.write(f"❌ 对齐退出码: {proc.returncode}") return False log.write("✅ 对齐完成") return True except Exception as e: log.write(f"❌ 启动对齐失败: {e}") return False def import_subtitle(timeline, srt_path): if not os.path.exists(srt_path): log.write(f"❌ SRT 文件不存在: {srt_path}") return False try: result = timeline.ImportSubtitleFromFile(srt_path) if result: log.write("✅ 字幕已导入时间线") else: log.write("❌ 导入失败，请在时间线上手动添加一个字幕轨道后重试") return result except Exception as e: log.write(f"❌ 导入异常: {e}") return False # ════════════════════════════════════════════════════════ # UI 构建 # ════════════════════════════════════════════════════════ def build_ui(dispatcher, ui, resolve, project, timeline): win_w, win_h = 800, 600 fusion_obj = globals().get("fusion") info = get_project_info(resolve) tracks = scan_audio_tracks(timeline) track_items = [f"{idx}|{name}" for idx, name in tracks] win = dispatcher.AddWindow({ "ID": "IndexTTS2XMLWindow", "WindowTitle": "春星开讲 · CXKJ 字幕识别（非破坏性变速）", "Geometry": [160, 60, 650, 680], }, ui.VGroup({"Spacing": 2, "Weight": 1}, [ ui.Label({"Text": "【春星开讲】达芬奇AI剪辑智能体@SCXWIN", "MinimumSize": [600, 36], "MaximumSize": [600, 36], "Alignment": {"AlignHCenter": True}, "Font": ui.Font({"PixelSize": 28, "Bold": True})}), ui.TabBar({"ID": "MainTabs", "MinimumSize": [200, 28], "MaximumSize": [9999, 28]}), ui.Stack({"ID": "PageStack", "Weight": 1, "Alignment": {"AlignTop": True}}, [ # ========== 标签页 1：字幕识别（主操作） ========== ui.VGroup({"ID": "PageMain", "Spacing": 8}, [ # ── 1. 项目信息 + 品牌 ── ui.HGroup({"Spacing": 8, "Weight": 0}, [ ui.Label({ "ID": "ProjectInfo", "Text": f"项目: {info['project']} | 时间线: {info['timeline']} | 帧率: {info['fps']} fps", "Weight": 1, }), ]), ui.VGap(2), # ── 2. 四个按钮（水平居中） ── ui.HGroup({"Spacing": 12, "Weight": 0, "Alignment": {"AlignCenter": True}}, [ ui.Button({"ID": "AddVideoBtn", "Text": "添加视频轨道", "MinimumSize": [150, 40]}), ui.Button({"ID": "AddAudioBtn", "Text": "添加音频轨道", "MinimumSize": [150, 40]}), ui.Button({"ID": "AddSubtitleBtn", "Text": "添加 subtitle 轨道", "MinimumSize": [160, 40]}), ui.Button({"ID": "BtnForceExit", "Text": "⏻ 强制退出", "MinimumSize": [120, 40]}), ]), ui.VGap(4), # ── 3. 音频轨道选择 + 输出目录 + 刷新按钮 ── ui.HGroup({"Spacing": 8, "Weight": 0}, [ # 音频轨道下拉 ui.VGroup({"Spacing": 3, "Weight": 1}, [ ui.Label({"Text": "音频轨道", "Alignment": {"AlignLeft": True}}), ui.ComboBox({"ID": "TrackCombo", "Text": track_items[0] if track_items else "无音频轨道"}), ]), # 刷新按钮 ui.VGroup({"Spacing": 3, "Weight": 0}, [ ui.Label({"Text": " ", "Alignment": {"AlignLeft": True}}), ui.Button({"ID": "BtnRefresh", "Text": "🔄 刷新项目", "MinimumSize": [100, 24]}), ]), # 输出目录 ui.VGroup({"Spacing": 3, "Weight": 2}, [ ui.Label({"Text": "输出目录", "Alignment": {"AlignLeft": True}}), ui.HGroup({"Spacing": 5, "Weight": 1}, [ ui.LineEdit({"ID": "OutputDir", "Text": TEMP_DIR, "MinimumSize": [120, 24], "Weight": ui.Expanding}), ui.Button({"ID": "PickOutputDir", "Text": "选择...", "MinimumSize": [60, 24], "MaximumSize": [60, 24]}), ]), ]), ]), # ── 3.5. 语言 + 每句字数 ── ui.HGroup({"Spacing": 16, "Weight": 0}, [ ui.VGroup({"Spacing": 3, "Weight": 1}, [ ui.Label({"Text": "语言", "Alignment": {"AlignLeft": True}}), ui.ComboBox({"ID": "LangCombo", "Text": "Chinese"}), ]), ui.VGroup({"Spacing": 3, "Weight": 0}, [ ui.Label({"Text": "每句字数", "Alignment": {"AlignLeft": True}}), ui.LineEdit({"ID": "MaxChars", "Text": "30", "MinimumSize": [60, 24], "MaximumSize": [60, 24]}), ]), ui.VGroup({"Spacing": 3, "Weight": 1}, [ ui.Label({"Text": " ", "Alignment": {"AlignLeft": True}}), ui.Label({"Text": "", "Weight": 0}), ]), ]), ui.VGap(4), # ── 4. 识别按钮 ── ui.HGroup({"Spacing": 0, "Weight": 0, "Alignment": {"AlignCenter": True}}, [ ui.Button({"ID": "BtnAlign", "Text": "▶ 识别字幕", "MinimumSize": [200, 44]}), ]), ui.VGap(2), # ── 三个进度条 ── ui.TextEdit({"ID": "ProgressAudio", "ReadOnly": True, "PlainText": "⏹ 音频输出: 等待开始...", "Weight": 0, "MinimumSize": [200, 20]}), ui.TextEdit({"ID": "ProgressQwen", "ReadOnly": True, "PlainText": "—", "Weight": 0, "MinimumSize": [200, 20]}), ui.VGap(4), # ── 5. 日志区 ── ui.Label({"Text": "日志：", "Weight": 0}), ui.TextEdit({"ID": "LogText", "PlainText": True, "ReadOnly": True, "Weight": 0.5}), ]), # ========== 标签页 2：AI 配置 ========== ui.VGroup({"ID": "PageConfig", "Spacing": 6, "Alignment": {"AlignTop": True}}, [ ui.Label({"Text": "粘贴 JSON 配置（填好 ai_root，其余留空脚本自动推导）", "MinimumSize": [600, 24], "MaximumSize": [600, 24]}), ui.TextEdit({"ID": "ConfigJson", "PlainText": "", "MinimumSize": [600, 380]}), ui.HGroup({"Spacing": 12, "Weight": 0}, [ ui.Button({"ID": "CopyConfigJson", "Text": "复制 JSON"}), ui.Button({"ID": "ApplyConfig", "Text": "应用配置"}), ui.Button({"ID": "OpenConfigFile", "Text": "打开配置文件"}), ui.Button({"ID": "OpenConfigDir", "Text": "打开文件位置"}), ui.Button({"ID": "CloseCfg", "Text": "关闭"}), ]), ui.Label({"ID": "CfgStatus", "Text": "", "WordWrap": True, "MinimumSize": [600, 24]}), ui.VGap({"Weight": 1.0}), ]), ]), ])) items = win.GetItems() # ── 标签页初始化 ── items["MainTabs"].AddTab("字幕识别") items["MainTabs"].AddTab("AI 配置") items["PageStack"].CurrentIndex = 0 # ── 配置页：预填模板或已有配置 ── try: if os.path.isfile(_CONFIG_PATH): with open(_CONFIG_PATH, "r", encoding="utf-8") as f: items["ConfigJson"].PlainText = f.read() else: items["ConfigJson"].PlainText = _CONFIG_TEMPLATE except Exception: items["ConfigJson"].PlainText = _CONFIG_TEMPLATE # ── 标签页切换 ── def on_tab_changed(ev): items["PageStack"].CurrentIndex = items["MainTabs"].CurrentIndex win.On.MainTabs.CurrentChanged = on_tab_changed # ── 填充轨道列表 ── combo = items.get("TrackCombo") if combo: try: combo.Clear() for item in track_items: combo.AddItem(item) if track_items: combo.CurrentIndex = 0 except Exception as e: log.write(f"⚠️ 填充轨道列表失败: {e}") # ── 填充语言列表 ── lang_combo = items.get("LangCombo") if lang_combo: try: lang_combo.Clear() for l in ["自动检测", "Chinese", "English", "Cantonese", "Japanese", "Korean", "German", "French", "Spanish", "Italian", "Russian"]: lang_combo.AddItem(l) lang_combo.CurrentIndex = 0 except Exception as e: log.write(f"⚠️ 填充语言列表失败: {e}") # ══════════════════════════════════════════════════ # 事件处理 # ══════════════════════════════════════════════════ # ---------- 添加轨道按钮 ---------- def on_add_video_track(ev): try: r = fusion_obj.GetResolve() p = r.GetProjectManager().GetCurrentProject() tl = p.GetCurrentTimeline() if tl: ok = tl.AddTrack("video") log.write(f"{'✅' if ok else '❌'} 视频轨道已添加" if ok else "❌ 视频轨道添加失败") except Exception as e: log.write(f"❌ 添加视频轨出错: {e}") win.On["AddVideoBtn"].Clicked = on_add_video_track def on_add_audio_track(ev): try: r = fusion_obj.GetResolve() p = r.GetProjectManager().GetCurrentProject() tl = p.GetCurrentTimeline() if tl: ok = tl.AddTrack("audio") log.write(f"{'✅' if ok else '❌'} 音频轨道已添加" if ok else "❌ 音频轨道添加失败") except Exception as e: log.write(f"❌ 添加音频轨出错: {e}") win.On["AddAudioBtn"].Clicked = on_add_audio_track def on_add_subtitle_track(ev): try: r = fusion_obj.GetResolve() p = r.GetProjectManager().GetCurrentProject() tl = p.GetCurrentTimeline() if tl: ok = tl.AddTrack("subtitle") log.write(f"{'✅' if ok else '❌'} subtitle 轨道已添加" if ok else "❌ subtitle 轨道添加失败") except Exception as e: log.write(f"❌ 添加 subtitle 轨出错: {e}") win.On["AddSubtitleBtn"].Clicked = on_add_subtitle_track # ---------- 刷新项目 ---------- def refresh_project_info_internal(): try: nonlocal project, timeline project = resolve.GetProjectManager().GetCurrentProject() if project: timeline = project.GetCurrentTimeline() info = get_project_info(resolve) p_label = items.get("ProjectInfo") if p_label: p_label.Text = f"项目: {info['project']} | 时间线: {info['timeline']} | 帧率: {info['fps']} fps" # 刷新轨道列表 if timeline: new_tracks = scan_audio_tracks(timeline) if combo: combo.Clear() for idx, name in new_tracks: combo.AddItem(f"{idx}|{name}") if new_tracks: combo.CurrentIndex = 0 except Exception as e: log.write(f"❌ 刷新项目信息失败: {e}") def on_refresh(ev): log.write("🔄 刷新项目信息...") refresh_project_info_internal() if timeline: log.write(f" ✅ 当前时间线: {timeline.GetName()}") win.On["BtnRefresh"].Clicked = on_refresh # ---------- 选择输出目录 ---------- def on_pick_output(ev): global _CFG try: default_path = items["OutputDir"].Text.strip() if not os.path.isdir(default_path): default_path = os.path.expanduser("~") picked = fusion_obj.RequestDir("选择输出目录", default_path) if picked: items["OutputDir"].Text = picked _CFG["temp_dir"] = picked with open(_CONFIG_PATH, "w", encoding="utf-8") as f: json.dump(_CFG, f, ensure_ascii=False, indent=2) log.write(f"📁 输出目录: {picked}（已写入配置）") except Exception as e: log.write(f"❌ 选择目录失败: {e}") win.On["PickOutputDir"].Clicked = on_pick_output # ---------- 文稿文本变化（已移除启用/禁用，按钮常亮）---------- # ---------- 文稿对齐SRT ---------- def get_track_idx(): if not combo: return "1" text = combo.Text if "|" in text: return text.split("|")[0] return "1" def on_align(ev): log.write(f"▶ 渲染按钮被点击") btn_align = items.get("BtnAlign") btn_refresh = items.get("BtnRefresh") prog_audio = items.get("ProgressAudio") if btn_align: btn_align.Enabled = False if btn_refresh: btn_refresh.Enabled = False if prog_audio: prog_audio.PlainText = "⏳ 渲染中..." win.Update() try: output_dir = items["OutputDir"].Text.strip() if not output_dir or not os.path.isdir(output_dir): log.write("⚠️ 请先选择输出目录") if prog_audio: prog_audio.PlainText = "⚠️ 请先选择输出目录" return timestamp = ts() audio_name = f"audio_{timestamp}" log.write(f"📁 输出: {output_dir}/{audio_name}") win.Update() job_id = start_render_job(project, output_dir, audio_name) if not job_id: log.write("❌ 渲染启动失败") if prog_audio: prog_audio.PlainText = "❌ 启动失败" return for i in range(60): try: fusion_obj.GetResolve().OpenPage("edit") except Exception: pass try: st = project.GetRenderJobStatus(job_id) if st: pct = st.get("CompletionPercentage", 0) js = st.get("JobStatus", "") bar_len = 40 filled = int(bar_len * min(100, pct) / 100) bar = "█" * filled + "░" * (bar_len - filled) if prog_audio: prog_audio.PlainText = f"⏳ 渲染 {bar} {pct}%" if js == "Complete" or pct >= 100: if prog_audio: prog_audio.PlainText = "✅ 渲染完成" log.write("✅ 渲染完成") break if js == "Failed": if prog_audio: prog_audio.PlainText = "❌ 渲染失败" log.write("❌ 渲染失败") return except Exception: pass time.sleep(1) win.Update() else: if prog_audio: prog_audio.PlainText = "✅ 已启动（后台）" # 渲染完成，冷却 2 秒避免 Metal 冲突 time.sleep(2) # ── 找音频文件 ── actual_audio = None for ext in [".mp3", ".wav"]: p = os.path.join(output_dir, f"audio_{timestamp}{ext}") if os.path.exists(p): actual_audio = p break if not actual_audio: log.write("⚠️ 未找到渲染输出音频") return # ── Step 2: Qwen 语音识别 ── srt_path = os.path.join(output_dir, f"srt_{timestamp}.srt") prog_qwen = items.get("ProgressQwen") if prog_qwen: prog_qwen.PlainText = "⏳ 语音识别 ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░" win.Update() sp = getattr(sys, "argv", [None])[0] if not sp or not os.path.exists(sp): sp = "/Library/Application Support/Blackmagic Design/DaVinci Resolve/Fusion/Scripts/Utility/CXKJ_字幕识别_Mac.py" sp = os.path.abspath(sp) lang = items.get("LangCombo") lang = (lang.Text or "Chinese") if lang else "Chinese" if lang == "自动检测": lang = "" # 空字符串让模型自动检测 mc = items.get("MaxChars") max_chars = (mc.Text or "30").strip() if mc else "30" if not max_chars.isdigit(): max_chars = "30" cmd = [CONDA_PYTHON, sp, "--worker", actual_audio, srt_path, lang, max_chars] try: proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env={**os.environ, "PYTHONIOENCODING": "utf-8", "PYTHONUTF8": "1", "PYTHONNOUSERSITE": "1", "PYTORCH_ENABLE_MPS_FALLBACK": "1", "PYTORCH_MPS_HIGH_WATERMARK_RATIO": "0.0"}) line_n = 0 for raw in iter(proc.stdout.readline, b""): line_n += 1 t = raw.decode("utf-8", "replace").rstrip() if t: log.write(f" {t}") bar_len = 40 if "加载" in t: filled = min(8, line_n) bar = "█" * filled + "░" * (bar_len - filled) if prog_qwen: prog_qwen.PlainText = f"⏳ 加载模型 {bar}" elif "开始识别" in t: filled = 20 bar = "█" * filled + "░" * (bar_len - filled) if prog_qwen: prog_qwen.PlainText = f"⏳ 识别中 {bar}" elif "完成" in t and ("字幕" in t or "条" in t): bar = "█" * bar_len if prog_qwen: prog_qwen.PlainText = f"✅ 语音识别 {bar}" proc.wait() win.Update() # 恢复界面状态 if proc.returncode != 0: log.write(f"❌ Qwen退出码: {proc.returncode}") if prog_qwen: prog_qwen.PlainText = "❌ 对齐失败" return except Exception as e2: log.write(f"❌ Qwen异常: {e2}") if prog_qwen: prog_qwen.PlainText = "❌ 异常" return # ── Step 3: 导入 SRT → 建轨 → 贴入 ── if os.path.exists(srt_path): mp = project.GetMediaPool() clips = mp.ImportMedia([srt_path]) if not clips: log.write("❌ 导入媒体池失败") return srt_clip = clips[0] log.write(f"✅ 已导入: {srt_clip.GetName()}") # 确保有字幕轨道 track_index = 1 while timeline.GetTrackCount("subtitle") < track_index: timeline.AddTrack("subtitle") sub_after = timeline.GetTrackCount("subtitle") log.write(f" 字幕轨已就绪: {sub_after} 条") # 贴入时间线 result = mp.AppendToTimeline([srt_clip]) log.write(f"{'✅ 字幕已贴入' if result else '❌ 贴入失败'}") else: log.write(f"⚠️ SRT未生成: {srt_path}") except Exception as e: log.write(f"❌ 异常: {e}") log.write(traceback.format_exc()) finally: if btn_align: btn_align.Enabled = True if btn_refresh: btn_refresh.Enabled = True win.On["BtnAlign"].Clicked = on_align # ---------- 强制退出 ---------- def on_force_exit(ev): try: project.StopRendering() except Exception: pass dispatcher.ExitLoop() os._exit(0) win.On["BtnForceExit"].Clicked = on_force_exit # ---------- 关闭事件 ---------- def on_close(ev): dispatcher.ExitLoop() win.On.IndexTTS2XMLWindow.Close = on_close win.On.CloseCfg.Clicked = on_close # ── 日志回调（把日志写入 TextEdit） ── def update_log(line, tag=None): w = items.get("LogText") if not w: return try: current = w.PlainText lines = current.split("\n") if len(lines) > 500: lines = lines[-400:] lines.append(line) w.PlainText = "\n".join(lines) except Exception: pass log.set_callback(update_log) # ========== 配置页事件处理 ========== def on_copy_config_json(ev): try: text = items["ConfigJson"].PlainText _copy_to_clipboard(text) items["CfgStatus"].Text = "✓ JSON 已复制到剪贴板" except Exception as e: items["CfgStatus"].Text = "复制失败: " + str(e) def on_apply_config(ev): global _CFG, CONDA_PYTHON, MODEL_PATH, ASR_MODEL_PATH, TEMP_DIR, LOG_FILE try: raw = items["ConfigJson"].PlainText cfg = json.loads(raw) if not isinstance(cfg, dict): items["CfgStatus"].Text = "JSON 格式错误：顶层不是对象" return with open(_CONFIG_PATH, "w", encoding="utf-8") as f: json.dump(cfg, f, ensure_ascii=False, indent=2) _CFG = cfg CONDA_PYTHON = _resolve("python", _CFG, "", lambda r: os.path.join(r, ".venv", "Scripts", "python.exe")) MODEL_PATH = _resolve("model_path", _CFG, "", lambda r: os.path.join(r, "Qwen3-ForcedAligner/model")) ASR_MODEL_PATH = _resolve("asr_model_path", _CFG, "", lambda r: os.path.join(r, "Qwen3-ASR-0.6B")) TEMP_DIR = os.path.expanduser(_resolve("temp_dir", _CFG, os.path.expanduser("~/Desktop/字幕识别缓存"), lambda r: os.path.join(r, "Desktop/字幕识别缓存"))) LOG_FILE = os.path.join(TEMP_DIR, "indextts_xml_resolve_generate.log") os.makedirs(TEMP_DIR, exist_ok=True) if "OutputDir" in items and TEMP_DIR: items["OutputDir"].Text = TEMP_DIR items["CfgStatus"].Text = "✓ 配置已保存：" + _CONFIG_PATH except json.JSONDecodeError as e: items["CfgStatus"].Text = "JSON 解析失败：" + str(e) except Exception as e: items["CfgStatus"].Text = "应用失败: " + str(e) def on_open_config_file(ev): try: subprocess.Popen(["open", _CONFIG_PATH]) items["CfgStatus"].Text = "✓ 已打开配置文件" except Exception as e: items["CfgStatus"].Text = "打开失败: " + str(e) def on_open_config_dir(ev): try: subprocess.Popen(["open", os.path.dirname(_CONFIG_PATH)]) items["CfgStatus"].Text = "✓ 已打开文件位置" except Exception as e: items["CfgStatus"].Text = "打开失败: " + str(e) win.On.CopyConfigJson.Clicked = on_copy_config_json win.On.ApplyConfig.Clicked = on_apply_config win.On.OpenConfigFile.Clicked = on_open_config_file win.On.OpenConfigDir.Clicked = on_open_config_dir return win # ════════════════════════════════════════════════════════ # UI 主入口 # ════════════════════════════════════════════════════════ def ui_main(): append_log(f"\n{'='*50}\n🚀 CXKJ_识别字幕: {datetime.datetime.now()}\n{'='*50}\n") resolve = get_resolve() if not resolve: log.write("❌ 无法连接 DaVinci Resolve") return project = resolve.GetProjectManager().GetCurrentProject() if not project: log.write("❌ 没有打开的项目") return timeline = project.GetCurrentTimeline() if not timeline: log.write("❌ 没有打开的时间线") return log.write(f"✅ {project.GetName()} / {timeline.GetName()}") fusion_obj = globals().get("fusion") bmd_obj = globals().get("bmd") if not bmd_obj or not fusion_obj: log.write("❌ 脚本必须从 DaVinci Resolve 的 Scripts 菜单运行") return ui = fusion_obj.UIManager if not ui: log.write("❌ 无法获取 UIManager") return try: dispatcher = bmd_obj.UIDispatcher(ui) except Exception as e: try: from fusionscript import UIDispatcher dispatcher = UIDispatcher(ui) except Exception as e2: log.write(f"❌ 创建 UIDispatcher 失败: {e2}") return win = build_ui(dispatcher, ui, resolve, project, timeline) log.write("=" * 50) # Show 后 Resize(oversize) → Resize(target) 强制 NSView 布局重算 win.Show() try: win.Resize([700, 730]) win.Resize([650, 680]) except Exception: pass dispatcher.RunLoop() # ════════════════════════════════════════════════════════ # 入口 # ════════════════════════════════════════════════════════ if __name__ == "__main__": if len(sys.argv) > 1 and sys.argv[1] == "--worker": worker_main() else: try: ui_main() except Exception as e: print(traceback.format_exc()) sys.exit(1)