diff --git a/.gitignore b/.gitignore index a4b46184..589797b4 100644 --- a/.gitignore +++ b/.gitignore @@ -54,3 +54,6 @@ translation_tool/core/*.txt # === OS === Thumbs.db .DS_Store + +# === Local secrets === +config.json diff --git a/app/services_impl/pipelines/_task_runner.py b/app/services_impl/pipelines/_task_runner.py index 31b874c1..73fb808d 100644 --- a/app/services_impl/pipelines/_task_runner.py +++ b/app/services_impl/pipelines/_task_runner.py @@ -11,14 +11,22 @@ logger = logging.getLogger(__name__) -def run_callable_task(*, session, task_name: str, func: Callable[..., Any], kwargs: dict, add_session_log_on_error: bool = False, ui_log_handler=UI_LOG_HANDLER): + +def run_callable_task( + *, + session, + task_name: str, + func: Callable[..., Any], + kwargs: dict, + add_session_log_on_error: bool = False, + ui_log_handler=UI_LOG_HANDLER, +): """執行可呼叫的流水線任務，並自動處理 Session 狀態切換、日誌紀錄及異常捕獲。""" ensure_pipeline_logging() try: session.start() ui_log_handler.set_session(session) result = func(**kwargs) - session.finish() return result except Exception as e: full_traceback = traceback.format_exc() @@ -28,5 +36,6 @@ def run_callable_task(*, session, task_name: str, func: Callable[..., Any], kwar session.set_error() return None finally: + # ⭐ session.finish() 一定會被執行，無論成功或失敗 + session.finish() ui_log_handler.set_session(None) - diff --git a/app/services_impl/pipelines/extract_service.py b/app/services_impl/pipelines/extract_service.py index ee9d10b2..d344ac42 100644 --- a/app/services_impl/pipelines/extract_service.py +++ b/app/services_impl/pipelines/extract_service.py @@ -21,6 +21,7 @@ logger = logging.getLogger(__name__) + def run_lang_extraction_service(mods_dir: str, output_dir: str, session): """執行語言檔擷取服務。""" ensure_pipeline_logging() @@ -53,10 +54,12 @@ def run_lang_extraction_service(mods_dir: str, output_dir: str, session): logger.error(f"[致命錯誤] Lang 檔案提取失敗：{e}\n{full_traceback}") session.add_log(f"[致命錯誤] Lang 檔案提取失敗：{e}\n{full_traceback}") session.set_error() + GLOBAL_LOG_LIMITER.flush() finally: # ⭐ 避免 handler 留著舊 session UI_LOG_HANDLER.set_session(None) + def run_book_extraction_service(mods_dir: str, output_dir: str, session): """執行書本檔擷取服務。""" ensure_pipeline_logging() @@ -89,6 +92,7 @@ def run_book_extraction_service(mods_dir: str, output_dir: str, session): logger.error(f"[致命錯誤] Book 檔案提取失敗：{e}\n{full_traceback}") session.add_log(f"[致命錯誤] Book 檔案提取失敗：{e}\n{full_traceback}") session.set_error() + GLOBAL_LOG_LIMITER.flush() finally: # ⭐ 避免 handler 留著舊 session diff --git a/app/services_impl/pipelines/lm_service.py b/app/services_impl/pipelines/lm_service.py index 86f12e99..c848a362 100644 --- a/app/services_impl/pipelines/lm_service.py +++ b/app/services_impl/pipelines/lm_service.py @@ -14,10 +14,13 @@ UI_LOG_HANDLER, ) from app.services_impl.pipelines._pipeline_logging import ensure_pipeline_logging -from translation_tool.core.lm_translator import translate_directory_generator as lm_translate_gen +from translation_tool.core.lm_translator import ( + translate_directory_generator as lm_translate_gen, +) logger = logging.getLogger(__name__) + def run_lm_translation_service( input_dir: str, output_dir: str, @@ -76,6 +79,7 @@ def run_lm_translation_service( logger.error(f"LM 服務失敗: {e}\n{full_traceback}") session.add_log(f"[致命錯誤] LM 翻譯服務失敗：{e}\n{full_traceback}") session.set_error() + GLOBAL_LOG_LIMITER.flush() finally: # ⭐ 避免 handler 留著舊 session UI_LOG_HANDLER.set_session(None) diff --git a/tests/test_cache_manager.py b/tests/test_cache_manager.py new file mode 100644 index 00000000..2384b342 --- /dev/null +++ b/tests/test_cache_manager.py @@ -0,0 +1,183 @@ +"""test_cache_manager.py + +測試 cache_manager 的執行緒安全與 dirty flag 行為。 +覆蓋： +1. initialize_translation_cache() 的 cache_lock 保護 +2. save_translation_cache() 的 clear_dirty() 時機（寫入成功後） +""" + +import threading +from pathlib import Path +from unittest.mock import patch + +import pytest + +from translation_tool.utils import cache_manager, cache_store + + +# ============================================================================= +# Fixtures +# ============================================================================= + + +@pytest.fixture +def fresh_state(): + """提供乾淨的 runtime state（每個測試獨立）。""" + cache_store.reset_runtime_state(cache_manager.CACHE_TYPES) + state = cache_store.get_runtime_state() + state.initialized = False + state.translation_cache = {k: {} for k in cache_manager.CACHE_TYPES} + state.session_new_entries = {k: {} for k in cache_manager.CACHE_TYPES} + state.is_dirty = {k: False for k in cache_manager.CACHE_TYPES} + yield state + # 測試結束重置，避免污染後續測試 + cache_store.reset_runtime_state(cache_manager.CACHE_TYPES) + + +@pytest.fixture +def mock_save_path(tmp_path: Path, fresh_state): + """設定假的 cache 檔案路徑（不碰真實檔案系統）。""" + cache_type = "lang" + type_dir = tmp_path / cache_type + type_dir.mkdir(parents=True, exist_ok=True) + fresh_state.cache_file_path = { + cache_type: type_dir / f"{cache_type}_cache_main.json" + } + return cache_type, type_dir + + +# ============================================================================= +# 測試 1: initialize_translation_cache() 的 cache_lock 保護 +# ============================================================================= + + +def test_initialize_translation_cache_uses_cache_lock(fresh_state): + """驗證 initialize_translation_cache() 在 cache_lock 保護下執行。 + + 情境：多執行緒同時呼叫 initialize_translation_cache()， + 確認第二次呼叫因為 lock 而被阻擋（initialized 已為 True）， + 不會造成重複載入。 + """ + call_count = 0 + + # Patch _load_cache_type 來計數呼叫 + def _load_cache_type_track(cache_type): + nonlocal call_count + call_count += 1 + + with patch.object( + cache_manager, "_load_cache_type", side_effect=_load_cache_type_track + ): + # 模擬兩執行緒同時進入 + def call_init(): + cache_manager.initialize_translation_cache() + + # 第一次呼叫 + t1 = threading.Thread(target=call_init) + t1.start() + t1.join() + + # 驗證：initialized 為 True + assert fresh_state.initialized is True + # 驗證：每個 cache type 只載入一次 + assert call_count == len(cache_manager.CACHE_TYPES) + + +def test_initialize_translation_cache_no_double_load_on_concurrent_calls(fresh_state): + """驗證 initialize_translation_cache() 重複呼叫不會造成 race condition。 + + 情境：多執行緒幾乎同時呼叫，確認只有一個執行緒真正執行初始化， + 其餘執行緒在 lock 處等待後直接返回（initialized=True）。 + """ + load_calls = [] + + def _load_cache_type_tracking(cache_type): + load_calls.append(cache_type) + + # 先把 initialized 設為 True，模擬已經初始化過 + fresh_state.initialized = True + + with patch.object( + cache_manager, "_load_cache_type", side_effect=_load_cache_type_tracking + ): + cache_manager.initialize_translation_cache() + + # 驗證：已初始化時不再呼叫 _load_cache_type + assert len(load_calls) == 0 + + +# ============================================================================= +# 測試 2: save_translation_cache() 的 clear_dirty() 時機 +# ============================================================================= + + +def test_save_translation_cache_dirty_True_when_save_fails(mock_save_path): + """驗證 save_translation_cache() 在寫入失敗後 dirty flag 仍為 True。 + + 情境：session_new_entries 有資料，is_dirty=True， + save_translation_cache() 嘗試儲存但 _save_entries_to_active_shards 失敗。 + 預期：is_dirty 保持 True（因為資料已從 session flush 但未成功寫入磁碟）。 + + 設計：此測試捕捉「crash 發生於寫入前」的場景——dirty flag 必須在 + 寫入真正成功後才能清除。 + """ + cache_type, _ = mock_save_path + + state = cache_store.get_runtime_state() + state.is_dirty[cache_type] = True + state.session_new_entries[cache_type] = {"key1": {"src": "Hello", "dst": "哈囉"}} + + with patch.object( + cache_manager, + "_save_entries_to_active_shards", + side_effect=RuntimeError("磁碟寫入失敗（模擬 crash）"), + ): + cache_manager.save_translation_cache(cache_type, write_new_shard=True) + + # 驗證：寫入失敗後，dirty flag 仍為 True + # （資料已從 session_new_entries flush，但寫入失敗，不能假設乾淨） + assert state.is_dirty[cache_type] is True, ( + "寫入失敗時 dirty 應保持 True，避免資料遺失後又被視為已同步" + ) + + +def test_save_translation_cache_dirty_cleared_when_save_succeeds(mock_save_path): + """驗證 save_translation_cache() 在寫入成功後 dirty flag 正確清除。""" + cache_type, _ = mock_save_path + + state = cache_store.get_runtime_state() + state.is_dirty[cache_type] = True + state.session_new_entries[cache_type] = {"key1": {"src": "Hello", "dst": "哈囉"}} + + saved_data = {} + + def _capture_save(_cache_type, entries, force_new_shard=False): + saved_data["cache_type"] = _cache_type + saved_data["entries"] = entries.copy() + + with patch.object( + cache_manager, "_save_entries_to_active_shards", side_effect=_capture_save + ): + cache_manager.save_translation_cache(cache_type, write_new_shard=True) + + # 驗證：寫入成功後，dirty 清除 + assert state.is_dirty[cache_type] is False + # 驗證：session_new_entries 已 flush + assert state.session_new_entries[cache_type] == {} + # 驗證：寫入函式被正確呼叫 + assert saved_data["entries"] == {"key1": {"src": "Hello", "dst": "哈囉"}} + + +def test_save_translation_cache_no_op_when_no_dirty_entries(mock_save_path): + """驗證無 dirty 資料時 save_translation_cache 不做任何事。""" + cache_type, _ = mock_save_path + + state = cache_store.get_runtime_state() + state.is_dirty[cache_type] = False + state.session_new_entries[cache_type] = {} + + with patch.object(cache_manager, "_save_entries_to_active_shards") as mock_save: + cache_manager.save_translation_cache(cache_type) + + # 驗證：無 session 資料時不呼叫儲存 + assert mock_save.call_count == 0 diff --git a/tests/test_cache_store.py b/tests/test_cache_store.py index c5ec975c..0f0cf704 100644 --- a/tests/test_cache_store.py +++ b/tests/test_cache_store.py @@ -1,4 +1,5 @@ from pathlib import Path +import threading from translation_tool.utils import cache_manager, cache_store @@ -12,7 +13,9 @@ def test_cache_store_entry_and_value_crud(): assert cache_store.get_entry(cache_dict, "k1") == {"src": "s1", "dst": "d1"} assert cache_store.get_value(cache_dict, "k1") == "d1" - changed_again = cache_store.add_entry(cache_dict, "k1", {"src": "s1-new", "dst": "d1"}) + changed_again = cache_store.add_entry( + cache_dict, "k1", {"src": "s1-new", "dst": "d1"} + ) assert changed_again is False # contract: dst 相同時不覆寫舊 entry assert cache_store.get_entry(cache_dict, "k1") == {"src": "s1", "dst": "d1"} @@ -55,12 +58,16 @@ def test_manager_add_save_reload_smoke(monkeypatch, tmp_path: Path): saved = {} - def _fake_save_entries(_cache_type: str, entries: dict, force_new_shard: bool = False): + def _fake_save_entries( + _cache_type: str, entries: dict, force_new_shard: bool = False + ): saved["cache_type"] = _cache_type saved["entries"] = entries.copy() saved["force_new_shard"] = force_new_shard - monkeypatch.setattr(cache_manager, "_save_entries_to_active_shards", _fake_save_entries) + monkeypatch.setattr( + cache_manager, "_save_entries_to_active_shards", _fake_save_entries + ) cache_manager.add_to_cache(cache_type, key, "Hello", "哈囉") assert cache_manager.get_from_cache(cache_type, key) == "哈囉" @@ -79,4 +86,105 @@ def _fake_load_cache_type(_cache_type: str): monkeypatch.setattr(cache_manager, "_load_cache_type", _fake_load_cache_type) cache_manager.reload_translation_cache_type(cache_type) - assert cache_manager.get_cache_entry(cache_type, key) == {"src": "Hello", "dst": "哈囉"} + assert cache_manager.get_cache_entry(cache_type, key) == { + "src": "Hello", + "dst": "哈囉", + } + + +# ============================================================================= +# 測試 3: add_entry() 執行緒安全保護 +# ============================================================================= + + +def test_add_entry_thread_safety_different_keys(): + """驗證 add_entry() 多執行緒同時寫入不同 key 時不造成資料遺失。 + + 情境：兩個執行緒同時對同一個 cache_dict 寫入不同的 key， + 不使用 cache_manager.add_to_cache() 的 lock（直接呼叫 add_entry）。 + 預期：所有 key 都正確寫入，無競爭條件導致 entry 遺失。 + + 此測試驗證 add_entry() 在多執行序並發呼叫時， + 對不同 key 的寫入能正確完成而不互相干擾。 + """ + # 使用獨立的 cache_dict，不走 manager 的 lock + cache_dict = {} + + results = [] + + def writer(thread_id, keys): + for k in keys: + entry = {"src": f"src_{thread_id}_{k}", "dst": f"dst_{thread_id}_{k}"} + changed = cache_store.add_entry(cache_dict, k, entry) + results.append((thread_id, k, changed)) + + # 建立兩組不同的 key，避免 key collision 測試混淆 + keys_a = [f"key_a_{i}" for i in range(50)] + keys_b = [f"key_b_{i}" for i in range(50)] + + t1 = threading.Thread(target=writer, args=(1, keys_a)) + t2 = threading.Thread(target=writer, args=(2, keys_b)) + + t1.start() + t2.start() + t1.join() + t2.join() + + # 驗證：所有 key 都成功寫入（無遺漏） + assert len(cache_dict) == 100, ( + f"預期 100 個 entry，實際只有 {len(cache_dict)} 個。" + "多執行序寫入不同 key 不應造成資料遺失。" + ) + + # 驗證：每個 key 都被寫入一次（changed=True） + changed_count = sum(1 for _, _, changed in results if changed is True) + assert changed_count == 100, ( + f"預期 100 次 changed=True，實際有 {changed_count} 次。" + ) + + +def test_add_entry_thread_safety_same_key_race(): + """驗證 add_entry() 多執行序同時寫入相同 key 的競爭行為。 + + 情境：兩個執行緒同時對同一個 key 寫入不同的值， + 模擬真實並發情境下的資料競爭。 + + 預期：最終 cache_dict[key] 為其中一個執行緒的寫入結果 + （add_entry 本身不做 internal lock，所以結果是 non-deterministic）。 + + 此測試記錄竞争結果，用於確認 add_entry() 在並發下 + 不會發生 dict 結構損壞或 exception。 + """ + cache_dict = {} + exceptions = [] + + def writer(thread_id, value): + try: + entry = {"src": f"src_{thread_id}", "dst": f"dst_{thread_id}_{value}"} + cache_store.add_entry(cache_dict, "shared_key", entry) + except Exception as e: + exceptions.append((thread_id, e)) + + t1 = threading.Thread(target=writer, args=(1, "value_a")) + t2 = threading.Thread(target=writer, args=(2, "value_b")) + + t1.start() + t2.start() + t1.join() + t2.join() + + # 驗證：無 exception（add_entry 不應拋出例外） + assert len(exceptions) == 0, f"add_entry 在並發下不應拋出例外: {exceptions}" + + # 驗證：cache_dict["shared_key"] 是其中一個執行緒的寫入結果 + # （dst 值是 "dst_1_value_a" 或 "dst_2_value_b"） + final_dst = cache_dict.get("shared_key", {}).get("dst", "") + assert final_dst in ( + "dst_1_value_a", + "dst_2_value_b", + ), f"最終值應為其中一個執行緒的寫入，實際: {final_dst}" + + # 驗證：entry 結構完整 + assert isinstance(cache_dict["shared_key"], dict) + assert "src" in cache_dict["shared_key"] + assert "dst" in cache_dict["shared_key"] diff --git a/tests/test_ftbquests_unshield_logic.py b/tests/test_ftbquests_unshield_logic.py new file mode 100644 index 00000000..fa1e9656 --- /dev/null +++ b/tests/test_ftbquests_unshield_logic.py @@ -0,0 +1,386 @@ +"""translation_tool/plugins/ftbquests/ftbquests_lmtranslator.py 單元測試：unshield_text 參數修正。 + +用途：驗證 on_translated_item 中 unshield_text 的呼叫使用 shielded_src.shields（list） +而非整個 ShieldedText 物件。 + +參考：PR #42 (pr/rich-text-shield) — 2026-03-23 +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +# 確保可以導入翻譯工具模組 +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from translation_tool.plugins.shared.rich_text_shield import ( # noqa: E402 + ShieldedText, + ShieldPiece, +) + + +# --------------------------------------------------------------------------- +# 測試：ftbquests_lmtranslator.on_translated_item — unshield 使用 .shields +# --------------------------------------------------------------------------- + + +def test_ftb_on_translated_item_unshield_uses_shields_list(): + """ + 驗證 ftbquests_lmtranslator 的 on_translated_item 回呼 + 呼叫 unshield_text(t, shielded_src.shields) 時傳入 list[ShieldPiece]， + 而非整個 ShieldedText 物件。 + + 這樣才能正確還原翻譯結果中的格式佔位符（如 $C0$）。 + """ + from translation_tool.plugins.ftbquests import ftbquests_lmtranslator + + # 建立假的 ShieldPiece 列表（模擬 shield_text 的輸出） + fake_shield_piece = ShieldPiece( + placeholder="$C0$", + original="&c", + category="color", + ) + fake_shields_list: list[ShieldPiece] = [fake_shield_piece] + + # 建立假的 ShieldedText（mock） + fake_shielded = MagicMock(spec=ShieldedText) + fake_shielded.shields = fake_shields_list # 關鍵：.shields 是 list + + # 用 MagicMock 模擬翻譯後文字（含佔位符） + translated_with_placeholder = "This is $C0$ important!" + + # 建立翻譯後的 item + translated_item = { + "path": "quest.1.title", + "text": translated_with_placeholder, # 翻譯後含 $C0$ 佔位符 + "source_text": "This is &c important!", # 原文含 &c 彩色碼 + } + + # 收集 unshield_text 的呼叫參數 + captured_calls = [] + + def mock_unshield_text(text: str, shields_arg) -> str: + """Mock unshield_text，記錄被呼叫時的第二個參數。""" + captured_calls.append({"text": text, "shields_arg": shields_arg}) + # 簡單還原：把 $C0$ 換回 &c + return text.replace("$C0$", "&c") + + # Patch 在 ftbquests_lmtranslator 命名空間中的 unshield_text + with ( + patch.object( + ftbquests_lmtranslator, + "unshield_text", + side_effect=mock_unshield_text, + ), + patch.object( + ftbquests_lmtranslator, + "shield_text", + return_value=fake_shielded, + ), + ): + # on_translated_item 是翻譯流程中的 nested callback， + # 無法直接呼叫。我們透過翻譯流程觸發它。 + # 為隔離測試，直接建構一個符合 on_translated_item 簽名的 closure 來測試。 + def closure_under_test(it: dict): + """重現 on_translated_item 的核心邏輯。""" + from translation_tool.plugins.ftbquests import ftbquests_lmtranslator as m + + p = it.get("path") + t = it.get("text") + src_text = str(it.get("source_text") or "") + if isinstance(p, str) and isinstance(t, str): + try: + shielded_src = m.shield_text(src_text) + t = m.unshield_text(t, shielded_src.shields) + except Exception: + pass + + closure_under_test(translated_item) + + # 斷言：unshield_text 被呼叫了 + assert len(captured_calls) == 1, "unshield_text 應該被呼叫一次" + + shields_arg = captured_calls[0]["shields_arg"] + + # 斷言：傳入的是 list[ShieldPiece]，不是 ShieldedText 整個物件 + assert isinstance(shields_arg, list), ( + f"unshield_text 第二參數應為 list，實際為 {type(shields_arg).__name__}。" + "使用 .shields 而非整個 ShieldedText 物件。" + ) + + # 斷言：list 中第一個元素是 ShieldPiece + assert len(shields_arg) == 1 + assert isinstance(shields_arg[0], ShieldPiece), ( + f"list[ShieldPiece] 中實際元素型別為 {type(shields_arg[0]).__name__}。" + ) + + # 斷言：ShieldPiece 的內容正確 + assert shields_arg[0].placeholder == "$C0$" + assert shields_arg[0].original == "&c" + assert shields_arg[0].category == "color" + + +def test_ftb_on_translated_item_unshield_rejects_whole_shielded_object(): + """ + 驗證：如果錯誤地傳入整個 ShieldedText 物件（而非 .shields）， + unshield_text 的第二參數會是非 list 型別，導致還原失敗。 + 此測試用來確認「錯誤版本」的呼叫模式確實會被偵測。 + """ + from translation_tool.plugins.ftbquests import ftbquests_lmtranslator + + # 建立假的 ShieldedText（mock），但不使用 .shields + fake_shielded = MagicMock(spec=ShieldedText) + fake_shielded.shields = [] # 空的 list + + # 建立翻譯後的 item（含佔位符） + translated_item = { + "path": "quest.1.title", + "text": "Result with $C0$ here", + "source_text": "Source &c text", + } + + # 收集傳入 unshield_text 的第二參數 + captured_second_arg_type = [] + + def mock_unshield_text(text: str, shields_arg): + captured_second_arg_type.append(type(shields_arg).__name__) + return text # 不做還原 + + with ( + patch.object( + ftbquests_lmtranslator, + "unshield_text", + side_effect=mock_unshield_text, + ), + patch.object( + ftbquests_lmtranslator, + "shield_text", + return_value=fake_shielded, + ), + ): + + def closure_under_test(it: dict): + from translation_tool.plugins.ftbquests import ftbquests_lmtranslator as m + + p = it.get("path") + t = it.get("text") + src_text = str(it.get("source_text") or "") + if isinstance(p, str) and isinstance(t, str): + try: + shielded_src = m.shield_text(src_text) + t = m.unshield_text(t, shielded_src.shields) + except Exception: + pass + + closure_under_test(translated_item) + + # 驗證：使用 .shields（list） 時，參數型別為 "list" + assert captured_second_arg_type[-1] == "list" + + +# --------------------------------------------------------------------------- +# 測試：md_lmtranslator.on_translated_item — unshield 使用 .shields +# --------------------------------------------------------------------------- + + +def test_md_on_translated_item_unshield_uses_shields_list(): + """ + 驗證 md_lmtranslator 的 on_translated_item 回呼 + 呼叫 unshield_text(dst, shielded.shields) 時傳入 list[ShieldPiece]。 + """ + from translation_tool.plugins.md import md_lmtranslator + + # 建立假的 ShieldPiece + fake_shield_piece = ShieldPiece( + placeholder="$P0$", + original="#minecraft:diamond", + category="item_id", + ) + fake_shields_list: list[ShieldPiece] = [fake_shield_piece] + + fake_shielded = MagicMock(spec=ShieldedText) + fake_shielded.shields = fake_shields_list + + # 翻譯後 item（含 item_id 佔位符） + translated_item = { + "path": "abc123", + "text": "You need $P0$ to craft this", + "source_text": "You need #minecraft:diamond to craft this", + "_shielded": fake_shielded, + } + + captured_calls = [] + + def mock_unshield_text(text: str, shields_arg) -> str: + captured_calls.append({"text": text, "shields_arg": shields_arg}) + return text.replace("$P0$", "#minecraft:diamond") + + with patch.object( + md_lmtranslator, + "unshield_text", + side_effect=mock_unshield_text, + ): + # 重現 md_lmtranslator 的 on_translated_item 核心邏輯 + def closure_md_on_translated_item(it: dict): + from translation_tool.plugins.md import md_lmtranslator as m + + h = str(it.get("path") or "") + dst = str(it.get("text") or "") + src_text = str(it.get("source_text") or "") + if h and dst: + shielded = it.get("_shielded") + if shielded is not None and getattr(shielded, "shields", None): + try: + dst = m.unshield_text(dst, shielded.shields) + except Exception: + pass + else: + try: + shielded_src = m.shield_text(src_text) + dst = m.unshield_text(dst, shielded_src.shields) + except Exception: + pass + + closure_md_on_translated_item(translated_item) + + assert len(captured_calls) == 1, "unshield_text 應該被呼叫一次" + shields_arg = captured_calls[0]["shields_arg"] + + assert isinstance(shields_arg, list), ( + f"unshield_text 第二參數應為 list，實際為 {type(shields_arg).__name__}。" + ) + assert len(shields_arg) == 1 + assert isinstance(shields_arg[0], ShieldPiece) + assert shields_arg[0].placeholder == "$P0$" + assert shields_arg[0].original == "#minecraft:diamond" + assert shields_arg[0].category == "item_id" + + +def test_md_on_translated_item_else_branch_uses_shields(): + """ + 驗證 md_lmtranslator 的 on_translated_item else 分支 + （當 item._shielded 為 None 時）仍使用 shield_text().shields。 + """ + from translation_tool.plugins.md import md_lmtranslator + + fake_shield_piece = ShieldPiece( + placeholder="$C1$", + original="&a", + category="color", + ) + fake_shields_list: list[ShieldPiece] = [fake_shield_piece] + + fake_shielded = MagicMock(spec=ShieldedText) + fake_shielded.shields = fake_shields_list + + # item._shielded 為 None，觸發 else 分支 + translated_item = { + "path": "xyz789", + "text": "Green text $C1$ here", + "source_text": "Green text &a here", + "_shielded": None, # 觸發 else 分支 + } + + captured_calls = [] + + def mock_unshield_text(text: str, shields_arg) -> str: + captured_calls.append({"text": text, "shields_arg": shields_arg}) + return text.replace("$C1$", "&a") + + with ( + patch.object( + md_lmtranslator, + "unshield_text", + side_effect=mock_unshield_text, + ), + patch.object( + md_lmtranslator, + "shield_text", + return_value=fake_shielded, + ), + ): + + def closure_md_on_translated_item(it: dict): + from translation_tool.plugins.md import md_lmtranslator as m + + h = str(it.get("path") or "") + dst = str(it.get("text") or "") + src_text = str(it.get("source_text") or "") + if h and dst: + shielded = it.get("_shielded") + if shielded is not None and getattr(shielded, "shields", None): + try: + dst = m.unshield_text(dst, shielded.shields) + except Exception: + pass + else: + try: + shielded_src = m.shield_text(src_text) + dst = m.unshield_text(dst, shielded_src.shields) + except Exception: + pass + + closure_md_on_translated_item(translated_item) + + assert len(captured_calls) == 1, "unshield_text 應在 else 分支被呼叫一次" + shields_arg = captured_calls[0]["shields_arg"] + + assert isinstance(shields_arg, list), ( + f"else 分支中 unshield_text 第二參數應為 list，實際為 {type(shields_arg).__name__}。" + ) + assert shields_arg[0].original == "&a" + + +def test_md_cache_hit_unshield_uses_shields(): + """ + 驗證 md_lmtranslator 中 cache hit 分支 + 呼叫 unshield_text(hash_to_dst, shielded.shields) 使用 .shields。 + """ + from translation_tool.plugins.md import md_lmtranslator + + fake_shield_piece = ShieldPiece( + placeholder="$P1$", + original="#minecraft:iron_ingot", + category="item_id", + ) + fake_shields_list: list[ShieldPiece] = [fake_shield_piece] + + fake_shielded = MagicMock(spec=ShieldedText) + fake_shielded.shields = fake_shields_list + + cached_item = { + "path": "cached_hash_001", + "text": "You need $P1$ to smelt", + "_shielded": fake_shielded, + } + + captured_calls = [] + + def mock_unshield_text(text: str, shields_arg) -> str: + captured_calls.append({"text": text, "shields_arg": shields_arg}) + return text.replace("$P1$", "#minecraft:iron_ingot") + + with patch.object( + md_lmtranslator, + "unshield_text", + side_effect=mock_unshield_text, + ): + # 重現 md_lmtranslator cache hit 區塊邏輯 + h = str(cached_item.get("path") or "") + dst = str(cached_item.get("text") or "") + if h and dst: + shielded = cached_item.get("_shielded") + if shielded is not None and getattr(shielded, "shields", None): + try: + dst = md_lmtranslator.unshield_text(dst, shielded.shields) + except Exception: + pass + + assert len(captured_calls) == 1 + shields_arg = captured_calls[0]["shields_arg"] + assert isinstance(shields_arg, list) + assert shields_arg[0].original == "#minecraft:iron_ingot" diff --git a/tests/test_kubejs_translator_clean.py b/tests/test_kubejs_translator_clean.py index c07030ef..9c2f6449 100644 --- a/tests/test_kubejs_translator_clean.py +++ b/tests/test_kubejs_translator_clean.py @@ -5,8 +5,8 @@ from __future__ import annotations -from pathlib import Path import sys +from pathlib import Path import orjson import pytest @@ -148,8 +148,253 @@ def test_empty_en_map(self): assert result == {} +class TestBuildReverseIndexImpl: + """測試 _build_reverse_index_impl 函式。 + + 驗證 reverse_index 為 dict[str, str] 而非 dict[str, list]， + 以及選擇 canonical key 的確定性邏輯（優先已翻譯，再取字母序最小）。 + """ + + def test_reverse_index_is_dict_str_str_not_list(self): + """reverse_index 必須是 dict[str, str]，不能是 dict[str, list]。""" + final_tw_lookup = { + "key_a": "翻譯值", + "key_b": "另一個翻譯", + } + result = _build_reverse_index_impl(final_tw_lookup) + + # 類型驗證：每個 value 都應該是 str，不是 list + for k, v in result.items(): + assert isinstance(k, str), f"key 應為 str，實際為 {type(k).__name__}" + assert isinstance(v, str), ( + f"value for key '{k}' 應為 str，實際為 {type(v).__name__}" + ) + + def test_prefers_translated_key_over_untranslated(self): + """當多個 key 有相同翻譯值時，應優先選擇「已翻譯」的 key。 + + 「已翻譯」定義：zh_tw 值與英文 key 名不同。 + """ + # key_a：翻譯值不同於 key 名（已翻譯） + # key_b：翻譯值等於 key 名（未翻譯） + final_tw_lookup = { + "apple": "蘋果", # 已翻譯（值 != key） + "蘋果": "蘋果", # 未翻譯（值 == key） + } + result = _build_reverse_index_impl(final_tw_lookup) + + # 對"蘋果"這個翻譯結果，應選擇 key "apple"（已翻譯）而非 "蘋果"（未翻譯） + assert result["蘋果"] == "apple" + + def test_prefers_alphabetically_smallest_among_same_priority(self): + """同優先級時（都是已翻譯或都是未翻譯），取字母序最小的 key。""" + # 多個 key 都已翻譯（值 != key），取字母序最小 + final_tw_lookup = { + "zebra": "動物", # 已翻譯，但字母序較大 + "ant": "動物", # 已翻譯，字母序最小 + "bee": "動物", # 已翻譯，字母序居中 + } + result = _build_reverse_index_impl(final_tw_lookup) + + assert result["動物"] == "ant" + + def test_mixed_translated_and_untranslated_chooses_correct(self): + """混合場景：已翻譯優先於未翻譯。""" + final_tw_lookup = { + "apple": "蘋果", # 已翻譯 + "banana": "香蕉", # 未翻譯 + "cherry": "櫻桃", # 已翻譯 + } + result = _build_reverse_index_impl(final_tw_lookup) + + assert result["蘋果"] == "apple" + assert result["香蕉"] == "banana" + assert result["櫻桃"] == "cherry" + + def test_stability_multiple_executions_same_input(self): + """多次執行同一組資料，結果必須完全一致（確定性）。""" + final_tw_lookup = { + "z_key": "翻譯Z", + "a_key": "翻譯A", + "m_key": "翻譯M", + "翻譯Z": "翻譯Z", # 未翻譯 + "翻譯A": "翻譯A", # 未翻譯 + } + + results = [_build_reverse_index_impl(final_tw_lookup) for _ in range(10)] + + # 所有結果應該完全相同 + first = results[0] + for i, r in enumerate(results[1:], 1): + assert r == first, f"第 {i} 次結果與第 1 次不同：{r} vs {first}" + + def test_stability_with_multiple_keys_same_translation(self): + """多個 key 映射到同一翻譯值時，選擇結果穩定。""" + final_tw_lookup = { + "zulu_item": "測試翻譯", + "alpha_item": "測試翻譯", + "測試翻譯": "測試翻譯", # 未翻譯 + } + + results = [_build_reverse_index_impl(final_tw_lookup) for _ in range(5)] + first = results[0] + for i, r in enumerate(results[1:], 1): + assert r == first, f"第 {i} 次結果與第 1 次不同" + + # 應選已翻譯且字母序最小的：alpha_item < zulu_item + assert first["測試翻譯"] == "alpha_item" + + def test_empty_final_tw_lookup_returns_empty_dict(self): + """空的 final_tw_lookup 回傳空字典。""" + result = _build_reverse_index_impl({}) + assert result == {} + + def test_non_filled_text_values_are_ignored(self): + """非填充文字值（如空字串、空白）不應進入 reverse_index。""" + final_tw_lookup = { + "key1": "有效翻譯", + "key2": "", # 空字串，應忽略 + "key3": " ", # 空白，應忽略 + "key4": "{ref}", # 語言參考，應忽略 + } + result = _build_reverse_index_impl(final_tw_lookup) + + assert "有效翻譯" in result + assert "" not in result + assert " " not in result + assert "{ref}" not in result + + def test_casefold_ascii_translation_detection(self): + """ASCII 翻譯使用 casefold() 判斷是否為「已翻譯」。""" + # "Copper Ingot" vs "copper ingot"：casefold 後相同，視為已翻譯 + # "copper ingot" vs "copper ingot"：完全相同，視為未翻譯 + final_tw_lookup = { + "copper_ingot": "Copper Ingot", # 已翻譯（casefold 不同） + "Copper Ingot": "Copper Ingot", # 未翻譯（casefold 相同） + } + result = _build_reverse_index_impl(final_tw_lookup) + + # 應選 key 名與值 casefold 後不同的 "copper_ingot" + assert result["Copper Ingot"] == "copper_ingot" + + def test_non_ascii_uses_direct_equality(self): + """非 ASCII 翻譯使用直接相等判斷是否為「已翻譯」。""" + final_tw_lookup = { + "蘋果": "蘋果", # 未翻譯 + "apple": "蘋果", # 已翻譯 + } + result = _build_reverse_index_impl(final_tw_lookup) + + assert result["蘋果"] == "apple" + + +class TestDedupPendingEnImpl: + """測試 _dedup_pending_en_impl 函式。 + + 驗證去重邏輯使用 `v in reverse_index` 而非 `k != reverse_index[v]`， + 以及跨命名空間比對的正確性。 + """ + + def test_dedup_removes_keys_with_value_in_reverse_index(self): + """當 pending_en 的 value 存在於 reverse_index 時，該 key 應被移除。""" + pending_en = { + "mod.item1": "Apple", + "mod.item2": "Banana", + "mod.item3": "Cherry", + } + reverse_index = { + "Apple": "final.apple", # Apple 已在 final 中 + "Banana": "final.banana", # Banana 已在 final 中 + } + + result = _dedup_pending_en_impl(pending_en, reverse_index) + + # Apple 和 Banana 已在 final，應被移除；Cherry 不在 reverse_index，應保留 + assert result == {"mod.item3": "Cherry"} + + def test_dedup_cross_namespace_bug_fixed(self): + """跨命名空間比對：raw/pending 的 k 與 final 的 key 名不同，但翻譯值相同時，應去重。 + + 這是原本 bug 的核心場景： + - pending 的 key: "raw_namespace:item_name"（value: "Apple"） + - final 的 key: "final_namespace:item_name"（value: "Apple"） + - 舊邏輯：`k != reverse_index[v]` → "raw_namespace:item_name" != "final_namespace:item_name" + → 判斷為「不相同」，導致不去重 ❌ + - 新邏輯：`v in reverse_index` → "Apple" in reverse_index → True → 去重 ✅ + """ + pending_en = { + "raw:item_a": "Apple", # value: Apple + "raw:item_b": "Banana", # value: Banana（不在 reverse_index） + "raw:item_c": "Cherry", # value: Cherry + } + reverse_index = { + # final 中有不同的 key 名，但相同的翻譯值 + "Apple": "final:item_x", + "Cherry": "final:item_y", + } + + result = _dedup_pending_en_impl(pending_en, reverse_index) + + # Apple 和 Cherry 的 key 名雖然與 reverse_index 中的不同， + # 但翻譯值存在於 reverse_index，仍應被去重 + assert result == {"raw:item_b": "Banana"} + + def test_dedup_non_filled_text_not_removed(self): + """非填充文字（如空字串、空白、語言參考）不受去重邏輯影響。""" + pending_en = { + "key1": "", # 空字串，應保留（即使 "" 在 reverse_index） + "key2": " ", # 空白，應保留 + "key3": "{ref}", # 語言參考，應保留 + "key4": "有效翻譯", # 有效文字，在 reverse_index 中，應移除 + } + reverse_index = { + "": "some_key", # reverse_index 中有 "" + " ": "some_key2", # reverse_index 中有空白 + "{ref}": "some_key3", # reverse_index 中有 ref + "有效翻譯": "tw_key", # 有效翻譯 + } + + result = _dedup_pending_en_impl(pending_en, reverse_index) + + # 只有 "有效翻譯" 應被移除；空字串、空白、ref 都應保留 + assert result == {"key1": "", "key2": " ", "key3": "{ref}"} + + def test_dedup_empty_pending_returns_empty(self): + """空的 pending_en 回傳空字典。""" + reverse_index = {"key": "value"} + result = _dedup_pending_en_impl({}, reverse_index) + assert result == {} + + def test_dedup_empty_reverse_index_keeps_all(self): + """空的 reverse_index 保留所有 pending_en。""" + pending_en = { + "key1": "Apple", + "key2": "Banana", + } + result = _dedup_pending_en_impl(pending_en, {}) + assert result == {"key1": "Apple", "key2": "Banana"} + + def test_dedup_stability_across_multiple_calls(self): + """同一組輸入，多次呼叫結果一致。""" + pending_en = { + "namespace:item1": "翻譯A", + "namespace:item2": "翻譯B", + "namespace:item3": "翻譯C", + } + reverse_index = { + "翻譯A": "final:key1", + "翻譯B": "final:key2", + } + + results = [_dedup_pending_en_impl(pending_en, reverse_index) for _ in range(10)] + + expected = {"namespace:item3": "翻譯C"} + for i, r in enumerate(results): + assert r == expected, f"第 {i} 次結果與預期不同" + + class TestCleanKubejsFromRawImpl: - """測試 clean_kubejs_from_raw_impl 函式。""" + """測試 clean_kubejs_from_raw_impl 函式（整合測試）。""" @pytest.fixture def mock_lang_files(self, tmp_path: Path): diff --git a/tests/test_lm_api_client.py b/tests/test_lm_api_client.py index 663bdb78..5e52b060 100644 --- a/tests/test_lm_api_client.py +++ b/tests/test_lm_api_client.py @@ -114,6 +114,50 @@ def test_custom_timeout(self, mock_config, mock_post): call_kwargs = mock_post.call_args.kwargs assert call_kwargs.get("timeout") == 120 + @patch("translation_tool.core.lm_api_client.requests.post") + @patch("translation_tool.core.lm_api_client.load_config") + def test_api_key_not_in_url(self, mock_config, mock_post): + """測試 API Key 不出現在 URL 中，而是放在 x-goog-api-key header。""" + from translation_tool.core.lm_api_client import call_gemini_requests + + # 使用假的 API key（長度 35-45 字，以 AIza 開頭） + fake_api_key = "AIza" + "a" * 37 # 共 41 字 + + mock_config.return_value = {"lm_translator": {"rate_limit": {"timeout": 60}}} + + mock_response = Mock() + mock_response.ok = True + mock_response.json.return_value = { + "candidates": [{"content": {"parts": [{"text": '{"result": "ok"}'}]}}] + } + mock_post.return_value = mock_response + + call_gemini_requests( + model_name="gemini-pro", + system_prompt="test prompt", + payload={"key": "value"}, + api_key=fake_api_key, + temperature=0.7, + ) + + # 驗證 URL 中不包含 API key + call_args = mock_post.call_args + called_url = ( + call_args.args[0] if call_args.args else call_args.kwargs.get("url", "") + ) + assert fake_api_key not in called_url, "API key 不應出現在 URL 中" + + # 驗證 x-goog-api-key header 存在（對照 Google 官方 REST 範例） + headers = call_args.kwargs.get("headers", {}) + assert "x-goog-api-key" in headers, "x-goog-api-key header 必須存在" + assert headers["x-goog-api-key"] == fake_api_key + assert "Authorization" not in headers, "不應再使用 Authorization: Bearer" + + # 確保 URL 中沒有 key=... 之類的 query string + assert "?" not in called_url or "key=" not in called_url, ( + "URL 中不應包含 key query parameter" + ) + class TestModuleImports: """測試模組導入。""" diff --git a/tests/test_lm_config_rules.py b/tests/test_lm_config_rules.py index 5122241e..a67f6a73 100644 --- a/tests/test_lm_config_rules.py +++ b/tests/test_lm_config_rules.py @@ -2,6 +2,7 @@ 用途：測試 LM 翻譯配置與規則相關功能。 """ + import pytest from unittest.mock import patch @@ -13,122 +14,133 @@ class TestAPIKeyManagement: def reset_key_tracker(self): """每個測試執行前重置 KeyIndexTracker 狀態，確保測試隔離。""" from translation_tool.core.lm_config_rules import reset_key_index + reset_key_index() # 重置在測試之前 yield reset_key_index() # 重置在測試之後（確保不影響後續測試） - @patch('translation_tool.core.lm_config_rules.load_config') + @patch("translation_tool.core.lm_config_rules.load_config") def test_get_current_api_key_with_keys(self, mock_load_config): """測試取得當前 API Key（有金鑰時）。""" from translation_tool.core.lm_config_rules import get_current_api_key - + mock_load_config.return_value = { "lm_translator": { - "keys": ["AIzaTestKey123", "AIzaTestKey456"] + "keys": [ + "AIzaSyctest123456789012345678901234567890", + "AIzaSydumm2222222222222222222222222222222222", + ] } } - + result = get_current_api_key() - - assert result == "AIzaTestKey123" - @patch('translation_tool.core.lm_config_rules.load_config') + assert result == "AIzaSyctest123456789012345678901234567890" + + @patch("translation_tool.core.lm_config_rules.load_config") def test_get_current_api_key_empty(self, mock_load_config): """測試取得當前 API Key（無金鑰時）。""" from translation_tool.core.lm_config_rules import get_current_api_key - + mock_load_config.return_value = {"lm_translator": {"keys": []}} - + result = get_current_api_key() - + assert result == "" - @patch('translation_tool.core.lm_config_rules.load_config') + @patch("translation_tool.core.lm_config_rules.load_config") def test_rotate_api_key_success(self, mock_load_config): """測試 API Key 輪換（成功）。""" - from translation_tool.core.lm_config_rules import rotate_api_key, reset_key_index, get_current_key_index - + from translation_tool.core.lm_config_rules import ( + rotate_api_key, + reset_key_index, + get_current_key_index, + ) + mock_load_config.return_value = { "lm_translator": { - "keys": ["AIzaTestKey123", "AIzaTestKey456"] + "keys": [ + "AIzaSyctest123456789012345678901234567890", + "AIzaSydumm2222222222222222222222222222222222", + ] } } - + # 重置索引 reset_key_index() - + result = rotate_api_key() - + assert result is True assert get_current_key_index() == 1 - @patch('translation_tool.core.lm_config_rules.load_config') + @patch("translation_tool.core.lm_config_rules.load_config") def test_rotate_api_key_no_more_keys(self, mock_load_config): """測試 API Key 輪換（無更多金鑰）。""" - from translation_tool.core.lm_config_rules import rotate_api_key, reset_key_index - + from translation_tool.core.lm_config_rules import ( + rotate_api_key, + reset_key_index, + ) + mock_load_config.return_value = { - "lm_translator": { - "keys": ["AIzaTestKey123"] - } + "lm_translator": {"keys": ["AIzaSyctest123456789012345678901234567890"]} } - + # 重置索引 reset_key_index() - + result = rotate_api_key() - + assert result is False - @patch('translation_tool.core.lm_config_rules.load_config') + @patch("translation_tool.core.lm_config_rules.load_config") def test_validate_api_keys_success(self, mock_load_config): """測試 API Key 驗證（成功）。""" from translation_tool.core.lm_config_rules import validate_api_keys - + mock_load_config.return_value = { "lm_translator": { - "keys": ["AIzaTestKey123", "AIzaTestKey456"] + "keys": [ + "AIzaSyctest123456789012345678901234567890", + "AIzaSydumm2222222222222222222222222222222222", + ] } } - + # 不應該拋出異常 validate_api_keys() - @patch('translation_tool.core.lm_config_rules.load_config') + @patch("translation_tool.core.lm_config_rules.load_config") def test_validate_api_keys_empty(self, mock_load_config): """測試 API Key 驗證（無金鑰）。""" from translation_tool.core.lm_config_rules import validate_api_keys - + mock_load_config.return_value = {"lm_translator": {"keys": []}} - + with pytest.raises(RuntimeError, match="沒有找到任何 API Key"): validate_api_keys() - @patch('translation_tool.core.lm_config_rules.load_config') + @patch("translation_tool.core.lm_config_rules.load_config") def test_validate_api_keys_invalid_format(self, mock_load_config): """測試 API Key 驗證（無效格式）。""" from translation_tool.core.lm_config_rules import validate_api_keys - - mock_load_config.return_value = { - "lm_translator": { - "keys": ["InvalidKey123"] - } - } - + + mock_load_config.return_value = {"lm_translator": {"keys": ["InvalidKey123"]}} + with pytest.raises(RuntimeError, match="無效的 API Key 格式"): validate_api_keys() def test_validate_api_keys_from_ui_success(self): """測試 UI API Key 驗證（成功）。""" from translation_tool.core.lm_config_rules import validate_api_keys_from_ui - + # 不應該拋出異常 - validate_api_keys_from_ui(["AIzaTestKey123"]) + validate_api_keys_from_ui(["AIzaSyctest123456789012345678901234567890"]) def test_validate_api_keys_from_ui_invalid(self): """測試 UI API Key 驗證（無效）。""" from translation_tool.core.lm_config_rules import validate_api_keys_from_ui - + with pytest.raises(RuntimeError, match="無效的 API Key 格式"): validate_api_keys_from_ui(["InvalidKey"]) @@ -139,37 +151,37 @@ class TestCJKDetection: def test_contains_cjk_chinese(self): """測試包含中文。""" from translation_tool.core.lm_config_rules import contains_cjk - + assert contains_cjk("你好世界") is True def test_contains_cjk_japanese(self): """測試包含日文。""" from translation_tool.core.lm_config_rules import contains_cjk - + assert contains_cjk("こんにちは") is True def test_contains_cjk_korean(self): """測試包含韓文。""" from translation_tool.core.lm_config_rules import contains_cjk - + assert contains_cjk("안녕하세요") is True def test_contains_cjk_english_only(self): """測試只有英文。""" from translation_tool.core.lm_config_rules import contains_cjk - + assert contains_cjk("Hello World") is False def test_contains_cjk_empty_string(self): """測試空字串。""" from translation_tool.core.lm_config_rules import contains_cjk - + assert contains_cjk("") is False def test_contains_cjk_none_input(self): """測試 None 輸入。""" from translation_tool.core.lm_config_rules import contains_cjk - + assert contains_cjk(None) is False @@ -179,43 +191,43 @@ class TestNeedsTranslationText: def test_needs_translation_empty(self): """測試空字串。""" from translation_tool.core.lm_config_rules import needs_translation_text - + assert needs_translation_text("") is False def test_needs_translation_none(self): """測試 None 輸入。""" from translation_tool.core.lm_config_rules import needs_translation_text - + assert needs_translation_text(None) is False def test_needs_translation_chinese(self): """測試中文（不需要翻譯）。""" from translation_tool.core.lm_config_rules import needs_translation_text - + assert needs_translation_text("你好") is False def test_needs_translation_english(self): """測試英文（需要翻譯）。""" from translation_tool.core.lm_config_rules import needs_translation_text - + assert needs_translation_text("Hello") is True def test_needs_translation_digit(self): """測試純數字。""" from translation_tool.core.lm_config_rules import needs_translation_text - + assert needs_translation_text("123") is False def test_needs_translation_section_symbol(self): """測試章節符號開頭。""" from translation_tool.core.lm_config_rules import needs_translation_text - + assert needs_translation_text("§lBold Text") is False def test_needs_translation_token(self): """測試 token 格式。""" from translation_tool.core.lm_config_rules import needs_translation_text - + assert needs_translation_text("$(some.token)") is False @@ -225,14 +237,14 @@ class TestValueFullyTranslated: def test_value_fully_translated_string(self): """測試字串翻譯狀態。""" from translation_tool.core.lm_config_rules import value_fully_translated - + assert value_fully_translated("已翻譯的文字") is True assert value_fully_translated("") is False def test_value_fully_translated_list(self): """測試列表翻譯狀態。""" from translation_tool.core.lm_config_rules import value_fully_translated - + # 全部非空視為已翻譯 assert value_fully_translated(["item1", "item2"]) is True # 包含空字串 @@ -241,7 +253,7 @@ def test_value_fully_translated_list(self): def test_value_fully_translated_other_types(self): """測試其他類型。""" from translation_tool.core.lm_config_rules import value_fully_translated - + assert value_fully_translated(123) is True assert value_fully_translated({"key": "value"}) is True assert value_fully_translated(None) is True @@ -253,9 +265,9 @@ class TestBuildSkipTermsPattern: def test_build_skip_terms_pattern_single(self): """測試單一術語。""" from translation_tool.core.lm_config_rules import build_skip_terms_pattern - + pattern = build_skip_terms_pattern(["discord"]) - + assert pattern is not None assert pattern.search("discord") is not None assert pattern.search("DISCORD") is not None @@ -263,9 +275,9 @@ def test_build_skip_terms_pattern_single(self): def test_build_skip_terms_pattern_multiple(self): """測試多個術語。""" from translation_tool.core.lm_config_rules import build_skip_terms_pattern - + pattern = build_skip_terms_pattern(["api", "discord", "github"]) - + assert pattern.search("api") is not None assert pattern.search("discord") is not None assert pattern.search("github") is not None @@ -273,144 +285,123 @@ def test_build_skip_terms_pattern_multiple(self): def test_build_skip_terms_pattern_escape(self): """測試特殊字元轉義。""" from translation_tool.core.lm_config_rules import build_skip_terms_pattern - + pattern = build_skip_terms_pattern(["test.key"]) - + assert pattern.search("test.key") is not None class TestIsValueTranslatable: """測試值是否可翻譯的判斷。""" - @patch('translation_tool.core.lm_config_rules.load_config') + @patch("translation_tool.core.lm_config_rules.load_config") def test_is_value_translatable_lang_true(self, mock_load_config): """測試 lang 值可翻譯。""" from translation_tool.core.lm_config_rules import is_value_translatable - + mock_load_config.return_value = { "lm_translator": { "translator": { "translatable_keywords": ["text", "name"], - "skip_terms": [] + "skip_terms": [], } } } - + assert is_value_translatable("Hello World", is_lang=True) is True - @patch('translation_tool.core.lm_config_rules.load_config') + @patch("translation_tool.core.lm_config_rules.load_config") def test_is_value_translatable_cjk(self, mock_load_config): """測試含 CJK 的值不可翻譯。""" from translation_tool.core.lm_config_rules import is_value_translatable - + mock_load_config.return_value = { "lm_translator": { - "translator": { - "translatable_keywords": ["text"], - "skip_terms": [] - } + "translator": {"translatable_keywords": ["text"], "skip_terms": []} } } - + assert is_value_translatable("你好", is_lang=True) is False - @patch('translation_tool.core.lm_config_rules.load_config') + @patch("translation_tool.core.lm_config_rules.load_config") def test_is_value_translatable_tech_pattern(self, mock_load_config): """測試技術模式不可翻譯。""" from translation_tool.core.lm_config_rules import is_value_translatable - + mock_load_config.return_value = { "lm_translator": { - "translator": { - "translatable_keywords": ["text"], - "skip_terms": [] - } + "translator": {"translatable_keywords": ["text"], "skip_terms": []} } } - + # minecraft:xxx 格式 assert is_value_translatable("minecraft:diamond", is_lang=True) is False - @patch('translation_tool.core.lm_config_rules.load_config') + @patch("translation_tool.core.lm_config_rules.load_config") def test_is_value_translatable_empty(self, mock_load_config): """測試空值不可翻譯。""" from translation_tool.core.lm_config_rules import is_value_translatable - + mock_load_config.return_value = { "lm_translator": { - "translator": { - "translatable_keywords": ["text"], - "skip_terms": [] - } + "translator": {"translatable_keywords": ["text"], "skip_terms": []} } } - + assert is_value_translatable("", is_lang=True) is False - @patch('translation_tool.core.lm_config_rules.load_config') + @patch("translation_tool.core.lm_config_rules.load_config") def test_is_value_translatable_roman_numeral(self, mock_load_config): """測試羅馬數字不可翻譯。""" from translation_tool.core.lm_config_rules import is_value_translatable - + mock_load_config.return_value = { "lm_translator": { - "translator": { - "translatable_keywords": ["text"], - "skip_terms": [] - } + "translator": {"translatable_keywords": ["text"], "skip_terms": []} } } - + assert is_value_translatable("III", is_lang=True) is False - @patch('translation_tool.core.lm_config_rules.load_config') + @patch("translation_tool.core.lm_config_rules.load_config") def test_is_value_translatable_digit(self, mock_load_config): """測試純數字不可翻譯。""" from translation_tool.core.lm_config_rules import is_value_translatable - + mock_load_config.return_value = { "lm_translator": { - "translator": { - "translatable_keywords": ["text"], - "skip_terms": [] - } + "translator": {"translatable_keywords": ["text"], "skip_terms": []} } } - + assert is_value_translatable("123", is_lang=True) is False class TestIsTranslatableField: """測試欄位是否可翻譯的判斷。""" - @patch('translation_tool.core.lm_config_rules.load_config') + @patch("translation_tool.core.lm_config_rules.load_config") def test_is_translatable_field_true(self, mock_load_config): """測試可翻譯欄位。""" from translation_tool.core.lm_config_rules import is_translatable_field - + mock_load_config.return_value = { "lm_translator": { - "translator": { - "translatable_keywords": ["text", "name", "description"] - } + "translator": {"translatable_keywords": ["text", "name", "description"]} } } - + assert is_translatable_field("item_text") is True assert is_translatable_field("display_name") is True - @patch('translation_tool.core.lm_config_rules.load_config') + @patch("translation_tool.core.lm_config_rules.load_config") def test_is_translatable_field_false(self, mock_load_config): """測試不可翻譯欄位。""" from translation_tool.core.lm_config_rules import is_translatable_field - + mock_load_config.return_value = { - "lm_translator": { - "translator": { - "translatable_keywords": ["text", "name"] - } - } + "lm_translator": {"translator": {"translatable_keywords": ["text", "name"]}} } - + assert is_translatable_field("id") is False assert is_translatable_field("damage") is False diff --git a/tests/test_lm_response_parser.py b/tests/test_lm_response_parser.py index 4b4bf717..ebcacb55 100644 --- a/tests/test_lm_response_parser.py +++ b/tests/test_lm_response_parser.py @@ -2,6 +2,7 @@ 用途：測試 LM 回應解析器相關功能。 """ + import pytest from translation_tool.core.lm_response_parser import ( safe_json_loads, @@ -51,7 +52,7 @@ def test_multiple_json_objects(self): def test_invalid_json_raises_error(self): """測試無效 JSON 拋出錯誤。""" - text = 'this is not valid json at all' + text = "this is not valid json at all" with pytest.raises(RuntimeError): safe_json_loads(text) @@ -112,6 +113,82 @@ def test_dicts(self): assert result == [[{"a": 1}, {"b": 2}], [{"c": 3}]] +class TestSafeJsonLoadsNonGreedy: + r"""測試 safe_json_loads 的 non-greedy regex 行為（Issue #12 修復）。 + + 驗證重點：non-greedy regex 不會吃太多內容（trailing text）， + 能正確解析多重 JSON 區塊。 + + non-greedy `\{[\s\S]*?\}` 匹配規則： + - 從左到右找到第一個完整 {...} 就停止 + - 不會 greedily 吃到 trailing text + - 若有多個巢狀 JSON，外層會被完整匹配（因為需要找配對的 }） + """ + + def test_json_with_trailing_text_non_greedy(self): + """測試 JSON 後有 trailing text 時，non-greedy 不會吃額外內容。 + + Issue #12 核心修復：greedy regex 吃到 "} extra text"， + 導致 json.loads() 失敗。non-greedy 只匹配到第一個完整 {}。 + """ + text = '{"items": [{"id": "0", "value": "你好"}]} extra text after' + result = safe_json_loads(text) + assert result == {"items": [{"id": "0", "value": "你好"}]} + + def test_json_surrounded_by_text_non_greedy(self): + """測試 JSON 被文字環繞時，non-greedy 只取第一個 JSON 區塊。""" + text = 'Some prefix {"key": "value"} some suffix' + result = safe_json_loads(text) + assert result == {"key": "value"} + + def test_multiple_json_blocks_takes_first(self): + """測試有多個 JSON 區塊時，取第一個（而非 greedy 吃到底）。""" + text = '{"first": 1} and then {"second": 2}' + result = safe_json_loads(text) + assert result == {"first": 1} + + def test_json_inside_code_fence_with_trailing_text(self): + """測試 Markdown code fence 中有多餘文字時，仍正確解析。""" + text = '```json\n{"key": "value"}\n```\nHere is some extra text' + result = safe_json_loads(text) + assert result == {"key": "value"} + + def test_code_block_with_multiple_json_blocks(self): + """測試 code block 內有多個 JSON 區塊時，取第一個完整 JSON。 + + re.findall 返回所有匹配，迭代時第一個可解析的成功。 + """ + text = '```\n{"items": [{"a": 1}]}\n{"extra": "data"}\n```' + result = safe_json_loads(text) + # non-greedy 第一個完整 match 是 {"items": [{"a": 1}]} + assert result == {"items": [{"a": 1}]} + + def test_deeply_nested_json_non_greedy(self): + """測試深度巢狀 JSON 能正確解析。 + + non-greedy `\{[\s\S]*?\}` 匹配時，regex engine 會擴展 `[\s\S]*?` + 直到找到一組平衡的 {...}。因此第一個完整 match 是外層物件， + 而非 inner（inner 雖然是完整 JSON，但需要更多 expansion 才能被確認）。 + """ + text = '{"outer": {"inner": {"deep": "value"}, "other": "skip"}} extra' + result = safe_json_loads(text) + # non-greedy 第一個完整 match 是外層物件（regex 擴展到所有內層都關閉） + assert result == {"outer": {"inner": {"deep": "value"}, "other": "skip"}} + + def test_realistic_gemini_response(self): + """測試模擬真實 Gemini 回應（含多餘內容）。""" + text = '```json\n{"items": [{"id": "0", "value": "翻譯結果"}]}\n```\n我認為這個翻譯是正確的。' + result = safe_json_loads(text) + assert result == {"items": [{"id": "0", "value": "翻譯結果"}]} + + def test_brace_balanced_nested_object(self): + """測試 brace-balanced 巢狀物件（最典型翻譯回應格式）。""" + # 這是最常見的 Gemini 回應格式：完整 JSON 物件 + text = '{"items": [{"id": "0", "translations": {"zh_tw": "你好"}}]}' + result = safe_json_loads(text) + assert result == {"items": [{"id": "0", "translations": {"zh_tw": "你好"}}]} + + class TestModuleExports: """測試模組導出。""" @@ -121,5 +198,6 @@ def test_exports(self): safe_json_loads, chunked, ) + assert callable(safe_json_loads) assert callable(chunked) diff --git a/tests/test_lm_translator_main.py b/tests/test_lm_translator_main.py index fd6b46ca..9aa46267 100644 --- a/tests/test_lm_translator_main.py +++ b/tests/test_lm_translator_main.py @@ -9,11 +9,11 @@ class TestTranslateBatchSmart: """translate_batch_smart 測試""" - @patch('translation_tool.core.lm_translator_main.safe_json_loads') - @patch('translation_tool.core.lm_translator_main.load_config') - @patch('translation_tool.core.lm_translator_main.get_current_api_key') - @patch('translation_tool.core.lm_translator_main.call_gemini_requests') - @patch('translation_tool.core.lm_translator_main.time.sleep') + @patch("translation_tool.core.lm_translator_main.safe_json_loads") + @patch("translation_tool.core.lm_translator_main.load_config") + @patch("translation_tool.core.lm_translator_main.get_current_api_key") + @patch("translation_tool.core.lm_translator_main.call_gemini_requests") + @patch("translation_tool.core.lm_translator_main.time.sleep") def test_translate_batch_smart_lang_success( self, mock_sleep, mock_call_api, mock_get_key, mock_config, mock_json_loads ): @@ -29,16 +29,14 @@ def test_translate_batch_smart_lang_success( "models": {"gemini-pro": {"enabled": True}}, "temperature": 0.2, "lang_system_prompt": "test", - "patchouli_system_prompt": "test" + "patchouli_system_prompt": "test", } } mock_get_key.return_value = "test_key" mock_call_api.return_value = '{"items": [{"id": "0", "value": "你好"}]}' mock_json_loads.return_value = {"items": [{"id": "0", "value": "你好"}]} - items = [ - {"path": "test.key", "text": "Hello", "cache_type": "lang"} - ] + items = [{"path": "test.key", "text": "Hello", "cache_type": "lang"}] result, status = translate_batch_smart(items, 1) @@ -46,11 +44,11 @@ def test_translate_batch_smart_lang_success( # 成功時 API 應該被調用一次 mock_call_api.assert_called_once() - @patch('translation_tool.core.lm_translator_main.safe_json_loads') - @patch('translation_tool.core.lm_translator_main.load_config') - @patch('translation_tool.core.lm_translator_main.get_current_api_key') - @patch('translation_tool.core.lm_translator_main.call_gemini_requests') - @patch('translation_tool.core.lm_translator_main.time.sleep') + @patch("translation_tool.core.lm_translator_main.safe_json_loads") + @patch("translation_tool.core.lm_translator_main.load_config") + @patch("translation_tool.core.lm_translator_main.get_current_api_key") + @patch("translation_tool.core.lm_translator_main.call_gemini_requests") + @patch("translation_tool.core.lm_translator_main.time.sleep") def test_translate_batch_smart_empty_batch( self, mock_sleep, mock_call_api, mock_get_key, mock_config, mock_json_loads ): @@ -71,11 +69,11 @@ def test_translate_batch_smart_empty_batch( assert status == "AUTO" mock_call_api.assert_not_called() - @patch('translation_tool.core.lm_translator_main.safe_json_loads') - @patch('translation_tool.core.lm_translator_main.load_config') - @patch('translation_tool.core.lm_translator_main.get_current_api_key') - @patch('translation_tool.core.lm_translator_main.call_gemini_requests') - @patch('translation_tool.core.lm_translator_main.time.sleep') + @patch("translation_tool.core.lm_translator_main.safe_json_loads") + @patch("translation_tool.core.lm_translator_main.load_config") + @patch("translation_tool.core.lm_translator_main.get_current_api_key") + @patch("translation_tool.core.lm_translator_main.call_gemini_requests") + @patch("translation_tool.core.lm_translator_main.time.sleep") def test_translate_batch_smart_api_error_with_retry( self, mock_sleep, mock_call_api, mock_get_key, mock_config, mock_json_loads ): @@ -100,14 +98,125 @@ def test_translate_batch_smart_api_error_with_retry( mock_sleep.assert_called() +class TestSystemPromptConversion: + """測試 System Prompt dict → string 轉換（PATCHOULI_SYSTEM_PROMPT / LANG_SYSTEM_PROMPT）。 + + 驗證設定檔中 system_prompt 無論是 dict 或 string， + 都會被正確轉為 string 傳入 API。 + """ + + @patch("translation_tool.core.lm_api_client.requests.post") + @patch("translation_tool.core.lm_translator_main.load_config") + @patch("translation_tool.core.lm_translator_main.get_current_api_key") + @patch("translation_tool.core.lm_translator_main.time.sleep") + def test_lang_prompt_dict_converted_to_string( + self, mock_sleep, mock_get_key, mock_config, mock_post + ): + """測試 lang_system_prompt 為 dict 時會被轉為 string。""" + from unittest.mock import Mock + from translation_tool.core.lm_translator_main import translate_batch_smart + + mock_response = Mock() + mock_response.ok = True + mock_response.json.return_value = { + "candidates": [ + { + "content": { + "parts": [{"text": '{"items": [{"id": "0", "value": "你好"}]}'}] + } + } + ] + } + mock_post.return_value = mock_response + + mock_config.return_value = { + "lm_translator": { + "initial_batch_size_lang": 300, + "batch_shrink_factor": 0.75, + "min_batch_size": 50, + "models": {"gemini-pro": {"enabled": True}}, + "temperature": 0.2, + "patchouli_system_prompt": "你是專業的 Minecraft Patchouli 翻譯員", + "lang_system_prompt": { + "role": "translator", + "content": "你正在翻譯 Minecraft 語言檔案", + }, + } + } + mock_get_key.return_value = "test_key" + + items = [{"path": "test.key", "text": "Hello", "cache_type": "lang"}] + + result, status = translate_batch_smart(items, 1) + + assert mock_post.call_count >= 1, "API 應該被調用至少一次" + call_kwargs = mock_post.call_args.kwargs + json_body = call_kwargs.get("json", {}) + system_instruction = json_body.get("systemInstruction", {}) + prompt_text = system_instruction.get("parts", [{}])[0].get("text", "") + assert isinstance(prompt_text, str), ( + "lang_system_prompt 必須是 string，而非 dict" + ) + + @patch("translation_tool.core.lm_api_client.requests.post") + @patch("translation_tool.core.lm_translator_main.load_config") + @patch("translation_tool.core.lm_translator_main.get_current_api_key") + @patch("translation_tool.core.lm_translator_main.time.sleep") + def test_prompt_already_string_unchanged( + self, mock_sleep, mock_get_key, mock_config, mock_post + ): + """測試 system_prompt 原本就是 string 時，內容保持不變。""" + from unittest.mock import Mock + from translation_tool.core.lm_translator_main import translate_batch_smart + + prompt_text = "你是一個專業的 Minecraft 翻譯員" + + mock_response = Mock() + mock_response.ok = True + mock_response.json.return_value = { + "candidates": [ + { + "content": { + "parts": [{"text": '{"items": [{"id": "0", "value": "結果"}]}'}] + } + } + ] + } + mock_post.return_value = mock_response + + mock_config.return_value = { + "lm_translator": { + "initial_batch_size_lang": 300, + "batch_shrink_factor": 0.75, + "min_batch_size": 50, + "models": {"gemini-pro": {"enabled": True}}, + "temperature": 0.2, + "patchouli_system_prompt": "另一個 prompt", + "lang_system_prompt": prompt_text, + } + } + mock_get_key.return_value = "test_key" + + items = [{"path": "test.key", "text": "Hello", "cache_type": "lang"}] + + result, status = translate_batch_smart(items, 1) + + assert mock_post.call_count >= 1, "API 應該被調用至少一次" + call_kwargs = mock_post.call_args.kwargs + json_body = call_kwargs.get("json", {}) + system_instruction = json_body.get("systemInstruction", {}) + actual_prompt = system_instruction.get("parts", [{}])[0].get("text", "") + assert actual_prompt == prompt_text, "string 類型的 system_prompt 應保持不變" + + class TestBatchProfileDetection: """批次設定偵測測試""" - @patch('translation_tool.core.lm_translator_main.safe_json_loads') - @patch('translation_tool.core.lm_translator_main.load_config') - @patch('translation_tool.core.lm_translator_main.get_current_api_key') - @patch('translation_tool.core.lm_translator_main.call_gemini_requests') - @patch('translation_tool.core.lm_translator_main.time.sleep') + @patch("translation_tool.core.lm_translator_main.safe_json_loads") + @patch("translation_tool.core.lm_translator_main.load_config") + @patch("translation_tool.core.lm_translator_main.get_current_api_key") + @patch("translation_tool.core.lm_translator_main.call_gemini_requests") + @patch("translation_tool.core.lm_translator_main.time.sleep") def test_detect_batch_profile_lang( self, mock_sleep, mock_call_api, mock_get_key, mock_config, mock_json_loads ): @@ -126,7 +235,10 @@ def test_detect_batch_profile_lang( mock_call_api.return_value = '{"items": []}' mock_json_loads.return_value = {"items": []} - items = [{"path": f"key.{i}", "text": f"text{i}", "cache_type": "lang"} for i in range(10)] + items = [ + {"path": f"key.{i}", "text": f"text{i}", "cache_type": "lang"} + for i in range(10) + ] result, status = translate_batch_smart(items, 1) assert status in ["AUTO", "PARTIAL", "FAILED"] diff --git a/tests/test_lm_translator_main_prompts.py b/tests/test_lm_translator_main_prompts.py new file mode 100644 index 00000000..bc8ab830 --- /dev/null +++ b/tests/test_lm_translator_main_prompts.py @@ -0,0 +1,19 @@ +"""測試 System Prompt dict → string 轉換（lm_translator_main.py）。""" + + +class TestSystemPromptConversion: + """測試 PATCHOULI_SYSTEM_PROMPT / LANG_SYSTEM_PROMPT 能正確處理 dict 輸入。""" + + def test_patchouli_prompt_dict_conversion_logic(self): + """測試 dict（含 content/text key）轉換邏輯。""" + raw = {"role": "system", "content": "測試內容"} + result = raw.get("content") or raw.get("text") or str(raw) + assert result == "測試內容" + assert isinstance(result, str) + + def test_lang_prompt_dict_conversion_logic(self): + """測試 lang dict 轉換邏輯。""" + raw = {"role": "system", "content": "Minecraft 翻譯中"} + result = raw.get("content") or raw.get("text") or str(raw) + assert result == "Minecraft 翻譯中" + assert isinstance(result, str) diff --git a/tests/test_pipeline_services_error_handling.py b/tests/test_pipeline_services_error_handling.py index 0eea2288..4121e446 100644 --- a/tests/test_pipeline_services_error_handling.py +++ b/tests/test_pipeline_services_error_handling.py @@ -6,38 +6,42 @@ def __init__(self): self.calls = [] def start(self): - self.calls.append('start') + self.calls.append("start") def finish(self): - self.calls.append('finish') + self.calls.append("finish") def set_error(self): - self.calls.append('set_error') + self.calls.append("set_error") def add_log(self, text): - self.calls.append(('add_log', text)) + self.calls.append(("add_log", text)) def test_run_callable_task_sets_error_and_optional_session_log(monkeypatch): session = _Session() seen = [] - monkeypatch.setattr(_task_runner, 'ensure_pipeline_logging', lambda: None) - monkeypatch.setattr(_task_runner.UI_LOG_HANDLER, 'set_session', lambda s: seen.append(s)) + monkeypatch.setattr(_task_runner, "ensure_pipeline_logging", lambda: None) + monkeypatch.setattr( + _task_runner.UI_LOG_HANDLER, "set_session", lambda s: seen.append(s) + ) def boom(**kwargs): - raise RuntimeError('boom') + raise RuntimeError("boom") result = _task_runner.run_callable_task( session=session, - task_name='task', + task_name="task", func=boom, kwargs={}, add_session_log_on_error=True, ) assert result is None - assert session.calls[0] == 'start' - assert session.calls[-1] == 'set_error' - assert any(isinstance(c, tuple) and c[0] == 'add_log' for c in session.calls) + assert session.calls[0] == "start" + assert ( + session.calls[-2] == "set_error" + ) # set_error 在倒數第二，finally 的 finish() 在最後 + assert any(isinstance(c, tuple) and c[0] == "add_log" for c in session.calls) assert seen == [session, None] diff --git a/translation_tool/core/lm_api_client.py b/translation_tool/core/lm_api_client.py index 23bf890a..85121d0a 100644 --- a/translation_tool/core/lm_api_client.py +++ b/translation_tool/core/lm_api_client.py @@ -12,6 +12,7 @@ from translation_tool.utils.config_manager import load_config + def call_gemini_requests( *, model_name: str, @@ -24,10 +25,12 @@ def call_gemini_requests( url = ( "https://generativelanguage.googleapis.com/" f"v1beta/models/{model_name}:generateContent" - f"?key={api_key}" ) - headers = {"Content-Type": "application/json"} + headers = { + "Content-Type": "application/json", + "x-goog-api-key": api_key, + } data = { "systemInstruction": {"parts": [{"text": system_prompt}]}, diff --git a/translation_tool/core/lm_config_rules.py b/translation_tool/core/lm_config_rules.py index dbf4fd9c..8b89497b 100644 --- a/translation_tool/core/lm_config_rules.py +++ b/translation_tool/core/lm_config_rules.py @@ -12,7 +12,6 @@ from ..utils.log_unit import log_info, log_error, log_debug - # ========================= # 1. 執行緒安全的 API Key 索引追蹤器 # ========================= @@ -21,7 +20,7 @@ class KeyIndexTracker: """ 執行緒安全的 API Key 索引追蹤器。 - + 用於解決多執行緒環境下全域變數 _current_key_index 的 race condition 問題。 透過 threading.Lock 確保並發存取的安全性。 """ @@ -83,6 +82,7 @@ def reset_key_index() -> None: # 2. 提示詞與配置 # ========================= + def _get_all_keys() -> list[str]: """ 私有輔助函式：統一代理從設定檔讀取並清理金鑰列表。 @@ -94,6 +94,7 @@ def _get_all_keys() -> list[str]: if isinstance(key, str) and key.strip() ] + def get_current_api_key() -> str: """ 從金鑰池中取得目前正在使用的 API 金鑰。 @@ -125,6 +126,7 @@ def get_current_api_key() -> str: _key_tracker._index = _key_tracker._index % len(keys) return key + def rotate_api_key(): """ 切換至下一個可用的 API Key。 @@ -168,6 +170,7 @@ def rotate_api_key(): log_info(f"🔁 切換 API Key → index {new_index}") return True + def validate_api_keys(): """ 驗證 API 金鑰格式。 @@ -187,9 +190,24 @@ def validate_api_keys(): f"❌ 無效的 API Key 格式：{k!r}\n" "Gemini API Key 應以 'AIza' 開頭，請檢查您的設定檔。" ) + # 2. 檢查金鑰長度（Google API Key 通常為 39-40 個字元） + if len(k) < 35: + log_error(f"❌ 偵測到過短的 API 金鑰: {k!r} (長度={len(k)})") + raise RuntimeError( + f"❌ API Key 長度異常：{k!r}\n" + f"長度為 {len(k)}，正常應為 35-45 個字元，請檢查是否輸入正確。" + ) + # 3. 檢查金鑰字元是否僅包含允許的字元（AIza + 英數字/ dash / underscore） + if not re.match(r"^AIza[a-zA-Z0-9_-]+$", k): + log_error(f"❌ 偵測到包含無效字元的 API 金鑰: {k!r}") + raise RuntimeError( + f"❌ API Key 包含無效字元：{k!r}\n" + "僅允許 'AIza' 開頭後接英文字母、數字、 dash(-) 或 underscore(_)。" + ) log_info(f"✅ 金鑰格式驗證通過，共載入 {len(keys)} 組金鑰。") + def validate_api_keys_from_ui(keys: list[str]): # ui 專用 """驗證 API Key 格式（UI 專用）。 @@ -197,12 +215,25 @@ def validate_api_keys_from_ui(keys: list[str]): # ui 專用 keys: API Key 列表 """ for k in keys: - if not k or not k.startswith("AIza"): + if not k: + raise RuntimeError("❌ API Key 不得為空，請輸入有效的 Gemini API Key。") + if not k.startswith("AIza"): raise RuntimeError( f"❌ 無效的 API Key 格式：{k!r}\n" "請使用 Google AI Studio 產生的 Gemini API Key，" "通常應以 'AIza' 字樣開頭。" ) + if len(k) < 35: + raise RuntimeError( + f"❌ API Key 長度異常：{k!r}\n" + f"長度為 {len(k)}，正常應為 35-45 個字元，請檢查是否輸入正確。" + ) + if not re.match(r"^AIza[a-zA-Z0-9_-]+$", k): + raise RuntimeError( + f"❌ API Key 包含無效字元：{k!r}\n" + "僅允許 'AIza' 開頭後接英文字母、數字、 dash(-) 或 underscore(_)。" + ) + # ========================= # 2. Regex 規則定義 @@ -238,10 +269,9 @@ def validate_api_keys_from_ui(keys: list[str]): # ui 專用 # 需要跳過翻譯的文字（你指定的類型） HASH_PREFIX_PATTERN = re.compile(r"^\s*#") # 任何 # 開頭（含前置空白） -def needs_translation_text(s: str) -> bool: - """ - """ +def needs_translation_text(s: str) -> bool: + """ """ if not s or not isinstance(s, str): return False @@ -260,6 +290,7 @@ def needs_translation_text(s: str) -> bool: # 還有英文 → 需要翻 return True + def value_fully_translated(value) -> bool: """ 判斷一個值是否「已完全翻譯完成」。 @@ -314,6 +345,7 @@ def value_fully_translated(value) -> bool: # 直接視為已完成翻譯 return True + def contains_cjk(s: str) -> bool: """ 檢查字串中是否包含 CJK（中 / 日 / 韓）文字。 @@ -349,6 +381,7 @@ def contains_cjk(s: str) -> bool: """ return isinstance(s, str) and CJK_RE.search(s) is not None + def build_skip_terms_pattern(terms: list[str]) -> re.Pattern: """ 將「需跳過翻譯的關鍵字清單」轉換為單一正規表達式（regex）。 @@ -398,12 +431,11 @@ def build_skip_terms_pattern(terms: list[str]) -> re.Pattern: # 編譯為不區分大小寫的正規表達式 return re.compile(pattern, re.IGNORECASE) + # ========================= # 值是否值得翻譯（核心判斷） def is_value_translatable(value: Any, *, is_lang: bool = False) -> bool: - """ - - """ + """ """ if not isinstance(value, str): return False @@ -463,6 +495,7 @@ def is_value_translatable(value: Any, *, is_lang: bool = False) -> bool: return True + # ========================= # 可翻譯欄位判斷 # ========================= diff --git a/translation_tool/core/lm_response_parser.py b/translation_tool/core/lm_response_parser.py index 0cfdeba3..17959c43 100644 --- a/translation_tool/core/lm_response_parser.py +++ b/translation_tool/core/lm_response_parser.py @@ -9,6 +9,7 @@ import json import re + def safe_json_loads(text: str): """將模型回傳的文字嘗試解析為 JSON，支援去除 Markdown code fence 並從雜訊文字中截取第一個合法 JSON 區塊。""" text = text.strip() @@ -23,15 +24,52 @@ def safe_json_loads(text: str): except json.JSONDecodeError: pass - matches = re.findall(r"\{[\s\S]*\}", text) - for m in matches: + # ✅ Issue #12 修復：使用 brace-counting parser 取代 non-greedy regex + # 正確處理巢狀 JSON 與多個相鄰 JSON 區塊 + blocks = _extract_json_blocks(text) + for block in blocks: try: - return json.loads(m) + return json.loads(block) except json.JSONDecodeError: continue raise RuntimeError("JSON 解析失敗：無法解析模型回傳內容") + +def _extract_json_blocks(text: str): + """使用 brace-counting 找出文字中所有完整的 JSON 區塊。 + + 演算法：從第一個 '{' 開始，計算深度（{ 和 [ +1，} 和 ] -1）。 + 當深度回到 0 時，該區塊為一個完整的 JSON。 + 遇到非 { 或 [ 時不影響（depth 不變）。 + """ + blocks = [] + i = 0 + n = len(text) + while i < n: + if text[i] == "{": + start = i + depth = 0 + j = i + while j < n: + c = text[j] + if c == "{" or c == "[": + depth += 1 + elif c == "}" or c == "]": + depth -= 1 + if depth == 0: + blocks.append(text[start : j + 1]) + i = j + 1 + break + j += 1 + else: + # 未找到匹配的結尾，結束 + break + else: + i += 1 + return blocks + + def chunked(lst, size): """將序列 lst 依指定大小 size 分塊，yield 每個 chunk（最後一塊可能較短）。""" for i in range(0, len(lst), size): diff --git a/translation_tool/core/lm_translator_main.py b/translation_tool/core/lm_translator_main.py index 7ae1d5fa..5fd67861 100644 --- a/translation_tool/core/lm_translator_main.py +++ b/translation_tool/core/lm_translator_main.py @@ -41,32 +41,38 @@ # 翻譯入口函數（新結構） # ========================================================= -def translate_batch_smart(batch_items, total=None, dry_run: bool = DEFAULT_DRY_RUN, export_cache_only: bool = DEFAULT_EXPORT_CACHE_ONLY): + +def translate_batch_smart( + batch_items, + total=None, + dry_run: bool = DEFAULT_DRY_RUN, + export_cache_only: bool = DEFAULT_EXPORT_CACHE_ONLY, +): """ 智慧批次翻譯函數（主入口） - + 參數: batch_items: 翻譯項目列表 total: 總項目數（可選） dry_run: True = 不呼叫API，只模擬流程（測試用） export_cache_only: True = 只輸出快取中的內容 - + 職責：協調各子流程，不直接處理細節 """ # 1. 驗證與正規化 items = _validate_batch_items(batch_items) if not items: return [], "AUTO" - + # 2. 偵測 profile（TODO: 舊函數會重新計算，目前是被丟棄的死碼） # batch_profile = _detect_batch_profile(items) - + # 3. 計算批次大小（TODO: 舊函數會重新計算，目前是被丟棄的死碼） # batch_size = _calculate_batch_size(batch_profile) - + # 4. 執行翻譯 results, status = _execute_translation(items, total, dry_run, export_cache_only) - + # 5. 處理輸出 return _process_output(results, status) @@ -74,7 +80,7 @@ def translate_batch_smart(batch_items, total=None, dry_run: bool = DEFAULT_DRY_R def _validate_batch_items(items): """ 驗證與正規化輸入資料 - + 參數： items: 原始項目列表 回傳： @@ -82,7 +88,7 @@ def _validate_batch_items(items): """ if not items: return [] - + validated = [] for item in items: # 跳过无效项目 @@ -92,13 +98,13 @@ def _validate_batch_items(items): text = item.get("text", "") if not text or not str(text).strip(): continue - + # 確保有 cache_type if "cache_type" not in item: item["cache_type"] = "patchouli" - + validated.append(item) - + return validated @@ -121,7 +127,7 @@ def _execute_translation(items, total, dry_run=False, export_cache_only=False): def _process_output(results, status): """ 處理輸出結果 - + 參數： results: 翻譯結果（可能是元組或列表） status: 翻譯狀態 @@ -131,11 +137,11 @@ def _process_output(results, status): # 處理元組情況（從舊函數返回） if isinstance(results, tuple): return results - + # 處理空結果 if not results: return [], "AUTO" - + return results, status @@ -143,7 +149,10 @@ def _process_output(results, status): # 舊翻譯函數（保留原邏輯） # ========================================================= -def translate_batch_smart_old(batch_items, total=None, dry_run=False, export_cache_only=False): + +def translate_batch_smart_old( + batch_items, total=None, dry_run=False, export_cache_only=False +): """ 智慧型分批翻譯函式 支援動態縮減 Batch Size、模型切換、以及自動處理輸出截斷問題。 @@ -255,19 +264,38 @@ def detect_batch_profile(items): # 模型溫度 MODEL_TEMP = load_config().get("lm_translator", {}).get("temperature", 0.2) - # 使用提示詞 手冊 - PATCHOUI_SYSTEM_PROMPT = ( + # 使用提示詞 手冊（確保為字串） + _patchouli_raw = ( load_config() .get("lm_translator", {}) - .get("patchouli_system_prompt", {"你是專業的 Minecraft Patchouli 手冊翻譯員"}) + .get("patchouli_system_prompt", "你是專業的 Minecraft Patchouli 手冊翻譯員") ) + if isinstance(_patchouli_raw, str): + PATCHOULI_SYSTEM_PROMPT = _patchouli_raw + elif isinstance(_patchouli_raw, dict): + # 支援 {"content": "..."} 或 {"text": "..."} 格式的 dict + PATCHOULI_SYSTEM_PROMPT = ( + _patchouli_raw.get("content") + or _patchouli_raw.get("text") + or str(_patchouli_raw) + ) + else: + PATCHOULI_SYSTEM_PROMPT = str(_patchouli_raw) - # 使用提示詞 lang - LANG_SYSTEM_PROMPT = ( + # 使用提示詞 lang（確保為字串） + _lang_raw = ( load_config() .get("lm_translator", {}) - .get("lang_system_prompt", {"你正在翻譯 Minecraft 語言檔案（JSON格式）。"}) + .get("lang_system_prompt", "你正在翻譯 Minecraft 語言檔案（JSON格式）。") ) + if isinstance(_lang_raw, str): + LANG_SYSTEM_PROMPT = _lang_raw + elif isinstance(_lang_raw, dict): + LANG_SYSTEM_PROMPT = ( + _lang_raw.get("content") or _lang_raw.get("text") or str(_lang_raw) + ) + else: + LANG_SYSTEM_PROMPT = str(_lang_raw) pinned_model_index = None # None = 正常模式，非 None = 鎖定指定模型 # 進入動態 Batch 迴圈 @@ -333,7 +361,7 @@ def detect_batch_profile(items): prompt = LANG_SYSTEM_PROMPT else: # ftb / patchouli / 其他 - prompt = PATCHOUI_SYSTEM_PROMPT + prompt = PATCHOULI_SYSTEM_PROMPT log_debug( "Batch profile=%s -> System Prompt=%s", @@ -366,6 +394,7 @@ def _is_truncated(text: str) -> bool: """ try: import json + json.loads(text) return False # 成功解析，代表沒截斷 except json.JSONDecodeError: @@ -373,9 +402,9 @@ def _is_truncated(text: str) -> bool: # 大括號平衡檢查 count = 0 for ch in text: - if ch == '{': + if ch == "{": count += 1 - elif ch == '}': + elif ch == "}": count -= 1 if count < 0: return True # } 比 { 先出現，代表截斷 @@ -383,7 +412,9 @@ def _is_truncated(text: str) -> bool: if _is_truncated(raw_text): overload_retry_count = 0 # 重置過載計數器 - log_info("[!] 偵測到 JSON 被截斷（結尾不完整或格式錯誤），將縮小 Batch 重試") + log_info( + "[!] 偵測到 JSON 被截斷（結尾不完整或格式錯誤），將縮小 Batch 重試" + ) break # 解析 JSON @@ -457,11 +488,17 @@ def _is_truncated(text: str) -> bool: # ATK-B-2: 翻譯品質驗證 # 1. 空翻譯 if not translated_text or translated_text.strip() == "": - log_warning("[⚠️ 空翻譯] path=%s：原文='%s'", original_item["path"], original_item["text"]) + log_warning( + "[⚠️ 空翻譯] path=%s：原文='%s'", + original_item["path"], + original_item["text"], + ) # 2. 異常長度（翻譯後長度是原文 3 倍以上） orig_len = len(original_item["text"]) if orig_len > 0 and len(translated_text) / orig_len > 3: - log_warning(f"[⚠️ 異常長度] {original_item['path']}：原文 {orig_len} 字，翻譯 {len(translated_text)} 字") + log_warning( + f"[⚠️ 異常長度] {original_item['path']}：原文 {orig_len} 字，翻譯 {len(translated_text)} 字" + ) new_item["text"] = translated_text merged_result.append(new_item) @@ -640,9 +677,7 @@ def _is_truncated(text: str) -> bool: except Exception as parse_err: # 備援比對邏輯 err_msg = str(e).upper() - log_error( - f"[⚠️] 無法解析 429 JSON，使用備援。錯誤: {parse_err}" - ) + log_error(f"[⚠️] 無法解析 429 JSON，使用備援。錯誤: {parse_err}") if "QUOTA" in err_msg or "EXCEEDED" in err_msg: # ⭐ 這裡之前會崩潰，現在這樣改就安全了 @@ -712,7 +747,9 @@ def _is_truncated(text: str) -> bool: log_info( "[✅] API Key 切換成功 → 原地重送同一 batch,等待12秒" ) - time.sleep(key_rotation_buffer_sec) # ⭐ 給新 Key 一點緩衝 + time.sleep( + key_rotation_buffer_sec + ) # ⭐ 給新 Key 一點緩衝 hit_overload_retry = True # ⭐ 重送同一 batch break # ← 跳出 model loop，回 while else: @@ -749,9 +786,7 @@ def _is_truncated(text: str) -> bool: # ======== 500 ========== if status == 500: - log_info( - "[⚠️] 500 INTERNAL：Gemini 後端錯誤，嘗試換模型或縮 batch" - ) + log_info("[⚠️] 500 INTERNAL：Gemini 後端錯誤，嘗試換模型或縮 batch") break # ========== requests timeout ========== @@ -826,7 +861,8 @@ def _is_truncated(text: str) -> bool: # 3. 如果還有剩下的，重置 batch_size if remaining_items: batch_size = min( - len(remaining_items), MIN_BATCH_SIZE if not is_lang else MIN_LANG_BATCH_SIZE + len(remaining_items), + MIN_BATCH_SIZE if not is_lang else MIN_LANG_BATCH_SIZE, ) # 4. 繼續 while 迴圈處理後面的東西 diff --git a/translation_tool/core/lm_translator_shared_loop.py b/translation_tool/core/lm_translator_shared_loop.py index 9cda5977..4f390a43 100644 --- a/translation_tool/core/lm_translator_shared_loop.py +++ b/translation_tool/core/lm_translator_shared_loop.py @@ -11,9 +11,17 @@ import time from translation_tool.utils.log_unit import log_info -from translation_tool.utils.cache_manager import add_to_cache, save_translation_cache, reload_translation_cache +from translation_tool.utils.cache_manager import ( + add_to_cache, + save_translation_cache, + reload_translation_cache, +) from translation_tool.utils.config_manager import load_config -from translation_tool.core.lm_translator_shared_cache import CacheRule, get_default_cache_rules +from translation_tool.core.lm_translator_shared_cache import ( + CacheRule, + get_default_cache_rules, +) + @dataclass class TranslateLoopResult: @@ -27,6 +35,7 @@ class TranslateLoopResult: exhausted: bool last_error: Optional[str] = None + def _get_default_batch_size( cache_type: str, batch_size_by_type: Optional[Dict[str, int]] ) -> int: @@ -47,6 +56,7 @@ def _get_default_batch_size( return int(lm_cfg.get("initial_batch_size_md", 100) or 100) return int(lm_cfg.get("initial_batch_size_lang", 300) or 300) + def translate_items_with_cache_loop( items_to_translate: List[Dict[str, Any]], *, @@ -94,8 +104,8 @@ def emit_progress(msg: str) -> None: else: eta_sec = 0.0 on_progress(progress, msg, eta_sec) - except Exception: - pass + except Exception as e: + log_info(f"[SharedLM] 進度回報失敗: {e}") emit_progress("🚀 [SharedLM] 準備開始翻譯工作...") @@ -146,28 +156,28 @@ def emit_progress(msg: str) -> None: if on_translated_item is not None: try: on_translated_item(it) - except Exception: - pass + except Exception as e: + log_info(f"[SharedLM] 處理翻譯結果失敗: {e}") rule = cache_rules.get(ctype) or CacheRule("path|source_text") cache_key = rule.make_key({"path": pth, "source_text": src}) try: add_to_cache(ctype, cache_key, src, txt) - except Exception: - pass + except Exception as e: + log_info(f"[SharedLM] 新增快取失敗: {e}") remaining = remaining[actual_processed_in_this_batch:] try: save_translation_cache(cache_type, write_new_shard=write_new_cache) - except Exception: - pass + except Exception as e: + log_info(f"[SharedLM] 儲存快取失敗: {e}") if on_batch_flushed is not None: try: on_batch_flushed() - except Exception: - pass + except Exception as e: + log_info(f"[SharedLM] 批次刷新回調失敗: {e}") emit_progress( f"✅ 批次完成 ({cache_type}) | 成功: {actual_processed_in_this_batch} | 總進度: {processed}/{total}" diff --git a/translation_tool/core/lm_translator_shared_recording.py b/translation_tool/core/lm_translator_shared_recording.py index a1deb148..e99a897f 100644 --- a/translation_tool/core/lm_translator_shared_recording.py +++ b/translation_tool/core/lm_translator_shared_recording.py @@ -12,6 +12,7 @@ import csv import json + @dataclass class TranslationRecorder: """收集翻譯紀錄並輸出 JSON/CSV。""" @@ -57,7 +58,10 @@ def export_csv(self, out_path: str | Path) -> Path: cols = cols + extra_cols with out_path.open("w", encoding="utf-8", newline="") as f: - w = csv.DictWriter(f, fieldnames=cols) + # ⭐ Use QUOTE_ALL to prevent CSV injection attacks + # All fields are quoted, preventing malicious values like + # "=cmd|'/C calc'!A0" or values with embedded newlines + w = csv.DictWriter(f, fieldnames=cols, quoting=csv.QUOTE_ALL) w.writeheader() for r in self.rows: w.writerow({k: r.get(k, "") for k in cols}) diff --git a/translation_tool/plugins/ftbquests/ftbquests_lmtranslator.py b/translation_tool/plugins/ftbquests/ftbquests_lmtranslator.py index a00a16b7..fe52943a 100644 --- a/translation_tool/plugins/ftbquests/ftbquests_lmtranslator.py +++ b/translation_tool/plugins/ftbquests/ftbquests_lmtranslator.py @@ -219,18 +219,20 @@ def set_prog(v: float): if not json_files: raise FileNotFoundError(f"找不到任何 .json：{in_dir}") - # ---- Global total keys (raw) ---- + # ---- Global total keys (raw) + cache mappings ---- per_file_counts: List[Tuple[Path, int]] = [] global_total_keys = 0 + # ✅ Issue #7 修復：緩存 JSON mapping 避免重複讀取 + src_mapping_cache: Dict[Path, Dict[str, Any]] = {} - def _count_one(src: Path) -> Tuple[Path, int]: - """讀取指定的 JSON 檔案並統計其中可翻譯的鍵值數量，若發生錯誤則返回 0。""" + def _count_one(src: Path) -> Tuple[Path, int, Dict[str, Any]]: + """讀取 JSON 並統計可翻譯鍵值數量，同時快取 mapping。""" try: mapping = read_json_dict(src) c = count_translatable_keys(mapping) - return src, int(c) + return src, int(c), mapping except Exception: - return src, 0 + return src, 0, {} # max_workers 你可以改成 config 的 parallel_execution_workers @@ -241,9 +243,11 @@ def _count_one(src: Path) -> Tuple[Path, int]: with ThreadPoolExecutor(max_workers=max_workers) as ex: futs = [ex.submit(_count_one, src) for src in json_files] for fu in as_completed(futs): - src, c = fu.result() + src, c, mapping = fu.result() per_file_counts.append((src, c)) global_total_keys += c + if mapping: # ✅ 只緩存非空的 mapping + src_mapping_cache[src] = mapping # 保持穩定順序（避免多執行緒導致排序亂） per_file_counts.sort(key=lambda x: x[0].as_posix()) @@ -263,7 +267,10 @@ def _count_one(src: Path) -> Tuple[Path, int]: if key_count == 0: continue try: - mapping = read_json_dict(src) + # ✅ Issue #7 修復：直接使用緩存的 mapping，不再重複讀取 + mapping = src_mapping_cache.get(src, {}) + if not mapping: + continue rel_src = src.relative_to(in_dir).as_posix() file_hint = f"config/ftbquests/quests/{rel_src}" all_items = map_to_items( @@ -339,7 +346,11 @@ def _writer(file_id: str) -> None: if key_count == 0: continue - mapping = read_json_dict(src) + # ✅ Issue #7 修復：直接使用緩存的 mapping，不再重複讀取 + mapping = src_mapping_cache.get(src, {}) + if not mapping: + log_error(f"⚠️ [FTB-LM] 找不到快取的 mapping：{src}") + continue rel_src = src.relative_to(in_dir).as_posix() # e.g. en_us/ftb_lang.json file_hint = ( @@ -532,14 +543,70 @@ def _fmt_eta(sec: float) -> str: return f"{m}m{s:02d}s" return f"{s}s" - def on_progress(p: float, msg: str, eta_sec: float) -> None: - """報告翻譯進度。""" - eta_txt = _fmt_eta(eta_sec) - if eta_txt: - log_info(f"⏳ [AI 翻譯中] {msg} | 預估剩餘時間：{eta_txt}") - else: - log_info(f"🚀 [AI 翻譯中] {msg}") - set_prog(p) + def make_on_progress(set_prog, _fmt_eta): + def on_progress(p: float, msg: str, eta_sec: float) -> None: + """報告翻譯進度。""" + eta_txt = _fmt_eta(eta_sec) + if eta_txt: + log_info(f"⏳ [AI 翻譯中] {msg} | 預估剩餘時間：{eta_txt}") + else: + log_info(f"🚀 [AI 翻譯中] {msg}") + set_prog(p) + + return on_progress + + def make_on_translated_item(rel_src, dst, out_map, rec, out_dir): + def on_translated_item(it: Dict[str, Any]) -> None: + """處理翻譯結果並寫入映射。""" + p = it.get("path") + t = it.get("text") + src_text = str(it.get("source_text") or "") + if isinstance(p, str) and isinstance(t, str): + try: + shielded_src = it.get("_shielded") or shield_text(src_text) + shields = getattr(shielded_src, "shields", []) + if shields: + t = unshield_text(t, shields) + except Exception: + pass + out_map[p] = t + try: + rec.record( + cache_type="ftbquests", + file_id=rel_src, + path=p, + src=src_text, + dst=t, + cache_hit=False, + extra={"dst_file": dst.relative_to(out_dir).as_posix()}, + ) + except Exception: + pass + + return on_translated_item + + def make_on_batch_flushed(file_id, touch, _writer, dst, out_map): + def on_batch_flushed() -> None: + """批量寫入翻譯結果。""" + try: + touch.touch(file_id) + touch.flush(_writer) # 最小改動：每批也照樣寫，避免中斷損失 + except Exception: + # fallback + write_json_dict(dst, out_map) + + return on_batch_flushed + + # ✅ 確保此檔案在翻譯路徑也有 file_id + file_id = dst.as_posix() + _file_write_table[file_id] = (dst, out_map) + + # ✅ Issue #8 修復：使用工廠函式創建 callbacks + on_translated_item = make_on_translated_item( + rel_src, dst, out_map, rec, out_dir + ) + on_batch_flushed = make_on_batch_flushed(file_id, touch, _writer, dst, out_map) + on_progress = make_on_progress(set_prog, _fmt_eta) res = translate_items_with_cache_loop( items_to_translate, diff --git a/translation_tool/plugins/kubejs/kubejs_tooltip_lmtranslator.py b/translation_tool/plugins/kubejs/kubejs_tooltip_lmtranslator.py index 8e5164df..e67daf2e 100644 --- a/translation_tool/plugins/kubejs/kubejs_tooltip_lmtranslator.py +++ b/translation_tool/plugins/kubejs/kubejs_tooltip_lmtranslator.py @@ -247,7 +247,8 @@ def _count_one(src: Path) -> Tuple[Path, int]: try: mapping = read_json_dict(src) return src, int(count_translatable_keys(mapping)) - except Exception: + except Exception as e: + log_warning(f"[KubeJS-LM] 讀取 JSON 失敗 {src}: {e}") return src, 0 max_workers = int( @@ -303,8 +304,8 @@ def _count_one(src: Path) -> Tuple[Path, int]: _split_off_tw_items(cached_items, items_to_translate) global_total_hit += len(cached_items) global_total_to_translate += len(items_to_translate) - except Exception: - pass + except Exception as e: + log_warning(f"[KubeJS-LM] 預掃描失敗 {src}: {e}") log_info( f"🔎 [KubeJS-LM] 待翻譯檔案數：{len(json_files)}；總 keys：{global_total_keys}\n" @@ -397,6 +398,10 @@ def _writer(file_id: str) -> None: p = it.get("path") t = it.get("text") if isinstance(p, str) and isinstance(t, str): + # ✅ Rich Text Shield：統一快取命中/miss 路徑 + shielded = it.get("_shielded") + if shielded is not None and shielded.shields: + t = unshield_text(t, shielded.shields) out_map[p] = t try: rec.record( @@ -408,8 +413,8 @@ def _writer(file_id: str) -> None: cache_hit=True, extra={"dst_file": dst.relative_to(out_dir).as_posix()}, ) - except Exception: - pass + except Exception as e: + log_warning(f"[KubeJS-LM] 記錄快取命中失敗: {e}") file_id = dst.as_posix() _file_write_table[file_id] = (dst, out_map) @@ -537,21 +542,21 @@ def on_translated_item(it: Dict[str, Any]) -> None: cache_hit=False, extra={"dst_file": st["dst"].relative_to(out_dir).as_posix()}, ) - except Exception: - pass + except Exception as e: + log_warning(f"[KubeJS-LM] 記錄翻譯結果失敗: {e}") try: touch.touch(st["file_id"]) - except Exception: - pass + except Exception as e: + log_warning(f"[KubeJS-LM] touch 失敗: {e}") def on_batch_flushed() -> None: # write touched files each batch """批量寫入翻譯結果。""" try: touch.flush(_writer) - except Exception: - # fallback: write all + except Exception as e: + log_warning(f"[KubeJS-LM] 批次刷新失敗，使用 fallback 寫入: {e}") for fid, (dstp, data) in _file_write_table.items(): write_json_dict(dstp, data) diff --git a/translation_tool/plugins/md/md_lmtranslator.py b/translation_tool/plugins/md/md_lmtranslator.py index fdfadaee..53a808b1 100644 --- a/translation_tool/plugins/md/md_lmtranslator.py +++ b/translation_tool/plugins/md/md_lmtranslator.py @@ -151,7 +151,8 @@ def translate_md_pending( for jp in json_files: try: _, items = load_pending_doc(jp) - except Exception: + except Exception as e: + log_warning(f"[MD-LM] 載入待翻譯文件失敗: {jp} ({e})") continue for it in items: @@ -200,6 +201,7 @@ def translate_md_pending( shielded = shield_text(src) if shielded.skip_reason is not None: + # 不應翻譯（圖片/URL/事件/空白），直接視為 cache hit skip_skipped += 1 all_unique_items.append( { @@ -214,6 +216,8 @@ def translate_md_pending( ) continue + translate_text = shielded.clean + all_unique_items.append( { "cache_type": "md", @@ -341,16 +345,16 @@ def on_translated_item(it: Dict[str, Any]) -> None: cache_hit=False, extra={}, ) - except Exception: - pass + except Exception as e: + log_warning(f"[MD-LM] 記錄翻譯結果失敗: {e}") def on_batch_flushed() -> None: """刷新批次緩衝區。""" try: touch.touch("noop") touch.flush(_writer) - except Exception: - pass + except Exception as e: + log_warning(f"[MD-LM] 批次刷新失敗: {e}") def _fmt_eta(sec: float) -> str: """格式化剩餘時間。""" @@ -456,8 +460,8 @@ def on_progress(p: float, msg: str, eta_sec: float) -> None: try: rec.export_json(out_root / "LM翻譯後" / "translation_map_md.json") rec.export_csv(out_root / "LM翻譯後" / "translation_map_md.csv") - except Exception: - pass + except Exception as e: + log_warning(f"[MD-LM] 匯出 translation_map 失敗: {e}") if missing: log_warning(f"⚠️ [MD-LM] 有 {missing} 個 item 沒拿到翻譯結果（已保留原文）。") diff --git a/translation_tool/utils/cache_loader.py b/translation_tool/utils/cache_loader.py index 6c7ffda4..880c3ca6 100644 --- a/translation_tool/utils/cache_loader.py +++ b/translation_tool/utils/cache_loader.py @@ -13,14 +13,26 @@ import orjson as json +logger = logging.getLogger(__name__) + + def load_shard_file(path: Path) -> dict[str, Any]: - """載入並解析單一分片（Shard）的 JSON 檔案，將其轉換為記憶體中的快取物件。""" + """載入並解析單一分片（Shard）的 JSON 檔案，將其轉換為記憶體中的快取物件。 + + 若 shard 檔案為空（0 bytes），會記錄警告並回傳空 dict。 + """ try: + file_size = path.stat().st_size + if file_size == 0: + logger.warning(f"空 shard 檔案（將跳過）: {path}") + return {} data = json.loads(path.read_bytes()) return data if isinstance(data, dict) else {} - except Exception: + except Exception as e: + logger.warning(f"載入分片失敗 {path}: {e}") return {} + def load_cache_type( cache_type: str, *, diff --git a/translation_tool/utils/cache_manager.py b/translation_tool/utils/cache_manager.py index 4eff8f2b..acbedc1d 100644 --- a/translation_tool/utils/cache_manager.py +++ b/translation_tool/utils/cache_manager.py @@ -8,7 +8,7 @@ import logging import threading from pathlib import Path -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional, Tuple from . import cache_shards, cache_store from .cache_loader import load_cache_type @@ -39,6 +39,7 @@ "reload_translation_cache_type", "save_translation_cache", "add_to_cache", + "add_to_cache_batch", "get_from_cache", "get_cache_entry", "get_cache_dict_ref", @@ -53,16 +54,19 @@ "find_similar_translations", ] + def _state(): """取得或建立快取執行期狀態實例""" return cache_store.ensure_runtime_maps(CACHE_TYPES) + def _get_cache_root() -> Path: """從設定取得快取根目錄路徑""" translation_config = load_config().get("translator", {}) cache_dir_name = translation_config.get("cache_directory", _CACHE_DIR_NAME) return resolve_project_path(cache_dir_name) + def _load_cache_type(cache_type: str): """載入指定類型的快取。""" state = _state() @@ -76,28 +80,48 @@ def _load_cache_type(cache_type: str): logger=log, ) + def initialize_translation_cache(): """初始化翻譯快取系統。""" state = _state() - if state.initialized: - return - try: - for cache_type in CACHE_TYPES: - _load_cache_type(cache_type) - state.initialized = True - except Exception as e: - log.error(f"快取系統初始化失敗: {e}", exc_info=True) + with state.cache_lock: + if state.initialized: + return + try: + for cache_type in CACHE_TYPES: + _load_cache_type(cache_type) + state.initialized = True + except Exception as e: + log.error(f"快取系統初始化失敗: {e}", exc_info=True) + def is_cache_initialized() -> bool: """檢查快取是否已初始化。""" return bool(_state().initialized) + def reload_translation_cache(): """重新載入翻譯快取。""" - state = cache_store.reset_runtime_state(CACHE_TYPES) + state = cache_store.get_runtime_state() with state.cache_lock: - pass - initialize_translation_cache() + cache_store.reset_runtime_state(CACHE_TYPES) + # re-fetch state after reset (reset_runtime_state modifies the global) + state = cache_store.get_runtime_state() + # 重新載入所有快取型別 + translation_config = load_config().get("translator", {}) + for cache_type in CACHE_TYPES: + load_cache_type( + cache_type, + translation_cache=state.translation_cache, + cache_file_path=state.cache_file_path, + cache_root=_get_cache_root(), + parallel_workers=translation_config.get( + "parallel_execution_workers", 4 + ), + logger=log, + ) + state.initialized = True + def reload_translation_cache_type(cache_type: str): """重新載入指定類型的翻譯快取。""" @@ -111,6 +135,7 @@ def reload_translation_cache_type(cache_type: str): cache_store.clear_dirty(state.is_dirty, cache_type) _load_cache_type(cache_type) + def _save_entries_to_active_shards( cache_type: str, entries: dict, force_new_shard: bool = False ): @@ -127,6 +152,7 @@ def _save_entries_to_active_shards( logger=log, ) + def save_translation_cache(cache_type: str, write_new_shard: bool = True): """儲存翻譯快取。""" if not load_config().get("translator", {}).get("enable_cache_saving", True): @@ -142,18 +168,19 @@ def save_translation_cache(cache_type: str, write_new_shard: bool = True): data_to_save = cache_store.flush_session_entries( state.session_new_entries, cache_type ) - cache_store.clear_dirty(state.is_dirty, cache_type) - try: save_path = state.cache_file_path.get(cache_type) if not save_path: + cache_store.clear_dirty(state.is_dirty, cache_type) return _save_entries_to_active_shards( cache_type, data_to_save, force_new_shard=write_new_shard ) + cache_store.clear_dirty(state.is_dirty, cache_type) except Exception as e: log.error(f"❌ 儲存 {cache_type} 失敗: {e}", exc_info=True) + def _get_active_shard_path(cache_type: str) -> Path: """取得目前作用中的分片檔案路徑""" state = _state() @@ -164,6 +191,7 @@ def _get_active_shard_path(cache_type: str) -> Path: active_shard_file=ACTIVE_SHARD_FILE, ) + def add_to_cache( cache_type: str, key: str, @@ -193,6 +221,50 @@ def add_to_cache( session_entries[key] = entry cache_store.mark_dirty(state.is_dirty, cache_type) + +def add_to_cache_batch( + cache_type: str, + entries: List[Tuple[str, str, str]], + *, + mods: Optional[List[Optional[str]]] = None, + paths: Optional[List[Optional[str]]] = None, +): + """批次新增翻譯到快取（單次鎖獲取，減少鎖競爭）。 + + Args: + cache_type: 快取類型 (lang, patchouli, ftbquests, kubejs, md) + entries: List of (key, src, dst) tuples + mods: Optional list of mod names (same length as entries) + paths: Optional list of paths (same length as entries) + """ + if not entries: + return + + state = _state() + with state.cache_lock: + cache = cache_store.get_cache_type_dict(state.translation_cache, cache_type) + session_entries = cache_store.get_session_entries( + state.session_new_entries, cache_type + ) + dirty = False + + for i, (key, src, dst) in enumerate(entries): + if not key or not dst: + continue + entry = {"src": src, "dst": dst} + if mods and i < len(mods) and mods[i]: + entry["mod"] = mods[i] + if paths and i < len(paths) and paths[i]: + entry["path"] = paths[i] + changed = cache_store.add_entry(cache, key, entry) + if changed: + session_entries[key] = entry + dirty = True + + if dirty: + cache_store.mark_dirty(state.is_dirty, cache_type) + + def get_from_cache(cache_type: str, key: str) -> Optional[str]: """從快取取得指定 key 的翻譯文字 (dst)。""" state = _state() @@ -203,6 +275,7 @@ def get_from_cache(cache_type: str, key: str) -> Optional[str]: return None return cache_store.get_value(cache, key) + def get_cache_entry(cache_type: str, key: str) -> Optional[Dict[str, Any]]: """取得指定 key 的完整快取項目（包含 src、dst、mod、path）。""" state = _state() @@ -213,6 +286,7 @@ def get_cache_entry(cache_type: str, key: str) -> Optional[Dict[str, Any]]: return None return cache_store.get_entry(cache, key) + def get_cache_dict_ref(cache_type: str) -> Dict[str, Dict[str, Any]]: """取得指定類型的快取字典參照。""" state = _state() @@ -221,6 +295,7 @@ def get_cache_dict_ref(cache_type: str) -> Dict[str, Dict[str, Any]]: cache = state.translation_cache.get(cache_type) return cache if isinstance(cache, dict) else {} + def get_session_new_count(cache_type: str) -> int: """取得本次 session 新增的項目數""" state = _state() @@ -229,6 +304,7 @@ def get_session_new_count(cache_type: str) -> int: cache_store.get_session_entries(state.session_new_entries, cache_type) ) + def get_active_shard_id(cache_type: str) -> str: """取得指定快取類型的目前作用中分片 ID""" state = _state() @@ -236,6 +312,7 @@ def get_active_shard_id(cache_type: str) -> str: state.cache_file_path, cache_type, ACTIVE_SHARD_FILE ) + def get_cache_overview() -> Dict[str, Any]: """取得所有快取類型的概覽（包含項目數與狀態）""" initialize_translation_cache() @@ -255,6 +332,7 @@ def get_cache_overview() -> Dict[str, Any]: resolve_project_path=resolve_project_path, ) + def force_rotate_shard(cache_type: str) -> bool: """強制輪轉至下一個分片。""" initialize_translation_cache() @@ -273,6 +351,7 @@ def force_rotate_shard(cache_type: str) -> bool: except Exception: return False + def _get_search_facade() -> CacheSearchFacade: """取得或建立快取搜尋外觀（惰性初始化）""" global _search_facade @@ -282,10 +361,12 @@ def _get_search_facade() -> CacheSearchFacade: _search_facade = CacheSearchFacade(_get_cache_root, log) return _search_facade + def get_search_engine(): """取得快取查詢用的搜尋引擎實例""" return _get_search_facade().get_search_engine() + def rebuild_search_index(): """重建所有快取類型的搜尋索引。""" state = _state() @@ -293,6 +374,7 @@ def rebuild_search_index(): CACHE_TYPES, state.translation_cache ) + def rebuild_search_index_for_type(cache_type: str): """重建指定快取類型的搜尋索引""" state = _state() @@ -300,6 +382,7 @@ def rebuild_search_index_for_type(cache_type: str): cache_type, CACHE_TYPES, state.translation_cache ) + def search_cache( query: str, cache_type: str = None, limit: int = 50, use_fuzzy: bool = True ) -> list: @@ -308,6 +391,7 @@ def search_cache( query=query, cache_type=cache_type, limit=limit, use_fuzzy=use_fuzzy ) + def find_similar_translations( text: str, cache_type: str = None, threshold: float = 0.6, limit: int = 20 ) -> list: @@ -316,6 +400,7 @@ def find_similar_translations( text=text, cache_type=cache_type, threshold=threshold, limit=limit ) + initialize_translation_cache() _state_obj = _state() log.info( diff --git a/translation_tool/utils/cache_search.py b/translation_tool/utils/cache_search.py index 650eb04d..8b0e8866 100644 --- a/translation_tool/utils/cache_search.py +++ b/translation_tool/utils/cache_search.py @@ -32,6 +32,7 @@ # 全文搜尋引擎 # ============================================================================= + class CacheSearchEngine: """快取全文搜尋引擎（使用 SQLite FTS5）""" @@ -53,14 +54,16 @@ def __init__(self, db_path: str | None = None): self.conn = sqlite3.connect(db_path, check_same_thread=False) self.conn.row_factory = sqlite3.Row # 讓結果可以用欄位名稱存取 self._lock = threading.RLock() - + # SQLite 效能優化（大幅提升大量寫入速度） with self._lock: self.conn.execute("PRAGMA journal_mode = WAL") - self.conn.execute("PRAGMA synchronous = NORMAL") # ATK-011: OFF 會導致 daemon crash 後 WAL recovery corruption + self.conn.execute( + "PRAGMA synchronous = NORMAL" + ) # ATK-011: OFF 會導致 daemon crash 後 WAL recovery corruption self.conn.execute("PRAGMA cache_size = 10000") self.conn.execute("PRAGMA temp_store = MEMORY") - + self._init_fts_table() def _init_fts_table(self): @@ -188,9 +191,9 @@ def index_batch(self, entries: List[dict], batch_size: int = 20000): """ if not entries: return - + t0 = time.time() - + data = [ ( e.get("key", ""), @@ -202,19 +205,19 @@ def index_batch(self, entries: List[dict], batch_size: int = 20000): ) for e in entries ] - + prepare_time = time.time() - t0 with self._lock: try: try: self.conn.execute("PRAGMA recursive_triggers = OFF") - except Exception: - pass - + except Exception as e: + log_debug(f"設定 PRAGMA recursive_triggers 失敗: {e}") + write_start = time.time() for i in range(0, len(data), batch_size): - batch = data[i:i + batch_size] + batch = data[i : i + batch_size] self.conn.executemany( """ INSERT INTO cache_fts (cache_key, source_text, translated_text, mod_name, file_path, cache_type) @@ -225,7 +228,7 @@ def index_batch(self, entries: List[dict], batch_size: int = 20000): except sqlite3.OperationalError: write_start = time.time() for i in range(0, len(data), batch_size): - batch = data[i:i + batch_size] + batch = data[i : i + batch_size] self.conn.executemany( """ INSERT INTO cache_basic (cache_key, source_text, translated_text, mod_name, file_path, cache_type) @@ -236,11 +239,15 @@ def index_batch(self, entries: List[dict], batch_size: int = 20000): write_time = time.time() - write_start self.conn.commit() - + total_time = time.time() - t0 - log_debug(f"index_batch: {len(entries)} entries, prepare={prepare_time:.2f}s, write={write_time:.2f}s, total={total_time:.2f}s") + log_debug( + f"index_batch: {len(entries)} entries, prepare={prepare_time:.2f}s, write={write_time:.2f}s, total={total_time:.2f}s" + ) - def search(self, query: str, limit: int = 50, cache_type: str | None = None) -> List[Dict]: + def search( + self, query: str, limit: int = 50, cache_type: str | None = None + ) -> List[Dict]: """搜尋快取（支援中英文、模糊比對） Args: @@ -368,10 +375,12 @@ def __exit__(self, exc_type, exc_val, exc_tb): """離開上下文時關閉搜尋引擎。""" self.close() + # ============================================================================= # 模糊比對器 # ============================================================================= + class FuzzyMatcher: """模糊比對器（相似度計算）""" @@ -454,10 +463,12 @@ def rank_results( # 按綜合分數排序 return sorted(scored, key=lambda x: x["combined_score"], reverse=True) + # ============================================================================= # 便利函式 # ============================================================================= + def search_cache( query: str, db_path: str = None, @@ -475,10 +486,12 @@ def search_cache( return results + # ============================================================================= # 搜尋協調輔助函式（PR12） # ============================================================================= + def _extract_path_from_composite_key(key: str, src: str = "") -> str: """從複合 key 拆出路徑段。 @@ -493,6 +506,7 @@ def _extract_path_from_composite_key(key: str, src: str = "") -> str: return key.split("|", 1)[0] return key + def _infer_search_path(cache_type: str, key: str, entry: Dict[str, Any] | None) -> str: """推導索引要寫入的 path 欄位。 @@ -513,6 +527,7 @@ def _infer_search_path(cache_type: str, key: str, entry: Dict[str, Any] | None) return _extract_path_from_composite_key(key, src) + def _infer_search_mod( cache_type: str, key: str, path: str, entry: Dict[str, Any] | None ) -> str: @@ -546,6 +561,7 @@ def _infer_search_mod( } return fallback.get(cache_type, "") + def _build_search_metadata( cache_type: str, key: str, entry: Dict[str, Any] | None ) -> Dict[str, str]: @@ -554,16 +570,19 @@ def _build_search_metadata( mod = _infer_search_mod(cache_type, key, path, entry) return {"mod": mod, "path": path} + def build_index_entries( cache_type: str, cache_dict: Dict[str, Any] ) -> List[Dict[str, Any]]: """把單一 cache_type 的記憶體字典轉成可批次索引的條目陣列（並行版本）。""" t0 = time.time() - items = [(key, entry) for key, entry in cache_dict.items() if isinstance(entry, dict)] - + items = [ + (key, entry) for key, entry in cache_dict.items() if isinstance(entry, dict) + ] + if not items: return [] - + # 使用多執行緒並行處理 metadata 建立 with ThreadPoolExecutor(max_workers=4) as executor: futures = { @@ -575,15 +594,20 @@ def build_index_entries( idx = futures[future] try: results[idx] = future.result() - except Exception: - pass - + except Exception as e: + log_debug(f"建構索引條目失敗: {e}") + elapsed = time.time() - t0 - log_debug(f"build_index_entries({cache_type}): {len(items)} entries in {elapsed:.2f}s") - + log_debug( + f"build_index_entries({cache_type}): {len(items)} entries in {elapsed:.2f}s" + ) + return [r for r in results if r is not None] -def _build_single_entry(cache_type: str, key: str, entry: Dict[str, Any]) -> Optional[Dict[str, Any]]: + +def _build_single_entry( + cache_type: str, key: str, entry: Dict[str, Any] +) -> Optional[Dict[str, Any]]: """建立單筆索引條目（供並行呼叫）。""" return { "key": key, @@ -593,6 +617,7 @@ def _build_single_entry(cache_type: str, key: str, entry: Dict[str, Any]) -> Opt **_build_search_metadata(cache_type, key, entry), } + def rebuild_from_cache_dicts( engine: CacheSearchEngine, cache_types: List[str], @@ -600,12 +625,16 @@ def rebuild_from_cache_dicts( ) -> int: """依序重建多個類型的索引，回傳實際索引筆數（並行版本）。""" total_indexed = 0 - + # 先並行處理所有 cache_type 的 entries 建立 all_entries: Dict[str, List[Dict[str, Any]]] = {} with ThreadPoolExecutor(max_workers=len(cache_types)) as executor: futures = { - executor.submit(build_index_entries, cache_type, cache_store.get_cache_type_dict(cache_state, cache_type)): cache_type + executor.submit( + build_index_entries, + cache_type, + cache_store.get_cache_type_dict(cache_state, cache_type), + ): cache_type for cache_type in cache_types } for future in as_completed(futures): @@ -613,16 +642,17 @@ def rebuild_from_cache_dicts( entries = future.result() if entries: all_entries[cache_type] = entries - + # 再依序寫入 SQLite（保持原有寫入邏輯） for cache_type in cache_types: entries = all_entries.get(cache_type, []) if entries: engine.index_batch(entries) total_indexed += len(entries) - + return total_indexed + class SearchOrchestrator: """快取搜尋協調器。 @@ -665,6 +695,7 @@ def rebuild_search_index( except PermissionError: if attempt < max_retries - 1: import time + time.sleep(retry_delay) continue else: @@ -672,17 +703,13 @@ def rebuild_search_index( raise def _do_rebuild_search_index( - self, - db_path, - cache_types: List[str], - cache_state: Dict[str, Dict[str, Any]] + self, db_path, cache_types: List[str], cache_state: Dict[str, Dict[str, Any]] ) -> int: """執行實際的索引重建作業。""" tmp_engine: Optional[CacheSearchEngine] = None old_engine: Optional[CacheSearchEngine] = None total_indexed = 0 - - + try: # 直接寫入目標資料庫（不使用 tmp 檔案，避免 Windows 檔案鎖問題） # 先關閉舊引擎 @@ -691,37 +718,38 @@ def _do_rebuild_search_index( if old_engine is not None: old_engine.close() self._engine = None - + del old_engine import gc + gc.collect() - + # 清理 WAL/SHM 檔案 for suffix in ["-wal", "-shm"]: wal_file = db_path.with_name(db_path.name + suffix) if wal_file.exists(): try: wal_file.unlink() - except Exception: - pass - + except Exception as e: + log_debug(f"刪除 WAL/SHM 檔案失敗: {e}") + # 刪除舊資料庫重新建立 if db_path.exists(): try: db_path.unlink() - except Exception: - pass - + except Exception as e: + log_debug(f"刪除舊資料庫檔案失敗: {e}") + # 建立新引擎並直接寫入 tmp_engine = CacheSearchEngine(str(db_path)) total_indexed = rebuild_from_cache_dicts( tmp_engine, cache_types, cache_state ) - + # 重新建立引擎 with self._lock: self._engine = CacheSearchEngine(str(db_path)) - + log_debug(f"索引重建完成: {total_indexed} 條") return total_indexed finally: diff --git a/translation_tool/utils/cache_shards.py b/translation_tool/utils/cache_shards.py index 89a11771..55676bdd 100644 --- a/translation_tool/utils/cache_shards.py +++ b/translation_tool/utils/cache_shards.py @@ -12,18 +12,54 @@ import orjson as json + +def _lock_file_fd(lock_fd: int) -> None: + """以跨平台方式鎖定 lock file descriptor。""" + if os.name == "nt": + import msvcrt + + msvcrt.locking(lock_fd, msvcrt.LK_LOCK, 1) + else: + import fcntl + + fcntl.flock(lock_fd, fcntl.LOCK_EX) + + +def _unlock_file_fd(lock_fd: int) -> None: + """以跨平台方式解鎖 lock file descriptor。""" + if os.name == "nt": + import msvcrt + + msvcrt.locking(lock_fd, msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(lock_fd, fcntl.LOCK_UN) + + def _write_json_atomic(path: Path, data: dict[str, Any]): """以原子方式將 JSON 內容覆寫到 ``path``。 目前此函式沒有具語意的回傳值； 呼叫端若選擇直接透傳回傳結果，可在未來新增成功/失敗回傳契約時 免於同步調整外層包裝介面。 + + 使用 fsync 確保資料寫入磁碟，避免作業系統緩衝區未 flush + 就執行 os.replace() 導致資料遺失。 """ tmp_path = path.with_suffix(".tmp") path.parent.mkdir(parents=True, exist_ok=True) + + # 寫入暫存檔 tmp_path.write_bytes(json.dumps(data, option=json.OPT_INDENT_2)) + + # 確保資料寫入磁碟（Windows 使用 FlushFileBuffers） + with open(tmp_path, "r+b") as f: + os.fsync(f.fileno()) + os.replace(tmp_path, path) + def _get_active_shard_path( *, type_dir: Path, @@ -51,6 +87,7 @@ def _get_active_shard_path( return type_dir / f"{cache_type}_{shard_id_str}.json" + def _rotate_shard_if_needed( *, type_dir: Path, @@ -60,26 +97,46 @@ def _rotate_shard_if_needed( active_shard_file: str, logger: logging.Logger | None = None, ) -> bool: - """當目前分片容量達上限時切到下一片，並回傳是否有旋轉。""" + """當目前分片容量達上限時切到下一片，並回傳是否有旋轉。 + + 使用檔案鎖確保旋轉操作的原子性，防止 TOCTOU Race Condition。 + """ if len(data) < rolling_shard_size: return False active_file = type_dir / active_shard_file - if not active_file.exists(): - _get_active_shard_path( - type_dir=type_dir, - cache_type=cache_type, - active_shard_file=active_shard_file, - ) + lock_file = type_dir / f"{active_shard_file}.lock" - cur_id = int((active_file.read_text(encoding="utf-8") or "1").strip()) - new_id = f"{cur_id + 1:05d}" - active_file.write_text(new_id, encoding="utf-8") + # 建立 lock 檔並取得獨占鎖，防止 TOCTOU race + type_dir.mkdir(parents=True, exist_ok=True) + lock_fd = os.open(str(lock_file), os.O_CREAT | os.O_RDWR) + try: + # 以跨平台 file lock 進行檔案鎖定 + _lock_file_fd(lock_fd) - if logger: - logger.info(f"🔁 {cache_type} rolling shard rotate → {new_id}") + # 再次確認容量（防止鎖競爭期間已被其他程序旋轉） + if len(data) < rolling_shard_size: + return False + + if not active_file.exists(): + _get_active_shard_path( + type_dir=type_dir, + cache_type=cache_type, + active_shard_file=active_shard_file, + ) + + cur_id = int((active_file.read_text(encoding="utf-8") or "1").strip()) + new_id = f"{cur_id + 1:05d}" + active_file.write_text(new_id, encoding="utf-8") + + if logger: + logger.info(f"🔁 {cache_type} rolling shard rotate → {new_id}") + + return True + finally: + _unlock_file_fd(lock_fd) + os.close(lock_fd) - return True def _save_entries_to_active_shards( *, @@ -91,13 +148,17 @@ def _save_entries_to_active_shards( force_new_shard: bool = False, logger: logging.Logger | None = None, ): - """把多筆條目分段寫入 active shard，必要時自動切片。""" + """把多筆條目分段寫入 active shard，必要時自動切片。 + + 使用檔案鎖確保讀取-修改-寫入循環的原子性，防止 TOCTOU race。 + """ if not entries: return active_file = type_dir / active_shard_file + lock_file = type_dir / f"{active_shard_file}.lock" + # 先確保 `.active` 指標檔存在，避免下方分支直接讀取時找不到檔案。 - # 這裡只需要副作用，不使用回傳路徑。 _get_active_shard_path( type_dir=type_dir, cache_type=cache_type, @@ -113,53 +174,79 @@ def _save_entries_to_active_shards( pending_items = list(entries.items()) while pending_items: - save_path = _get_active_shard_path( - type_dir=type_dir, - cache_type=cache_type, - active_shard_file=active_shard_file, - ) - - current_data: dict[str, Any] = {} - if save_path.exists(): + # 建立 lock 檔並取得獨占鎖，防止 TOCTOU race + type_dir.mkdir(parents=True, exist_ok=True) + lock_fd = os.open(str(lock_file), os.O_CREAT | os.O_RDWR) + rotated = False + try: + _lock_file_fd(lock_fd) + + # 在鎖保護下讀取 active shard path（避免 TOCTOU） + save_path = _get_active_shard_path( + type_dir=type_dir, + cache_type=cache_type, + active_shard_file=active_shard_file, + ) + + current_data: dict[str, Any] = {} try: old_data = json.loads(save_path.read_bytes()) if isinstance(old_data, dict): current_data = old_data + except FileNotFoundError: + current_data = {} except Exception as e: if logger: logger.warning(f"⚠️ 讀取舊分片失敗，將以空白分片續寫: {e}") - if _rotate_shard_if_needed( - type_dir=type_dir, - cache_type=cache_type, - data=current_data, - rolling_shard_size=rolling_shard_size, - active_shard_file=active_shard_file, - logger=logger, - ): - continue - - capacity = max(0, rolling_shard_size - len(current_data)) - chunk = pending_items[:capacity] - - for k, v in chunk: - current_data[k] = v - - _write_json_atomic(save_path, current_data) - if logger: - logger.info( - f"💾 {cache_type} saved: {save_path.name} (+{len(chunk)} / total={len(current_data)})" - ) - - pending_items = pending_items[capacity:] - if pending_items: - # 若目前分片已滿，先預轉到下一片，讓下次迴圈可直接續寫。 - # 此處只依賴副作用，刻意忽略布林回傳值。 - _ = _rotate_shard_if_needed( - type_dir=type_dir, - cache_type=cache_type, - data=current_data, - rolling_shard_size=rolling_shard_size, - active_shard_file=active_shard_file, - logger=logger, - ) + # 在鎖保護下檢查是否需要旋轉 + if len(current_data) >= rolling_shard_size: + # 需要旋轉：釋放當前鎖，讓旋轉邏輯取得鎖 + _unlock_file_fd(lock_fd) + os.close(lock_fd) + lock_fd = -1 + + _rotate_shard_if_needed( + type_dir=type_dir, + cache_type=cache_type, + data=current_data, + rolling_shard_size=rolling_shard_size, + active_shard_file=active_shard_file, + logger=logger, + ) + rotated = True + continue # 重新取得路徑和資料 + + finally: + if lock_fd != -1: + try: + _unlock_file_fd(lock_fd) + except Exception: + pass + os.close(lock_fd) + + if not rotated: + capacity = max(0, rolling_shard_size - len(current_data)) + chunk = pending_items[:capacity] + + for k, v in chunk: + current_data[k] = v + + _write_json_atomic(save_path, current_data) + if logger: + logger.info( + f"💾 {cache_type} saved: {save_path.name} (+{len(chunk)} / total={len(current_data)})" + ) + + pending_items = pending_items[capacity:] + + if pending_items: + # 若目前分片已滿，預旋轉到下一片 + _rotate_shard_if_needed( + type_dir=type_dir, + cache_type=cache_type, + data=current_data, + rolling_shard_size=rolling_shard_size, + active_shard_file=active_shard_file, + logger=logger, + ) diff --git a/workspace/patch_md_lmtranslator.py b/workspace/patch_md_lmtranslator.py new file mode 100644 index 00000000..ca1feda5 --- /dev/null +++ b/workspace/patch_md_lmtranslator.py @@ -0,0 +1,107 @@ +from pathlib import Path +import re + +p = Path(r"translation_tool/plugins/md/md_lmtranslator.py") +text = p.read_text(encoding="utf-8") + +pattern = re.compile( + r"for h, src in hash_to_src\.items\(\):\n" + r"\s+if is_already_zh\(src\):\n" + r"\s+already_zh_skipped \+= 1\n" + r"\s+continue\n" + r"\s+all_unique_items\.append\(\n" + r"\s+\{\n" + r'\s+"cache_type": "md",\n' + r'\s+"file": "md_pending_blocks",\n' + r'\s+"path": h, # .*?\n' + r'\s+"source_text": src,\n' + r'\s+"text": src,\n' + r"\s+\}\n" + r"\s+\)", + re.S, +) +replacement = """for h, src in hash_to_src.items(): + if is_already_zh(src): + already_zh_skipped += 1 + continue + + shielded = shield_text(src) + translate_text = shielded.clean + if shielded.skip_reason is not None: + translate_text = src + + all_unique_items.append( + { + "cache_type": "md", + "file": "md_pending_blocks", + "path": h, # ✅ 用 content_hash 當 path（去重 + 快取 key 的一部分） + "source_text": src, + "text": translate_text, + "_shielded": shielded, + } + )""" +text, n = pattern.subn(replacement, text, count=1) +if n != 1: + raise SystemExit(f"pattern replace 1 failed: {n}") + +text = text.replace( + """ hash_to_dst: Dict[str, str] = {} + for it in cached_items: + h = str(it.get("path") or "") + dst = str(it.get("text") or "") + if h and dst: + hash_to_dst[h] = dst +""", + """ hash_to_dst: Dict[str, str] = {} + for it in cached_items: + h = str(it.get("path") or "") + dst = str(it.get("text") or "") + if h and dst: + shielded = it.get("_shielded") + if shielded is not None and getattr(shielded, "shields", None): + try: + dst = unshield_text(dst, shielded.shields) + except Exception: + pass + hash_to_dst[h] = dst +""", +) + +text = text.replace( + ''' def on_translated_item(it: Dict[str, Any]) -> None: + """處理翻譯結果。""" + h = str(it.get("path") or "") + dst = str(it.get("text") or "") + src_text = str(it.get("source_text") or "") + if h and dst: + try: + shielded_src = shield_text(src_text) + dst = unshield_text(dst, shielded_src.shields) + except Exception: + pass + hash_to_dst[h] = dst +''', + ''' def on_translated_item(it: Dict[str, Any]) -> None: + """處理翻譯結果。""" + h = str(it.get("path") or "") + dst = str(it.get("text") or "") + src_text = str(it.get("source_text") or "") + if h and dst: + shielded = it.get("_shielded") + if shielded is not None and getattr(shielded, "shields", None): + try: + dst = unshield_text(dst, shielded.shields) + except Exception: + pass + else: + try: + shielded_src = shield_text(src_text) + dst = unshield_text(dst, shielded_src.shields) + except Exception: + pass + hash_to_dst[h] = dst +''', +) + +p.write_text(text, encoding="utf-8") +print("patched md_lmtranslator")