261 lines
9.1 KiB
Python
261 lines
9.1 KiB
Python
"""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 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()
|