"""ConfigRepository chạy trên file JSON — R02-T02. Thay cho ``config.py::AppConfig``. Hai khác biệt duy nhất về hành vi, cả hai đều là thứ ta muốn: 1. Ghi qua :class:`AtomicJsonFile` — mất điện giữa lúc lưu không còn làm hỏng cấu hình (R02-T01). 2. API key đọc từ :class:`SecretStore` rồi **ghép vào** dict do ``provider_conf()`` trả về — đúng đường A đã chốt 21/08 (``docs/refactor/GammaTeam_decisions.md``). Nhờ vậy 5 nơi đang đọc ``conf["api_key"]`` không phải sửa dòng nào, trong đó 3 nơi thuộc Team Duy. Mọi thứ còn lại giữ nguyên có chủ đích: trộn sâu với mặc định, đọc biến môi trường, ``ms365.unlocked`` không bao giờ chạm đĩa. Đây là refactor — hành vi nhìn từ ngoài phải y hệt. """ from __future__ import annotations import copy from pathlib import Path from typing import Any, Dict from ..persistence.json.atomic_json_file import AtomicJsonFile from ..secrets.secret_store import SecretStore, provider_key class JsonConfigRepository: """Cấu hình đọc/ghi từ một file JSON, bí mật để trong ``SecretStore``. ``secrets`` để None nghĩa là không có kho bí mật — mọi thứ vẫn chạy, chỉ là ``api_key`` lấy nguyên từ file như trước. Cần vậy để chuyển dần (R02-T05) chứ không phải đổi một phát cả app. """ def __init__(self, path: Path, *, secrets: SecretStore | None = None, defaults: Dict[str, Any] | None = None, env_overrides=None): self._file = AtomicJsonFile(path) self._secrets = secrets # Lấy thẳng từ config.py để hai bên không lệch nhau trong lúc chuyển. if defaults is None or env_overrides is None: from ... import config as legacy defaults = defaults if defaults is not None else legacy.DEFAULT_CONFIG env_overrides = env_overrides or legacy._apply_env_overrides self._defaults = defaults self._env_overrides = env_overrides self.data: Dict[str, Any] = self._load() # ---- nạp ------------------------------------------------------------ def _load(self) -> Dict[str, Any]: merged = copy.deepcopy(self._defaults) stored = self._file.read(default=None) if isinstance(stored, dict): merged = _deep_merge(merged, stored) merged = self._env_overrides(merged) # Trạng thái mở khoá ms365 chỉ tồn tại lúc chạy — mỗi lần mở app đều # bắt đầu ở trạng thái khoá, không tin giá trị đọc từ đĩa. merged.setdefault("ms365", {})["unlocked"] = False return merged def reload(self) -> None: self.data = self._load() # ---- provider -------------------------------------------------------- @property def active_provider(self) -> str: return self.data.get("active_provider", "") def set_active_provider(self, name: str) -> None: self.data["active_provider"] = name def provider_conf(self, name: str | None = None) -> Dict[str, Any]: """Cấu hình provider, có sẵn ``api_key``. Trả về BẢN SAO: chỗ gọi sửa dict này thì không được âm thầm ghi ngược vào cấu hình — và quan trọng hơn, khoá vừa ghép vào không được lẫn ngược vào ``self.data`` rồi theo ``save()`` xuống đĩa. """ name = name or self.active_provider conf = dict(self.data.get("providers", {}).get(name, {})) if self._secrets is not None: stored = self._secrets.get(provider_key(name)) if stored: conf["api_key"] = stored return conf def set_api_key(self, name: str, value: str) -> None: """Lưu khoá vào kho bí mật, và xoá khỏi cấu hình trên đĩa. Đây là nửa còn lại của đường A: dict *đọc ra* vẫn có ``api_key``, nhưng file JSON *trên đĩa* thì không — điều kiện để qua CASAN Check 1. """ if self._secrets is not None: self._secrets.set(provider_key(name), value) self.data.setdefault("providers", {}).setdefault(name, {})["api_key"] = "" else: self.data.setdefault("providers", {}).setdefault(name, {})["api_key"] = value # ---- đường dẫn ------------------------------------------------------- @property def shared_dir(self) -> str: return self.data.get("shared_dir", "") def history_dir(self) -> Path: rt = self.data.get("_project_history_dir") if rt: return Path(rt) custom = (self.data.get("history", {}).get("custom_dir") or "").strip() if custom: return Path(custom).expanduser() from ...config import CONFIG_DIR return CONFIG_DIR / "history" def cowork_output_dir(self) -> Path: custom = (self.data.get("cowork", {}).get("output_dir") or "").strip() if custom: return Path(custom).expanduser() from ... import paths from ...config import CONFIG_DIR root = paths.primary_onedrive_root() if root is not None: return root / "CoworkLocal" / "output" return CONFIG_DIR / "output" / "cowork" # ---- giao diện ------------------------------------------------------- @property def theme(self) -> str: return self.data.get("theme", "dark") def set_theme(self, value: str) -> None: self.data["theme"] = value @property def language(self) -> str: return self.data.get("language", "vi") def set_language(self, value: str) -> None: self.data["language"] = value # ---- nhóm cấu hình --------------------------------------------------- @property def routing(self) -> Dict[str, Any]: return self.data.setdefault("routing", {}) @property def auth(self) -> Dict[str, Any]: return self.data.setdefault("auth", {}) @property def agent_security(self) -> Dict[str, Any]: return self.data.setdefault("agent_security", {}) @property def tools_disabled(self) -> list[str]: return list(self.data.get("tools_disabled", [])) def set_tool_enabled(self, name: str, enabled: bool) -> None: disabled = list(self.data.get("tools_disabled", [])) if enabled: disabled = [t for t in disabled if t != name] elif name not in disabled: disabled.append(name) self.data["tools_disabled"] = disabled # ---- ghi ------------------------------------------------------------- def save(self) -> None: """Ghi nguyên tử. Không bao giờ để lộ trạng thái mở khoá ms365.""" to_write = self.data if self.data.get("ms365", {}).get("unlocked"): to_write = copy.deepcopy(self.data) to_write["ms365"]["unlocked"] = False to_write.pop("_project_history_dir", None) self._file.write(to_write) def _deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]: """Trộn sâu — giống hệt ``config.py::_deep_merge``. Không import lại từ đó vì file này phải sống được sau khi ``config.py`` biến mất; giữ bản sao 6 dòng còn hơn giữ một sợi dây phụ thuộc. """ out = copy.deepcopy(base) for key, value in (override or {}).items(): if isinstance(value, dict) and isinstance(out.get(key), dict): out[key] = _deep_merge(out[key], value) else: out[key] = value return out