Files
mediaplayer/sentence_api/translation_jobs.py
2026-10-25 10:30:22 +08:00

307 lines
11 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, translation_needs_repair
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 repair_missing(self, video_hash: str) -> AdminSubtitleTranslationResponse:
"""Re-translate missing or still-English sentences, keeping good Chinese."""
if self.repository.get_video(video_hash) is None:
raise KeyError(video_hash)
document = self.repository.get_document(video_hash)
if document is None:
raise KeyError(video_hash)
missing = missing_translation_indexes(document)
if missing:
self.invalidate(video_hash)
self.repository.clear_sentence_translations(video_hash, missing)
snapshot = self.ensure(video_hash)
else:
snapshot = self.snapshot(document)
result = self._with_terms(snapshot)
return result.model_copy(update={"attempted_count": len(missing)})
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)
]
document = self.repository.get_document(snapshot.video_hash)
missing_count = 0 if document is None else len(missing_translation_indexes(document))
payload = snapshot.model_dump()
payload["terms"] = terms
payload["missing_count"] = missing_count
payload["attempted_count"] = 0
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()
def missing_translation_indexes(document: SentenceBoundaryDocument) -> list[int]:
return [
sentence.index
for sentence in document.sentences
if translation_needs_repair(sentence.text, sentence.zh_text)
]