翻訳モジュールのドキュメントを更新し、セットアップ手順やAPI使用例を追加。型注釈を強化し、関数の戻り値を明示化。エラーハンドリングを改善し、コードの可読性を向上。

src-python/models/translation/translation_languages.py

@@ -1,4 +1,13 @@
-translation_lang = {}
+"""Language code mappings for supported translation backends.
+
+Provides `translation_lang` mapping keyed by backend name with `source` and
+`target` maps used by `Translator.getLanguageCode`.
+"""
+
+from typing import Dict
+
+translation_lang: Dict[str, Dict[str, Dict[str, str]]] = {}
+
 dict_deepl_languages = {
     "Arabic":"ar",
     "Bulgarian":"bg",
@@ -37,10 +46,7 @@ dict_deepl_languages = {
 	"Chinese Simplified":"zh",
     "Chinese Traditional":"zh"
 }
-translation_lang["DeepL"] = {
-    "source":dict_deepl_languages,
-    "target":dict_deepl_languages,
-}
+translation_lang["DeepL"] = {"source": dict_deepl_languages, "target": dict_deepl_languages}
 
 dict_deepl_api_source_languages = {
     "Japanese":"ja",
@@ -109,10 +115,7 @@ dict_deepl_api_target_languages = {
     "Chinese Simplified":"zh",
     "Chinese Traditional":"zh"
 }
-translation_lang["DeepL_API"] = {
-    "source": dict_deepl_api_source_languages,
-    "target": dict_deepl_api_target_languages,
-}
+translation_lang["DeepL_API"] = {"source": dict_deepl_api_source_languages, "target": dict_deepl_api_target_languages}
 
 dict_google_languages = {
     "Japanese":"ja",
@@ -179,10 +182,7 @@ dict_google_languages = {
     # "Basque":"eu",
     "Irish":"ga"
 }
-translation_lang["Google"] = {
-    "source":dict_google_languages,
-    "target":dict_google_languages,
-}
+translation_lang["Google"] = {"source": dict_google_languages, "target": dict_google_languages}
 
 dict_bing_languages = {
     "Japanese":"ja",
@@ -247,10 +247,7 @@ dict_bing_languages = {
     "Punjabi":"pa",
     "Irish":"ga"
 }
-translation_lang["Bing"] = {
-    "source":dict_bing_languages,
-    "target":dict_bing_languages,
-}
+translation_lang["Bing"] = {"source": dict_bing_languages, "target": dict_bing_languages}
 
 dict_papago_languages = {
     "German": "de",
@@ -270,10 +267,7 @@ dict_papago_languages = {
     "Chinese Traditional":"zh-TW",
 }
 
-translation_lang["Papago"] = {
-    "source":dict_papago_languages,
-    "target":dict_papago_languages,
-}
+translation_lang["Papago"] = {"source": dict_papago_languages, "target": dict_papago_languages}
 
 dict_ctranslate2_languages = {
     "English": "en",
@@ -378,7 +372,4 @@ dict_ctranslate2_languages = {
     "Sundanese": "su"
 }
 
-translation_lang["CTranslate2"] = {
-    "source":dict_ctranslate2_languages,
-    "target":dict_ctranslate2_languages,
-}
+translation_lang["CTranslate2"] = {"source": dict_ctranslate2_languages, "target": dict_ctranslate2_languages}

src-python/models/translation/translation_translator.py

@@ -4,6 +4,7 @@ try:
     from translators import translate_text as other_web_Translator
     ENABLE_TRANSLATORS = True
 except Exception:
+    other_web_Translator = None  # type: ignore
     ENABLE_TRANSLATORS = False
 
 from .translation_languages import translation_lang
@@ -14,22 +15,37 @@ import transformers
 from utils import errorLogging, getBestComputeType
 
 import warnings
+from typing import Any, Optional, Tuple
+
 warnings.filterwarnings("ignore")
 
-# Translator
-class Translator():
-    def __init__(self):
-        self.deepl_client = None
-        self.ctranslate2_translator = None
-        self.ctranslate2_tokenizer = None
-        self.is_loaded_ctranslate2_model = False
-        self.is_changed_translator_parameters = False
-        self.is_enable_translators = ENABLE_TRANSLATORS
 
-    def authenticationDeepLAuthKey(self, authkey):
+class Translator:
+    """High-level translator facade.
+
+    This class wraps multiple backends (DeepL, DeepL API, Google, Bing, Papago,
+    and CTranslate2 local models). Optional dependencies may be unavailable at
+    runtime; methods degrade gracefully and return False or an empty string on
+    failure (kept compatible with existing behavior).
+    """
+
+    def __init__(self) -> None:
+        self.deepl_client: Optional[DeepLClient] = None
+        self.ctranslate2_translator: Any = None
+        self.ctranslate2_tokenizer: Any = None
+        self.is_loaded_ctranslate2_model: bool = False
+        self.is_changed_translator_parameters: bool = False
+        self.is_enable_translators: bool = ENABLE_TRANSLATORS
+
+    def authenticationDeepLAuthKey(self, authkey: str) -> bool:
+        """Authenticate DeepL API with the provided key.
+
+        Returns True on success, False on failure.
+        """
         result = True
         try:
             self.deepl_client = DeepLClient(authkey)
+            # quick smoke test
             self.deepl_client.translate_text(" ", target_lang="EN-US")
         except Exception:
             errorLogging()
@@ -37,7 +53,12 @@ class Translator():
             result = False
         return result
 
-    def changeCTranslate2Model(self, path, model_type, device="cpu", device_index=0, compute_type="auto"):
+    def changeCTranslate2Model(self, path: str, model_type: str, device: str = "cpu", device_index: int = 0, compute_type: str = "auto") -> None:
+        """Load a CTranslate2 model from weights.
+
+        This sets internal translator/tokenizer objects and flips
+        ``is_loaded_ctranslate2_model`` on success.
+        """
         self.is_loaded_ctranslate2_model = False
         directory_name = ctranslate2_weights[model_type]["directory_name"]
         tokenizer = ctranslate2_weights[model_type]["tokenizer"]
@@ -52,7 +73,7 @@ class Translator():
             device_index=device_index,
             compute_type=compute_type,
             inter_threads=1,
-            intra_threads=4
+            intra_threads=4,
         )
         try:
             self.ctranslate2_tokenizer = transformers.AutoTokenizer.from_pretrained(tokenizer, cache_dir=tokenizer_path)
@@ -62,17 +83,21 @@ class Translator():
             self.ctranslate2_tokenizer = transformers.AutoTokenizer.from_pretrained(tokenizer, cache_dir=tokenizer_path)
         self.is_loaded_ctranslate2_model = True
 
-    def isLoadedCTranslate2Model(self):
+    def isLoadedCTranslate2Model(self) -> bool:
         return self.is_loaded_ctranslate2_model
 
-    def isChangedTranslatorParameters(self):
+    def isChangedTranslatorParameters(self) -> bool:
         return self.is_changed_translator_parameters
 
-    def setChangedTranslatorParameters(self, is_changed):
+    def setChangedTranslatorParameters(self, is_changed: bool) -> None:
         self.is_changed_translator_parameters = is_changed
 
-    def translateCTranslate2(self, message, source_language, target_language):
-        result = False
+    def translateCTranslate2(self, message: str, source_language: str, target_language: str) -> Any:
+        """Translate using a loaded CTranslate2 model.
+
+        Returns a string on success or False on failure (keeps legacy behavior).
+        """
+        result: Any = False
         if self.is_loaded_ctranslate2_model is True:
             try:
                 self.ctranslate2_tokenizer.src_lang = source_language
@@ -86,7 +111,11 @@ class Translator():
         return result
 
     @staticmethod
-    def getLanguageCode(translator_name, target_country, source_language, target_language):
+    def getLanguageCode(translator_name: str, target_country: str, source_language: str, target_language: str) -> Tuple[str, str]:
+        """Resolve a friendly language name to translator-specific codes.
+
+        Returns (source_code, target_code).
+        """
         match translator_name:
             case "DeepL_API":
                 if target_language == "English":
@@ -101,66 +130,63 @@ class Translator():
                         target_language = "Portuguese Brazilian"
             case _:
                 pass
-        source_language=translation_lang[translator_name]["source"][source_language]
-        target_language=translation_lang[translator_name]["target"][target_language]
+        source_language = translation_lang[translator_name]["source"][source_language]
+        target_language = translation_lang[translator_name]["target"][target_language]
         return source_language, target_language
 
-    def translate(self, translator_name, source_language, target_language, target_country, message):
+    def translate(self, translator_name: str, source_language: str, target_language: str, target_country: str, message: str) -> Any:
+        """Translate `message` using the named translator backend.
+
+        Returns translated string on success, or False on failure. When
+        source_language == target_language the original message is returned.
+        """
         try:
             if source_language == target_language:
                 return message
 
-            result = ""
+            result: Any = ""
             source_language, target_language = self.getLanguageCode(translator_name, target_country, source_language, target_language)
             match translator_name:
                 case "DeepL":
-                    if self.is_enable_translators is True:
+                    if self.is_enable_translators is True and other_web_Translator is not None:
                         result = other_web_Translator(
                             query_text=message,
                             translator="deepl",
                             from_language=source_language,
                             to_language=target_language,
-                            )
+                        )
                 case "DeepL_API":
                     if self.is_enable_translators is True:
                         if self.deepl_client is None:
                             result = False
                         else:
-                            result = self.deepl_client.translate_text(
-                                message,
-                                source_lang=source_language,
-                                target_lang=target_language,
-                                ).text
+                            result = self.deepl_client.translate_text(message, source_lang=source_language, target_lang=target_language).text
                 case "Google":
-                    if self.is_enable_translators is True:
+                    if self.is_enable_translators is True and other_web_Translator is not None:
                         result = other_web_Translator(
                             query_text=message,
                             translator="google",
                             from_language=source_language,
                             to_language=target_language,
-                            )
+                        )
                 case "Bing":
-                    if self.is_enable_translators is True:
+                    if self.is_enable_translators is True and other_web_Translator is not None:
                         result = other_web_Translator(
                             query_text=message,
                             translator="bing",
                             from_language=source_language,
                             to_language=target_language,
-                            )
+                        )
                 case "Papago":
-                    if self.is_enable_translators is True:
+                    if self.is_enable_translators is True and other_web_Translator is not None:
                         result = other_web_Translator(
                             query_text=message,
                             translator="papago",
                             from_language=source_language,
                             to_language=target_language,
-                            )
-                case "CTranslate2":
-                    result = self.translateCTranslate2(
-                        message=message,
-                        source_language=source_language,
-                        target_language=target_language,
                         )
+                case "CTranslate2":
+                    result = self.translateCTranslate2(message=message, source_language=source_language, target_language=target_language)
         except Exception:
             errorLogging()
             result = False

src-python/models/translation/translation_utils.py

@@ -3,13 +3,22 @@ from zipfile import ZipFile
 from os import path as os_path
 from os import makedirs as os_makedirs
 from requests import get as requests_get
-from typing import Callable
+from typing import Callable, Optional
 import hashlib
 import transformers
 from utils import errorLogging
 
+
+"""Utilities for downloading and verifying CTranslate2 weights and tokenizers.
+
+This module provides a small, dependency-light set of helpers used by the
+translation layer. It purposely keeps behavior resilient: network errors are
+logged (via utils.errorLogging) and the functions return/complete without
+raising, which matches the repository's defensive style.
+"""
+
 ctranslate2_weights = {
-    "small": { # M2M-100 418M-parameter model
+    "small": {
         "url": "https://github.com/misyaguziya/VRCT-weights/releases/download/v1.0/m2m100_418m.zip",
         "directory_name": "m2m100_418m",
         "tokenizer": "facebook/m2m100_418M",
@@ -17,9 +26,9 @@ ctranslate2_weights = {
             "model.bin": "e7c26a9abb5260abd0268fbe3040714070dec254a990b4d7fd3f74c5230e3acb",
             "sentencepiece.model": "d8f7c76ed2a5e0822be39f0a4f95a55eb19c78f4593ce609e2edbc2aea4d380a",
             "shared_vocabulary.txt": "bd440aa21b8ca3453fc792a0018a1f3fe68b3464aadddd4d16a4b72f73c86d8c",
-        }
+        },
     },
-    "large": { # M2M-100 1.2B-parameter model
+    "large": {
         "url": "https://github.com/misyaguziya/VRCT-weights/releases/download/v1.0/m2m100_12b.zip",
         "directory_name": "m2m100_12b",
         "tokenizer": "facebook/m2m100_1.2b",
@@ -27,77 +36,107 @@ ctranslate2_weights = {
             "model.bin": "abb7bf4ba7e5e016b6e3ed480c752459b2f783ac8fca372e7587675e5bf3a919",
             "sentencepiece.model": "d8f7c76ed2a5e0822be39f0a4f95a55eb19c78f4593ce609e2edbc2aea4d380a",
             "shared_vocabulary.txt": "bd440aa21b8ca3453fc792a0018a1f3fe68b3464aadddd4d16a4b72f73c86d8c",
-        }
+        },
     },
 }
 
-def calculate_file_hash(file_path, block_size=65536):
+
+def calculate_file_hash(file_path: str, block_size: int = 65536) -> str:
     hash_object = hashlib.sha256()
-
-    with open(file_path, 'rb') as file:
-        for block in iter(lambda: file.read(block_size), b''):
+    with open(file_path, "rb") as f:
+        for block in iter(lambda: f.read(block_size), b""):
             hash_object.update(block)
-
     return hash_object.hexdigest()
 
-def checkCTranslate2Weight(root, weight_type="small"):
-    weight_directory_name = ctranslate2_weights[weight_type]["directory_name"]
-    hash_data = ctranslate2_weights[weight_type]["hash"]
-    files = [
-        "model.bin",
-        "sentencepiece.model",
-        "shared_vocabulary.txt"
-    ]
-    path = os_path.join(root, "weights", "ctranslate2")
 
-    # check already downloaded
-    already_downloaded = False
-    if all(os_path.exists(os_path.join(path, weight_directory_name, file)) for file in files):
-        # check hash
-        for file in files:
-            original_hash = hash_data[file]
-            current_hash = calculate_file_hash(os_path.join(path, weight_directory_name, file))
-            if original_hash != current_hash:
-                break
-        already_downloaded = True
-    return already_downloaded
+def checkCTranslate2Weight(root: str, weight_type: str = "small") -> bool:
+    """Return True if the requested weight files exist and match their hashes.
 
-def downloadCTranslate2Weight(root, weight_type="small", callback=None, end_callback=None):
-    url = ctranslate2_weights[weight_type]["url"]
-    filename = "weight.zip"
-    path = os_path.join(root, "weights", "ctranslate2")
-    os_makedirs(path, exist_ok=True)
-
-    if checkCTranslate2Weight(root, weight_type) is False:
+    This function intentionally avoids raising: callers use the boolean to
+    decide whether to (re)download weights.
+    """
+    weight_info = ctranslate2_weights.get(weight_type)
+    if weight_info is None:
+        return False
+    weight_directory_name = weight_info["directory_name"]
+    hash_data = weight_info["hash"]
+    files = ["model.bin", "sentencepiece.model", "shared_vocabulary.txt"]
+    base_path = os_path.join(root, "weights", "ctranslate2")
+    # quick existence check
+    for f in files:
+        p = os_path.join(base_path, weight_directory_name, f)
+        if not os_path.exists(p):
+            return False
+    # verify hashes
+    for f in files:
+        p = os_path.join(base_path, weight_directory_name, f)
         try:
-            with tempfile.TemporaryDirectory() as tmp_path:
-                res = requests_get(url, stream=True)
-                file_size = int(res.headers.get('content-length', 0))
-                total_chunk = 0
-                with open(os_path.join(tmp_path, filename), 'wb') as file:
-                    for chunk in res.iter_content(chunk_size=1024*2000):
-                        file.write(chunk)
-                        if isinstance(callback, Callable):
-                            total_chunk += len(chunk)
-                            callback(total_chunk/file_size)
-
-                with ZipFile(os_path.join(tmp_path, filename)) as zf:
-                    zf.extractall(path)
+            if calculate_file_hash(p) != hash_data[f]:
+                return False
         except Exception:
             errorLogging()
+            return False
+    return True
 
-    if isinstance(end_callback, Callable):
-        end_callback()
 
-def downloadCTranslate2Tokenizer(path, weight_type="small"):
-    directory_name = ctranslate2_weights[weight_type]["directory_name"]
-    tokenizer = ctranslate2_weights[weight_type]["tokenizer"]
-    tokenizer_path = os_path.join(path, "weights", "ctranslate2", directory_name, "tokenizer")
+def downloadCTranslate2Weight(root: str, weight_type: str = "small", callback: Optional[Callable[[float], None]] = None, end_callback: Optional[Callable[[], None]] = None) -> None:
+    """Download and extract ctranslate2 weights for the given type.
 
+    callback receives a float between 0 and 1 for progress when available.
+    end_callback is invoked after success or failure to allow caller cleanup.
+    """
+    weight_info = ctranslate2_weights.get(weight_type)
+    if weight_info is None:
+        return
+    url = weight_info["url"]
+    filename = "weight.zip"
+    dst_path = os_path.join(root, "weights", "ctranslate2")
+    os_makedirs(dst_path, exist_ok=True)
+    if checkCTranslate2Weight(root, weight_type):
+        if callable(end_callback):
+            end_callback()
+        return
     try:
-        os_makedirs(tokenizer_path, exist_ok=True)
-        transformers.AutoTokenizer.from_pretrained(tokenizer, cache_dir=tokenizer_path)
+        with tempfile.TemporaryDirectory() as tmp_path:
+            res = requests_get(url, stream=True, timeout=30)
+            total = int(res.headers.get("content-length", 0) or 0)
+            written = 0
+            out_path = os_path.join(tmp_path, filename)
+            with open(out_path, "wb") as out:
+                for chunk in res.iter_content(chunk_size=1024 * 1024):
+                    if not chunk:
+                        continue
+                    out.write(chunk)
+                    written += len(chunk)
+                    if callable(callback) and total:
+                        try:
+                            callback(written / total)
+                        except Exception:
+                            errorLogging()
+            with ZipFile(out_path) as zf:
+                zf.extractall(dst_path)
     except Exception:
         errorLogging()
-        tokenizer_path = os_path.join("./weights", "ctranslate2", directory_name, "tokenizer")
-        transformers.AutoTokenizer.from_pretrained(tokenizer, cache_dir=tokenizer_path)
+    finally:
+        if callable(end_callback):
+            end_callback()
+
+
+def downloadCTranslate2Tokenizer(root: str, weight_type: str = "small") -> None:
+    """Ensure a tokenizer for the requested weight is available (cached).
+
+    This will attempt to download the tokenizer via Hugging Face's transformers
+    and cache it under the weights directory. It logs failures instead of
+    raising to keep runtime resilient during startup.
+    """
+    weight_info = ctranslate2_weights.get(weight_type)
+    if weight_info is None:
+        return
+    directory_name = weight_info["directory_name"]
+    tokenizer_name = weight_info["tokenizer"]
+    tokenizer_cache = os_path.join(root, "weights", "ctranslate2", directory_name, "tokenizer")
+    try:
+        os_makedirs(tokenizer_cache, exist_ok=True)
+        transformers.AutoTokenizer.from_pretrained(tokenizer_name, cache_dir=tokenizer_cache)
+    except Exception:
+        errorLogging()