chore: offline mode paring code deduplication

Author: bsbds

deepdoc/common/misc_utils.py

@@ -15,6 +15,7 @@
 #
 
 import logging
+import os
 
 logger = logging.getLogger(__name__)
 
@@ -31,3 +32,13 @@ def pip_install_torch():
         return True
     except ImportError:
         return False
+
+
+def parse_bool(value: str | None, default: bool = False) -> bool:
+    if value is None:
+        return default
+    return value.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def offline_mode_or_from_env(offline: bool | None = None) -> bool:
+    return offline if offline is not None else parse_bool(os.getenv("DEEPDOC_OFFLINE"), default=False)

deepdoc/common/model_store.py

@@ -24,19 +24,14 @@
 from dataclasses import dataclass
 from importlib import resources
 from pathlib import Path
+from ..common.misc_utils import offline_mode_or_from_env
 
 
 GLOBAL_MODELSCOPE_REPO_ENV = "DEEPDOC_MODELSCOPE_REPO"
 GLOBAL_MODELSCOPE_REVISION_ENV = "DEEPDOC_MODELSCOPE_REVISION"
 TOKENIZER_MODEL_DIR_ENV = "DEEPDOC_TOKENIZER_MODEL_DIR"
 
 
-def _parse_bool(value: str | None, default: bool = False) -> bool:
-    if value is None:
-        return default
-    return value.strip().lower() in {"1", "true", "yes", "on"}
-
-
 def _normalize_provider(provider: str | None) -> str:
     normalized = (provider or os.getenv("DEEPDOC_MODEL_PROVIDER", "auto")).strip().lower()
     aliases = {
@@ -66,10 +61,7 @@ def _resolve_tokenizer_dict_path() -> Path:
         dictionary = Path(str(resources.files("deepdoc").joinpath("dict", "huqie.txt"))).resolve()
 
     if not dictionary.exists():
-        raise FileNotFoundError(
-            "Tokenizer dictionary not found: {}. Set {} to a directory containing huqie.txt."
-            .format(dictionary, TOKENIZER_MODEL_DIR_ENV)
-        )
+        raise FileNotFoundError("Tokenizer dictionary not found: {}. Set {} to a directory containing huqie.txt.".format(dictionary, TOKENIZER_MODEL_DIR_ENV))
     return dictionary
 
 
@@ -260,7 +252,7 @@ def resolve_bundle_dir(
 
     spec = BUNDLES[bundle]
     provider_name = _normalize_provider(provider)
-    offline_mode = offline if offline is not None else _parse_bool(os.getenv("DEEPDOC_OFFLINE"), default=False)
+    offline_mode = offline_mode_or_from_env(offline)
 
     explicit_local = os.getenv(spec.local_dir_env)
     if explicit_local:

deepdoc/config.py

@@ -11,16 +11,11 @@
     resolve_xgb_model_dir,
     validate_bundle_dir,
 )
+from .common.misc_utils import offline_mode_or_from_env
 
 ProviderType = Literal["local", "modelscope", "auto"]
 
 
-def _parse_bool(value: str | None, default: bool = False) -> bool:
-    if value is None:
-        return default
-    return value.strip().lower() in {"1", "true", "yes", "on"}
-
-
 def _normalize_provider(provider: str) -> ProviderType:
     normalized = provider.strip().lower()
     aliases = {
@@ -31,9 +26,7 @@ def _normalize_provider(provider: str) -> ProviderType:
     }
     normalized = aliases.get(normalized, normalized)
     if normalized not in {"local", "modelscope", "auto"}:
-        raise ValueError(
-            "Unsupported model provider '{}'. Use one of: local, modelscope, auto.".format(provider)
-        )
+        raise ValueError("Unsupported model provider '{}'. Use one of: local, modelscope, auto.".format(provider))
     return normalized  # type: ignore[return-value]
 
 
@@ -64,13 +57,10 @@ def resolve_dict_path(self) -> str:
             if dictionary.is_dir():
                 dictionary = dictionary.joinpath("huqie.txt")
             if dictionary.suffix != ".txt":
-                raise ValueError(
-                    "TokenizerConfig.dict_path must point to a '.txt' dictionary file, got: {}".format(dictionary)
-                )
+                raise ValueError("TokenizerConfig.dict_path must point to a '.txt' dictionary file, got: {}".format(dictionary))
             _require_file(
                 dictionary,
-                "Tokenizer dictionary not found: {}. Provide a valid TokenizerConfig.dict_path."
-                .format(dictionary),
+                "Tokenizer dictionary not found: {}. Provide a valid TokenizerConfig.dict_path.".format(dictionary),
             )
             return str(dictionary)
 
@@ -86,7 +76,7 @@ def from_env(cls) -> "TokenizerConfig":
 
         return cls(
             dict_path=dict_path,
-            offline=_parse_bool(os.getenv("DEEPDOC_OFFLINE"), default=False),
+            offline=offline_mode_or_from_env(None)
             nltk_data_dir=os.getenv("DEEPDOC_NLTK_DATA_DIR"),
         )
 
@@ -107,10 +97,7 @@ def _resolve_bundle_dir(self, bundle: str, explicit_dir: str | None) -> str:
             candidate = Path(explicit_dir).expanduser().resolve()
             exists, missing = validate_bundle_dir(bundle, candidate)
             if not exists:
-                raise FileNotFoundError(
-                    "Missing required files for '{}' bundle in {}: {}"
-                    .format(bundle, candidate, ", ".join(missing))
-                )
+                raise FileNotFoundError("Missing required files for '{}' bundle in {}: {}".format(bundle, candidate, ", ".join(missing)))
             return str(candidate)
 
         model_provider = self.normalized_provider()

deepdoc/depend/nltk_manager.py

@@ -2,6 +2,7 @@
 import os
 import threading
 from pathlib import Path
+from ..common.misc_utils import offline_mode_or_from_env
 
 logger = logging.getLogger(__name__)
 
@@ -24,12 +25,6 @@
 _ensured_keys: set[tuple[str, bool]] = set()
 
 
-def _parse_bool(value: str | None, default: bool = False) -> bool:
-    if value is None:
-        return default
-    return value.strip().lower() in {"1", "true", "yes", "on"}
-
-
 def _resolve_nltk_data_dir(data_dir: str | None) -> Path | None:
     # Resolution precedence:
     # 1) explicit arg
@@ -81,7 +76,7 @@ def ensure_nltk_data(
     import nltk
 
     resolved_dir = _resolve_nltk_data_dir(data_dir)
-    offline_mode = offline if offline is not None else _parse_bool(os.getenv("DEEPDOC_OFFLINE"), default=False)
+    offline_mode = offline_mode_or_from_env(offline)
     auto_download_mode = not offline_mode
 
     _ensure_search_path(nltk, resolved_dir)
@@ -110,10 +105,7 @@ def ensure_nltk_data(
         if missing_packages:
             searched_paths = ", ".join(nltk.data.path)
             raise RuntimeError(
-                "Missing required NLTK packages: {}. Searched paths: {}. "
-                "Set DEEPDOC_NLTK_DATA_DIR to a local NLTK data path, or disable offline mode by setting "
-                "DEEPDOC_OFFLINE=0."
-                .format(
+                "Missing required NLTK packages: {}. Searched paths: {}. Set DEEPDOC_NLTK_DATA_DIR to a local NLTK data path, or disable offline mode by setting DEEPDOC_OFFLINE=0.".format(
                     ", ".join(missing_packages),
                     searched_paths,
                 )