200 lines
6.1 KiB
Python
200 lines
6.1 KiB
Python
"""
|
|
配置文件加载和验证工具
|
|
"""
|
|
|
|
import yaml
|
|
from pathlib import Path
|
|
from typing import Dict, Any
|
|
|
|
|
|
def load_config(config_path: str) -> Dict[str, Any]:
|
|
"""
|
|
加载配置文件
|
|
|
|
Args:
|
|
config_path: 配置文件路径
|
|
|
|
Returns:
|
|
配置字典
|
|
|
|
Raises:
|
|
FileNotFoundError: 配置文件不存在
|
|
yaml.YAMLError: 配置文件格式错误
|
|
"""
|
|
config_file = Path(config_path)
|
|
|
|
if not config_file.exists():
|
|
raise FileNotFoundError(f"配置文件不存在: {config_path}")
|
|
|
|
try:
|
|
with open(config_file, 'r', encoding='utf-8') as f:
|
|
config = yaml.safe_load(f)
|
|
return config
|
|
except yaml.YAMLError as e:
|
|
raise yaml.YAMLError(f"配置文件格式错误: {str(e)}")
|
|
|
|
|
|
def validate_config(config: Dict[str, Any]) -> None:
|
|
"""
|
|
验证配置文件的完整性和有效性
|
|
|
|
Args:
|
|
config: 配置字典
|
|
|
|
Raises:
|
|
ValueError: 配置验证失败
|
|
"""
|
|
# 检查必需的顶级配置块
|
|
required_sections = ['ai', 'feishu', 'system']
|
|
for section in required_sections:
|
|
if section not in config:
|
|
raise ValueError(f"缺少必需的配置块: {section}")
|
|
|
|
# 验证AI配置
|
|
ai_config = config['ai']
|
|
if not ai_config.get('api_key'):
|
|
raise ValueError("AI配置缺少 api_key")
|
|
if not ai_config.get('base_url'):
|
|
raise ValueError("AI配置缺少 base_url")
|
|
if not ai_config.get('model'):
|
|
raise ValueError("AI配置缺少 model")
|
|
|
|
# 验证飞书配置
|
|
feishu_config = config['feishu']
|
|
if not feishu_config.get('app_id'):
|
|
raise ValueError("飞书配置缺少 app_id")
|
|
if not feishu_config.get('app_secret'):
|
|
raise ValueError("飞书配置缺少 app_secret")
|
|
if not feishu_config.get('app_token'):
|
|
raise ValueError("飞书配置缺少 app_token")
|
|
if not feishu_config.get('table_id'):
|
|
raise ValueError("飞书配置缺少 table_id")
|
|
|
|
# 验证系统配置
|
|
system_config = config['system']
|
|
if not system_config.get('watch_folder'):
|
|
raise ValueError("系统配置缺少 watch_folder")
|
|
|
|
# 验证后处理配置
|
|
post_process = system_config.get('post_process', 'keep')
|
|
valid_post_process = ['delete', 'move', 'keep']
|
|
if post_process not in valid_post_process:
|
|
raise ValueError(f"post_process 必须是 {valid_post_process} 之一，当前值: {post_process}")
|
|
|
|
if post_process == 'move' and not system_config.get('processed_folder'):
|
|
raise ValueError("当 post_process 为 'move' 时，必须配置 processed_folder")
|
|
|
|
# 验证路径格式
|
|
watch_folder = Path(system_config['watch_folder'])
|
|
if not watch_folder.is_absolute():
|
|
# 如果是相对路径，转换为绝对路径
|
|
watch_folder = Path.cwd() / watch_folder
|
|
|
|
# 检查监控文件夹是否存在（允许不存在，会在运行时创建）
|
|
if watch_folder.exists() and not watch_folder.is_dir():
|
|
raise ValueError(f"watch_folder 必须是一个目录: {watch_folder}")
|
|
|
|
# 如果配置了移动目标文件夹，检查其路径
|
|
if post_process == 'move':
|
|
processed_folder = Path(system_config['processed_folder'])
|
|
if not processed_folder.is_absolute():
|
|
processed_folder = Path.cwd() / processed_folder
|
|
|
|
if processed_folder.exists() and not processed_folder.is_dir():
|
|
raise ValueError(f"processed_folder 必须是一个目录: {processed_folder}")
|
|
|
|
|
|
def save_config(config: Dict[str, Any], config_path: str) -> None:
|
|
"""
|
|
保存配置文件
|
|
|
|
Args:
|
|
config: 配置字典
|
|
config_path: 配置文件路径
|
|
"""
|
|
config_file = Path(config_path)
|
|
|
|
# 确保目录存在
|
|
config_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
with open(config_file, 'w', encoding='utf-8') as f:
|
|
yaml.dump(config, f, default_flow_style=False, allow_unicode=True)
|
|
|
|
|
|
def create_default_config() -> Dict[str, Any]:
|
|
"""
|
|
创建默认配置
|
|
|
|
Returns:
|
|
默认配置字典
|
|
"""
|
|
return {
|
|
'ai': {
|
|
'api_key': 'sk-xxxxxxxxxxxxxxxxxxxxxxxx',
|
|
'base_url': 'https://api.siliconflow.cn/v1',
|
|
'model': 'gemini-2.5-pro'
|
|
},
|
|
'feishu': {
|
|
'app_id': 'cli_xxxxxxxxxxxx',
|
|
'app_secret': 'xxxxxxxxxxxxxxxxxxxxxxxx',
|
|
'app_token': 'bascnxxxxxxxxxxxxxxx',
|
|
'table_id': 'tblxxxxxxxxxxxx'
|
|
},
|
|
'system': {
|
|
'watch_folder': './monitor_images',
|
|
'post_process': 'move',
|
|
'processed_folder': './processed_images'
|
|
}
|
|
}
|
|
|
|
|
|
def create_sample_config(config_path: str = 'config.example.yaml') -> None:
|
|
"""
|
|
创建示例配置文件
|
|
|
|
Args:
|
|
config_path: 示例配置文件路径
|
|
"""
|
|
default_config = create_default_config()
|
|
save_config(default_config, config_path)
|
|
print(f"示例配置文件已创建: {config_path}")
|
|
|
|
|
|
def get_config_summary(config: Dict[str, Any]) -> str:
|
|
"""
|
|
获取配置摘要信息
|
|
|
|
Args:
|
|
config: 配置字典
|
|
|
|
Returns:
|
|
配置摘要字符串
|
|
"""
|
|
summary = []
|
|
summary.append("配置摘要:")
|
|
summary.append(f" AI模型: {config['ai']['model']}")
|
|
summary.append(f" AI服务: {config['ai']['base_url']}")
|
|
summary.append(f" 监控文件夹: {config['system']['watch_folder']}")
|
|
summary.append(f" 后处理方式: {config['system'].get('post_process', 'keep')}")
|
|
|
|
if config['system'].get('post_process') == 'move':
|
|
summary.append(f" 处理后文件夹: {config['system']['processed_folder']}")
|
|
|
|
return '\n'.join(summary)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 测试配置验证
|
|
try:
|
|
# 创建示例配置
|
|
create_sample_config('config.example.yaml')
|
|
print("示例配置创建成功!")
|
|
|
|
# 加载并验证示例配置
|
|
config = load_config('config.example.yaml')
|
|
validate_config(config)
|
|
print("配置验证通过!")
|
|
print(get_config_summary(config))
|
|
|
|
except Exception as e:
|
|
print(f"配置测试失败: {e}") |