add a retranslate chinese function

This commit is contained in:
2026-10-06 14:47:22 +08:00
parent 4fd0fd36a4
commit b28f809c46
11 changed files with 448 additions and 9 deletions

View File

@@ -7,7 +7,13 @@ import threading
from dataclasses import dataclass
from .ai_tutor import AITutor, AITutorError
from .models import SentenceBoundaryDocument, SubtitleTranslationItem, SubtitleTranslationResponse
from .models import (
AdminSubtitleTranslationResponse,
SentenceBoundaryDocument,
SubtitleTranslationItem,
SubtitleTranslationResponse,
TranslationTerm,
)
from .repository import VideoRepository
@@ -110,6 +116,34 @@ class SubtitleTranslationService:
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(self, video_hash: str) -> SubtitleTranslationResponse:
if not getattr(self.ai_tutor, "available", False):
raise AITutorError("AI translation is not configured.")
@@ -200,6 +234,7 @@ class SubtitleTranslationService:
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)