#!/usr/bin/env python3 # -*- coding: utf-8 -*- import logging import asyncio import signal from pathlib import Path from typing import Dict, Any, List import yaml import os logger = logging.getLogger(__name__) def _get_script_processes() -> List: """获取全局自动启动脚本进程列表（模块级单例）""" if not hasattr(_get_script_processes, "_procs"): _get_script_processes._procs = [] return _get_script_processes._procs class InitService: """初始化服务""" def __init__(self, config_path: str = "config/framework"): self.config_path = Path(config_path) self.configs: Dict[str, Any] = {} self.script_processes: List = _get_script_processes() logger.debug("InitService初始化开始") async def initialize_framework(self): """初始化框架""" try: logger.info("开始初始化框架") # 1. 加载配置 await self._load_configs() # 2. 创建必要目录 await self._create_directories() # 3. 验证初始化状态 await self._validate_init() logger.info("框架初始化完成") return self.configs except Exception as e: logger.error(f"框架初始化失败: {str(e)}", exc_info=True) raise async def start_auto_scripts(self): """启动配置中声明的自动启动脚本（在日志服务就绪后调用）""" try: cfg = self.configs.get("base", {}) scripts_cfg = cfg.get("auto_start_scripts", {}) if not scripts_cfg.get("enabled", True): logger.info("自动启动脚本已禁用") return entries = scripts_cfg.get("scripts", []) if not entries: logger.debug("没有配置自动启动脚本") return logger.info(f"检查自动启动脚本 ({len(entries)} 个)...") for entry in entries: if not entry.get("enabled", True): logger.debug(f" 跳过已禁用的脚本: {entry.get('name', entry.get('path', '?'))}") continue script_path = os.path.expanduser(entry["path"]) if not os.path.isfile(script_path): logger.info( f" 自动启动脚本不存在，跳过: {entry.get('name', script_path)}" ) continue script_args = entry.get("args", []) script_name = entry.get("name", os.path.basename(script_path)) cwd = entry.get("cwd") or os.path.dirname(script_path) try: proc = await asyncio.create_subprocess_exec( "python3" if script_path.endswith(".py") else script_path, script_path if script_path.endswith(".py") else None, *script_args, cwd=cwd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, preexec_fn=os.setsid, ) # Filter out None (when script_path is the executable itself) self.script_processes.append( {"name": script_name, "proc": proc, "path": script_path} ) logger.info(f"✅ 自动启动脚本已拉起: {script_name} (pid={proc.pid})") except Exception as start_err: logger.warning( f"⚠️ 自动启动脚本失败 {script_name}: {start_err}" ) except Exception as e: logger.warning(f"自动启动脚本扫描出错 (不影响框架): {e}") async def _load_configs(self): """加载配置文件""" try: logger.debug("开始加载配置文件") if not self.config_path.exists(): logger.warning(f"配置路径不存在: {self.config_path}，将创建默认配置") self.config_path.mkdir(parents=True, exist_ok=True) # 加载基础配置 base_config_file = self.config_path / "base_config.yaml" if base_config_file.exists(): with open(base_config_file, 'r', encoding='utf-8') as f: loaded = yaml.safe_load(f) or {} # 合并模板中新增的字段 (不影响用户已修改的值) template_file = self.config_path / "base_config.yaml.example" if template_file.exists(): try: with open(template_file, 'r', encoding='utf-8') as tf: template = yaml.safe_load(tf) or {} def _deep_merge(base, tpl): for k, v in tpl.items(): if k not in base: base[k] = v elif isinstance(v, dict) and isinstance(base.get(k), dict): _deep_merge(base[k], v) _deep_merge(loaded, template) self._save_config(base_config_file, loaded) except Exception: pass self.configs['base'] = loaded logger.debug("基础配置加载成功") else: # 从模板复制 template = self.config_path / "base_config.yaml.example" if template.exists(): import shutil shutil.copy(template, base_config_file) logger.info(f"📋 从模板创建配置文件: {base_config_file}") with open(base_config_file, 'r', encoding='utf-8') as f: self.configs['base'] = yaml.safe_load(f) else: logger.warning("基础配置文件不存在，使用默认配置") self.configs['base'] = self._get_default_base_config() self._save_config(base_config_file, self.configs['base']) # 加载权限规则 permission_file = self.config_path / "permission_rules.yaml" if permission_file.exists(): with open(permission_file, 'r', encoding='utf-8') as f: self.configs['permission_rules'] = yaml.safe_load(f) logger.debug("权限规则配置加载成功") else: logger.warning("权限规则文件不存在，使用默认配置") self.configs['permission_rules'] = self._get_default_permission_rules() self._save_config(permission_file, self.configs['permission_rules']) logger.debug(f"配置文件加载完成，共加载 {len(self.configs)} 个配置集") except Exception as e: logger.error(f"加载配置文件时出错: {str(e)}", exc_info=True) raise async def _create_directories(self): """创建必要目录""" try: logger.debug("开始创建必要目录") directories = [ "config/plugins", "config/services", "config/permissions", "logs/runtime", "logs/debug", "plugins", "utils" ] for dir_path in directories: path = Path(dir_path) path.mkdir(parents=True, exist_ok=True) logger.debug(f"创建目录: {dir_path}") logger.debug("目录创建完成") except Exception as e: logger.error(f"创建目录时出错: {str(e)}", exc_info=True) raise async def _validate_init(self): """验证初始化状态""" try: logger.debug("开始验证初始化状态") required_configs = ['base', 'permission_rules'] for config_name in required_configs: if config_name not in self.configs: logger.error(f"缺少必要配置: {config_name}") raise ValueError(f"缺少必要配置: {config_name}") required_dirs = ['config', 'logs', 'plugins'] for dir_name in required_dirs: if not Path(dir_name).exists(): logger.error(f"必要目录不存在: {dir_name}") raise ValueError(f"必要目录不存在: {dir_name}") logger.debug("初始化状态验证通过") except Exception as e: logger.error(f"验证初始化状态时出错: {str(e)}", exc_info=True) raise def _get_default_base_config(self) -> Dict: """获取默认基础配置""" return { 'framework': { 'name': 'SenSu', 'version': 'Alpha_0.2.0', 'debug': True }, 'logging': { 'level': 'INFO', 'debug_level_file': True, 'max_log_files': 20, 'max_file_size': '10MB' }, 'tui': { 'layout': { 'grid-rows': '4fr 5fr 1fr' } }, 'services': { 'internet': { 'ws_port': 8765, 'api_port': 8000, 'enable_reverse_proxy': False } }, 'plugins': { 'auto_load': True, 'hot_reload': True, 'max_retry_count': 3, 'isolation': False, # 默认不隔离 (需要 Web 页面的插件必须非隔离) }, 'auto_start_scripts': { 'enabled': True, 'scripts': [ # 示例: 自动拉起 cyrene_debug_server.py # {'name': 'cyrene_debug', 'path': '~/cyrene_debug_server.py', # 'enabled': True, 'args': [], 'cwd': '~'}, ] } } def _get_default_permission_rules(self) -> Dict: """获取默认权限规则""" return { 'permission_levels': ['read', 'write', 'execute', 'admin'], 'default_permissions': [ 'framework.status.read', 'plugin.self.info.read' ], 'admin_permissions': [ 'framework.*', 'plugin.*', 'service.*' ] } def _save_config(self, file_path: Path, config: Dict): """保存配置到文件""" try: with open(file_path, 'w', encoding='utf-8') as f: yaml.dump(config, f, default_flow_style=False, allow_unicode=True) logger.debug(f"配置保存到: {file_path}") except Exception as e: logger.error(f"保存配置到 {file_path} 时出错: {str(e)}", exc_info=True) def get_config(self, config_name: str) -> Dict: """获取配置""" try: config = self.configs.get(config_name) if not config: logger.error(f"配置不存在: {config_name}") raise ValueError(f"配置 {config_name} 不存在") logger.debug(f"获取配置: {config_name}") return config except Exception as e: logger.error(f"获取配置 {config_name} 时出错: {str(e)}", exc_info=True) raise def shutdown(self): """关闭初始化服务""" try: logger.info("关闭初始化服务") # 终止所有自动启动的脚本 for entry in self.script_processes: proc = entry.get("proc") if proc and proc.returncode is None: try: os.killpg(os.getpgid(proc.pid), signal.SIGTERM) logger.info(f" 已终止: {entry.get('name', '?')} (pid={proc.pid})") except ProcessLookupError: pass except Exception as kill_err: logger.warning( f" 终止失败 {entry.get('name', '?')}: {kill_err}" ) self.script_processes.clear() self.configs.clear() logger.debug("初始化服务关闭完成") except Exception as e: logger.error(f"关闭初始化服务时出错: {str(e)}", exc_info=True)