MCP Code Indexer
by zxfgds
- mcp_code_indexer
"""
配置管理模块
负责加载、验证和提供应用配置
"""
import os
import yaml
from typing import Dict, Any, Optional, List
from pathlib import Path
import logging
logger = logging.getLogger(__name__)
class Config:
"""
配置管理类
负责加载和管理应用配置，支持从配置文件、环境变量和默认值获取配置
"""
# 默认配置
DEFAULT_CONFIG = {
"server": {
"host": "127.0.0.1",
"port": 5000,
"debug": False,
"workers": 4
},
"indexer": {
"embedding_model": "sentence-transformers/all-MiniLM-L6-v2",
"batch_size": 32,
"max_tokens": 8192,
"chunk_size": 1000,
"chunk_overlap": 200
},
"storage": {
"vector_db_path": "./vector_db",
"project_data_path": "./project_data"
},
"exclude_patterns": [
"node_modules/",
"vendor/",
".git/",
"venv/",
"__pycache__/",
"dist/",
"build/",
".next/",
"target/",
".idea/",
".vscode/",
"logs/"
],
"file_extensions": [
".py", ".js", ".ts", ".java", ".c", ".cpp", ".h", ".hpp",
".cs", ".go", ".rb", ".php", ".swift", ".kt", ".rs", ".sh"
],
"max_file_size_kb": 1024,
"logging": {
"level": "INFO",
"file": "mcp_indexer.log"
}
}
def __init__(self, config_path: Optional[str] = None):
"""
初始化配置管理器
Args:
config_path: 配置文件路径，如果为None则使用默认配置
Returns:
无返回值
"""
self.config = self.DEFAULT_CONFIG.copy()
# 加载配置文件
if config_path and os.path.exists(config_path):
self._load_from_file(config_path)
# 从环境变量加载配置
self._load_from_env()
# 验证配置
self._validate_config()
logger.info(f"配置加载完成: {len(self.config)} 个配置项")
def _load_from_file(self, config_path: str) -> None:
"""
从YAML配置文件加载配置
Args:
config_path: 配置文件路径
Returns:
无返回值
"""
try:
with open(config_path, 'r', encoding='utf-8') as f:
file_config = yaml.safe_load(f)
if file_config:
self._merge_config(self.config, file_config)
logger.info(f"从配置文件加载配置: {config_path}")
except Exception as e:
logger.error(f"加载配置文件失败: {str(e)}")
def _load_from_env(self) -> None:
"""
从环境变量加载配置
环境变量格式: MCP_SECTION_KEY=value
Returns:
无返回值
"""
prefix = "MCP_"
for key, value in os.environ.items():
if key.startswith(prefix):
parts = key[len(prefix):].lower().split('_', 1)
if len(parts) == 2:
section, option = parts
if section in self.config:
if option in self.config[section]:
# 尝试转换值类型
try:
orig_type = type(self.config[section][option])
if orig_type == bool:
self.config[section][option] = value.lower() in ('true', 'yes', '1')
elif orig_type == int:
self.config[section][option] = int(value)
elif orig_type == float:
self.config[section][option] = float(value)
else:
self.config[section][option] = value
except Exception:
self.config[section][option] = value
logger.debug(f"从环境变量加载配置: {key}={value}")
def _merge_config(self, target: Dict[str, Any], source: Dict[str, Any]) -> None:
"""
递归合并配置字典
Args:
target: 目标配置字典
source: 源配置字典
Returns:
无返回值
"""
for key, value in source.items():
if key in target and isinstance(target[key], dict) and isinstance(value, dict):
self._merge_config(target[key], value)
else:
target[key] = value
def _validate_config(self) -> None:
"""
验证配置有效性
Returns:
无返回值
"""
# 确保必要的目录存在
os.makedirs(self.get("storage.vector_db_path"), exist_ok=True)
os.makedirs(self.get("storage.project_data_path"), exist_ok=True)
# 验证嵌入模型配置
embedding_model = self.get("indexer.embedding_model")
if not embedding_model:
logger.warning("未配置嵌入模型，将使用默认模型")
self.config["indexer"]["embedding_model"] = self.DEFAULT_CONFIG["indexer"]["embedding_model"]
def get(self, key_path: str, default: Any = None) -> Any:
"""
获取配置值
Args:
key_path: 配置键路径，格式为"section.key"
default: 默认值，如果配置不存在则返回此值
Returns:
配置值或默认值
"""
parts = key_path.split('.')
config = self.config
for part in parts:
if isinstance(config, dict) and part in config:
config = config[part]
else:
return default
return config
def set(self, key_path: str, value: Any) -> None:
"""
设置配置值
Args:
key_path: 配置键路径，格式为"section.key"
value: 要设置的值
Returns:
无返回值
"""
parts = key_path.split('.')
config = self.config
for i, part in enumerate(parts[:-1]):
if part not in config:
config[part] = {}
config = config[part]
config[parts[-1]] = value
def get_exclude_patterns(self) -> List[str]:
"""
获取排除模式列表
Returns:
排除模式字符串列表
"""
return self.get("exclude_patterns", [])
def get_file_extensions(self) -> List[str]:
"""
获取支持的文件扩展名列表
Returns:
文件扩展名字符串列表
"""
return self.get("file_extensions", [])
def save_to_file(self, file_path: str) -> bool:
"""
将当前配置保存到文件
Args:
file_path: 保存的文件路径
Returns:
保存是否成功
"""
try:
with open(file_path, 'w', encoding='utf-8') as f:
yaml.dump(self.config, f, default_flow_style=False)
return True
except Exception as e:
logger.error(f"保存配置文件失败: {str(e)}")
return False