198 lines
7.2 KiB
Python
198 lines
7.2 KiB
Python
"""Application configuration and per-user preferences.
|
|
|
|
Secrets are intentionally excluded: the desktop client only receives a short-lived
|
|
TRTC UserSig from the authenticated backend and never stores an SDKSecretKey.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from contextlib import suppress
|
|
from dataclasses import asdict, dataclass, fields, replace
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
from doctor_workstation import DEBUG_MODE, ONLINE_API_BASE_URL
|
|
|
|
try:
|
|
from dotenv import load_dotenv
|
|
except ImportError: # pragma: no cover - optional during pure unit tests
|
|
load_dotenv = None
|
|
|
|
try:
|
|
from platformdirs import user_config_dir, user_log_dir
|
|
except ImportError: # pragma: no cover - deterministic fallback
|
|
user_config_dir = None
|
|
user_log_dir = None
|
|
|
|
|
|
APP_NAME = "ZhenyangDoctor"
|
|
APP_AUTHOR = "Zhenyangtang"
|
|
|
|
|
|
def _as_bool(value: Any, default: bool) -> bool:
|
|
if isinstance(value, bool):
|
|
return value
|
|
if value is None:
|
|
return default
|
|
return str(value).strip().lower() in {"1", "true", "yes", "on", "y"}
|
|
|
|
|
|
def _safe_timeout(value: str | int | float | None, default: float = 30.0) -> float:
|
|
try:
|
|
parsed = float(value) # type: ignore[arg-type]
|
|
except (TypeError, ValueError):
|
|
return default
|
|
return max(3.0, min(parsed, 120.0))
|
|
|
|
|
|
def _config_home() -> Path:
|
|
override = os.getenv("DOCTOR_CONFIG_DIR", "").strip()
|
|
if override:
|
|
return Path(override).expanduser()
|
|
if user_config_dir is not None:
|
|
return Path(user_config_dir(APP_NAME, APP_AUTHOR))
|
|
return Path.home() / f".{APP_NAME.lower()}"
|
|
|
|
|
|
def _log_home() -> Path:
|
|
override = os.getenv("DOCTOR_LOG_DIR", "").strip()
|
|
if override:
|
|
return Path(override).expanduser()
|
|
if user_log_dir is not None:
|
|
return Path(user_log_dir(APP_NAME, APP_AUTHOR))
|
|
return _config_home() / "logs"
|
|
|
|
|
|
def normalize_api_base_url(value: str) -> str:
|
|
"""Return a normalized HTTP(S) base URL ending in ``/adminapi``.
|
|
|
|
Empty values are accepted for demo mode. Credentials, fragments and query
|
|
strings are rejected to prevent accidentally persisting tokens in settings.
|
|
"""
|
|
|
|
raw = (value or "").strip().rstrip("/")
|
|
if not raw:
|
|
return ""
|
|
parts = urlsplit(raw)
|
|
if parts.scheme not in {"http", "https"} or not parts.netloc:
|
|
raise ValueError("服务器地址必须是完整的 http:// 或 https:// 地址")
|
|
if parts.username or parts.password or parts.query or parts.fragment:
|
|
raise ValueError("服务器地址不能包含账号、密码、查询参数或片段")
|
|
path = parts.path.rstrip("/")
|
|
if not path.endswith("/adminapi"):
|
|
path = f"{path}/adminapi" if path else "/adminapi"
|
|
return urlunsplit((parts.scheme, parts.netloc, path, "", ""))
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class AppConfig:
|
|
"""Runtime configuration loaded from environment and user preferences."""
|
|
|
|
api_base_url: str = ""
|
|
demo_mode: bool = False
|
|
debug_mode: bool = DEBUG_MODE
|
|
video_mode: str = "embedded"
|
|
video_web_url: str = ""
|
|
verify_ssl: bool = True
|
|
request_timeout: float = 30.0
|
|
log_level: str = "INFO"
|
|
remembered_account: str = ""
|
|
|
|
@property
|
|
def config_dir(self) -> Path:
|
|
return _config_home()
|
|
|
|
@property
|
|
def log_dir(self) -> Path:
|
|
return _log_home()
|
|
|
|
@property
|
|
def preferences_file(self) -> Path:
|
|
return self.config_dir / "preferences.json"
|
|
|
|
@classmethod
|
|
def load(cls, env_file: Path | None = None) -> AppConfig:
|
|
if load_dotenv is not None:
|
|
load_dotenv(dotenv_path=env_file, override=False)
|
|
|
|
debug_mode = bool(DEBUG_MODE)
|
|
raw_url = (
|
|
os.getenv("DOCTOR_API_BASE_URL", "") if debug_mode else ONLINE_API_BASE_URL
|
|
)
|
|
try:
|
|
api_url = normalize_api_base_url(raw_url)
|
|
except ValueError as error:
|
|
if not debug_mode:
|
|
raise ValueError("ONLINE_API_BASE_URL 必须是有效的 HTTP(S) 域名") from error
|
|
api_url = ""
|
|
if not debug_mode and not api_url:
|
|
raise ValueError("正式模式下 ONLINE_API_BASE_URL 不能为空")
|
|
|
|
config = cls(
|
|
api_base_url=api_url,
|
|
demo_mode=_as_bool(os.getenv("DOCTOR_DEMO_MODE"), False) if debug_mode else False,
|
|
debug_mode=debug_mode,
|
|
video_mode=os.getenv("DOCTOR_VIDEO_MODE", "embedded").strip().lower(),
|
|
video_web_url=os.getenv("DOCTOR_VIDEO_WEB_URL", "").strip(),
|
|
verify_ssl=(
|
|
_as_bool(os.getenv("DOCTOR_VERIFY_SSL"), True) if debug_mode else True
|
|
),
|
|
request_timeout=_safe_timeout(os.getenv("DOCTOR_REQUEST_TIMEOUT")),
|
|
log_level=os.getenv("DOCTOR_LOG_LEVEL", "INFO").strip().upper(),
|
|
)
|
|
return config._merge_preferences()
|
|
|
|
def _merge_preferences(self) -> AppConfig:
|
|
try:
|
|
payload = json.loads(self.preferences_file.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError, TypeError):
|
|
return self
|
|
allowed = {item.name for item in fields(self)} - {"debug_mode"}
|
|
if not self.debug_mode:
|
|
allowed -= {"api_base_url", "demo_mode", "verify_ssl"}
|
|
clean: dict[str, Any] = {key: value for key, value in payload.items() if key in allowed}
|
|
if "api_base_url" in clean:
|
|
try:
|
|
clean["api_base_url"] = normalize_api_base_url(str(clean["api_base_url"]))
|
|
except ValueError:
|
|
clean.pop("api_base_url", None)
|
|
if "video_mode" in clean and clean["video_mode"] not in {"embedded", "browser"}:
|
|
clean.pop("video_mode", None)
|
|
if "request_timeout" in clean:
|
|
clean["request_timeout"] = _safe_timeout(clean["request_timeout"])
|
|
if "verify_ssl" in clean:
|
|
clean["verify_ssl"] = _as_bool(clean["verify_ssl"], True)
|
|
return replace(self, **clean)
|
|
|
|
def save_preferences(self) -> None:
|
|
"""Persist non-secret preferences atomically with user-only intent."""
|
|
|
|
self.config_dir.mkdir(parents=True, exist_ok=True)
|
|
target = self.preferences_file
|
|
temporary = target.with_suffix(".tmp")
|
|
payload = asdict(self)
|
|
payload.pop("debug_mode", None)
|
|
temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
with suppress(OSError):
|
|
os.chmod(temporary, 0o600)
|
|
temporary.replace(target)
|
|
|
|
def with_updates(self, **changes: Any) -> AppConfig:
|
|
changes.pop("debug_mode", None)
|
|
if not self.debug_mode:
|
|
changes.pop("api_base_url", None)
|
|
changes.pop("demo_mode", None)
|
|
changes.pop("verify_ssl", None)
|
|
if "api_base_url" in changes:
|
|
changes["api_base_url"] = normalize_api_base_url(str(changes["api_base_url"]))
|
|
if "video_mode" in changes and changes["video_mode"] not in {"embedded", "browser"}:
|
|
raise ValueError("视频模式只能是 embedded 或 browser")
|
|
if "request_timeout" in changes:
|
|
changes["request_timeout"] = _safe_timeout(changes["request_timeout"])
|
|
if "verify_ssl" in changes:
|
|
changes["verify_ssl"] = _as_bool(changes["verify_ssl"], True)
|
|
return replace(self, **changes)
|