389 lines
16 KiB
Python
389 lines
16 KiB
Python
"""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
|
|
from .schema_migration import CURRENT_VERSION, migrate
|
|
|
|
|
|
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()
|
|
|
|
@classmethod
|
|
def from_data(cls, data: Dict[str, Any], path: Path):
|
|
"""Dựng từ dict có sẵn — KHÔNG đọc đĩa, KHÔNG nâng cấp schema.
|
|
|
|
Dành cho test: chúng dựng cấu hình trong bộ nhớ rồi mới ghi. Đi qua
|
|
``__init__`` thường thì nó đọc file (chưa có) và có thể chạy migration
|
|
trên dữ liệu test, tức là test đo nhầm thứ khác.
|
|
"""
|
|
obj = cls.__new__(cls)
|
|
obj._file = AtomicJsonFile(Path(path))
|
|
obj._secrets = None
|
|
from ... import config as legacy
|
|
obj._defaults = legacy.DEFAULT_CONFIG
|
|
obj._env_overrides = legacy._apply_env_overrides
|
|
obj.data = data
|
|
return obj
|
|
|
|
# ---- nạp ------------------------------------------------------------
|
|
def _load(self) -> Dict[str, Any]:
|
|
merged = copy.deepcopy(self._defaults)
|
|
stored = self._file.read(default=None)
|
|
if isinstance(stored, dict):
|
|
# Nâng cấp TRƯỚC khi trộn với mặc định: bước v1→v2 gỡ api_key khỏi
|
|
# đĩa, mà mặc định thì không có khoá nào để gỡ.
|
|
stored, changed = migrate(stored, secrets=self._secrets,
|
|
path=self._file.path)
|
|
merged = _deep_merge(merged, stored)
|
|
if changed:
|
|
self.data = merged
|
|
self.save() # ghi ngay, để lần sau khỏi chuyển lại
|
|
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", "")
|
|
|
|
@active_provider.setter
|
|
def active_provider(self, name: str) -> None:
|
|
"""``AppConfig`` cũ cho gán thẳng, và 3 chỗ trong app.py đang gán. Bỏ
|
|
setter đi thì Qt nuốt AttributeError trong slot và triệu chứng là
|
|
"bấm không ăn", không có lỗi nào hiện ra — mất hẳn một buổi mới truy
|
|
ra. Refactor thì hành vi nhìn từ ngoài phải y hệt."""
|
|
self.data["active_provider"] = name
|
|
|
|
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 = getattr(self, "_project_history_dir", None) or 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")
|
|
|
|
@theme.setter
|
|
def theme(self, value: str) -> None:
|
|
self.data["theme"] = value
|
|
|
|
def set_theme(self, value: str) -> None:
|
|
self.data["theme"] = value
|
|
|
|
@property
|
|
def language(self) -> str:
|
|
return self.data.get("language", "vi")
|
|
|
|
@language.setter
|
|
def language(self, value: str) -> None:
|
|
self.data["language"] = value
|
|
|
|
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
|
|
|
|
|
|
# ---- phần bù để thay được AppConfig ----------------------------------
|
|
# 21 thành viên dưới đây chép nguyên ngữ nghĩa từ ``config.py::AppConfig``.
|
|
# Không phải thiết kế mới: chừng nào 29 file còn gọi qua ``ctx.config`` thì
|
|
# repository phải trả lời được đúng những câu hỏi cũ, nếu không thì không
|
|
# tráo được. Dọn lại là việc của các R sau, không phải của R02.
|
|
|
|
#: Các chế độ định tuyến. Delta thêm "fallback" ở R03-T03. Định nghĩa ở đây
|
|
#: là bản chính; ``tests/test_config_repository.py`` có bài đối chiếu với
|
|
#: ``config.py`` để hai bên lệch nhau là đỏ ngay.
|
|
ROUTING_MODES = ("off", "auto", "manual", "fallback")
|
|
|
|
@classmethod
|
|
def load(cls, path: Path | None = None, *, secrets: SecretStore | None = None):
|
|
"""Dựng repository từ đường dẫn mặc định — thay ``AppConfig.load()``."""
|
|
if path is None:
|
|
from ... import config as legacy
|
|
path = legacy.CONFIG_PATH
|
|
return cls(Path(path), secrets=secrets)
|
|
|
|
@property
|
|
def path(self) -> Path:
|
|
return self._file.path
|
|
|
|
# ---- TLS -------------------------------------------------------------
|
|
|
|
@property
|
|
def ca_bundle(self) -> str:
|
|
"""Đường dẫn file PEM riêng, hoặc '' để kiểm chứng chỉ như bình thường.
|
|
|
|
Dùng làm tham số ``verify=`` của ``requests`` cho mọi lượt gọi HTTPS."""
|
|
return (self.data.get("tls_ca_bundle") or "").strip()
|
|
|
|
@ca_bundle.setter
|
|
def ca_bundle(self, value: str) -> None:
|
|
self.data["tls_ca_bundle"] = (value or "").strip()
|
|
|
|
# ---- MS365 -----------------------------------------------------------
|
|
|
|
@property
|
|
def ms365(self) -> Dict[str, Any]:
|
|
return self.data.setdefault("ms365", copy.deepcopy(self._defaults["ms365"]))
|
|
|
|
def ms365_try_unlock(self, code: str) -> bool:
|
|
"""Mở khoá nhóm MS365 trong Cài đặt cho phiên này.
|
|
|
|
Đây là khoá phía giao diện (chặn bấm nhầm vào một mục nhạy cảm), KHÔNG
|
|
phải xác thực Microsoft. Không bao giờ được lưu ở trạng thái đã mở."""
|
|
if (code or "") and code == self.ms365.get("unlock_code", ""):
|
|
self.data["ms365"]["unlocked"] = True
|
|
return True
|
|
return False
|
|
|
|
def ms365_lock(self) -> None:
|
|
self.data.setdefault("ms365", {})["unlocked"] = False
|
|
|
|
# ---- nhóm cấu hình đọc thẳng ------------------------------------------
|
|
|
|
@property
|
|
def code(self) -> Dict[str, Any]:
|
|
return self.data["code"]
|
|
|
|
@property
|
|
def teams(self) -> Dict[str, Any]:
|
|
return self.data["teams"]
|
|
|
|
@property
|
|
def history(self) -> Dict[str, Any]:
|
|
return self.data["history"]
|
|
|
|
@property
|
|
def codebase_memory(self) -> Dict[str, Any]:
|
|
return self.data["codebase_memory"]
|
|
|
|
@property
|
|
def cowork(self) -> Dict[str, Any]:
|
|
return self.data["cowork"]
|
|
|
|
@property
|
|
def mcp_servers(self) -> list:
|
|
return self.data.setdefault("mcp_servers", [])
|
|
|
|
@property
|
|
def structure(self) -> Dict[str, Any]:
|
|
return self.data.setdefault("structure", {"max_nodes": 400, "max_edges": 400})
|
|
|
|
@property
|
|
def monitoring_visibility(self) -> Dict[str, bool]:
|
|
return self.data.setdefault(
|
|
"monitoring_visibility",
|
|
copy.deepcopy(self._defaults["monitoring_visibility"]))
|
|
|
|
@property
|
|
def ext_connectors(self) -> Dict[str, list]:
|
|
"""Connector (MCP) gom theo nhóm CAD/CAE/MS365/Other."""
|
|
d = self.data.setdefault(
|
|
"ext_connectors", {"cad": [], "cae": [], "ms365": [], "other": []})
|
|
for cat in ("cad", "cae", "ms365", "other"):
|
|
d.setdefault(cat, [])
|
|
return d
|
|
|
|
# ---- công tắc tổng cho connector -------------------------------------
|
|
|
|
@property
|
|
def connect_external(self) -> bool:
|
|
"""Tắt cái này là agent không nối tới connector ngoài nào cả. Mặc định
|
|
BẬT để cấu hình đang chạy không đổi hành vi."""
|
|
return bool(self.data.setdefault("tools", {}).get("connect_external", True))
|
|
|
|
def set_connect_external(self, enabled: bool) -> None:
|
|
self.data.setdefault("tools", {})["connect_external"] = bool(enabled)
|
|
self.save()
|
|
|
|
# ---- những thứ đã gieo sẵn -------------------------------------------
|
|
|
|
@property
|
|
def seeded_library_skills(self) -> list:
|
|
"""Slug của skill thư viện đã gieo — để cái người dùng xoá đi không bị
|
|
lặng lẽ gieo lại."""
|
|
return list(self.data.setdefault("seeded_library_skills", []))
|
|
|
|
@seeded_library_skills.setter
|
|
def seeded_library_skills(self, slugs) -> None:
|
|
self.data["seeded_library_skills"] = list(dict.fromkeys(slugs or []))
|
|
|
|
@property
|
|
def seeded_builtin_flows(self) -> list:
|
|
"""Id của flow Co4E dựng sẵn đã gieo (cùng quy tắc tôn trọng việc người
|
|
dùng đã xoá như seeded_library_skills)."""
|
|
return list(self.data.setdefault("seeded_builtin_flows", []))
|
|
|
|
@seeded_builtin_flows.setter
|
|
def seeded_builtin_flows(self, ids) -> None:
|
|
self.data["seeded_builtin_flows"] = list(dict.fromkeys(ids or []))
|
|
|
|
# ---- định tuyến theo từng bề mặt chat --------------------------------
|
|
|
|
def routing_mode_for(self, surface: str) -> str:
|
|
"""Chế độ có hiệu lực cho một bề mặt chat.
|
|
|
|
Đặt riêng cho bề mặt thì thắng; để trống thì lấy ``switch_mode`` chung.
|
|
Giá trị lạ rơi về "off" — định tuyến luôn là thứ phải bật, kể cả khi
|
|
có người sửa tay file cấu hình."""
|
|
routing = self.routing
|
|
override = (routing.get("surface_modes", {}) or {}).get(surface, "")
|
|
mode = override or routing.get("switch_mode", "off")
|
|
return mode if mode in self.ROUTING_MODES else "off"
|
|
|
|
def set_routing_mode_for(self, surface: str, mode: str) -> None:
|
|
mode = mode if mode in self.ROUTING_MODES else "off"
|
|
self.routing.setdefault("surface_modes", {})[surface] = mode
|
|
self.save()
|
|
|
|
# ---- tiện ích --------------------------------------------------------
|
|
|
|
def model_label(self) -> str:
|
|
return str(self.provider_conf().get("model", "?"))
|
|
|
|
# ---- 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)
|
|
to_write["schema_version"] = CURRENT_VERSION
|
|
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
|