268 lines
10 KiB
Python
268 lines
10 KiB
Python
"""AI-backed sentence explanations and subtitle translation."""
|
||
|
||
from dataclasses import dataclass
|
||
import json
|
||
import logging
|
||
import re
|
||
from urllib.parse import urlsplit
|
||
|
||
import requests
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_JSON_FENCE_RE = re.compile(r"^```(?:json)?\s*|\s*```$", re.IGNORECASE)
|
||
_NUMBERED_LINE_RE = re.compile(r"^\s*(\d+)[\.、:：]\s*(.+?)\s*$")
|
||
_TRANSLATE_BATCH_CHARS = 12_000
|
||
|
||
|
||
class AITutorError(RuntimeError):
|
||
"""Raised when the configured AI service cannot answer a request."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class AITutor:
|
||
endpoint: str
|
||
api_key: str
|
||
model: str
|
||
timeout_seconds: float
|
||
|
||
@property
|
||
def available(self) -> bool:
|
||
return bool(self.endpoint and self.model)
|
||
|
||
@property
|
||
def request_url(self) -> str:
|
||
"""Resolve either a full chat URL or an OpenAI-compatible base URL."""
|
||
parsed = urlsplit(self.endpoint)
|
||
path = parsed.path.rstrip("/")
|
||
if parsed.query or path.endswith("/chat/completions"):
|
||
return self.endpoint
|
||
if path.endswith("/v1") or path.endswith("/v1/openai"):
|
||
return self.endpoint + "/chat/completions"
|
||
if not path:
|
||
return self.endpoint + "/v1/chat/completions"
|
||
return self.endpoint
|
||
|
||
def explain(
|
||
self,
|
||
sentence_text: str,
|
||
*,
|
||
course_title: str = "",
|
||
sentence_index: int | None = None,
|
||
language: str = "",
|
||
) -> str:
|
||
if not self.available:
|
||
raise AITutorError("AI explanation is not configured.")
|
||
|
||
context = course_title.strip() or "口语训练课程"
|
||
position = f"第 {sentence_index + 1} 句" if sentence_index is not None else "当前句子"
|
||
language_note = f"句子语言：{language.strip()}。" if language.strip() else ""
|
||
system_prompt = (
|
||
"你是一位耐心的外语口语老师。请用简体中文讲解用户提供的影视/课程句子。"
|
||
"先给出自然的中文翻译，再解释关键词、常用搭配、语法要点，并简要说明这句话适合怎样跟读。"
|
||
"回答使用简明清晰的分段短文，总长度不超过 600 字，不要编造视频里没有的信息。"
|
||
)
|
||
user_prompt = (
|
||
f"课程：{context}\n位置：{position}\n{language_note}句子：{sentence_text}"
|
||
)
|
||
return self._complete(system_prompt, user_prompt, temperature=0.3)
|
||
|
||
def translate_sentences(
|
||
self,
|
||
sentences: list[tuple[int, str]],
|
||
*,
|
||
course_title: str = "",
|
||
) -> dict[int, str]:
|
||
if not self.available:
|
||
raise AITutorError("AI translation is not configured.")
|
||
if not sentences:
|
||
raise AITutorError("No sentences were provided for translation.")
|
||
|
||
translations: dict[int, str] = {}
|
||
numbered = [(index, text.strip()) for index, text in sentences if text.strip()]
|
||
if not numbered:
|
||
raise AITutorError("No sentence text was provided for translation.")
|
||
|
||
for batch, context_rows in _translation_batches(numbered):
|
||
translations.update(
|
||
self._translate_batch(
|
||
batch,
|
||
context_rows=context_rows,
|
||
course_title=course_title,
|
||
total_count=len(numbered),
|
||
)
|
||
)
|
||
missing = [index for index, _ in numbered if not translations.get(index, "").strip()]
|
||
if missing:
|
||
raise AITutorError("AI translation did not cover every sentence.")
|
||
return translations
|
||
|
||
def _translate_batch(
|
||
self,
|
||
batch: list[tuple[int, str]],
|
||
*,
|
||
context_rows: list[tuple[int, str]],
|
||
course_title: str,
|
||
total_count: int,
|
||
) -> dict[int, str]:
|
||
context = course_title.strip() or "口语训练课程"
|
||
display_offset = {index: position for position, (index, _) in enumerate(batch, start=1)}
|
||
system_prompt = (
|
||
"你是专业的影视字幕翻译。请结合全文语境，把指定的外语句子翻译成自然、口语化的简体中文。"
|
||
"必须一句对一句，不能合并、拆分、跳过或改写编号。"
|
||
"只输出 JSON 数组，数组长度必须等于待翻译句子数，第 i 项对应第 i 句中文。"
|
||
"不要输出解释、markdown 或编号。"
|
||
)
|
||
lines = [f"{display_offset[index]}. {text}" for index, text in batch]
|
||
user_parts = [
|
||
f"课程：{context}",
|
||
f"全片共 {total_count} 句，本次需要翻译 {len(batch)} 句。",
|
||
]
|
||
if context_rows:
|
||
user_parts.append("前文（只供理解语境，不要翻译）：")
|
||
user_parts.extend(f"{index + 1}. {text}" for index, text in context_rows)
|
||
user_parts.append("待翻译句子：")
|
||
user_parts.extend(lines)
|
||
content = self._complete(
|
||
system_prompt,
|
||
"\n".join(user_parts),
|
||
temperature=0.2,
|
||
timeout_seconds=max(self.timeout_seconds, 90.0),
|
||
)
|
||
zh_rows = _parse_translation_payload(content, len(batch))
|
||
return {
|
||
index: zh_rows[offset - 1]
|
||
for index, offset in display_offset.items()
|
||
}
|
||
|
||
def _complete(
|
||
self,
|
||
system_prompt: str,
|
||
user_prompt: str,
|
||
*,
|
||
temperature: float,
|
||
timeout_seconds: float | None = None,
|
||
) -> str:
|
||
headers = {"Content-Type": "application/json"}
|
||
if self.api_key:
|
||
headers["Authorization"] = f"Bearer {self.api_key}"
|
||
timeout = self.timeout_seconds if timeout_seconds is None else timeout_seconds
|
||
try:
|
||
response = requests.post(
|
||
self.request_url,
|
||
headers=headers,
|
||
json={
|
||
"model": self.model,
|
||
"messages": [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt},
|
||
],
|
||
"temperature": temperature,
|
||
"stream": False,
|
||
},
|
||
timeout=timeout,
|
||
)
|
||
except requests.RequestException as exc:
|
||
logger.exception(
|
||
"AI request failed: endpoint=%s model=%s",
|
||
self.request_url,
|
||
self.model,
|
||
)
|
||
raise AITutorError("AI service request failed.") from exc
|
||
|
||
if response.status_code < 200 or response.status_code >= 300:
|
||
logger.warning(
|
||
"AI service returned HTTP %s: endpoint=%s model=%s body=%s",
|
||
response.status_code,
|
||
self.request_url,
|
||
self.model,
|
||
response.text[:1000],
|
||
)
|
||
raise AITutorError(f"AI service returned HTTP {response.status_code}.")
|
||
try:
|
||
body = response.json()
|
||
content = body["choices"][0]["message"]["content"]
|
||
except (KeyError, IndexError, TypeError, ValueError) as exc:
|
||
logger.warning(
|
||
"AI service returned an unexpected body: endpoint=%s body=%s",
|
||
self.request_url,
|
||
response.text[:1000],
|
||
)
|
||
raise AITutorError("AI service returned an unexpected response.") from exc
|
||
if not isinstance(content, str) or not content.strip():
|
||
logger.warning(
|
||
"AI service returned empty content: endpoint=%s body=%s",
|
||
self.request_url,
|
||
response.text[:1000],
|
||
)
|
||
raise AITutorError("AI service returned an empty explanation.")
|
||
return content.strip()
|
||
|
||
|
||
def _translation_batches(
|
||
sentences: list[tuple[int, str]],
|
||
) -> list[tuple[list[tuple[int, str]], list[tuple[int, str]]]]:
|
||
batches: list[list[tuple[int, str]]] = []
|
||
current: list[tuple[int, str]] = []
|
||
current_chars = 0
|
||
for item in sentences:
|
||
extra = len(item[1]) + 8
|
||
if current and current_chars + extra > _TRANSLATE_BATCH_CHARS:
|
||
batches.append(current)
|
||
current = []
|
||
current_chars = 0
|
||
current.append(item)
|
||
current_chars += extra
|
||
if current:
|
||
batches.append(current)
|
||
result: list[tuple[list[tuple[int, str]], list[tuple[int, str]]]] = []
|
||
consumed = 0
|
||
for batch in batches:
|
||
context_rows = sentences[max(0, consumed - 8):consumed]
|
||
result.append((batch, context_rows))
|
||
consumed += len(batch)
|
||
return result
|
||
|
||
|
||
def _parse_translation_payload(content: str, expected_count: int) -> list[str]:
|
||
text = _JSON_FENCE_RE.sub("", content.strip()).strip()
|
||
try:
|
||
data = json.loads(text)
|
||
parsed = _translations_from_json(data, expected_count)
|
||
if parsed is not None:
|
||
return parsed
|
||
except (TypeError, ValueError):
|
||
pass
|
||
numbered: dict[int, str] = {}
|
||
for line in text.splitlines():
|
||
match = _NUMBERED_LINE_RE.match(line)
|
||
if match is None:
|
||
continue
|
||
numbered[int(match.group(1))] = match.group(2).strip()
|
||
if len(numbered) == expected_count and all(
|
||
index in numbered and numbered[index] for index in range(1, expected_count + 1)
|
||
):
|
||
return [numbered[index] for index in range(1, expected_count + 1)]
|
||
raise AITutorError("AI translation returned an unexpected response.")
|
||
|
||
|
||
def _translations_from_json(data: object, expected_count: int) -> list[str] | None:
|
||
if isinstance(data, dict):
|
||
for key in ("translations", "sentences", "zh", "items"):
|
||
if key in data:
|
||
data = data[key]
|
||
break
|
||
if isinstance(data, list) and len(data) == expected_count:
|
||
if all(isinstance(item, str) and item.strip() for item in data):
|
||
return [item.strip() for item in data]
|
||
if all(isinstance(item, dict) for item in data):
|
||
values: list[str] = []
|
||
for item in data:
|
||
zh = item.get("zh_text") or item.get("zh") or item.get("text")
|
||
if not isinstance(zh, str) or not zh.strip():
|
||
return None
|
||
values.append(zh.strip())
|
||
return values
|
||
return None
|