"""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 ( AdminSubtitleTranslationResponse, SentenceBoundaryDocument, SubtitleTranslationItem, SubtitleTranslationResponse, TranslationTerm, ) from .repository import VideoRepository logger = logging.getLogger(__name__) @dataclass class _Job: video_hash: str done: threading.Event lock: threading.Lock running: bool = True cancelled: bool = False 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 invalidate(self, video_hash: str) -> None: """Stop an in-flight job so stale translations cannot be written back.""" with self._lock: job = self._jobs.pop(video_hash, None) if job is None: return with job.lock: job.cancelled = True def refresh(self, video_hash: str) -> None: """Drop saved Chinese subtitles and translate the current sentences again.""" self.invalidate(video_hash) self.repository.clear_sentence_translations(video_hash) if not getattr(self.ai_tutor, "available", False): return try: self._start_if_needed(video_hash) except (KeyError, ValueError, AITutorError): logger.exception("Could not restart subtitle translation: video=%s", video_hash) def start_missing(self, video_hash: str) -> None: """Translate sentences that still lack Chinese text, keeping existing ones.""" if not getattr(self.ai_tutor, "available", False): return document = self.repository.get_document(video_hash) if document is None: return if self.snapshot(document).complete: return try: self._start_if_needed(video_hash) except (KeyError, ValueError, AITutorError): logger.exception( "Could not start subtitle translation after gap repair: video=%s", video_hash, ) def admin_snapshot(self, video_hash: str) -> AdminSubtitleTranslationResponse | None: current = self.current(video_hash) if current is None: return None return self._with_terms(current) def retranslate( self, video_hash: str, terms: list[tuple[str, str]] | None = None, ) -> AdminSubtitleTranslationResponse: if self.repository.get_video(video_hash) is None: raise KeyError(video_hash) if terms is not None: self.repository.save_translation_glossary(video_hash, terms) self.invalidate(video_hash) self.repository.clear_sentence_translations(video_hash) return self._with_terms(self.ensure(video_hash)) def _with_terms(self, snapshot: SubtitleTranslationResponse) -> AdminSubtitleTranslationResponse: terms = [ TranslationTerm(source=source, zh=zh) for source, zh in self.repository.get_translation_glossary(snapshot.video_hash) ] payload = snapshot.model_dump() 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.") 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 = ( not existing.cancelled and ( 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 _persist(self, job: _Job, translations: dict[int, str]) -> None: with job.lock: if job.cancelled: return self.repository.save_sentence_translations(job.video_hash, translations) def _run_job(self, job: _Job) -> None: try: with job.lock: if job.cancelled: return 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._persist(job, part), terms=self.repository.get_translation_glossary(job.video_hash), ) if translated: self._persist(job, translated) with job.lock: if not job.cancelled: job.error = None except Exception as exc: # noqa: BLE001 - surface job failure to clients with job.lock: cancelled = job.cancelled if cancelled: return 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()