BBIT-Kai
132 lines
4.2 KiB
Python
132 lines
4.2 KiB
Python
"""
|
|
配置管理工具
|
|
"""
|
|
|
|
import os
|
|
import configparser
|
|
import uuid
|
|
from typing import Optional
|
|
from pathlib import Path
|
|
|
|
|
|
class ConfigManager:
|
|
"""配置管理器"""
|
|
|
|
def __init__(self, config_file: str = "data/.mcp-endpoint-server.cfg"):
|
|
self.config_file = config_file
|
|
self.config = configparser.ConfigParser()
|
|
self._load_config()
|
|
|
|
def _load_config(self):
|
|
"""加载配置文件"""
|
|
if os.path.exists(self.config_file):
|
|
self.config.read(self.config_file, encoding="utf-8")
|
|
# 检查并生成key
|
|
self._check_and_generate_key()
|
|
else:
|
|
# 如果配置文件不存在，从根目录拷贝
|
|
self._copy_config_from_root()
|
|
|
|
def _copy_config_from_root(self):
|
|
"""从根目录拷贝配置文件到data目录"""
|
|
root_config = "mcp-endpoint-server.cfg"
|
|
if os.path.exists(root_config):
|
|
# 确保data目录存在
|
|
os.makedirs(os.path.dirname(self.config_file), exist_ok=True)
|
|
|
|
# 拷贝配置文件
|
|
import shutil
|
|
|
|
shutil.copy2(root_config, self.config_file)
|
|
|
|
# 重新加载配置
|
|
self.config.read(self.config_file, encoding="utf-8")
|
|
# 检查并生成key
|
|
self._check_and_generate_key()
|
|
else:
|
|
# 如果根目录也没有配置文件，则创建默认配置
|
|
self._create_default_config()
|
|
|
|
def _check_and_generate_key(self):
|
|
"""检查key是否存在且长度足够，如果不足则生成新的"""
|
|
try:
|
|
current_key = self.config.get("server", "key", fallback="")
|
|
if not current_key or len(current_key) < 32:
|
|
# 生成32位随机密码
|
|
new_key = self._generate_random_key()
|
|
self.config.set("server", "key", new_key)
|
|
|
|
# 保存到配置文件
|
|
with open(self.config_file, "w", encoding="utf-8") as f:
|
|
self.config.write(f)
|
|
|
|
print(f"已自动生成新的32位密钥: {new_key}")
|
|
except Exception as e:
|
|
print(f"检查密钥时发生错误: {e}")
|
|
|
|
def _generate_random_key(self) -> str:
|
|
"""生成指定长度的随机密钥"""
|
|
# 使用UUID生成密钥，移除连字符
|
|
return str(uuid.uuid4()).replace("-", "")
|
|
|
|
def _create_default_config(self):
|
|
"""创建默认配置"""
|
|
self.config["server"] = {
|
|
"host": "127.0.0.1",
|
|
"port": "8004",
|
|
"debug": "false",
|
|
"log_level": "INFO",
|
|
"key": self._generate_random_key(), # 生成默认密钥
|
|
}
|
|
|
|
self.config["websocket"] = {
|
|
"max_connections": "1000",
|
|
"ping_interval": "30",
|
|
"ping_timeout": "10",
|
|
"close_timeout": "10",
|
|
}
|
|
|
|
self.config["security"] = {"allowed_origins": "*", "enable_cors": "true"}
|
|
|
|
self.config["logging"] = {
|
|
"log_file": "logs/mcp_server.log",
|
|
"max_file_size": "10MB",
|
|
"backup_count": "5",
|
|
}
|
|
|
|
# 确保目录存在
|
|
os.makedirs(os.path.dirname(self.config_file), exist_ok=True)
|
|
|
|
# 保存默认配置
|
|
with open(self.config_file, "w", encoding="utf-8") as f:
|
|
self.config.write(f)
|
|
|
|
def get(self, section: str, key: str, default: Optional[str] = None) -> str:
|
|
"""获取配置值"""
|
|
try:
|
|
return self.config.get(section, key)
|
|
except (configparser.NoSectionError, configparser.NoOptionError):
|
|
return default
|
|
|
|
def getint(self, section: str, key: str, default: int = 0) -> int:
|
|
"""获取整数配置值"""
|
|
try:
|
|
return self.config.getint(section, key)
|
|
except (configparser.NoSectionError, configparser.NoOptionError, ValueError):
|
|
return default
|
|
|
|
def getboolean(self, section: str, key: str, default: bool = False) -> bool:
|
|
"""获取布尔配置值"""
|
|
try:
|
|
return self.config.getboolean(section, key)
|
|
except (configparser.NoSectionError, configparser.NoOptionError, ValueError):
|
|
return default
|
|
|
|
def reload(self):
|
|
"""重新加载配置"""
|
|
self._load_config()
|
|
|
|
|
|
# 全局配置实例
|
|
config = ConfigManager()
|