#!/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()