diff --git a/android/sample-app/src/main/java/cn/learningpad/oraltrainer/sample/MainActivity.kt b/android/sample-app/src/main/java/cn/learningpad/oraltrainer/sample/MainActivity.kt index 384c406..f36181d 100644 --- a/android/sample-app/src/main/java/cn/learningpad/oraltrainer/sample/MainActivity.kt +++ b/android/sample-app/src/main/java/cn/learningpad/oraltrainer/sample/MainActivity.kt @@ -357,6 +357,20 @@ class MainActivity : Activity() { } } + private fun applyTranslatedSentences(itemId: String, translations: Map) { + if (translations.isEmpty()) return + val current = controller.currentTrainingItem() ?: return + if (current.id != itemId) return + val updated = current.sentences.map { sentence -> + sentence.copy(zhText = translations[sentence.index] ?: sentence.zhText) + } + controller.updateSentences(current.id, updated) + screenState.courseBoundaries[current.id]?.let { cached -> + screenState.courseBoundaries[current.id] = cached.copy(sentences = updated) + } + applyChineseSubtitle(controller.currentSentence()) + } + private fun ensureChineseTranslations() { if (screenState.activeModule != Module.TRAIN || !screenState.chineseSubtitlesEnabled) { return @@ -382,18 +396,16 @@ class MainActivity : Activity() { item.id, expectedIndexes, object : SubtitleTranslationCallback { + override fun onProgress(result: SubtitleTranslationResult) { + if (!screenState.isCurrentTranslationRequest(generation, item.id)) return + applyTranslatedSentences(item.id, result.translations) + } + override fun onSuccess(result: SubtitleTranslationResult) { if (!screenState.isCurrentTranslationRequest(generation, item.id)) return + applyTranslatedSentences(item.id, result.translations) screenState.chineseTranslationLoading = false - val current = controller.currentTrainingItem() ?: return - if (current.id != item.id) return - val updated = current.sentences.map { sentence -> - sentence.copy(zhText = result.translations[sentence.index] ?: sentence.zhText) - } - controller.updateSentences(current.id, updated) - screenState.courseBoundaries[current.id]?.let { cached -> - screenState.courseBoundaries[current.id] = cached.copy(sentences = updated) - } + screenState.chineseTranslationError = "" applyChineseSubtitle(controller.currentSentence()) if (bindings.hasPlayerChrome) { bindings.statusText.text = "已加载中文字幕" diff --git a/android/sample-app/src/main/java/cn/learningpad/oraltrainer/sample/SubtitleTranslationApi.kt b/android/sample-app/src/main/java/cn/learningpad/oraltrainer/sample/SubtitleTranslationApi.kt index c4ecb15..3db0d0b 100644 --- a/android/sample-app/src/main/java/cn/learningpad/oraltrainer/sample/SubtitleTranslationApi.kt +++ b/android/sample-app/src/main/java/cn/learningpad/oraltrainer/sample/SubtitleTranslationApi.kt @@ -3,9 +3,11 @@ package cn.learningpad.oraltrainer.sample import android.content.Context import android.os.Handler import android.os.Looper +import android.os.SystemClock import org.json.JSONObject import java.io.IOException import java.net.HttpURLConnection +import java.net.SocketTimeoutException import java.net.URL import java.util.concurrent.ExecutorService import java.util.concurrent.Executors @@ -14,9 +16,13 @@ import java.util.concurrent.Future data class SubtitleTranslationResult( val videoHash: String, val translations: Map, + val complete: Boolean = false, + val status: String = "", + val error: String = "", ) interface SubtitleTranslationCallback { + fun onProgress(result: SubtitleTranslationResult) {} fun onSuccess(result: SubtitleTranslationResult) fun onError(error: Throwable) } @@ -46,7 +52,11 @@ internal class SubtitleTranslationApi( ): CancellableTranslationRequest { val future = executor.submit { try { - val result = loadBlocking(videoHash, expectedIndexes) + val result = loadBlocking(videoHash, expectedIndexes) { partial -> + if (!Thread.currentThread().isInterrupted) { + mainHandler.post { callback.onProgress(partial) } + } + } if (!Thread.currentThread().isInterrupted) { mainHandler.post { callback.onSuccess(result) } } @@ -66,15 +76,71 @@ internal class SubtitleTranslationApi( private fun loadBlocking( videoHash: String, expectedIndexes: Collection, + onProgress: (SubtitleTranslationResult) -> Unit, ): SubtitleTranslationResult { - val stored = runCatching { - requestBlocking(videoHash, method = "GET", readTimeoutMs = 15_000) - }.getOrNull() val expected = expectedIndexes.toSet() - if (stored != null && (expected.isEmpty() || expected.all { stored.translations.containsKey(it) })) { - return stored + val deadline = SystemClock.elapsedRealtime() + TRANSLATION_DEADLINE_MS + var kickJob = true + var lastError: Throwable? = null + var lastCount = -1 + var stalledRounds = 0 + while (SystemClock.elapsedRealtime() < deadline) { + if (Thread.currentThread().isInterrupted) { + throw InterruptedException("Subtitle translation cancelled") + } + val stored = runCatching { + requestBlocking(videoHash, method = "GET", readTimeoutMs = 15_000) + }.onFailure { lastError = it }.getOrNull() + if (stored == null) { + kickJob = true + } else { + onProgress(stored) + if (isComplete(stored, expected)) { + return stored + } + if (stored.translations.size > lastCount) { + lastCount = stored.translations.size + stalledRounds = 0 + } else { + stalledRounds += 1 + } + if (stored.status != "running" && stalledRounds >= 2) { + kickJob = true + } + } + if (kickJob) { + val posted = runCatching { + requestBlocking(videoHash, method = "POST", readTimeoutMs = 30_000) + }.onFailure { error -> + lastError = error + if (error !is SocketTimeoutException) { + stalledRounds += 1 + } + }.getOrNull() + kickJob = false + if (posted != null) { + onProgress(posted) + if (isComplete(posted, expected)) { + return posted + } + if (posted.status == "error") { + lastError = IOException(posted.error.ifBlank { "Subtitle translation failed" }) + } + } + } + Thread.sleep(POLL_INTERVAL_MS) } - return requestBlocking(videoHash, method = "POST", readTimeoutMs = 180_000) + throw lastError ?: IOException("Subtitle translation timed out") + } + + private fun isComplete( + result: SubtitleTranslationResult, + expected: Set, + ): Boolean { + if (expected.isNotEmpty()) { + return expected.all { result.translations.containsKey(it) } + } + return result.complete } private fun requestBlocking( @@ -115,9 +181,17 @@ internal class SubtitleTranslationApi( return SubtitleTranslationResult( videoHash = root.getString("video_hash"), translations = translations, + complete = root.optBoolean("complete", false), + status = root.optString("status"), + error = root.optString("error"), ) } finally { connection.disconnect() } } + + private companion object { + const val POLL_INTERVAL_MS = 1_500L + const val TRANSLATION_DEADLINE_MS = 15 * 60 * 1000L + } } diff --git a/ios/OralTrainer/Config.swift b/ios/OralTrainer/Config.swift index 37cc0d2..a7c3141 100644 --- a/ios/OralTrainer/Config.swift +++ b/ios/OralTrainer/Config.swift @@ -15,7 +15,9 @@ enum AppConfig { static let connectTimeout: TimeInterval = 15 static let readTimeout: TimeInterval = 30 static let assessmentTimeout: TimeInterval = 180 - static let translationTimeout: TimeInterval = 180 + static let translationTimeout: TimeInterval = 30 + static let translationDeadline: TimeInterval = 900 + static let translationPollIntervalNs: UInt64 = 1_500_000_000 static let sha256Pattern = #"^[0-9a-fA-F]{64}$"# diff --git a/ios/OralTrainer/Services/SentenceBoundaryAPI.swift b/ios/OralTrainer/Services/SentenceBoundaryAPI.swift index fe15581..ec20ab7 100644 --- a/ios/OralTrainer/Services/SentenceBoundaryAPI.swift +++ b/ios/OralTrainer/Services/SentenceBoundaryAPI.swift @@ -59,10 +59,24 @@ struct SubtitleTranslationItem: Decodable { struct SubtitleTranslationResponse: Decodable { let videoHash: String let translations: [SubtitleTranslationItem] + let complete: Bool? + let status: String? + let error: String? private enum CodingKeys: String, CodingKey { case translations case videoHash = "video_hash" + case complete + case status + case error + } + + var mapped: [Int: String] { + var values: [Int: String] = [:] + for item in translations where !item.zhText.isEmpty { + values[item.index] = item.zhText + } + return values } } @@ -100,29 +114,90 @@ extension SentenceBoundaryAPI { .explanation } - static func fetchStoredSubtitles(videoHash: String) async throws -> [Int: String] { + static func fetchStoredSubtitles(videoHash: String) async throws -> SubtitleTranslationResponse { try await requestSubtitles(videoHash: videoHash, method: "GET", timeout: AppConfig.readTimeout) } - static func translateSubtitles(videoHash: String) async throws -> [Int: String] { + static func translateSubtitles(videoHash: String) async throws -> SubtitleTranslationResponse { try await requestSubtitles(videoHash: videoHash, method: "POST", timeout: AppConfig.translationTimeout) } - static func loadSubtitles(videoHash: String, expectedIndexes: [Int]) async throws -> [Int: String] { - if let stored = try? await fetchStoredSubtitles(videoHash: videoHash) { - let expected = Set(expectedIndexes) - if expected.isEmpty || expected.isSubset(of: Set(stored.keys)) { - return stored + static func loadSubtitles( + videoHash: String, + expectedIndexes: [Int], + onProgress: (([Int: String]) -> Void)? = nil + ) async throws -> [Int: String] { + let expected = Set(expectedIndexes) + let deadline = Date().addingTimeInterval(AppConfig.translationDeadline) + var kickJob = true + var lastError: Error? + var lastCount = -1 + var stalledRounds = 0 + while Date() < deadline { + try Task.checkCancellation() + if let stored = try? await fetchStoredSubtitles(videoHash: videoHash) { + let mapped = stored.mapped + onProgress?(mapped) + if isComplete(mapped, expected: expected, payload: stored) { + return mapped + } + if mapped.count > lastCount { + lastCount = mapped.count + stalledRounds = 0 + } else { + stalledRounds += 1 + } + if stored.status != "running" && stalledRounds >= 2 { + kickJob = true + } + } else { + kickJob = true } + if kickJob { + do { + let posted = try await translateSubtitles(videoHash: videoHash) + let mapped = posted.mapped + onProgress?(mapped) + if isComplete(mapped, expected: expected, payload: posted) { + return mapped + } + if posted.status == "error" { + lastError = HTTPClientError.badStatus(503, posted.error ?? "Subtitle translation failed") + } + } catch { + lastError = error + if !isTimeout(error) { + stalledRounds += 1 + } + } + kickJob = false + } + try await Task.sleep(nanoseconds: AppConfig.translationPollIntervalNs) } - return try await translateSubtitles(videoHash: videoHash) + throw lastError ?? HTTPClientError.badStatus(408, "Subtitle translation timed out") + } + + private static func isComplete( + _ mapped: [Int: String], + expected: Set, + payload: SubtitleTranslationResponse + ) -> Bool { + if !expected.isEmpty { + return expected.isSubset(of: Set(mapped.keys)) + } + return payload.complete == true + } + + private static func isTimeout(_ error: Error) -> Bool { + let nsError = error as NSError + return nsError.domain == NSURLErrorDomain && nsError.code == NSURLErrorTimedOut } private static func requestSubtitles( videoHash: String, method: String, timeout: TimeInterval - ) async throws -> [Int: String] { + ) async throws -> SubtitleTranslationResponse { let hash = videoHash.lowercased() guard hash.range(of: AppConfig.sha256Pattern, options: .regularExpression) != nil else { throw SentenceBoundaryAPIError.invalidHash @@ -142,11 +217,6 @@ extension SentenceBoundaryAPI { let detail = String(data: result.body, encoding: .utf8) ?? "" throw HTTPClientError.badStatus(result.statusCode, detail) } - let payload = try JSONDecoder().decode(SubtitleTranslationResponse.self, from: result.body) - var mapped: [Int: String] = [:] - for item in payload.translations where !item.zhText.isEmpty { - mapped[item.index] = item.zhText - } - return mapped + return try JSONDecoder().decode(SubtitleTranslationResponse.self, from: result.body) } } diff --git a/ios/OralTrainer/Views/RootView.swift b/ios/OralTrainer/Views/RootView.swift index f5a2ff9..9758ea3 100644 --- a/ios/OralTrainer/Views/RootView.swift +++ b/ios/OralTrainer/Views/RootView.swift @@ -627,6 +627,15 @@ struct RootView: View { return "暂无中文字幕" } + @MainActor + private func applyChineseTranslations(_ map: [Int: String], mediaId: String) { + guard controller.currentItem?.id == mediaId else { return } + let updated = (controller.currentItem?.sentences ?? []).map { sentence in + sentence.withZhText(map[sentence.index] ?? sentence.zhText) + } + controller.updateBoundaries(updated) + } + private func ensureChineseTranslations() { guard showChineseSubtitles, activeModule == .train else { return @@ -656,14 +665,14 @@ struct RootView: View { let map = try await SentenceBoundaryAPI.loadSubtitles( videoHash: mediaId, expectedIndexes: expectedIndexes - ) + ) { partial in + Task { @MainActor in + applyChineseTranslations(partial, mediaId: mediaId) + } + } guard !Task.isCancelled else { return } await MainActor.run { - guard controller.currentItem?.id == mediaId else { return } - let updated = (controller.currentItem?.sentences ?? []).map { sentence in - sentence.withZhText(map[sentence.index] ?? sentence.zhText) - } - controller.updateBoundaries(updated) + applyChineseTranslations(map, mediaId: mediaId) chineseTranslationLoading = false chineseTranslationError = "" controller.showTransientMessage("已加载中文字幕") diff --git a/sentence_api/README.md b/sentence_api/README.md index bee32d1..f6629e0 100644 --- a/sentence_api/README.md +++ b/sentence_api/README.md @@ -102,11 +102,12 @@ to the mobile clients. Chinese captions are stored on the course in the server database and shared by every user. `GET /api/v1/videos/{sha256}/subtitle-translation` returns the saved -captions without calling the AI. `POST` generates any missing captions in small -windows with neighboring context, runs several windows in parallel, retries and -splits a window if the model response is invalid, and writes each finished -window immediately so later viewers can reuse it. English sentence edits clear -the saved Chinese captions for that course. +captions without calling the AI, including `complete` and `status` so clients can +poll long videos. `POST` starts missing-caption generation in the background and +returns quickly; it translates in small windows with neighboring context, runs +several windows in parallel, retries and splits a window if the model response is +invalid, and writes each finished window immediately so later viewers can reuse +it. English sentence edits clear the saved Chinese captions for that course. ## Assessment diff --git a/sentence_api/main.py b/sentence_api/main.py index cdf2ddc..cbf4789 100644 --- a/sentence_api/main.py +++ b/sentence_api/main.py @@ -29,6 +29,7 @@ from fastapi.staticfiles import StaticFiles from starlette.concurrency import run_in_threadpool from starlette.requests import ClientDisconnect from .ai_tutor import AITutor, AITutorError +from .translation_jobs import SubtitleTranslationService from .assessment import AssessmentService from .auth import AuthenticatedUser @@ -45,7 +46,6 @@ from .models import ( SentenceBoundaryDocument, SentenceExplanationRequest, SentenceExplanationResponse, - SubtitleTranslationItem, SubtitleTranslationResponse, PasswordResetRequest, SentenceTextUpdate, @@ -129,6 +129,10 @@ def create_app( application.state.oss_uploader = oss application.state.video_processor = processor application.state.assessment_service = assessment_service + application.state.subtitle_translations = SubtitleTranslationService( + video_repository, + lambda: application.state.ai_tutor, + ) static_dir = Path(__file__).resolve().parent / "static" if static_dir.is_dir(): @@ -338,14 +342,7 @@ def create_app( def stored_subtitle_translations( document: SentenceBoundaryDocument, ) -> SubtitleTranslationResponse: - return SubtitleTranslationResponse( - video_hash=document.video_hash, - translations=[ - SubtitleTranslationItem(index=sentence.index, zh_text=sentence.zh_text) - for sentence in document.sentences - if sentence.zh_text and sentence.zh_text.strip() - ], - ) + return application.state.subtitle_translations.snapshot(document) @application.get( "/api/v1/videos/{video_hash}/subtitle-translation", @@ -356,10 +353,10 @@ def create_app( user: AuthenticatedUser = Depends(current_user), ) -> SubtitleTranslationResponse: del user - document = video_repository.get_document(video_hash) - if document is None: + snapshot = application.state.subtitle_translations.current(video_hash) + if snapshot is None: raise HTTPException(status_code=404, detail="Course is not available.") - return stored_subtitle_translations(document) + return snapshot @application.post( "/api/v1/videos/{video_hash}/subtitle-translation", @@ -370,52 +367,14 @@ def create_app( user: AuthenticatedUser = Depends(current_user), ) -> SubtitleTranslationResponse: del user - document = video_repository.get_document(video_hash) - if document is None: - raise HTTPException(status_code=404, detail="Course is not available.") - payload = [ - (sentence.index, sentence.text or "") - for sentence in document.sentences - if (sentence.text or "").strip() - ] - if not payload: - raise HTTPException(status_code=422, detail="No sentence text is available to translate.") - cached = stored_subtitle_translations(document).translations - complete = len(cached) == len(payload) and all( - (sentence.zh_text or "").strip() - for sentence in document.sentences - if (sentence.text or "").strip() - ) - if complete: - return stored_subtitle_translations(document) - video = video_repository.get_video(video_hash) - existing = { - item.index: item.zh_text - for item in cached - } try: - translated = application.state.ai_tutor.translate_sentences( - payload, - course_title=(video["title"] if video is not None else "") or "", - existing=existing, - persist=lambda part: video_repository.save_sentence_translations( - video_hash, - part, - ), - ) + return application.state.subtitle_translations.ensure(video_hash) + except KeyError: + raise HTTPException(status_code=404, detail="Course is not available.") from None + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc except AITutorError as exc: raise HTTPException(status_code=503, detail=str(exc)) from exc - saved = video_repository.save_sentence_translations(video_hash, translated) - if saved is None: - raise HTTPException(status_code=404, detail="Course is not available.") - return SubtitleTranslationResponse( - video_hash=saved.video_hash, - translations=[ - SubtitleTranslationItem(index=sentence.index, zh_text=sentence.zh_text) - for sentence in saved.sentences - if sentence.zh_text and sentence.zh_text.strip() - ], - ) @application.post( "/api/v1/audio/sentence-boundaries", diff --git a/sentence_api/models.py b/sentence_api/models.py index d59f57a..5893dfd 100644 --- a/sentence_api/models.py +++ b/sentence_api/models.py @@ -139,6 +139,9 @@ class SubtitleTranslationResponse(BaseModel): video_hash: str translations: List[SubtitleTranslationItem] + complete: bool = True + status: str = "ready" + error: Optional[str] = None class SentenceBoundaryAdjust(BaseModel): diff --git a/sentence_api/tests/test_api.py b/sentence_api/tests/test_api.py index 0984a08..22f129a 100644 --- a/sentence_api/tests/test_api.py +++ b/sentence_api/tests/test_api.py @@ -1,4 +1,6 @@ import json +import threading +import time from dataclasses import replace from fastapi.testclient import TestClient @@ -14,6 +16,18 @@ from sentence_api.store import BoundaryStore VIDEO_HASH = "a" * 64 +def wait_for_subtitle_translation(client, video_hash, timeout=5.0): + deadline = time.time() + timeout + last = None + while time.time() < deadline: + last = client.get(f"/api/v1/videos/{video_hash}/subtitle-translation") + if last.status_code == 200 and last.json().get("complete"): + return last + time.sleep(0.05) + raise AssertionError(f"subtitle translation did not complete: {getattr(last, 'text', last)}") + + + def make_client(tmp_path, admin_api_key=""): index_path = tmp_path / "boundaries.json" @@ -172,7 +186,7 @@ def test_sentence_explanation_uses_server_side_ai_client(tmp_path, monkeypatch): return "这句话表示打招呼。" class StubAITutor: - pass + available = True stub = StubAITutor() stub.explain = explain @@ -244,7 +258,7 @@ def test_subtitle_translation_uses_full_document_and_is_cached(tmp_path, monkeyp return {index: f"中文{index}" for index, _text in sentences} class StubAITutor: - pass + available = True stub = StubAITutor() stub.translate_sentences = translate_sentences @@ -252,8 +266,12 @@ def test_subtitle_translation_uses_full_document_and_is_cached(tmp_path, monkeyp response = client.post(f"/api/v1/videos/{VIDEO_HASH}/subtitle-translation") assert response.status_code == 200, response.text + if not response.json().get("complete"): + response = wait_for_subtitle_translation(client, VIDEO_HASH) payload = response.json() assert payload["video_hash"] == VIDEO_HASH + assert payload["complete"] is True + assert payload["status"] == "ready" assert payload["translations"] == [ {"index": 0, "zh_text": "中文0"}, {"index": 1, "zh_text": "中文1"}, @@ -263,7 +281,8 @@ def test_subtitle_translation_uses_full_document_and_is_cached(tmp_path, monkeyp second = client.post(f"/api/v1/videos/{VIDEO_HASH}/subtitle-translation") assert second.status_code == 200 - assert second.json() == payload + assert second.json()["translations"] == payload["translations"] + assert second.json()["complete"] is True assert len(requested) == 1 assert requested[0][0] == [(0, "Hello."), (1, "Nice to meet you.")] @@ -278,7 +297,7 @@ def test_subtitle_translation_is_shared_with_other_users(tmp_path, monkeypatch): return {index: f"中文{index}" for index, _text in sentences} class StubAITutor: - pass + available = True stub = StubAITutor() stub.translate_sentences = translate_sentences @@ -286,10 +305,13 @@ def test_subtitle_translation_is_shared_with_other_users(tmp_path, monkeypatch): first = client.post(f"/api/v1/videos/{VIDEO_HASH}/subtitle-translation") assert first.status_code == 200, first.text + if not first.json().get("complete"): + first = wait_for_subtitle_translation(client, VIDEO_HASH) stored = client.get(f"/api/v1/videos/{VIDEO_HASH}/subtitle-translation") assert stored.status_code == 200 - assert stored.json() == first.json() + assert stored.json()["translations"] == first.json()["translations"] + assert stored.json()["complete"] is True other = TestClient(client.app) signup = other.post( @@ -311,7 +333,61 @@ def test_subtitle_translation_is_shared_with_other_users(tmp_path, monkeypatch): reused = other.post(f"/api/v1/videos/{VIDEO_HASH}/subtitle-translation") assert reused.status_code == 200 - assert reused.json() == first.json() + assert reused.json()["translations"] == first.json()["translations"] + assert reused.json()["complete"] is True + assert len(requested) == 1 + + + +def test_subtitle_translation_does_not_block_http_on_long_jobs(tmp_path, monkeypatch): + client = make_client(tmp_path) + _ready_video(client.app.state.video_repository) + started = threading.Event() + release = threading.Event() + requested = [] + + def translate_sentences(sentences, **kwargs): + requested.append(sentences) + persist = kwargs.get("persist") + if persist is not None: + persist({0: "中文0"}) + started.set() + assert release.wait(timeout=5) + if persist is not None: + persist({1: "中文1"}) + return {0: "中文0", 1: "中文1"} + + class StubAITutor: + available = True + + stub = StubAITutor() + stub.translate_sentences = translate_sentences + monkeypatch.setattr(client.app.state, "ai_tutor", stub) + client.app.state.subtitle_translations.ai_tutor = stub + client.app.state.subtitle_translations.post_wait_seconds = 0.2 + + response = client.post(f"/api/v1/videos/{VIDEO_HASH}/subtitle-translation") + assert started.wait(timeout=2) + assert response.status_code == 200, response.text + payload = response.json() + assert payload["complete"] is False + assert payload["status"] == "running" + if {"index": 0, "zh_text": "中文0"} not in payload["translations"]: + payload = client.get(f"/api/v1/videos/{VIDEO_HASH}/subtitle-translation").json() + assert {"index": 0, "zh_text": "中文0"} in payload["translations"] + + stored = client.get(f"/api/v1/videos/{VIDEO_HASH}/subtitle-translation") + assert stored.status_code == 200 + assert stored.json()["status"] == "running" + assert stored.json()["complete"] is False + + release.set() + finished = wait_for_subtitle_translation(client, VIDEO_HASH) + assert finished.json()["complete"] is True + assert finished.json()["translations"] == [ + {"index": 0, "zh_text": "中文0"}, + {"index": 1, "zh_text": "中文1"}, + ] assert len(requested) == 1 diff --git a/sentence_api/translation_jobs.py b/sentence_api/translation_jobs.py new file mode 100644 index 0000000..c31e638 --- /dev/null +++ b/sentence_api/translation_jobs.py @@ -0,0 +1,181 @@ +"""Background subtitle translation so long videos do not block HTTP.""" + +from __future__ import annotations + +import logging +import threading +from dataclasses import dataclass + +from .ai_tutor import AITutor, AITutorError +from .models import SentenceBoundaryDocument, SubtitleTranslationItem, SubtitleTranslationResponse +from .repository import VideoRepository + + +logger = logging.getLogger(__name__) + + +@dataclass +class _Job: + video_hash: str + done: threading.Event + lock: threading.Lock + running: bool = True + error: str | None = None + thread: threading.Thread | None = None + + +class SubtitleTranslationService: + def __init__( + self, + repository: VideoRepository, + ai_tutor: AITutor, + *, + post_wait_seconds: float = 8.0, + ) -> None: + self.repository = repository + self._ai_tutor = ai_tutor + self.post_wait_seconds = post_wait_seconds + self._lock = threading.Lock() + self._jobs: dict[str, _Job] = {} + + @property + def ai_tutor(self) -> AITutor: + value = self._ai_tutor + return value() if callable(value) else value + + @ai_tutor.setter + def ai_tutor(self, value) -> None: + self._ai_tutor = value + + def snapshot(self, document: SentenceBoundaryDocument) -> SubtitleTranslationResponse: + translations = [ + SubtitleTranslationItem(index=sentence.index, zh_text=sentence.zh_text) + for sentence in document.sentences + if sentence.zh_text and sentence.zh_text.strip() + ] + needed = [ + sentence + for sentence in document.sentences + if (sentence.text or "").strip() + ] + complete = bool(needed) and all((sentence.zh_text or "").strip() for sentence in needed) + job = self._jobs.get(document.video_hash) + running = False + error: str | None = None + if job is not None: + with job.lock: + running = job.running + error = job.error + if complete: + status = "ready" + error = None + elif running: + status = "running" + elif error: + status = "error" + else: + status = "idle" + return SubtitleTranslationResponse( + video_hash=document.video_hash, + translations=translations, + complete=complete, + status=status, + error=error, + ) + + def current(self, video_hash: str) -> SubtitleTranslationResponse | None: + document = self.repository.get_document(video_hash) + if document is None: + return None + return self.snapshot(document) + + def ensure(self, video_hash: str) -> SubtitleTranslationResponse: + if not getattr(self.ai_tutor, "available", False): + raise AITutorError("AI translation is not configured.") + document = self.repository.get_document(video_hash) + if document is None: + raise KeyError(video_hash) + current = self.snapshot(document) + if current.complete: + return current + payload = [ + (sentence.index, sentence.text or "") + for sentence in document.sentences + if (sentence.text or "").strip() + ] + if not payload: + raise ValueError("No sentence text is available to translate.") + job = self._start_if_needed(video_hash) + wait_seconds = max(0.0, float(self.post_wait_seconds)) + if wait_seconds > 0: + job.done.wait(timeout=wait_seconds) + latest = self.current(video_hash) + if latest is None: + raise KeyError(video_hash) + return latest + + def _start_if_needed(self, video_hash: str) -> _Job: + with self._lock: + existing = self._jobs.get(video_hash) + if existing is not None: + with existing.lock: + still_running = existing.running or ( + existing.thread is not None and existing.thread.is_alive() + ) + if still_running: + return existing + job = _Job( + video_hash=video_hash, + done=threading.Event(), + lock=threading.Lock(), + ) + thread = threading.Thread( + target=self._run_job, + args=(job,), + name=f"subtitle-translate-{video_hash[:8]}", + daemon=True, + ) + job.thread = thread + self._jobs[video_hash] = job + thread.start() + return job + + def _run_job(self, job: _Job) -> None: + try: + document = self.repository.get_document(job.video_hash) + if document is None: + raise AITutorError("Course is not available.") + payload = [ + (sentence.index, sentence.text or "") + for sentence in document.sentences + if (sentence.text or "").strip() + ] + if not payload: + return + existing = { + sentence.index: sentence.zh_text + for sentence in document.sentences + if sentence.zh_text and sentence.zh_text.strip() + } + video = self.repository.get_video(job.video_hash) + translated = self.ai_tutor.translate_sentences( + payload, + course_title=(video["title"] if video is not None else "") or "", + existing=existing, + persist=lambda part: self.repository.save_sentence_translations( + job.video_hash, + part, + ), + ) + if translated: + self.repository.save_sentence_translations(job.video_hash, translated) + with job.lock: + job.error = None + except Exception as exc: # noqa: BLE001 - surface job failure to clients + logger.exception("Subtitle translation job failed: video=%s", job.video_hash) + with job.lock: + job.error = str(exc) + finally: + with job.lock: + job.running = False + job.done.set()