fixed a bug:translate maybe timeout
This commit is contained in:
@@ -1,9 +1,11 @@
|
||||
"""AI-backed sentence explanations and subtitle translation."""
|
||||
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from dataclasses import dataclass
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import threading
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
import requests
|
||||
@@ -12,8 +14,14 @@ 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
|
||||
_NUMBERED_LINE_RE = re.compile(r"^\s*[\[(]?(\d+)[\])]?[\.、:：\)]\s*(.+?)\s*$")
|
||||
_TRANSLATE_BATCH_SENTENCES = 12
|
||||
_TRANSLATE_BATCH_CHARS = 2_200
|
||||
_TRANSLATE_CONTEXT_BEFORE = 8
|
||||
_TRANSLATE_CONTEXT_AFTER = 4
|
||||
_TRANSLATE_WORKERS = 4
|
||||
_TRANSLATE_TIMEOUT_SECONDS = 45.0
|
||||
_TRANSLATE_SPLIT_THRESHOLD = 4
|
||||
|
||||
|
||||
class AITutorError(RuntimeError):
|
||||
@@ -73,68 +81,210 @@ class AITutor:
|
||||
sentences: list[tuple[int, str]],
|
||||
*,
|
||||
course_title: str = "",
|
||||
existing: dict[int, str] | None = None,
|
||||
persist=None,
|
||||
) -> 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:
|
||||
ordered = [(index, text.strip()) for index, text in sentences if text.strip()]
|
||||
if not ordered:
|
||||
raise AITutorError("No sentence text was provided for translation.")
|
||||
|
||||
for batch, context_rows in _translation_batches(numbered):
|
||||
translations.update(
|
||||
self._translate_batch(
|
||||
translations = {
|
||||
index: text.strip()
|
||||
for index, text in (existing or {}).items()
|
||||
if isinstance(text, str) and text.strip()
|
||||
}
|
||||
pending = [(index, text) for index, text in ordered if index not in translations]
|
||||
if not pending:
|
||||
return translations
|
||||
|
||||
persist_lock = threading.Lock()
|
||||
|
||||
def save_partial(partial: dict[int, str]) -> None:
|
||||
if not partial:
|
||||
return
|
||||
translations.update(partial)
|
||||
if persist is None:
|
||||
return
|
||||
with persist_lock:
|
||||
persist(partial)
|
||||
|
||||
batches = _translation_batches(pending, ordered)
|
||||
workers = min(_TRANSLATE_WORKERS, len(batches)) or 1
|
||||
failures: list[tuple[list[tuple[int, str]], Exception]] = []
|
||||
|
||||
def run_batch(batch: list[tuple[int, str]]) -> dict[int, str]:
|
||||
return self._translate_batch_resilient(
|
||||
batch,
|
||||
all_sentences=ordered,
|
||||
existing=translations,
|
||||
course_title=course_title,
|
||||
)
|
||||
|
||||
if workers == 1:
|
||||
for batch in batches:
|
||||
try:
|
||||
save_partial(run_batch(batch))
|
||||
except Exception as exc: # noqa: BLE001 - collect and retry below
|
||||
failures.append((batch, exc))
|
||||
else:
|
||||
with ThreadPoolExecutor(max_workers=workers) as pool:
|
||||
futures = {pool.submit(run_batch, batch): batch for batch in batches}
|
||||
for future in as_completed(futures):
|
||||
batch = futures[future]
|
||||
try:
|
||||
save_partial(future.result())
|
||||
except Exception as exc: # noqa: BLE001
|
||||
failures.append((batch, exc))
|
||||
|
||||
for batch, error in failures:
|
||||
logger.warning(
|
||||
"Translation batch failed, retrying with smaller windows: size=%s error=%s",
|
||||
len(batch),
|
||||
error,
|
||||
)
|
||||
save_partial(
|
||||
self._translate_batch_resilient(
|
||||
batch,
|
||||
context_rows=context_rows,
|
||||
all_sentences=ordered,
|
||||
existing=translations,
|
||||
course_title=course_title,
|
||||
total_count=len(numbered),
|
||||
force_split=len(batch) > 1,
|
||||
)
|
||||
)
|
||||
missing = [index for index, _ in numbered if not translations.get(index, "").strip()]
|
||||
|
||||
missing = [index for index, _ in pending if not translations.get(index, "").strip()]
|
||||
if missing:
|
||||
raise AITutorError("AI translation did not cover every sentence.")
|
||||
return translations
|
||||
|
||||
def _translate_batch_resilient(
|
||||
self,
|
||||
batch: list[tuple[int, str]],
|
||||
*,
|
||||
all_sentences: list[tuple[int, str]],
|
||||
existing: dict[int, str],
|
||||
course_title: str,
|
||||
force_split: bool = False,
|
||||
) -> dict[int, str]:
|
||||
if not batch:
|
||||
return {}
|
||||
if force_split and len(batch) > 1:
|
||||
return self._translate_split_batch(
|
||||
batch,
|
||||
all_sentences=all_sentences,
|
||||
existing=existing,
|
||||
course_title=course_title,
|
||||
)
|
||||
|
||||
last_error: Exception | None = None
|
||||
attempts = 2 if len(batch) <= _TRANSLATE_SPLIT_THRESHOLD else 1
|
||||
for _ in range(attempts):
|
||||
try:
|
||||
return self._translate_batch(
|
||||
batch,
|
||||
all_sentences=all_sentences,
|
||||
existing=existing,
|
||||
course_title=course_title,
|
||||
)
|
||||
except AITutorError as exc:
|
||||
last_error = exc
|
||||
logger.warning(
|
||||
"Translation batch attempt failed: size=%s indexes=%s error=%s",
|
||||
len(batch),
|
||||
[index for index, _ in batch],
|
||||
exc,
|
||||
)
|
||||
|
||||
if len(batch) > 1:
|
||||
return self._translate_split_batch(
|
||||
batch,
|
||||
all_sentences=all_sentences,
|
||||
existing=existing,
|
||||
course_title=course_title,
|
||||
)
|
||||
if last_error is not None:
|
||||
raise last_error
|
||||
raise AITutorError("AI translation did not cover every sentence.")
|
||||
|
||||
def _translate_split_batch(
|
||||
self,
|
||||
batch: list[tuple[int, str]],
|
||||
*,
|
||||
all_sentences: list[tuple[int, str]],
|
||||
existing: dict[int, str],
|
||||
course_title: str,
|
||||
) -> dict[int, str]:
|
||||
mid = max(1, len(batch) // 2)
|
||||
merged: dict[int, str] = {}
|
||||
for part in (batch[:mid], batch[mid:]):
|
||||
merged.update(
|
||||
self._translate_batch_resilient(
|
||||
part,
|
||||
all_sentences=all_sentences,
|
||||
existing=existing,
|
||||
course_title=course_title,
|
||||
)
|
||||
)
|
||||
return merged
|
||||
|
||||
def _translate_batch(
|
||||
self,
|
||||
batch: list[tuple[int, str]],
|
||||
*,
|
||||
context_rows: list[tuple[int, str]],
|
||||
all_sentences: list[tuple[int, str]],
|
||||
existing: dict[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)}
|
||||
existing = dict(existing)
|
||||
before, after = _neighbor_context(batch, all_sentences)
|
||||
expected_indexes = [index for index, _ in batch]
|
||||
system_prompt = (
|
||||
"你是专业的影视字幕翻译。请结合全文语境，把指定的外语句子翻译成自然、口语化的简体中文。"
|
||||
"你是专业的影视字幕翻译。请结合前后文语境，把指定编号的外语句子翻译成自然、口语化的简体中文。"
|
||||
"必须一句对一句，不能合并、拆分、跳过或改写编号。"
|
||||
"只输出 JSON 数组，数组长度必须等于待翻译句子数，第 i 项对应第 i 句中文。"
|
||||
"不要输出解释、markdown 或编号。"
|
||||
"只输出 JSON 对象，键是句子编号，值是对应中文，不要输出解释或 markdown。"
|
||||
)
|
||||
lines = [f"{display_offset[index]}. {text}" for index, text in batch]
|
||||
user_parts = [
|
||||
f"课程：{context}",
|
||||
f"全片共 {total_count} 句，本次需要翻译 {len(batch)} 句。",
|
||||
f"本次需要翻译 {len(batch)} 句，编号必须与输入完全一致。",
|
||||
]
|
||||
if context_rows:
|
||||
glossary = [
|
||||
(index, source, existing[index])
|
||||
for index, source in before
|
||||
if existing.get(index, "").strip()
|
||||
]
|
||||
if glossary:
|
||||
user_parts.append("已有译文（供用词一致，不要重复输出）：")
|
||||
user_parts.extend(
|
||||
f"[{index}] {source} => {zh}" for index, source, zh in glossary[-6:]
|
||||
)
|
||||
elif before:
|
||||
user_parts.append("前文（只供理解语境，不要翻译）：")
|
||||
user_parts.extend(f"{index + 1}. {text}" for index, text in context_rows)
|
||||
user_parts.extend(f"[{index}] {text}" for index, text in before)
|
||||
if after:
|
||||
user_parts.append("后文（只供理解语境，不要翻译）：")
|
||||
user_parts.extend(f"[{index}] {text}" for index, text in after)
|
||||
user_parts.append("待翻译句子：")
|
||||
user_parts.extend(lines)
|
||||
user_parts.extend(f"[{index}] {text}" for index, text in batch)
|
||||
user_parts.append(
|
||||
"输出示例：{" + ",".join(f'"{index}":"中文"' for index in expected_indexes[:2]) + "}"
|
||||
)
|
||||
content = self._complete(
|
||||
system_prompt,
|
||||
"\n".join(user_parts),
|
||||
temperature=0.2,
|
||||
timeout_seconds=max(self.timeout_seconds, 90.0),
|
||||
timeout_seconds=max(self.timeout_seconds, _TRANSLATE_TIMEOUT_SECONDS),
|
||||
)
|
||||
zh_rows = _parse_translation_payload(content, len(batch))
|
||||
return {
|
||||
index: zh_rows[offset - 1]
|
||||
for index, offset in display_offset.items()
|
||||
}
|
||||
parsed = _parse_translation_payload(content, expected_indexes)
|
||||
missing = [index for index in expected_indexes if not parsed.get(index, "").strip()]
|
||||
if missing:
|
||||
raise AITutorError("AI translation returned an unexpected response.")
|
||||
return {index: parsed[index] for index in expected_indexes}
|
||||
|
||||
def _complete(
|
||||
self,
|
||||
@@ -201,67 +351,164 @@ class AITutor:
|
||||
|
||||
|
||||
def _translation_batches(
|
||||
sentences: list[tuple[int, str]],
|
||||
) -> list[tuple[list[tuple[int, str]], list[tuple[int, str]]]]:
|
||||
batches: list[list[tuple[int, str]]] = []
|
||||
pending: list[tuple[int, str]],
|
||||
all_sentences: list[tuple[int, str]],
|
||||
) -> list[list[tuple[int, str]]]:
|
||||
position = {index: offset for offset, (index, _) in enumerate(all_sentences)}
|
||||
ordered_pending = sorted(pending, key=lambda item: position[item[0]])
|
||||
runs: 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)
|
||||
last_pos: int | None = None
|
||||
for item in ordered_pending:
|
||||
pos = position[item[0]]
|
||||
if current and last_pos is not None and pos != last_pos + 1:
|
||||
runs.append(current)
|
||||
current = []
|
||||
current_chars = 0
|
||||
current.append(item)
|
||||
current_chars += extra
|
||||
last_pos = pos
|
||||
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
|
||||
runs.append(current)
|
||||
|
||||
batches: list[list[tuple[int, str]]] = []
|
||||
for run in runs:
|
||||
batch: list[tuple[int, str]] = []
|
||||
batch_chars = 0
|
||||
for item in run:
|
||||
extra = len(item[1]) + 8
|
||||
too_many = len(batch) >= _TRANSLATE_BATCH_SENTENCES
|
||||
too_long = batch and batch_chars + extra > _TRANSLATE_BATCH_CHARS
|
||||
if too_many or too_long:
|
||||
batches.append(batch)
|
||||
batch = []
|
||||
batch_chars = 0
|
||||
batch.append(item)
|
||||
batch_chars += extra
|
||||
if batch:
|
||||
batches.append(batch)
|
||||
return batches
|
||||
|
||||
|
||||
def _parse_translation_payload(content: str, expected_count: int) -> list[str]:
|
||||
def _neighbor_context(
|
||||
batch: list[tuple[int, str]],
|
||||
all_sentences: list[tuple[int, str]],
|
||||
) -> tuple[list[tuple[int, str]], list[tuple[int, str]]]:
|
||||
position = {index: offset for offset, (index, _) in enumerate(all_sentences)}
|
||||
first = position[batch[0][0]]
|
||||
last = position[batch[-1][0]]
|
||||
before = all_sentences[max(0, first - _TRANSLATE_CONTEXT_BEFORE):first]
|
||||
after = all_sentences[last + 1:last + 1 + _TRANSLATE_CONTEXT_AFTER]
|
||||
return before, after
|
||||
|
||||
|
||||
def _parse_translation_payload(content: str, expected_indexes: list[int] | int) -> dict[int, str]:
|
||||
if isinstance(expected_indexes, int):
|
||||
expected = list(range(expected_indexes))
|
||||
else:
|
||||
expected = list(expected_indexes)
|
||||
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:
|
||||
for candidate in (text, *_json_snippets(text)):
|
||||
try:
|
||||
parsed = _translations_from_json(json.loads(candidate), expected)
|
||||
except (TypeError, ValueError):
|
||||
parsed = None
|
||||
if parsed:
|
||||
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)]
|
||||
by_original = {
|
||||
index: numbered[index]
|
||||
for index in expected
|
||||
if numbered.get(index, "").strip()
|
||||
}
|
||||
if len(by_original) == len(expected):
|
||||
return by_original
|
||||
by_order = {
|
||||
expected[offset]: numbered[offset + 1]
|
||||
for offset in range(len(expected))
|
||||
if numbered.get(offset + 1, "").strip()
|
||||
}
|
||||
if len(by_order) == len(expected):
|
||||
return by_order
|
||||
raise AITutorError("AI translation returned an unexpected response.")
|
||||
|
||||
|
||||
def _translations_from_json(data: object, expected_count: int) -> list[str] | None:
|
||||
def _json_snippets(text: str) -> list[str]:
|
||||
snippets: list[str] = []
|
||||
for opener, closer in (("{", "}"), ("[", "]")):
|
||||
start = text.find(opener)
|
||||
while start != -1:
|
||||
depth = 0
|
||||
in_string = False
|
||||
escape = False
|
||||
for offset, char in enumerate(text[start:], start=start):
|
||||
if in_string:
|
||||
if escape:
|
||||
escape = False
|
||||
elif char == "\\":
|
||||
escape = True
|
||||
elif char == '"':
|
||||
in_string = False
|
||||
continue
|
||||
if char == '"':
|
||||
in_string = True
|
||||
elif char == opener:
|
||||
depth += 1
|
||||
elif char == closer:
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
snippets.append(text[start:offset + 1])
|
||||
break
|
||||
start = text.find(opener, start + 1)
|
||||
return snippets
|
||||
|
||||
|
||||
def _translations_from_json(data: object, expected: list[int]) -> dict[int, str] | None:
|
||||
expected_set = set(expected)
|
||||
if isinstance(data, dict):
|
||||
nested = None
|
||||
for key in ("translations", "sentences", "zh", "items"):
|
||||
if key in data:
|
||||
data = data[key]
|
||||
nested = 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())
|
||||
if nested is not None:
|
||||
parsed = _translations_from_json(nested, expected)
|
||||
if parsed:
|
||||
return parsed
|
||||
values: dict[int, str] = {}
|
||||
for raw_key, raw_value in data.items():
|
||||
try:
|
||||
index = int(raw_key)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
if index not in expected_set or not isinstance(raw_value, str) or not raw_value.strip():
|
||||
continue
|
||||
values[index] = raw_value.strip()
|
||||
if len(values) == len(expected):
|
||||
return values
|
||||
return None
|
||||
if isinstance(data, list) and len(data) == len(expected):
|
||||
if all(isinstance(item, str) and item.strip() for item in data):
|
||||
return {
|
||||
index: item.strip()
|
||||
for index, item in zip(expected, data)
|
||||
}
|
||||
if all(isinstance(item, dict) for item in data):
|
||||
values = {}
|
||||
for item in data:
|
||||
raw_index = item.get("index")
|
||||
zh = item.get("zh_text") or item.get("zh") or item.get("text")
|
||||
try:
|
||||
index = int(raw_index)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
if index not in expected_set or not isinstance(zh, str) or not zh.strip():
|
||||
continue
|
||||
values[index] = zh.strip()
|
||||
if len(values) == len(expected):
|
||||
return values
|
||||
return None
|
||||
|
||||
Reference in New Issue
Block a user