339 lines
13 KiB
Python
339 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
r"""
|
||
translate.py — перевод .nsv файлов через OpenAI-совместимый API.
|
||
Включает глобальный кэш переводов, сырое логирование запросов,
|
||
все системные файлы аккуратно складываются в папку _translate_data.
|
||
"""
|
||
|
||
import json
|
||
import re
|
||
import sys
|
||
import time
|
||
import urllib.request
|
||
import urllib.error
|
||
import logging
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from nsv_io import load_nsv_file, save_nsv_file
|
||
|
||
BASE_DIR = Path(__file__).resolve().parent
|
||
CONFIG_PATH = BASE_DIR / "config.json"
|
||
PROMPT_PATH = BASE_DIR / "prompt.txt"
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Настройка системных папок (кэш, логи)
|
||
# --------------------------------------------------------------------------
|
||
SYS_DIR = BASE_DIR / "_translate_data"
|
||
SYS_DIR.mkdir(exist_ok=True)
|
||
|
||
CACHE_PATH = SYS_DIR / "translation_cache.json"
|
||
|
||
LOG_DIR = SYS_DIR / "translate_logs"
|
||
LOG_DIR.mkdir(exist_ok=True)
|
||
log_filename = LOG_DIR / f"translation_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
|
||
|
||
RAW_LOG_DIR = SYS_DIR / "raw_api_logs"
|
||
RAW_LOG_DIR.mkdir(exist_ok=True)
|
||
raw_log_filename = RAW_LOG_DIR / f"raw_llm_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jsonl"
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(message)s",
|
||
handlers=[
|
||
logging.FileHandler(log_filename, encoding="utf-8"),
|
||
logging.StreamHandler(sys.stdout)
|
||
]
|
||
)
|
||
|
||
def log_raw_api(request_body, raw_response_dict, error_msg=None):
|
||
"""Пишет сырой запрос и ответ в jsonl файл для дебага."""
|
||
entry = {
|
||
"timestamp": datetime.now().isoformat(),
|
||
"request": request_body,
|
||
"response": raw_response_dict,
|
||
"error": error_msg
|
||
}
|
||
with open(raw_log_filename, "a", encoding="utf-8") as f:
|
||
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Загрузка конфига, глоссария, промпта и кэша
|
||
# --------------------------------------------------------------------------
|
||
def load_config():
|
||
if not CONFIG_PATH.exists():
|
||
logging.error(f"Ошибка: не найден файл конфигурации {CONFIG_PATH}")
|
||
sys.exit(1)
|
||
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
|
||
def load_glossary(glossary_path: Path) -> str:
|
||
if not glossary_path.exists():
|
||
return ""
|
||
with open(glossary_path, "r", encoding="utf-8") as f:
|
||
return f.read().strip()
|
||
|
||
def load_prompt() -> str:
|
||
if not PROMPT_PATH.exists():
|
||
logging.error(f"Ошибка: не найден файл промпта {PROMPT_PATH}. Создайте его.")
|
||
sys.exit(1)
|
||
with open(PROMPT_PATH, "r", encoding="utf-8") as f:
|
||
return f.read()
|
||
|
||
def load_cache() -> dict:
|
||
if CACHE_PATH.exists():
|
||
try:
|
||
with open(CACHE_PATH, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except json.JSONDecodeError:
|
||
logging.warning("Предупреждение: файл кэша поврежден, создаем новый.")
|
||
return {}
|
||
|
||
def save_cache(cache_data: dict):
|
||
with open(CACHE_PATH, "w", encoding="utf-8") as f:
|
||
json.dump(cache_data, f, ensure_ascii=False, indent=2)
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Фильтр текста
|
||
# --------------------------------------------------------------------------
|
||
def sanitize_text(text: str) -> str:
|
||
"""Очищает текст от символов, несовместимых со стандартным Shift-JIS (cp932)."""
|
||
if not text:
|
||
return ""
|
||
replacements = {
|
||
'—': '―', '–': '-', '−': '-',
|
||
'\xa0': ' ', '\u200b': '', '\u200c': '', '\u200d': '', '\ufeff': '',
|
||
'…': '...', '№': '#', '́': ''
|
||
}
|
||
for bad_char, good_char in replacements.items():
|
||
text = text.replace(bad_char, good_char)
|
||
return text
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Построение промпта
|
||
# --------------------------------------------------------------------------
|
||
def format_glossary_block(glossary_text: str) -> str:
|
||
if not glossary_text:
|
||
return ""
|
||
return f"\n=== GLOSSARY ===\n{glossary_text}\n=== END GLOSSARY ===\n"
|
||
|
||
def format_context_block(label: str, entries) -> str:
|
||
if not entries:
|
||
return ""
|
||
lines_str = "\n".join(f'[{e["line"]}] {e["text"]}' for e in entries)
|
||
return f"\n=== {label} (reference only, do not translate) ===\n{lines_str}\n"
|
||
|
||
def format_batch_block(entries) -> str:
|
||
lines_str = "\n".join(f'[{e["line"]}] {e["text"]}' for e in entries)
|
||
return f"\n=== LINES TO TRANSLATE ===\n{lines_str}\n"
|
||
|
||
def build_messages(cfg, glossary_text, context_before_entries, batch_entries, context_after_entries):
|
||
tr_cfg = cfg["translation"]
|
||
system_prompt = load_prompt().format(source_lang=tr_cfg["source_language"], target_lang=tr_cfg["target_language"])
|
||
system_prompt += format_glossary_block(glossary_text)
|
||
|
||
user_parts = [
|
||
format_context_block("CONTEXT BEFORE", context_before_entries),
|
||
format_context_block("CONTEXT AFTER", context_after_entries),
|
||
format_batch_block(batch_entries)
|
||
]
|
||
return [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": "".join(p for p in user_parts if p)}
|
||
]
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Вызов API
|
||
# --------------------------------------------------------------------------
|
||
def call_llm(cfg, messages):
|
||
api_cfg = cfg["api"]
|
||
url = api_cfg["base_url"].rstrip("/") + "/chat/completions"
|
||
|
||
body = {
|
||
"model": api_cfg["model"],
|
||
"messages": messages,
|
||
"max_tokens": api_cfg.get("max_tokens", 4000),
|
||
"temperature": api_cfg.get("temperature", 0.3),
|
||
}
|
||
|
||
if "reasoning_effort" in api_cfg:
|
||
body["reasoning_effort"] = api_cfg["reasoning_effort"]
|
||
|
||
data = json.dumps(body).encode("utf-8")
|
||
req = urllib.request.Request(
|
||
url, data=data,
|
||
headers={"Content-Type": "application/json", "Authorization": f"Bearer {api_cfg['api_key']}"},
|
||
method="POST"
|
||
)
|
||
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=api_cfg.get("timeout_seconds", 120)) as resp:
|
||
resp_body = resp.read().decode("utf-8")
|
||
resp_json = json.loads(resp_body)
|
||
|
||
log_raw_api(body, resp_json)
|
||
|
||
return resp_json["choices"][0]["message"]["content"]
|
||
except urllib.error.URLError as e:
|
||
log_raw_api(body, None, str(e))
|
||
raise
|
||
|
||
def extract_json_array(text: str):
|
||
original_text = text
|
||
text = text.strip()
|
||
fence_match = re.search(r"```(?:json)?\s*(.*?)```", text, re.DOTALL)
|
||
if fence_match:
|
||
text = fence_match.group(1).strip()
|
||
if not text.startswith("["):
|
||
start, end = text.find("["), text.rfind("]")
|
||
if start != -1 and end != -1 and end > start:
|
||
text = text[start:end + 1]
|
||
|
||
try:
|
||
return json.loads(text)
|
||
except json.decoder.JSONDecodeError as e:
|
||
logging.error(f"\n [!] Ошибка парсинга JSON: {e}")
|
||
logging.error(f" [!] Сырой ответ:\n{original_text}\n")
|
||
raise
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Основная логика перевода
|
||
# --------------------------------------------------------------------------
|
||
def get_translated_map(entries):
|
||
return {e["line"]: e["translated_text"] for e in entries if e.get("translated_text")}
|
||
|
||
def build_context_before(entries, idx, n, translated_map):
|
||
return [{"line": e["line"], "text": translated_map.get(e["line"], e["text"])} for e in entries[max(0, idx - n):idx]]
|
||
|
||
def build_context_after(entries, idx_end, n):
|
||
return [{"line": e["line"], "text": e["text"]} for e in entries[idx_end:idx_end + n]]
|
||
|
||
def translate_batch(cfg, glossary_text, entries, batch_idx_list, translated_map):
|
||
tr_cfg = cfg["translation"]
|
||
batch_entries = [entries[idx] for idx in batch_idx_list]
|
||
context_before = build_context_before(entries, batch_idx_list[0], tr_cfg["context_before"], translated_map)
|
||
context_after = build_context_after(entries, batch_idx_list[-1] + 1, tr_cfg["context_after"])
|
||
messages = build_messages(cfg, glossary_text, context_before, batch_entries, context_after)
|
||
expected_lines = {e["line"] for e in batch_entries}
|
||
max_retries = tr_cfg.get("max_retries", 3)
|
||
retry_delay = tr_cfg.get("retry_delay_seconds", 5)
|
||
last_error = None
|
||
|
||
for attempt in range(1, max_retries + 1):
|
||
try:
|
||
raw_response = call_llm(cfg, messages)
|
||
parsed = extract_json_array(raw_response)
|
||
|
||
result_map = {}
|
||
for item in parsed:
|
||
if "line" not in item or "text" not in item:
|
||
raise ValueError(f"Элемент без line/text: {item}")
|
||
result_map[item["line"]] = sanitize_text(item["text"])
|
||
|
||
if set(result_map.keys()) != expected_lines:
|
||
raise ValueError("Несовпадение строк в ответе.")
|
||
return result_map, None
|
||
except Exception as e:
|
||
last_error = e
|
||
if attempt < max_retries:
|
||
logging.warning(f" [!] Попытка {attempt}/{max_retries} ошибка: {e}. Повтор через {retry_delay}с...")
|
||
time.sleep(retry_delay)
|
||
else:
|
||
logging.error(f" [!] Все попытки исчерпаны: {e}")
|
||
return None, last_error
|
||
|
||
def translate_file(cfg, glossary_text, nsv_path: Path, global_cache: dict):
|
||
entries = load_nsv_file(nsv_path)
|
||
|
||
cache_applied = 0
|
||
for e in entries:
|
||
if not e.get("translated_text"):
|
||
orig_text = e.get("text", "")
|
||
if orig_text in global_cache:
|
||
e["translated_text"] = global_cache[orig_text]
|
||
cache_applied += 1
|
||
|
||
if cache_applied > 0:
|
||
logging.info(f" Восстановлено из кэша: {cache_applied} строк.")
|
||
save_nsv_file(nsv_path, entries)
|
||
|
||
translated_map = get_translated_map(entries)
|
||
line_to_idx = {e["line"]: i for i, e in enumerate(entries)}
|
||
pending_indices = [i for i, e in enumerate(entries) if not e.get("translated_text")]
|
||
|
||
if not pending_indices:
|
||
logging.info(" Все строки переведены, пропускаем.")
|
||
return []
|
||
|
||
batch_size = cfg["translation"]["batch_size"]
|
||
total_pending = len(pending_indices)
|
||
done_count = 0
|
||
problem_batches = []
|
||
logging.info(f" К переводу через API: {total_pending} строк")
|
||
|
||
i = 0
|
||
while i < len(pending_indices):
|
||
batch_idx_list = pending_indices[i:i + batch_size]
|
||
result_map, error = translate_batch(cfg, glossary_text, entries, batch_idx_list, translated_map)
|
||
|
||
if result_map is not None:
|
||
for line_no, translated_text in result_map.items():
|
||
idx = line_to_idx.get(line_no)
|
||
if idx is not None:
|
||
entries[idx]["translated_text"] = translated_text
|
||
translated_map[line_no] = translated_text
|
||
global_cache[entries[idx]["text"]] = translated_text
|
||
done_count += len(batch_idx_list)
|
||
else:
|
||
problem_batches.append({"lines": [entries[idx]["line"] for idx in batch_idx_list], "error": str(error)})
|
||
|
||
logging.info(f" Прогресс API: {min(done_count + sum(len(b['lines']) for b in problem_batches), total_pending)}/{total_pending} строк")
|
||
|
||
save_nsv_file(nsv_path, entries)
|
||
save_cache(global_cache)
|
||
|
||
i += batch_size
|
||
|
||
return problem_batches
|
||
|
||
# --------------------------------------------------------------------------
|
||
# main
|
||
# --------------------------------------------------------------------------
|
||
def main():
|
||
cfg = load_config()
|
||
paths_cfg = cfg["paths"]
|
||
global_cache = load_cache()
|
||
|
||
nsv_dir = BASE_DIR / paths_cfg["json_dir"]
|
||
glossary_path = BASE_DIR / paths_cfg["glossary_file"]
|
||
|
||
if not nsv_dir.is_dir():
|
||
logging.error(f"Ошибка: папка не найдена: {nsv_dir}")
|
||
sys.exit(1)
|
||
|
||
glossary_text = load_glossary(glossary_path)
|
||
all_nsv_files = sorted(nsv_dir.glob("*.nsv"))
|
||
|
||
if not all_nsv_files:
|
||
logging.error(f"В {nsv_dir} нет .nsv файлов.")
|
||
sys.exit(0)
|
||
|
||
choice = input(f"\nНайдено {len(all_nsv_files)} файлов. Имена через запятую (или 'all'): ").strip()
|
||
selected_files = all_nsv_files if choice.lower() == "all" else [nsv_dir / f"{n.strip()}.nsv" for n in choice.split(",") if (nsv_dir / f"{n.strip()}.nsv").exists()]
|
||
|
||
all_problems = {}
|
||
for nsv_path in selected_files:
|
||
logging.info(f"\n=== {nsv_path.name} ===")
|
||
problems = translate_file(cfg, glossary_text, nsv_path, global_cache)
|
||
if problems:
|
||
all_problems[nsv_path.name] = problems
|
||
|
||
logging.info("\n" + "=" * 60)
|
||
if all_problems:
|
||
logging.warning("Перевод завершён с проблемами в некоторых пачках. Запустите скрипт повторно.")
|
||
else:
|
||
logging.info("Перевод завершён успешно.")
|
||
|
||
if __name__ == "__main__":
|
||
main() |