Coverage for src\core\config.py: 0%

1"""

2配置管理模块

3管理 PAO 系统的配置和设置

4"""

6import os

7import json

8import logging

9from pathlib import Path

10from typing import Dict, Any, Optional

11from dataclasses import dataclass, field, asdict

12from enum import Enum

14import yaml

15from pydantic import BaseModel, Field

18class LogLevel(str, Enum):

19 """日志级别枚举"""

20 DEBUG = "DEBUG"

21 INFO = "INFO"

22 WARNING = "WARNING"

23 ERROR = "ERROR"

24 CRITICAL = "CRITICAL"

27class NetworkMode(str, Enum):

28 """网络模式枚举"""

29 LAN_ONLY = "lan_only" # 仅局域网

30 P2P_RELAY = "p2p_relay" # P2P带中继

31 CLOUD_SYNC = "cloud_sync" # 云同步

34class PrivacyLevel(str, Enum):

35 """隐私级别枚举"""

36 MINIMAL = "minimal" # 最小隐私，共享基本信息

37 STANDARD = "standard" # 标准隐私，加密通信

38 STRICT = "strict" # 严格隐私，本地优先

39 MAXIMUM = "maximum" # 最大隐私，无网络

42@dataclass

43class NetworkConfig:

44 """网络配置"""

45 mode: NetworkMode = NetworkMode.LAN_ONLY

46 discovery_port: int = 8765

47 data_port: int = 8766

48 relay_servers: list = field(default_factory=lambda: [

49 "relay1.pao-system.dev:8767",

50 "relay2.pao-system.dev:8767"

51 ])

52 max_connections: int = 10

53 connection_timeout: int = 30 # 秒

54 heartbeat_interval: int = 60 # 秒

55 enable_upnp: bool = True

56 enable_ipv6: bool = True

59@dataclass

60class SecurityConfig:

61 """安全配置"""

62 privacy_level: PrivacyLevel = PrivacyLevel.STANDARD

63 encryption_enabled: bool = True

64 encryption_algorithm: str = "AES256-GCM"

65 require_authentication: bool = True

66 authentication_timeout: int = 300 # 秒

67 max_auth_attempts: int = 3

68 enable_firewall: bool = True

69 allowed_networks: list = field(default_factory=lambda: [

70 "192.168.0.0/16",

71 "10.0.0.0/8",

72 "172.16.0.0/12"

73 ])

76@dataclass

77class StorageConfig:

78 """存储配置"""

79 type: str = "sqlite" # 存储类型

80 path: str = field(default_factory=lambda: str(Path.home() / ".pao" / "data" / "pao.db"))

81 data_dir: str = field(default_factory=lambda: str(Path.home() / ".pao" / "data"))

82 table_name: str = "memories"

83 max_storage_mb: int = 1024 # MB

84 backup_enabled: bool = True

85 backup_interval: int = 3600 # 秒

86 backup_count: int = 7

87 compression_enabled: bool = True

88 encryption_at_rest: bool = True

91@dataclass

92class MemoryConfig:

93 """记忆系统配置"""

94 enable_persistent_memory: bool = True

95 memory_ttl_days: int = 30 # 记忆保存天数

96 max_memory_items: int = 10000

97 memory_compression: bool = True

98 enable_semantic_search: bool = True

99 semantic_search_threshold: float = 0.7

100 enable_context_association: bool = True

101 context_window_size: int = 10

102

103

104@dataclass

105class LoggingConfig:

106 """日志配置"""

107 level: LogLevel = LogLevel.INFO

108 enable_file_logging: bool = True

109 log_dir: str = field(default_factory=lambda: str(Path.home() / ".pao" / "logs"))

110 max_log_size_mb: int = 10

111 log_backup_count: int = 5

112 enable_json_logs: bool = False

113 enable_console_logging: bool = True

114

115

116@dataclass

117class PerformanceConfig:

118 """性能配置"""

119 max_workers: int = 4

120 worker_timeout: int = 300 # 秒

121 cache_size_mb: int = 100

122 enable_gzip: bool = True

123 enable_connection_pooling: bool = True

124 pool_size: int = 5

125 pool_timeout: int = 30 # 秒

126

127

128class PAOConfig(BaseModel):

129 """PAO 系统主配置"""

130

131 # 基本配置

132 device_name: str = Field(default_factory=lambda: f"pao-{os.getlogin()}")

133 device_id: str = Field(default_factory=lambda: str(hash(os.getlogin() + os.environ.get('COMPUTERNAME', ''))))

134 version: str = "0.1.0"

135

136 # 子系统配置

137 network: NetworkConfig = Field(default_factory=NetworkConfig)

138 security: SecurityConfig = Field(default_factory=SecurityConfig)

139 storage: StorageConfig = Field(default_factory=StorageConfig)

140 memory: MemoryConfig = Field(default_factory=MemoryConfig)

141 logging: LoggingConfig = Field(default_factory=LoggingConfig)

142 performance: PerformanceConfig = Field(default_factory=PerformanceConfig)

143

144 # 功能开关

145 enable_discovery: bool = True

146 enable_sync: bool = True

147 enable_memory_sharing: bool = True

148 enable_skill_evolution: bool = False # 第一阶段不启用

149 enable_context_awareness: bool = False # 第一阶段不启用

150

151 # 高级设置

152 auto_start: bool = False

153 debug_mode: bool = False

154 developer_mode: bool = False

155

156 class Config:

157 json_encoders = {

158 # 处理枚举类型

159 LogLevel: lambda v: v.value,

160 NetworkMode: lambda v: v.value,

161 PrivacyLevel: lambda v: v.value

162 }

163

164

165class ConfigManager:

166 """配置管理器"""

167

168 def __init__(self, config_dir: Optional[str] = None):

169 if config_dir is None:

170 config_dir = str(Path.home() / ".pao")

171

172 self.config_dir = Path(config_dir)

173 self.config_file = self.config_dir / "config.yaml"

174 self.secrets_file = self.config_dir / "secrets.yaml"

175

176 # 确保目录存在

177 self.config_dir.mkdir(parents=True, exist_ok=True)

178

179 # 加载配置

180 self.config = self._load_or_create_config()

181 self.secrets = self._load_or_create_secrets()

182

183 def _load_or_create_config(self) -> PAOConfig:

184 """加载或创建配置"""

185 if self.config_file.exists():

186 try:

187 with open(self.config_file, 'r', encoding='utf-8') as f:

188 config_data = yaml.safe_load(f) or {}

189 return PAOConfig(**config_data)

190 except Exception as e:

191 logging.warning(f"加载配置文件失败，使用默认配置: {e}")

192 return PAOConfig()

193 else:

194 config = PAOConfig()

195 self.save_config(config)

196 return config

197

198 def _load_or_create_secrets(self) -> Dict[str, Any]:

199 """加载或创建密钥文件"""

200 if self.secrets_file.exists():

201 try:

202 with open(self.secrets_file, 'r', encoding='utf-8') as f:

203 return yaml.safe_load(f) or {}

204 except Exception as e:

205 logging.warning(f"加载密钥文件失败: {e}")

206 return {}

207 else:

208 secrets = self._generate_default_secrets()

209 self.save_secrets(secrets)

210 return secrets

211

212 def _generate_default_secrets(self) -> Dict[str, Any]:

213 """生成默认密钥"""

214 import secrets

215 import base64

216

217 return {

218 "encryption_key": base64.b64encode(secrets.token_bytes(32)).decode('utf-8'),

219 "auth_token": secrets.token_hex(32),

220 "device_secret": secrets.token_hex(16),

221 "relay_tokens": {}

222 }

223

224 def save_config(self, config: Optional[PAOConfig] = None) -> None:

225 """保存配置"""

226 if config is None:

227 config = self.config

228

229 # 转换为字典

230 config_dict = config.dict()

231

232 # 保存到文件

233 with open(self.config_file, 'w', encoding='utf-8') as f:

234 yaml.dump(config_dict, f, default_flow_style=False, allow_unicode=True, indent=2)

235

236 logging.info(f"配置已保存到: {self.config_file}")

237

238 def save_secrets(self, secrets: Optional[Dict[str, Any]] = None) -> None:

239 """保存密钥"""

240 if secrets is None:

241 secrets = self.secrets

242

243 # 保存到文件

244 with open(self.secrets_file, 'w', encoding='utf-8') as f:

245 yaml.dump(secrets, f, default_flow_style=False, allow_unicode=True)

246

247 # 设置文件权限（仅在Unix系统）

248 if os.name != 'nt':

249 os.chmod(self.secrets_file, 0o600)

250

251 logging.info(f"密钥已保存到: {self.secrets_file}")

252

253 def update_config(self, updates: Dict[str, Any]) -> PAOConfig:

254 """更新配置"""

255 # 深度合并更新

256 import copy

257

258 current_dict = self.config.dict()

259 updated_dict = self._deep_merge(current_dict, updates)

260

261 # 创建新配置

262 self.config = PAOConfig(**updated_dict)

263 self.save_config()

264

265 return self.config

266

267 def _deep_merge(self, base: Dict[str, Any], updates: Dict[str, Any]) -> Dict[str, Any]:

268 """深度合并字典"""

269 result = copy.deepcopy(base)

270

271 for key, value in updates.items():

272 if key in result and isinstance(result[key], dict) and isinstance(value, dict):

273 result[key] = self._deep_merge(result[key], value)

274 else:

275 result[key] = value

276

277 return result

278

279 def get_config_path(self, *args) -> Path:

280 """获取配置目录下的路径"""

281 return self.config_dir.joinpath(*args)

282

283 def get_data_path(self, *args) -> Path:

284 """获取数据目录下的路径"""

285 data_dir = Path(self.config.storage.data_dir)

286 data_dir.mkdir(parents=True, exist_ok=True)

287 return data_dir.joinpath(*args)

288

289 def get_log_path(self, *args) -> Path:

290 """获取日志目录下的路径"""

291 log_dir = Path(self.config.logging.log_dir)

292 log_dir.mkdir(parents=True, exist_ok=True)

293 return log_dir.joinpath(*args)

294

295 def setup_logging(self) -> None:

296 """设置日志系统"""

297 import sys

298

299 log_config = self.config.logging

300

301 # 配置根日志记录器

302 root_logger = logging.getLogger()

303 root_logger.setLevel(getattr(logging, log_config.level.value))

304

305 # 清除现有处理器

306 root_logger.handlers.clear()

307

308 # 控制台处理器

309 if log_config.enable_console_logging:

310 console_handler = logging.StreamHandler(sys.stdout)

311 console_handler.setLevel(getattr(logging, log_config.level.value))

312 console_format = logging.Formatter(

313 '%(asctime)s - %(name)s - %(levelname)s - %(message)s'

314 )

315 console_handler.setFormatter(console_format)

316 root_logger.addHandler(console_handler)

317

318 # 文件处理器

319 if log_config.enable_file_logging:

320 log_file = self.get_log_path("pao.log")

321 file_handler = logging.FileHandler(log_file, encoding='utf-8')

322 file_handler.setLevel(getattr(logging, log_config.level.value))

323

324 if log_config.enable_json_logs:

325 import json_log_formatter

326 formatter = json_log_formatter.JSONFormatter()

327 else:

328 formatter = logging.Formatter(

329 '%(asctime)s - %(name)s - %(levelname)s - %(message)s'

330 )

331

332 file_handler.setFormatter(formatter)

333 root_logger.addHandler(file_handler)

334

335

336def load_config(config_dir: Optional[str] = None) -> PAOConfig:

337 """快速加载配置"""

338 manager = ConfigManager(config_dir)

339 return manager.config

340

341

342def get_config_manager(config_dir: Optional[str] = None) -> ConfigManager:

343 """获取配置管理器实例"""

344 return ConfigManager(config_dir)

345

346

347def create_default_config() -> PAOConfig:

348 """创建默认配置"""

349 return PAOConfig()

350

351

352if __name__ == "__main__":

353 # 测试配置管理

354 config = create_default_config()

355 print("默认配置:", json.dumps(config.dict(), indent=2, ensure_ascii=False))

356

357 # 测试配置管理器

358 manager = get_config_manager()

359 print(f"配置目录: {manager.config_dir}")