103 lines
3.3 KiB
Python
103 lines
3.3 KiB
Python
"""采购工具的本地运行目录策略。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RuntimePaths:
|
|
"""仅包含应用生成物的本地目录。
|
|
|
|
日志和证据产物不落在源码树中，避免把含有现场信息的运行数据误提交到 Git。
|
|
"""
|
|
|
|
root: Path
|
|
logs: Path
|
|
artifacts: Path
|
|
state: Path
|
|
database: Path
|
|
|
|
@classmethod
|
|
def from_root(cls, root: Path) -> "RuntimePaths":
|
|
# 路径在进程启动时一次性固化；之后 cwd 改变不能打开第二套数据库或绕过原 mutex。
|
|
resolved_root = root.expanduser().resolve(strict=False)
|
|
state = resolved_root / "state"
|
|
return cls(
|
|
root=resolved_root,
|
|
logs=resolved_root / "logs",
|
|
artifacts=resolved_root / "artifacts",
|
|
state=state,
|
|
database=state / "client-state.sqlite3",
|
|
)
|
|
|
|
@classmethod
|
|
def default(cls) -> "RuntimePaths":
|
|
local_app_data = os.environ.get("LOCALAPPDATA")
|
|
if local_app_data:
|
|
local_root = Path(local_app_data).expanduser()
|
|
if not local_root.is_absolute():
|
|
raise RuntimeError("local_app_data_must_be_absolute")
|
|
return cls.from_root(local_root / "cmbuyer")
|
|
|
|
if os.name == "nt":
|
|
# Windows 上回退到 home 会悄悄创建第二套状态库并绕开同一 mutex，必须失败闭合。
|
|
raise RuntimeError("local_app_data_required")
|
|
|
|
return cls.from_root(Path.home() / ".local" / "share" / "cmbuyer")
|
|
|
|
def ensure_exists(self) -> None:
|
|
"""创建运行目录；调用方负责向用户呈现无法创建目录的错误。"""
|
|
|
|
self.logs.mkdir(parents=True, exist_ok=True)
|
|
self.artifacts.mkdir(parents=True, exist_ok=True)
|
|
self.state.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
@dataclass
|
|
class LocalStateRuntime:
|
|
"""持有 named mutex 与本地状态库，保证 mutex 总是先取得。"""
|
|
|
|
paths: RuntimePaths
|
|
mutex: Any
|
|
store: Any
|
|
|
|
@classmethod
|
|
def open(
|
|
cls,
|
|
paths: RuntimePaths | None = None,
|
|
*,
|
|
mutex_factory: Callable[[Path], Any] | None = None,
|
|
protector_factory: Callable[[], Any] | None = None,
|
|
store_factory: Callable[[Path, Any], Any] | None = None,
|
|
) -> "LocalStateRuntime":
|
|
from .localstate.protection import DpapiProtector
|
|
from .localstate.single_instance import NamedMutex
|
|
from .localstate.store import LocalStateStore
|
|
|
|
selected = paths or RuntimePaths.default()
|
|
selected.ensure_exists()
|
|
make_mutex = mutex_factory or NamedMutex
|
|
make_protector = protector_factory or DpapiProtector
|
|
make_store = store_factory or LocalStateStore
|
|
mutex = make_mutex(selected.database)
|
|
try:
|
|
protector = make_protector()
|
|
store = make_store(selected.database, protector)
|
|
except Exception:
|
|
mutex.close()
|
|
raise
|
|
return cls(selected, mutex, store)
|
|
|
|
def close(self) -> None:
|
|
self.mutex.close()
|
|
|
|
def __enter__(self) -> "LocalStateRuntime":
|
|
return self
|
|
|
|
def __exit__(self, exc_type: object, exc: object, traceback: object) -> None:
|
|
self.close()
|