From 474633151561fe05a9d9ea83d4e7a20779dc43fe Mon Sep 17 00:00:00 2001 From: Shuming Liu Date: Thu, 8 Oct 2026 10:08:16 +0800 Subject: [PATCH] pc translate sentences --- README.md | 6 +- media_player.py | 484 ++++++++++++++++++++++++++++++- sentence_api/README.md | 6 + sentence_api/main.py | 20 +- sentence_api/models.py | 12 +- sentence_api/repository.py | 100 +++++++ sentence_api/tests/test_api.py | 85 ++++++ sentence_api/translation_jobs.py | 5 + 8 files changed, 714 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 394a956..02982af 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,8 @@ pip 安装依赖后即可直接播放视频。 视频的句边界；有则直接取回，没有时才本地提取音频并上传。生成完成后才能 勾选“按句跳转”，之后 `, / .` 或 `← / →` 可跳句。PC 会按文件 SHA-256 缓存结果；生成过程有进度条。 +- 画面字幕：句边界就绪后，当前句会叠在视频画面底部。可勾选“中文字幕” + （快捷键 `C`）显示或关闭中文翻译；翻译由服务端生成并缓存。 - AI 讲解：播放中按 `E` 或点击“AI讲解 (E)”，由服务端 AI 讲解当前句子。 - 服务配置：点击“服务…”可填写服务器地址，或使用账号密码登录后保存令牌。 - 调试日志：点击“调试日志”可查看/复制请求状态、耗时和错误摘要；日志保存在 @@ -83,6 +85,7 @@ pip install -r requirements.txt | `N` / `P` | 下一首 / 上一首 | | `[` / `]` | 减速 / 加速（0.5x - 2.0x） | | `L` | 切换循环模式（关 / 单曲 / 列表） | +| `C` | 开 / 关中文字幕 | | 双击画面 | 进入 / 退出全屏 | | `F` | 视频全屏切换 | | `Esc` | 退出全屏 | @@ -105,7 +108,8 @@ pip install -r requirements.txt 静音阈值会根据每个文件的背景噪声自动调整，带环境噪音的录音也能正常分句。 分析在后台进行，首次播放某文件时状态栏会提示“正在分析句边界…”，完成后 自动缓存；无法识别（如纯音乐、无音频轨）时会提示并回退为普通步进跳转。 -“按句跳转”的开关状态同样会自动保存。 +“按句跳转”和“中文字幕”的开关状态同样会自动保存。句边界就绪后，原文会叠在 +视频画面底部；打开中文字幕时会向服务器请求翻译并一并显示。 ## 常见问题 diff --git a/media_player.py b/media_player.py index d087103..4adb109 100644 --- a/media_player.py +++ b/media_player.py @@ -16,6 +16,7 @@ Ctrl+↑ / ↓ 快进步长 +1 / -1 秒（macOS 亦支持 Command+↑ / ↓） , / . 上一句 / 下一句 勾选“按句跳转”后，←/→ 也会自动播放上一句/下一句 + C 开 / 关中文字幕 E 讲解当前句子（AI） M 静音 / 恢复 Ctrl+← / → 上一首 / 下一首（macOS 也支持 Command+← / →） @@ -225,6 +226,15 @@ def fmt_time(ms): return f"{minutes:02d}:{seconds:02d}" +def _srt_timestamp(ms): + """把毫秒格式化为 SRT 时间戳。""" + total = max(0, int(ms)) + hours, remainder = divmod(total, 3_600_000) + minutes, remainder = divmod(remainder, 60_000) + seconds, millis = divmod(remainder, 1000) + return f"{hours:02d}:{minutes:02d}:{seconds:02d},{millis:03d}" + + class _BaseEngine: """播放内核统一接口。""" @@ -270,6 +280,9 @@ class _BaseEngine: def embed(self, window_id): return False + def set_subtitle_file(self, path): + return False + def get_frame(self): """返回当前应显示的帧（PIL 图像），没有新帧时返回 None。""" return None @@ -371,6 +384,31 @@ class VlcEngine(_BaseEngine): self.player.set_xwindow(int(window_id)) return True + def set_subtitle_file(self, path): + if not self.is_active(): + return False + try: + if not path: + try: + self.player.video_set_spu(-1) + except Exception: + pass + return True + uri = Path(path).resolve().as_uri() + slave_type = getattr(getattr(vlc, "MediaSlaveType", None), "subtitle", 0) + try: + result = self.player.add_slave(slave_type, uri, True) + if result == 0: + return True + except Exception: + pass + setter = getattr(self.player, "video_set_subtitle_file", None) + if setter is None: + return False + return bool(setter(str(path))) + except Exception: + return False + def release(self): try: self.player.stop() @@ -977,6 +1015,7 @@ class MediaPlayerApp(tk.Tk): self.seek_step = settings["seek_step"] # Sentence mode is always re-enabled manually after boundaries exist. self.sentence_mode = tk.BooleanVar(value=False) + self.show_chinese_subtitles = tk.BooleanVar(value=bool(settings.get("show_chinese_subtitles", False))) self.server_url = settings["server_url"].rstrip("/") self.auth_token = settings["auth_token"] self.auth_username = settings["auth_username"] @@ -999,6 +1038,12 @@ class MediaPlayerApp(tk.Tk): self._explanation_text = None self._explanation_debug_var = None self._explanation_generation = 0 + self._translation_generation = 0 + self._translation_running = set() + self._translation_loading = False + self._translation_error = "" + self._subtitle_srt_key = None + self._overlay_key = None self.engine = self._create_engine() if self.engine is None: @@ -1097,7 +1142,33 @@ class MediaPlayerApp(tk.Tk): hint_text = "打开文件开始播放\n（视频将在独立窗口中显示）" self.video_hint = tk.Label(self.video_frame, text=hint_text, bg="#101010", fg="#8a8a8a") self.video_hint.place(relx=0.5, rely=0.5, anchor="center") + self.subtitle_overlay = tk.Frame(self.video_frame, bg="#000000", padx=10, pady=8) + self.subtitle_en = tk.Label( + self.subtitle_overlay, + text="", + fg="#ffffff", + bg="#000000", + font=("Helvetica", 16, "bold"), + wraplength=720, + justify="center", + ) + self.subtitle_en.pack(fill="x") + self.subtitle_zh = tk.Label( + self.subtitle_overlay, + text="", + fg="#ffd54a", + bg="#000000", + font=("Helvetica", 14), + wraplength=720, + justify="center", + ) + self.subtitle_zh.pack(fill="x", pady=(4, 0)) self.video_frame.bind("", self._on_video_double_click) + self.video_frame.bind("", self._on_video_resize) + if getattr(self, "video_canvas", None) is not None: + self.video_canvas.bind("", self._on_video_double_click) + for widget in (self.subtitle_overlay, self.subtitle_en, self.subtitle_zh): + widget.bind("", self._on_video_double_click) side = ttk.Frame(main, width=260) side.grid(row=0, column=1, sticky="ns") @@ -1165,6 +1236,15 @@ class MediaPlayerApp(tk.Tk): # 点击 Checkbutton 后 tk 仍会把焦点给它；把空格键重定向到播放/暂停， # 避免误改复选按钮状态。 self.sentence_mode_btn.bind("", self._on_space) + self.chinese_subtitle_btn = ttk.Checkbutton( + controls, + text="中文字幕", + variable=self.show_chinese_subtitles, + command=self._on_chinese_subtitles, + takefocus=0, + ) + self.chinese_subtitle_btn.pack(side="left", padx=(12, 0)) + self.chinese_subtitle_btn.bind("", self._on_space) self.btn_generate_boundaries = ttk.Button( controls, text="生成句边界", @@ -1230,6 +1310,8 @@ class MediaPlayerApp(tk.Tk): ("", lambda _e: self.step_speed(1)), ("", lambda _e: self.seek_sentence(-1)), ("", lambda _e: self.seek_sentence(1)), + ("", lambda _e: self.toggle_chinese_subtitles()), + ("", lambda _e: self.toggle_chinese_subtitles()), ("", lambda _e: self.show_ai_explanation()), ("", lambda _e: self.show_ai_explanation()), ("", lambda _e: self.show_ai_explanation()), @@ -1324,6 +1406,7 @@ class MediaPlayerApp(tk.Tk): sentence = self._current_sentence() text = str(sentence.get("text") or "").strip() if sentence else "" self.sentence_var.set(text) + self._update_on_screen_subtitles() playing = self.engine.is_playing() self.btn_play.config(text="暂停" if playing else "播放") self.after(33, self._tick) @@ -1348,6 +1431,8 @@ class MediaPlayerApp(tk.Tk): self._photo = ImageTk.PhotoImage(img) canvas.delete("all") canvas.create_image(width // 2, height // 2, image=self._photo, anchor="center") + if getattr(self, "subtitle_overlay", None) is not None: + self.subtitle_overlay.lift() def open_files(self): filetypes = [ @@ -1411,6 +1496,7 @@ class MediaPlayerApp(tk.Tk): return { "seek_step": step, "sentence_mode": bool(data.get("sentence_mode", False)), + "show_chinese_subtitles": bool(data.get("show_chinese_subtitles", False)), "server_url": server_url, "auth_token": str(data.get("auth_token") or "").strip(), "auth_username": str(data.get("auth_username") or "").strip(), @@ -1424,6 +1510,7 @@ class MediaPlayerApp(tk.Tk): { "seek_step": self.seek_step, "sentence_mode": bool(self.sentence_mode.get()), + "show_chinese_subtitles": bool(self.show_chinese_subtitles.get()), "server_url": self.server_url, "auth_token": self.auth_token, "auth_username": self.auth_username, @@ -1510,6 +1597,13 @@ class MediaPlayerApp(tk.Tk): self.status_var.set("请先生成句子边界") self._load_local_boundaries_async(path) self._update_sentence_controls() + self._translation_generation += 1 + self._translation_loading = False + self._translation_error = "" + self._overlay_key = None + self._subtitle_srt_key = None + self._ensure_chinese_translations() + self.after(400, self._sync_vlc_subtitles) def toggle_play(self): if not self.playlist: @@ -1526,6 +1620,9 @@ class MediaPlayerApp(tk.Tk): self._seeking = False self.time_var.set("00:00 / --:--") self.sentence_var.set("") + self._overlay_key = None + self._subtitle_srt_key = None + self._update_on_screen_subtitles() self.seekbar.set_frac(0.0) self.btn_play.config(text="播放") canvas = getattr(self, "video_canvas", None) @@ -1707,11 +1804,13 @@ class MediaPlayerApp(tk.Tk): def _finish_local_boundary_load(self, path, message): self._local_cache_loading.discard(path) - if path in self._analysis_running or path in self._boundaries: + if path in self._analysis_running: return if path == self._current_path(): self.status_var.set(message) self._update_sentence_controls() + self._ensure_chinese_translations() + self._sync_vlc_subtitles() def _update_sentence_controls(self): path = self._current_path() @@ -1762,6 +1861,22 @@ class MediaPlayerApp(tk.Tk): count = len(self._boundaries[path]["sentences"]) self.status_var.set(f"按句跳转已开启（识别到 {count} 句）") + def toggle_chinese_subtitles(self): + self.show_chinese_subtitles.set(not self.show_chinese_subtitles.get()) + self._on_chinese_subtitles() + + def _on_chinese_subtitles(self): + self._save_settings() + if self.show_chinese_subtitles.get(): + self._translation_error = "" + self._ensure_chinese_translations() + self.status_var.set("中文字幕已开启") + else: + self.status_var.set("中文字幕已关闭") + self._overlay_key = None + self._update_on_screen_subtitles() + self._sync_vlc_subtitles(force=True) + def generate_sentence_boundaries(self): path = self._current_path() if path is None: @@ -1901,6 +2016,8 @@ class MediaPlayerApp(tk.Tk): ) self._report_progress(None, None, indeterminate=False) self._ui_events.put(lambda: self._update_sentence_controls()) + self._ui_events.put(self._ensure_chinese_translations) + self._ui_events.put(self._sync_vlc_subtitles) self._analysis_done.put((path, message)) def _boundary_cache_path(self, media_hash: str) -> Path: @@ -1920,12 +2037,14 @@ class MediaPlayerApp(tk.Tk): end = int(item["end_ms"]) if end <= start or start < previous_end: raise ValueError("invalid cached boundary") + zh_text = item.get("zh_text") cleaned.append({ "index": index, "start_ms": start, "end_ms": end, "text": (str(item["text"]) if item.get("text") is not None else None), "language": item.get("language"), + "zh_text": (str(zh_text).strip() if zh_text else None) or None, }) previous_end = end if not cleaned: @@ -2094,6 +2213,368 @@ class MediaPlayerApp(tk.Tk): detail = str(detail) return detail if len(detail) <= 300 else detail[:297] + "..." + def _on_video_resize(self, event): + if event.widget is not getattr(self, "video_frame", None): + return + wrap = max(200, int(event.width) - 48) + if getattr(self, "subtitle_en", None) is not None: + self.subtitle_en.configure(wraplength=wrap) + if getattr(self, "subtitle_zh", None) is not None: + self.subtitle_zh.configure(wraplength=wrap) + + def _subtitle_sentence(self) -> Optional[Dict[str, Any]]: + sentence = self._current_sentence() + if not sentence: + return None + pos = self.engine.get_time() + if pos < 0: + return None + start_ms = int(sentence.get("start_ms") or 0) + end_ms = int(sentence.get("end_ms") or 0) + if pos + 180 < start_ms or pos > end_ms + 180: + return None + return sentence + + def _update_on_screen_subtitles(self): + overlay = getattr(self, "subtitle_overlay", None) + if overlay is None: + return + sentence = self._subtitle_sentence() + english = str(sentence.get("text") or "").strip() if sentence else "" + chinese = "" + if self.show_chinese_subtitles.get() and english: + chinese = str((sentence or {}).get("zh_text") or "").strip() + if not chinese: + if self._translation_error: + chinese = self._translation_error + elif self._translation_loading: + chinese = "正在翻译中文字幕…" + else: + chinese = "暂无中文字幕" + key = (english, chinese, bool(self.show_chinese_subtitles.get())) + if key == self._overlay_key: + return + self._overlay_key = key + try: + self.subtitle_en.configure(text=english) + if chinese: + self.subtitle_zh.configure(text=chinese) + if not self.subtitle_zh.winfo_ismapped(): + self.subtitle_zh.pack(fill="x", pady=(4, 0)) + elif self.subtitle_zh.winfo_ismapped(): + self.subtitle_zh.pack_forget() + if english or chinese: + overlay.place(relx=0.5, rely=0.97, anchor="s", relwidth=0.92) + overlay.lift() + else: + overlay.place_forget() + except tk.TclError: + pass + + def _build_srt_content(self, sentences, include_chinese: bool) -> str: + blocks = [] + number = 1 + for sentence in sentences: + english = str(sentence.get("text") or "").strip() + chinese = str(sentence.get("zh_text") or "").strip() if include_chinese else "" + if not english and not chinese: + continue + lines = [] + if english: + lines.append(english) + if chinese: + lines.append(chinese) + start = _srt_timestamp(sentence.get("start_ms") or 0) + end = _srt_timestamp(sentence.get("end_ms") or 0) + blocks.append(f"{number}\n{start} --> {end}\n" + "\n".join(lines)) + number += 1 + return "\n\n".join(blocks) + ("\n" if blocks else "") + + def _sync_vlc_subtitles(self, force=False): + if not isinstance(self.engine, VlcEngine): + return + path = self._current_path() + sentences = (self._boundaries.get(path) or {}).get("sentences") or [] + include_chinese = bool(self.show_chinese_subtitles.get()) + fingerprint = ( + path, + include_chinese, + tuple( + ( + sentence.get("index"), + sentence.get("start_ms"), + sentence.get("end_ms"), + sentence.get("text"), + sentence.get("zh_text") if include_chinese else None, + ) + for sentence in sentences + ), + ) + if not force and fingerprint == self._subtitle_srt_key: + return + self._subtitle_srt_key = fingerprint + media_hash = self._media_hashes.get(path) if path else None + if not sentences or not media_hash: + self.engine.set_subtitle_file(None) + return + srt_path = Path.home() / ".mediaplayer_sentence_boundaries" / f"{media_hash}.srt" + try: + srt_path.parent.mkdir(parents=True, exist_ok=True) + content = self._build_srt_content(sentences, include_chinese) + srt_path.write_text("\ufeff" + content, encoding="utf-8") + self.engine.set_subtitle_file(str(srt_path)) + except Exception as exc: + debug_log("subtitle_srt_failed", error=str(exc)) + + def _ensure_chinese_translations(self): + if not self.show_chinese_subtitles.get(): + return + path = self._current_path() + if not path: + return + sentences = (self._boundaries.get(path) or {}).get("sentences") or [] + if not sentences: + return + missing = [ + sentence + for sentence in sentences + if str(sentence.get("text") or "").strip() and not str(sentence.get("zh_text") or "").strip() + ] + if not missing: + self._translation_loading = False + self._translation_error = "" + return + media_hash = self._media_hashes.get(path) + if not media_hash: + return + if not self.auth_token: + self._translation_error = "请先登录后再加载中文字幕" + self._overlay_key = None + self._update_on_screen_subtitles() + return + if path in self._translation_running: + return + self._translation_generation += 1 + generation = self._translation_generation + self._translation_running.add(path) + self._translation_loading = True + self._translation_error = "" + self._overlay_key = None + self._update_on_screen_subtitles() + expected = [int(sentence["index"]) for sentence in missing] + local_payload = self._local_subtitle_payload(path) + threading.Thread( + target=self._load_translations_worker, + args=(path, media_hash, expected, generation, local_payload), + daemon=True, + ).start() + + def _local_subtitle_payload(self, path) -> Optional[Dict[str, Any]]: + document = self._boundaries.get(path) or {} + sentences = document.get("sentences") or [] + cleaned = [] + previous_end = 0 + last_end = 0 + for expected_index, sentence in enumerate(sentences): + try: + start_ms = max(0, int(sentence.get("start_ms") or 0)) + end_ms = int(sentence.get("end_ms") or 0) + except (TypeError, ValueError): + return None + if end_ms <= start_ms or start_ms < previous_end: + return None + item = { + "index": expected_index, + "start_ms": start_ms, + "end_ms": end_ms, + "text": (str(sentence.get("text") or "").strip() or None), + } + language = sentence.get("language") + if language: + item["language"] = str(language) + cleaned.append(item) + previous_end = end_ms + last_end = end_ms + if not cleaned: + return None + try: + duration_ms = int(document.get("duration_ms") or 0) + except (TypeError, ValueError): + duration_ms = 0 + if duration_ms < last_end: + duration_ms = last_end + return { + "duration_ms": duration_ms, + "algorithm_version": str(document.get("algorithm_version") or "local-pc"), + "sentences": cleaned, + } + + def _load_translations_worker(self, path, media_hash, expected, generation, local_payload=None): + translations = {} + error = "" + complete = False + cancelled = False + try: + deadline = time.monotonic() + 15 * 60 + kick_job = True + expected_set = set(expected) + last_count = -1 + stalled = 0 + while time.monotonic() < deadline: + if generation != self._translation_generation: + cancelled = True + break + payload = self._request_subtitle_translation(media_hash, "GET") + posted = None + if payload is not None: + translations.update(payload.get("translations") or {}) + complete = bool(payload.get("complete")) or ( + expected_set and expected_set.issubset(translations) + ) + self._ui_events.put( + lambda trans=dict(translations), done=complete: self._apply_translations( + path, trans, done, generation=generation + ) + ) + if complete: + break + if payload.get("status") == "error": + error = str(payload.get("error") or "中文字幕翻译失败") + if len(translations) > last_count: + last_count = len(translations) + stalled = 0 + else: + stalled += 1 + if payload.get("status") != "running" and stalled >= 2: + kick_job = True + else: + kick_job = True + if kick_job: + posted = self._request_subtitle_translation(media_hash, "POST") + if posted is None and local_payload: + posted = self._request_subtitle_translation( + media_hash, "POST", payload=local_payload + ) + kick_job = False + if posted is not None: + translations.update(posted.get("translations") or {}) + complete = bool(posted.get("complete")) or ( + expected_set and expected_set.issubset(translations) + ) + self._ui_events.put( + lambda trans=dict(translations), done=complete: self._apply_translations( + path, trans, done, generation=generation + ) + ) + if complete: + break + if posted.get("status") == "error": + error = str(posted.get("error") or "中文字幕翻译失败") + if payload is None and posted is None: + error = "无法上传英文字幕到服务器" + break + if complete: + break + time.sleep(1.5) + if not complete and not cancelled: + error = error or "中文字幕翻译超时" + except Exception as exc: + error = str(exc) or exc.__class__.__name__ + debug_log( + "subtitle_translation_failed", + media_hash=media_hash, + error_class=exc.__class__.__name__, + error=error, + ) + self._ui_events.put(lambda: self._translation_running.discard(path)) + if cancelled or generation != self._translation_generation: + return + if not complete: + message = error or "中文字幕翻译失败" + self._ui_events.put( + lambda: self._apply_translations( + path, translations, False, error=message, generation=generation + ) + ) + + def _request_subtitle_translation(self, media_hash, method: str, payload=None): + request_url = self.server_url + f"/api/v1/videos/{media_hash}/subtitle-translation" + headers = {"User-Agent": CLIENT_NAME, "Accept": "application/json"} + if self.auth_token: + headers["Authorization"] = f"Bearer {self.auth_token}" + timeout = (15, 30 if method == "POST" else 15) + request_kwargs = { + "method": method, + "url": request_url, + "headers": headers, + "timeout": timeout, + } + if payload is not None: + request_kwargs["json"] = payload + response = requests.request(**request_kwargs) + if response.status_code == 401: + raise RuntimeError("请先在“服务…”中登录") + if response.status_code == 404: + return None + if response.status_code not in range(200, 300): + raise RuntimeError( + f"中文字幕接口返回 {response.status_code}：{self._server_error_detail(response)}" + ) + data = response.json() + items = data.get("translations") or [] + translations = {} + for item in items: + try: + index = int(item.get("index")) + except (TypeError, ValueError, AttributeError): + continue + zh_text = str(item.get("zh_text") or "").strip() + if zh_text: + translations[index] = zh_text + return { + "translations": translations, + "complete": bool(data.get("complete")), + "status": str(data.get("status") or ""), + "error": str(data.get("error") or ""), + } + + def _apply_translations(self, path, translations, complete, error="", generation=None): + if generation is not None and generation != self._translation_generation: + return + document = self._boundaries.get(path) + if not document: + return + sentences = document.get("sentences") or [] + changed = False + for sentence in sentences: + try: + index = int(sentence.get("index")) + except (TypeError, ValueError): + continue + zh_text = str(translations.get(index) or "").strip() + if zh_text and sentence.get("zh_text") != zh_text: + sentence["zh_text"] = zh_text + changed = True + if changed: + media_hash = self._media_hashes.get(path) + if media_hash: + try: + self._save_cached_boundaries(media_hash, document) + except Exception: + pass + self._sync_vlc_subtitles(force=True) + if complete: + self._translation_loading = False + self._translation_error = "" + if path == self._current_path() and self.show_chinese_subtitles.get(): + self.status_var.set("已加载中文字幕") + elif error: + self._translation_loading = False + self._translation_error = error + self._overlay_key = None + if path == self._current_path(): + self._update_on_screen_subtitles() + def _current_sentence(self) -> Optional[Dict[str, Any]]: path = self._current_path() if path is None: @@ -2753,6 +3234,7 @@ class MediaPlayerApp(tk.Tk): self.status_var.set("播放结束") def _on_close(self): + self._translation_generation += 1 self._save_playlist() self.engine.release() self.destroy() diff --git a/sentence_api/README.md b/sentence_api/README.md index 0092d21..501ef8a 100644 --- a/sentence_api/README.md +++ b/sentence_api/README.md @@ -111,6 +111,12 @@ several windows in parallel, retries and splits a window if the model response i invalid, and writes each finished window immediately so later viewers can reuse it. English sentence edits clear the saved Chinese captions for that course. +Desktop playback of a local file can `POST` the same endpoint with the locally +generated English cues (`duration_ms`, `algorithm_version`, `sentences`). If that +SHA-256 is not already a course, the server stores the English subtitles and +translates them; later `GET`/`POST` calls poll and reuse the saved Chinese +captions. Official course sentences are never overwritten by a local upload. + ## Assessment ```bash diff --git a/sentence_api/main.py b/sentence_api/main.py index 48e8d8b..25ac8b6 100644 --- a/sentence_api/main.py +++ b/sentence_api/main.py @@ -24,6 +24,7 @@ from fastapi import ( Request, UploadFile, ) +from pydantic import ValidationError from fastapi.responses import FileResponse, RedirectResponse, Response from fastapi.staticfiles import StaticFiles from starlette.concurrency import run_in_threadpool @@ -47,6 +48,7 @@ from .models import ( SentenceExplanationRequest, SentenceExplanationResponse, AdminSubtitleTranslationResponse, + LocalSubtitleTranslationRequest, SubtitleRetranslateRequest, SubtitleTranslationResponse, PasswordResetRequest, @@ -364,12 +366,28 @@ def create_app( "/api/v1/videos/{video_hash}/subtitle-translation", response_model=SubtitleTranslationResponse, ) - def translate_subtitles( + async def translate_subtitles( + request: Request, video_hash: str = SHA256_PATH, user: AuthenticatedUser = Depends(current_user), ) -> SubtitleTranslationResponse: del user + raw = await request.body() + local_document = None + if raw.strip(): + try: + payload = LocalSubtitleTranslationRequest.model_validate_json(raw) + local_document = SentenceBoundaryDocument( + video_hash=video_hash, + duration_ms=payload.duration_ms, + algorithm_version=payload.algorithm_version, + sentences=payload.sentences, + ) + except ValidationError as exc: + raise HTTPException(status_code=422, detail=exc.errors()) from exc try: + if local_document is not None: + return application.state.subtitle_translations.ensure_from_local(local_document) return application.state.subtitle_translations.ensure(video_hash) except KeyError: raise HTTPException(status_code=404, detail="Course is not available.") from None diff --git a/sentence_api/models.py b/sentence_api/models.py index 85f8796..b428897 100644 --- a/sentence_api/models.py +++ b/sentence_api/models.py @@ -52,7 +52,7 @@ class SentenceBoundaryDocument(BaseModel): return self -VideoStatus = Literal["uploaded", "processing", "ready", "failed"] +VideoStatus = Literal["uploaded", "processing", "ready", "failed", "local_subtitles"] class VideoSummary(BaseModel): @@ -144,6 +144,16 @@ class SubtitleTranslationResponse(BaseModel): error: Optional[str] = None +class LocalSubtitleTranslationRequest(BaseModel): + """Upload locally generated English subtitles so the server can translate them.""" + + model_config = ConfigDict(extra="forbid") + + duration_ms: int = Field(gt=0) + algorithm_version: str = Field(default="local-pc", min_length=1, max_length=64) + sentences: List[SentenceBoundary] = Field(min_length=1) + + class TranslationTerm(BaseModel): model_config = ConfigDict(extra="forbid") diff --git a/sentence_api/repository.py b/sentence_api/repository.py index 9d81665..c251f0d 100644 --- a/sentence_api/repository.py +++ b/sentence_api/repository.py @@ -355,6 +355,98 @@ class VideoRepository: ], ) + def import_local_subtitle_document( + self, + document: SentenceBoundaryDocument, + ) -> SentenceBoundaryDocument: + """Save locally generated English subtitles for translation. + + Official course videos are left unchanged. A previous local-only + import is replaced when the English cues changed. + """ + normalized_hash = normalize_video_hash(document.video_hash) + document = document.model_copy(update={"video_hash": normalized_hash}) + existing_video = self.get_video(normalized_hash) + existing_document = self.get_document(normalized_hash) + if existing_video is not None and existing_video.get("status") != "local_subtitles": + if existing_document is None: + raise KeyError(normalized_hash) + return existing_document + if existing_document is not None and _english_fingerprint(existing_document) == _english_fingerprint(document): + return existing_document + + now = utc_now() + language = next((sentence.language for sentence in document.sentences if sentence.language), None) + with self._connect() as connection: + if existing_video is None: + connection.execute( + """ + INSERT INTO videos ( + video_hash, title, filename, stored_filename, content_type, + size_bytes, duration_ms, language, status, error_message, + algorithm_version, transcription, remote_url, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'local_subtitles', NULL, ?, NULL, NULL, ?, ?) + """, + ( + normalized_hash, + "Local subtitles", + "local-subtitles.json", + "", + "application/json", + 0, + document.duration_ms, + language, + document.algorithm_version, + now, + now, + ), + ) + else: + connection.execute( + """ + UPDATE videos + SET duration_ms = ?, language = COALESCE(?, language), + algorithm_version = ?, status = 'local_subtitles', + error_message = NULL, updated_at = ? + WHERE video_hash = ? + """, + ( + document.duration_ms, + language, + document.algorithm_version, + now, + normalized_hash, + ), + ) + connection.execute( + "DELETE FROM sentences WHERE video_hash = ?", + (normalized_hash,), + ) + connection.executemany( + """ + INSERT INTO sentences ( + video_hash, sentence_index, start_ms, end_ms, text, + language, zh_text, reference_speech_duration_ms + ) VALUES (?, ?, ?, ?, ?, ?, NULL, ?) + """, + [ + ( + normalized_hash, + sentence.index, + sentence.start_ms, + sentence.end_ms, + sentence.text, + sentence.language, + sentence.reference_speech_duration_ms, + ) + for sentence in document.sentences + ], + ) + imported = self.get_document(normalized_hash) + if imported is None: + raise KeyError(normalized_hash) + return imported + def mark_oss_uploaded(self, video_hash: str, remote_url: str) -> None: with self._connect() as connection: connection.execute( @@ -389,6 +481,7 @@ class VideoRepository: SELECT v.*, COUNT(s.sentence_index) AS sentence_count FROM videos v LEFT JOIN sentences s ON s.video_hash = v.video_hash + WHERE v.status != 'local_subtitles' GROUP BY v.video_hash ORDER BY v.created_at DESC """ @@ -1316,6 +1409,13 @@ class VideoRepository: +def _english_fingerprint(document: SentenceBoundaryDocument) -> tuple: + return tuple( + (sentence.index, sentence.start_ms, sentence.end_ms, (sentence.text or "").strip()) + for sentence in document.sentences + ) + + def _sentence_from_row(row: sqlite3.Row) -> SentenceBoundary: keys = set(row.keys()) zh_text = row["zh_text"] if "zh_text" in keys else None diff --git a/sentence_api/tests/test_api.py b/sentence_api/tests/test_api.py index 571879b..b9976da 100644 --- a/sentence_api/tests/test_api.py +++ b/sentence_api/tests/test_api.py @@ -248,6 +248,91 @@ def test_subtitle_translation_requires_login(tmp_path): assert response.status_code == 401 +LOCAL_SUBTITLE_HASH = "b" * 64 + + +def _stub_translator(monkeypatch, client, requested=None): + requested = requested if requested is not None else [] + + def translate_sentences(sentences, **kwargs): + requested.append((sentences, kwargs)) + return {index: f"中文{index}" for index, _text in sentences} + + class StubAITutor: + available = True + + stub = StubAITutor() + stub.translate_sentences = translate_sentences + monkeypatch.setattr(client.app.state, "ai_tutor", stub) + return requested + + +def test_local_english_subtitles_can_be_uploaded_for_translation(tmp_path, monkeypatch): + client = make_client(tmp_path) + requested = _stub_translator(monkeypatch, client) + + missing = client.post(f"/api/v1/videos/{LOCAL_SUBTITLE_HASH}/subtitle-translation") + assert missing.status_code == 404 + + response = client.post( + f"/api/v1/videos/{LOCAL_SUBTITLE_HASH}/subtitle-translation", + json={ + "duration_ms": 9000, + "algorithm_version": "local-pc", + "sentences": [ + {"index": 0, "start_ms": 0, "end_ms": 4000, "text": "Hello.", "language": "en"}, + {"index": 1, "start_ms": 4000, "end_ms": 9000, "text": "Nice to meet you.", "language": "en"}, + ], + }, + ) + assert response.status_code == 200, response.text + if not response.json().get("complete"): + response = wait_for_subtitle_translation(client, LOCAL_SUBTITLE_HASH) + payload = response.json() + assert payload["complete"] is True + assert payload["translations"] == [ + {"index": 0, "zh_text": "中文0"}, + {"index": 1, "zh_text": "中文1"}, + ] + assert requested[0][0] == [(0, "Hello."), (1, "Nice to meet you.")] + + stored = client.get(f"/api/v1/videos/{LOCAL_SUBTITLE_HASH}/subtitle-translation") + assert stored.status_code == 200 + assert stored.json()["translations"] == payload["translations"] + + courses = client.get("/api/v1/courses") + assert courses.status_code == 200 + assert all(item["video_hash"] != LOCAL_SUBTITLE_HASH for item in courses.json()) + + videos = client.get("/api/v1/videos") + assert videos.status_code == 200 + assert all(item["video_hash"] != LOCAL_SUBTITLE_HASH for item in videos.json()["videos"]) + + +def test_local_subtitle_upload_does_not_overwrite_course_sentences(tmp_path, monkeypatch): + client = make_client(tmp_path) + _ready_video(client.app.state.video_repository) + requested = _stub_translator(monkeypatch, client) + + response = client.post( + f"/api/v1/videos/{VIDEO_HASH}/subtitle-translation", + json={ + "duration_ms": 9000, + "algorithm_version": "local-pc", + "sentences": [ + {"index": 0, "start_ms": 0, "end_ms": 4000, "text": "Changed.", "language": "en"}, + {"index": 1, "start_ms": 4000, "end_ms": 9000, "text": "Also changed.", "language": "en"}, + ], + }, + ) + assert response.status_code == 200, response.text + if not response.json().get("complete"): + response = wait_for_subtitle_translation(client, VIDEO_HASH) + assert requested[0][0] == [(0, "Hello."), (1, "Nice to meet you.")] + document = client.app.state.video_repository.get_document(VIDEO_HASH) + assert [sentence.text for sentence in document.sentences] == ["Hello.", "Nice to meet you."] + + def test_subtitle_translation_uses_full_document_and_is_cached(tmp_path, monkeypatch): client = make_client(tmp_path) _ready_video(client.app.state.video_repository) diff --git a/sentence_api/translation_jobs.py b/sentence_api/translation_jobs.py index 53d4caf..383dd58 100644 --- a/sentence_api/translation_jobs.py +++ b/sentence_api/translation_jobs.py @@ -144,6 +144,11 @@ class SubtitleTranslationService: payload["terms"] = terms return AdminSubtitleTranslationResponse(**payload) + def ensure_from_local(self, document: SentenceBoundaryDocument) -> SubtitleTranslationResponse: + """Accept locally generated English subtitles, then translate like a course.""" + self.repository.import_local_subtitle_document(document) + return self.ensure(document.video_hash) + def ensure(self, video_hash: str) -> SubtitleTranslationResponse: if not getattr(self.ai_tutor, "available", False): raise AITutorError("AI translation is not configured.")