config_mgmt.py 2.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283
  1. """
  2. 配置管理工具
  3. 实现配置查询和管理功能。
  4. """
  5. from typing import Dict, Optional, Any, TypedDict
  6. from ..services.data_service import DataService
  7. from ..utils.validators import validate_config_section
  8. from ..utils.errors import MCPError
  9. class ErrorInfo(TypedDict, total=False):
  10. """错误信息结构"""
  11. code: str
  12. message: str
  13. suggestion: str
  14. class ConfigResult(TypedDict):
  15. """配置查询结果 - success 字段必需,其他字段可选"""
  16. success: bool
  17. config: Optional[Dict[str, Any]]
  18. section: Optional[str]
  19. error: Optional[ErrorInfo]
  20. class ConfigManagementTools:
  21. """配置管理工具类"""
  22. def __init__(self, project_root: str = None):
  23. """
  24. 初始化配置管理工具
  25. Args:
  26. project_root: 项目根目录
  27. """
  28. self.data_service = DataService(project_root)
  29. def get_current_config(self, section: Optional[str] = None) -> ConfigResult:
  30. """
  31. 获取当前系统配置
  32. Args:
  33. section: 配置节 - all/crawler/push/keywords/weights,默认all
  34. Returns:
  35. 配置字典
  36. Example:
  37. >>> tools = ConfigManagementTools()
  38. >>> result = tools.get_current_config(section="crawler")
  39. >>> print(result['crawler']['platforms'])
  40. """
  41. try:
  42. # 参数验证
  43. section = validate_config_section(section)
  44. # 获取配置
  45. config = self.data_service.get_current_config(section=section)
  46. return ConfigResult(
  47. success=True,
  48. config=config,
  49. section=section,
  50. error=None
  51. )
  52. except MCPError as e:
  53. return ConfigResult(
  54. success=False,
  55. config=None,
  56. section=None,
  57. error=e.to_dict()
  58. )
  59. except Exception as e:
  60. return ConfigResult(
  61. success=False,
  62. config=None,
  63. section=None,
  64. error={"code": "INTERNAL_ERROR", "message": str(e), "suggestion": "请查看服务日志获取详细信息"}
  65. )