first pull
This commit is contained in:
@@ -0,0 +1,452 @@
|
||||
"""
|
||||
零初始化 JSON 配置管理
|
||||
用法：from src.config import cfg # 直接访问，自动初始化
|
||||
"""
|
||||
import json
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional, TypeVar
|
||||
from dataclasses import dataclass, field, asdict
|
||||
|
||||
def _project_root() -> Path:
|
||||
"""自动查找项目根目录"""
|
||||
markers = ['pyproject.toml', 'settings.json', '.gitignore', 'main.py', '.python-version']
|
||||
current = Path(__file__).resolve().parent.parent # src的父目录
|
||||
|
||||
for path in [current, *current.parents]:
|
||||
if any((path / m).exists() for m in markers):
|
||||
return path
|
||||
if path == path.parent:
|
||||
break
|
||||
return current
|
||||
|
||||
# 配置定义
|
||||
|
||||
@dataclass
|
||||
class AIConfig:
|
||||
api_key: Optional[str] = "sk-xxxxx"
|
||||
base_url: str = "http://localhost:1234/v1"
|
||||
model_name: str = "qwen/qwen3-4b-2507"
|
||||
timeout: int = 30
|
||||
temperature: float = 0.4
|
||||
max_tokens: int = 8192
|
||||
|
||||
|
||||
@dataclass
|
||||
class TTSConfig:
|
||||
enabled: bool = True
|
||||
api_key: Optional[str] = None
|
||||
gpt_model_name: str = "GPT_weights_v2Pro/Yosuga_Airi-e32.ckpt"
|
||||
sovits_model_name: str = "SoVITS_weights_v2Pro/Yosuga_Airi_e16_s864.pth"
|
||||
host: str = "localhost"
|
||||
port: int = 20261
|
||||
reference_audio: str = "./using/reference.wav"
|
||||
streaming: bool = True
|
||||
speed: float = 1.0
|
||||
|
||||
@dataclass
|
||||
class ASRConfig:
|
||||
enabled: bool = True
|
||||
api_key: Optional[str] = None
|
||||
model_name: str = "fast-whisper"
|
||||
url: str = "http://localhost:20260/"
|
||||
|
||||
@dataclass
|
||||
class AutoAgentConfig:
|
||||
enabled: bool = True
|
||||
api_key: Optional[str] = None
|
||||
deployment_type: str = "lmstudio"
|
||||
model_name: str = "ui-tars-1.5-7b@q4_k_m"
|
||||
base_url: str = "http://localhost:1234/v1"
|
||||
temperature: float = 0.1
|
||||
max_tokens: int = 16384
|
||||
|
||||
@dataclass
|
||||
class LLMConfig:
|
||||
enabled: bool = True
|
||||
role_character: str = "你是由Misakiotoha开发的助手稲葉愛理ちゃん，可以和用户一起玩游戏，聊天，做各种事情，性格抽象，没事爱整整活。"
|
||||
max_context_tokens: int = 2048
|
||||
enable_history: bool = True
|
||||
language: str = "日本语"
|
||||
|
||||
@dataclass
|
||||
class PathsConfig:
|
||||
temp: str = "./tmp/"
|
||||
log: str = "./log/"
|
||||
using: str = "./using/"
|
||||
|
||||
|
||||
class AppConfig:
|
||||
"""
|
||||
应用主配置
|
||||
新增配置分组：1) 上方新建 dataclass 2) 下方 __init__ 添加字段 3) 完成
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
version: str = "1.0.0",
|
||||
debug: bool = False,
|
||||
ai: Optional[AIConfig] = None,
|
||||
tts: Optional[TTSConfig] = None,
|
||||
asr: Optional[ASRConfig] = None,
|
||||
auto_agent: Optional[AutoAgentConfig] = None,
|
||||
llm_core: Optional[LLMConfig] = None,
|
||||
paths: Optional[PathsConfig] = None,
|
||||
_config_path: Optional[Path] = None,
|
||||
**kwargs
|
||||
):
|
||||
# 基础字段
|
||||
self.version = version
|
||||
self.debug = debug
|
||||
self.ai = ai if ai is not None else AIConfig()
|
||||
self.tts = tts if tts is not None else TTSConfig()
|
||||
self.asr = asr if asr is not None else ASRConfig()
|
||||
self.auto_agent = auto_agent if auto_agent is not None else AutoAgentConfig()
|
||||
self.llm_core = llm_core if llm_core is not None else LLMConfig()
|
||||
self.paths = paths if paths is not None else PathsConfig()
|
||||
|
||||
# 内部状态（非 dataclass 字段，不会被序列化）
|
||||
self._config_path = _config_path
|
||||
self._lock = threading.RLock() # 普通属性，非 dataclass 字段
|
||||
|
||||
# 应用其他字段（用于从 JSON 加载时）
|
||||
for k, v in kwargs.items():
|
||||
if hasattr(self, k):
|
||||
setattr(self, k, v)
|
||||
|
||||
# 路径解析为绝对路径
|
||||
if self._config_path:
|
||||
root = self._config_path.parent
|
||||
for field_name in ['temp', 'log', 'using']:
|
||||
rel_path = getattr(self.paths, field_name)
|
||||
if not Path(rel_path).is_absolute():
|
||||
abs_path = (root / Path(rel_path)).resolve()
|
||||
setattr(self.paths, field_name, str(abs_path) + '/')
|
||||
|
||||
# 便捷属性
|
||||
|
||||
@property
|
||||
def temp_dir(self) -> Path:
|
||||
return Path(self.paths.temp)
|
||||
|
||||
@property
|
||||
def log_dir(self) -> Path:
|
||||
return Path(self.paths.log)
|
||||
|
||||
@property
|
||||
def using_dir(self) -> Path:
|
||||
return Path(self.paths.using)
|
||||
|
||||
# 核心方法
|
||||
|
||||
def get(self, key: str, default: Any = None) -> Any:
|
||||
"""
|
||||
点号路径访问：cfg.get("ai.timeout") / cfg.get("tts.enabled")
|
||||
"""
|
||||
try:
|
||||
keys = key.split('.')
|
||||
value = self
|
||||
for k in keys:
|
||||
value = getattr(value, k) if not isinstance(value, dict) else value[k]
|
||||
return value
|
||||
except (AttributeError, KeyError):
|
||||
return default
|
||||
|
||||
def set(self, key: str, value: Any, save: bool = True) -> 'AppConfig':
|
||||
"""
|
||||
点号路径设置，支持链式调用
|
||||
cfg.set("ai.timeout", 60).set("debug", True)
|
||||
"""
|
||||
with self._lock:
|
||||
keys = key.split('.')
|
||||
target = self
|
||||
for k in keys[:-1]:
|
||||
target = getattr(target, k)
|
||||
setattr(target, keys[-1], value)
|
||||
|
||||
if save:
|
||||
self._save()
|
||||
return self
|
||||
|
||||
def update(self, updates: Dict[str, Any], save: bool = True) -> 'AppConfig':
|
||||
"""
|
||||
批量更新：cfg.update({"ai": {"timeout": 60}, "debug": True})
|
||||
"""
|
||||
def deep_update(obj: Any, data: dict):
|
||||
for k, v in data.items():
|
||||
if hasattr(obj, k):
|
||||
current = getattr(obj, k)
|
||||
if isinstance(v, dict) and hasattr(current, '__dataclass_fields__'):
|
||||
deep_update(current, v)
|
||||
else:
|
||||
setattr(obj, k, v)
|
||||
|
||||
with self._lock:
|
||||
deep_update(self, updates)
|
||||
|
||||
if save:
|
||||
self._save()
|
||||
return self
|
||||
|
||||
def reload(self) -> 'AppConfig':
|
||||
"""热重载配置"""
|
||||
if self._config_path and self._config_path.exists():
|
||||
with self._lock:
|
||||
data = json.loads(self._config_path.read_text(encoding='utf-8'))
|
||||
|
||||
# 配置项名 -> dataclass 类的映射（和 _load 保持一致）
|
||||
config_classes = {
|
||||
'ai': AIConfig,
|
||||
'tts': TTSConfig,
|
||||
'asr': ASRConfig,
|
||||
'auto_agent': AutoAgentConfig,
|
||||
'llm_core': LLMConfig,
|
||||
'paths': PathsConfig,
|
||||
}
|
||||
|
||||
for k, v in data.items():
|
||||
if hasattr(self, k) and not k.startswith('_'):
|
||||
# 如果是配置项且是 dict，转换为 dataclass
|
||||
if k in config_classes and isinstance(v, dict):
|
||||
setattr(self, k, config_classes[k](**v))
|
||||
else:
|
||||
setattr(self, k, v)
|
||||
print(f"配置重载: {self._config_path}")
|
||||
return self
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""导出为字典（手动实现，排除内部属性）"""
|
||||
result = {
|
||||
'version': self.version,
|
||||
'debug': self.debug,
|
||||
'ai': asdict(self.ai) if hasattr(self.ai, '__dataclass_fields__') else self.ai,
|
||||
'tts': asdict(self.tts) if hasattr(self.tts, '__dataclass_fields__') else self.tts,
|
||||
'asr': asdict(self.asr) if hasattr(self.asr, '__dataclass_fields__') else self.asr,
|
||||
'auto_agent': asdict(self.auto_agent) if hasattr(self.auto_agent, '__dataclass_fields__') else self.auto_agent,
|
||||
'llm_core': asdict(self.llm_core) if hasattr(self.llm_core, '__dataclass_fields__') else self.llm_core,
|
||||
'paths': asdict(self.paths) if hasattr(self.paths, '__dataclass_fields__') else self.paths,
|
||||
}
|
||||
return result
|
||||
|
||||
def _save(self) -> None:
|
||||
"""保存到文件"""
|
||||
if self._config_path:
|
||||
with self._lock:
|
||||
json_str = json.dumps(self.to_dict(), indent=2, ensure_ascii=False)
|
||||
self._config_path.write_text(json_str, encoding='utf-8')
|
||||
|
||||
@classmethod
|
||||
def _load(cls, path: Path) -> 'AppConfig':
|
||||
"""从文件加载"""
|
||||
data = json.loads(path.read_text(encoding='utf-8'))
|
||||
|
||||
# 配置项名 -> dataclass 类的映射
|
||||
config_classes = {
|
||||
'ai': AIConfig,
|
||||
'tts': TTSConfig,
|
||||
'asr': ASRConfig,
|
||||
'auto_agent': AutoAgentConfig,
|
||||
'llm_core': LLMConfig,
|
||||
'paths': PathsConfig,
|
||||
}
|
||||
|
||||
# 自动转换 dict 为对应 dataclass
|
||||
for key, config_class in config_classes.items():
|
||||
if key in data and isinstance(data[key], dict):
|
||||
data[key] = config_class(**data[key])
|
||||
|
||||
return cls(_config_path=path, **data)
|
||||
|
||||
@classmethod
|
||||
def _create_default(cls, path: Path) -> 'AppConfig':
|
||||
"""创建默认配置"""
|
||||
instance = cls(_config_path=path)
|
||||
instance._save()
|
||||
print(f"默认配置已创建: {path}")
|
||||
return instance
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""友好的打印格式"""
|
||||
lines = ["AppConfig("]
|
||||
for k, v in self.to_dict().items():
|
||||
lines.append(f" {k}={v!r},")
|
||||
lines.append(")")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# 延迟初始化机制
|
||||
|
||||
_root: Path = _project_root()
|
||||
_config_path: Path = _root / "settings.json"
|
||||
_config_instance: Optional[AppConfig] = None
|
||||
_init_lock: threading.Lock = threading.Lock()
|
||||
|
||||
|
||||
def _ensure_initialized() -> AppConfig:
|
||||
"""
|
||||
确保配置已初始化（线程安全的延迟初始化）
|
||||
"""
|
||||
global _config_instance
|
||||
|
||||
if _config_instance is not None:
|
||||
return _config_instance
|
||||
|
||||
with _init_lock:
|
||||
if _config_instance is not None:
|
||||
return _config_instance
|
||||
|
||||
# 自动加载或创建
|
||||
if _config_path.exists():
|
||||
_config_instance = AppConfig._load(_config_path)
|
||||
print(f"配置加载: {_config_path}")
|
||||
else:
|
||||
_config_instance = AppConfig._create_default(_config_path)
|
||||
|
||||
# 确保目录存在
|
||||
for d in [_config_instance.temp_dir,
|
||||
_config_instance.log_dir,
|
||||
_config_instance.using_dir]:
|
||||
d.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
return _config_instance
|
||||
|
||||
|
||||
class _LazyConfig:
|
||||
"""
|
||||
配置代理类：拦截所有属性访问，第一次使用时自动初始化
|
||||
"""
|
||||
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
"""拦截属性访问，延迟初始化"""
|
||||
instance = _ensure_initialized()
|
||||
return getattr(instance, name)
|
||||
|
||||
def __setattr__(self, name: str, value: Any) -> None:
|
||||
"""拦截属性设置"""
|
||||
# 特殊属性直接设置到代理对象本身
|
||||
if name.startswith('_'):
|
||||
super().__setattr__(name, value)
|
||||
else:
|
||||
instance = _ensure_initialized()
|
||||
setattr(instance, name, value)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
instance = _ensure_initialized()
|
||||
return repr(instance)
|
||||
|
||||
def __dir__(self) -> list:
|
||||
"""支持 IDE 自动补全"""
|
||||
instance = _ensure_initialized()
|
||||
return dir(instance)
|
||||
|
||||
# 显式代理必要方法
|
||||
def get(self, key: str, default: Any = None) -> Any:
|
||||
return _ensure_initialized().get(key, default)
|
||||
|
||||
def set(self, key: str, value: Any, save: bool = True) -> AppConfig:
|
||||
return _ensure_initialized().set(key, value, save)
|
||||
|
||||
def update(self, updates: Dict[str, Any], save: bool = True) -> AppConfig:
|
||||
return _ensure_initialized().update(updates, save)
|
||||
|
||||
def reload(self) -> AppConfig:
|
||||
return _ensure_initialized().reload()
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return _ensure_initialized().to_dict()
|
||||
|
||||
def save(self) -> None:
|
||||
_ensure_initialized()._save()
|
||||
|
||||
# 属性代理
|
||||
@property
|
||||
def ai(self) -> AIConfig:
|
||||
return _ensure_initialized().ai
|
||||
|
||||
@property
|
||||
def tts(self) -> TTSConfig:
|
||||
return _ensure_initialized().tts
|
||||
|
||||
@property
|
||||
def asr(self) -> ASRConfig:
|
||||
return _ensure_initialized().asr
|
||||
|
||||
@property
|
||||
def auto_agent(self) -> AutoAgentConfig:
|
||||
return _ensure_initialized().auto_agent
|
||||
|
||||
@property
|
||||
def llm_core(self) -> LLMConfig:
|
||||
return _ensure_initialized().llm_core
|
||||
|
||||
@property
|
||||
def paths(self) -> PathsConfig:
|
||||
return _ensure_initialized().paths
|
||||
|
||||
@property
|
||||
def temp_dir(self) -> Path:
|
||||
return _ensure_initialized().temp_dir
|
||||
|
||||
@property
|
||||
def log_dir(self) -> Path:
|
||||
return _ensure_initialized().log_dir
|
||||
|
||||
@property
|
||||
def using_dir(self) -> Path:
|
||||
return _ensure_initialized().using_dir
|
||||
|
||||
|
||||
# 全局配置对象：导入即用，自动初始化
|
||||
cfg: AppConfig = _LazyConfig() # type: ignore
|
||||
|
||||
|
||||
# 工具函数
|
||||
|
||||
def generate_example(path: Path = _root / "settings.example.json") -> None:
|
||||
"""生成示例配置文件"""
|
||||
example = {
|
||||
"version": "1.0.0",
|
||||
"debug": False,
|
||||
"ai": {
|
||||
"api_key": "sk-your-api-key",
|
||||
"base_url": "https://api.deepseek.com",
|
||||
"model": "deepseek-chat",
|
||||
"timeout": 30
|
||||
},
|
||||
"tts": {
|
||||
"enabled": True,
|
||||
"model": "GPT_SoVITS",
|
||||
"url": "http://localhost:12458/",
|
||||
"reference_audio": "./using/reference.wav",
|
||||
"streaming": True,
|
||||
"speed": 1.0
|
||||
},
|
||||
"paths": {
|
||||
"temp": "./tmp/",
|
||||
"log": "./log/",
|
||||
"data": "./data/"
|
||||
}
|
||||
}
|
||||
path.write_text(json.dumps(example, indent=2, ensure_ascii=False), encoding='utf-8')
|
||||
print(f"示例配置已生成: {path}")
|
||||
|
||||
|
||||
# 测试代码
|
||||
if __name__ == "__main__":
|
||||
# 测试：直接访问，自动初始化
|
||||
print("第一次访问 cfg.ai.model:")
|
||||
print(f" → {cfg.ai.model_name}")
|
||||
|
||||
print(f"\n配置详情:")
|
||||
print(cfg)
|
||||
|
||||
print(f"\n测试修改:")
|
||||
cfg.set("ai.timeout", 60)
|
||||
print(f" ai.timeout = {cfg.ai.timeout}")
|
||||
|
||||
print(f"\n测试批量更新:")
|
||||
cfg.update({"debug": True, "tts": {"speed": 1.5}})
|
||||
print(f" debug = {cfg.debug}, tts.speed = {cfg.tts.speed}")
|
||||
|
||||
print(f"\n测试热重载:")
|
||||
cfg.reload()
|
||||
Reference in New Issue
Block a user