Files
pingtai_agent/steering_agent/src/server/extractors/config.py
T

223 lines
8.6 KiB
Python

"""
配置规范提取器
提取项目的配置规范:环境变量、配置文件、密钥处理等。
"""
import os
import re
from typing import Dict, List, Any, Optional
class ConfigExtractor:
"""配置规范提取器"""
def extract(self, project_path: str) -> Dict[str, Any]:
"""
提取项目的配置规范
Args:
project_path: 项目根目录路径
Returns:
配置规范信息字典
"""
python_files = self._find_python_files(project_path)
return {
"env_variables": self._extract_env_variables(python_files),
"config_files": self._analyze_config_files(project_path),
"secrets_handling": self._analyze_secrets_handling(python_files)
}
def _find_python_files(self, project_path: str) -> List[str]:
"""查找所有 Python 文件"""
python_files = []
ignore_dirs = {'__pycache__', '.git', '.venv', 'venv', 'node_modules', '.pytest_cache'}
for root, dirs, files in os.walk(project_path):
dirs[:] = [d for d in dirs if d not in ignore_dirs]
for f in files:
if f.endswith('.py'):
python_files.append(os.path.join(root, f))
return python_files
def _read_file_content(self, file_path: str) -> Optional[str]:
"""读取文件内容"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
return f.read()
except Exception:
return None
def _extract_env_variables(self, python_files: List[str]) -> List[Dict[str, Any]]:
"""从代码中提取使用的环境变量"""
env_vars = {}
# 匹配模式
patterns = [
# os.getenv('VAR', 'default')
(r"os\.getenv\s*\(\s*['\"](\w+)['\"]\s*(?:,\s*['\"]([^'\"]*)['\"])?\s*\)", 'getenv'),
# os.environ.get('VAR', 'default')
(r"os\.environ\.get\s*\(\s*['\"](\w+)['\"]\s*(?:,\s*['\"]([^'\"]*)['\"])?\s*\)", 'environ_get'),
# os.environ['VAR']
(r"os\.environ\s*\[\s*['\"](\w+)['\"]\s*\]", 'environ_bracket'),
# os.environ.setdefault('VAR', 'default')
(r"os\.environ\.setdefault\s*\(\s*['\"](\w+)['\"]\s*,\s*['\"]([^'\"]*)['\"]", 'setdefault'),
]
for file_path in python_files:
content = self._read_file_content(file_path)
if not content:
continue
for pattern, pattern_type in patterns:
matches = re.findall(pattern, content)
for match in matches:
if isinstance(match, tuple):
var_name = match[0]
default_value = match[1] if len(match) > 1 else None
else:
var_name = match
default_value = None
if var_name not in env_vars:
env_vars[var_name] = {
"name": var_name,
"required": pattern_type == 'environ_bracket',
"default": default_value if default_value else None,
"description": self._guess_env_description(var_name)
}
elif default_value and not env_vars[var_name]["default"]:
env_vars[var_name]["default"] = default_value
return list(env_vars.values())
def _guess_env_description(self, var_name: str) -> str:
"""根据变量名猜测描述"""
descriptions = {
"OPENAI_API_KEY": "OpenAI API Key",
"OPENAI_BASE_URL": "OpenAI API Base URL",
"API_KEY": "API Key",
"API_PORT": "服务端口",
"API_HOST": "服务主机地址",
"MODEL_NAME": "模型名称",
"LITELLM_MODEL": "LiteLLM 模型名称",
"LITELLM_GATEWAY_URL": "LiteLLM Gateway URL",
"LLM_BASE_URL": "LLM API Base URL",
"DATABASE_URL": "数据库连接 URL",
"REDIS_URL": "Redis 连接 URL",
"SECRET_KEY": "密钥",
"DEBUG": "调试模式",
"LOG_LEVEL": "日志级别",
}
if var_name in descriptions:
return descriptions[var_name]
# 根据命名模式猜测
if "KEY" in var_name or "SECRET" in var_name or "TOKEN" in var_name:
return "密钥/令牌"
elif "URL" in var_name or "HOST" in var_name:
return "服务地址"
elif "PORT" in var_name:
return "端口号"
elif "PATH" in var_name or "DIR" in var_name:
return "路径"
elif "NAME" in var_name:
return "名称"
elif "TIMEOUT" in var_name:
return "超时时间"
return "配置项"
def _analyze_config_files(self, project_path: str) -> List[Dict[str, Any]]:
"""分析配置文件"""
config_files = []
# 配置文件类型定义
config_types = {
"Dockerfile": ("docker", "Docker 容器配置"),
"docker-compose.yml": ("docker-compose", "Docker Compose 配置"),
"docker-compose.yaml": ("docker-compose", "Docker Compose 配置"),
"requirements.txt": ("python_deps", "Python 依赖"),
"requirements-dev.txt": ("python_deps_dev", "Python 开发依赖"),
"pyproject.toml": ("python_project", "Python 项目配置"),
"setup.py": ("python_setup", "Python 包配置"),
"setup.cfg": ("python_setup", "Python 包配置"),
"package.json": ("node_deps", "Node.js 依赖"),
".env": ("env", "环境变量"),
".env.example": ("env_example", "环境变量示例"),
"config.json": ("json_config", "JSON 配置"),
"config.yaml": ("yaml_config", "YAML 配置"),
"config.yml": ("yaml_config", "YAML 配置"),
".gitignore": ("git", "Git 忽略规则"),
"Makefile": ("make", "Make 构建配置"),
}
for filename, (file_type, purpose) in config_types.items():
file_path = os.path.join(project_path, filename)
if os.path.exists(file_path):
config_files.append({
"file": filename,
"type": file_type,
"purpose": purpose
})
return config_files
def _analyze_secrets_handling(self, python_files: List[str]) -> Dict[str, Any]:
"""分析密钥处理方式"""
hardcoded_secrets = False
env_based = False
# 检测硬编码密钥的模式
hardcoded_patterns = [
r'api_key\s*=\s*["\'][a-zA-Z0-9_-]{20,}["\']',
r'secret\s*=\s*["\'][a-zA-Z0-9_-]{20,}["\']',
r'password\s*=\s*["\'][^"\']{8,}["\']',
r'token\s*=\s*["\'][a-zA-Z0-9_-]{20,}["\']',
]
# 检测环境变量获取密钥的模式
env_patterns = [
r'os\.getenv\s*\(\s*["\'].*(?:KEY|SECRET|TOKEN|PASSWORD)',
r'os\.environ\.get\s*\(\s*["\'].*(?:KEY|SECRET|TOKEN|PASSWORD)',
r'os\.environ\s*\[\s*["\'].*(?:KEY|SECRET|TOKEN|PASSWORD)',
]
for file_path in python_files:
content = self._read_file_content(file_path)
if not content:
continue
# 检查硬编码
for pattern in hardcoded_patterns:
if re.search(pattern, content, re.IGNORECASE):
# 排除示例和测试
if 'example' not in file_path.lower() and 'test' not in file_path.lower():
hardcoded_secrets = True
break
# 检查环境变量方式
for pattern in env_patterns:
if re.search(pattern, content, re.IGNORECASE):
env_based = True
break
# 确定处理模式
if env_based and not hardcoded_secrets:
pattern = "os.getenv with fallback"
elif hardcoded_secrets:
pattern = "hardcoded (not recommended)"
else:
pattern = "unknown"
return {
"hardcoded_secrets": hardcoded_secrets,
"env_based": env_based,
"pattern": pattern,
"recommendation": "使用环境变量管理敏感信息" if hardcoded_secrets else None
}