简道云V2.0
This commit is contained in:
+16
-5
@@ -1,17 +1,28 @@
|
||||
"""
|
||||
核心模块初始化
|
||||
统一初始化请求头管理器、模块注册表等核心组件
|
||||
|
||||
本模块统一初始化和管理核心组件，包括：
|
||||
- ModuleRegistry: 模块注册表
|
||||
- CoreManager: 核心管理器
|
||||
|
||||
提供统一的接口来管理这些核心组件。
|
||||
"""
|
||||
from typing import Dict, Any, Callable
|
||||
from app.core.header_manager import HeaderManager
|
||||
from app.core.module_registry import ModuleRegistry, registry
|
||||
|
||||
|
||||
class CoreManager:
|
||||
"""核心管理器 - 统一管理所有核心组件"""
|
||||
"""
|
||||
核心管理器
|
||||
|
||||
统一管理所有核心组件，提供便捷的方法来初始化和注册模块。
|
||||
|
||||
属性:
|
||||
registry: 模块注册表实例
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.header_manager = HeaderManager
|
||||
"""初始化核心管理器"""
|
||||
self.registry = registry
|
||||
|
||||
def initialize_modules(self, modules: Dict[str, Any]):
|
||||
@@ -49,12 +60,12 @@ class CoreManager:
|
||||
|
||||
|
||||
# 全局核心管理器实例
|
||||
# 在应用启动时使用此实例来注册模块和操作
|
||||
core_manager = CoreManager()
|
||||
|
||||
# 导出常用类和函数
|
||||
__all__ = [
|
||||
'core_manager',
|
||||
'HeaderManager',
|
||||
'ModuleRegistry',
|
||||
'registry',
|
||||
]
|
||||
|
||||
@@ -1,118 +0,0 @@
|
||||
"""
|
||||
请求头管理器
|
||||
统一管理不同模块的请求头配置
|
||||
"""
|
||||
from typing import Dict, Optional
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
|
||||
@dataclass
|
||||
class HeaderConfig:
|
||||
"""请求头配置"""
|
||||
referer: Optional[str] = None
|
||||
user_agent: Optional[str] = None
|
||||
content_type: Optional[str] = None
|
||||
authorization: Optional[str] = None
|
||||
custom_headers: Dict[str, str] = field(default_factory=dict)
|
||||
|
||||
def to_dict(self) -> Dict[str, str]:
|
||||
"""转换为字典格式"""
|
||||
headers = {}
|
||||
|
||||
if self.referer:
|
||||
headers['Referer'] = self.referer
|
||||
if self.user_agent:
|
||||
headers['User-Agent'] = self.user_agent
|
||||
if self.content_type:
|
||||
headers['Content-Type'] = self.content_type
|
||||
if self.authorization:
|
||||
headers['Authorization'] = self.authorization
|
||||
|
||||
# 添加自定义请求头
|
||||
headers.update(self.custom_headers)
|
||||
|
||||
return headers
|
||||
|
||||
|
||||
class HeaderManager:
|
||||
"""请求头管理器"""
|
||||
|
||||
# 默认请求头配置
|
||||
DEFAULT_HEADERS = HeaderConfig(
|
||||
referer='https://yunxiu.f6car.cn/erp/view/index.html',
|
||||
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36 Edg/130.0.0.0'
|
||||
)
|
||||
|
||||
# F6系统登录请求头
|
||||
F6_LOGIN_HEADERS = HeaderConfig(
|
||||
referer='https://yunxiu.f6car.com/kzf6/login/confirm',
|
||||
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36 Edg/129.0.0.0'
|
||||
)
|
||||
|
||||
# 简道云API请求头
|
||||
JIANDAOYUN_API_HEADERS = HeaderConfig(
|
||||
content_type='application/json',
|
||||
# authorization 应该从配置中获取，这里只是示例
|
||||
)
|
||||
|
||||
# 模块特定的请求头配置
|
||||
_module_headers: Dict[str, HeaderConfig] = {
|
||||
'default': DEFAULT_HEADERS,
|
||||
'f6_login': F6_LOGIN_HEADERS,
|
||||
'jiandaoyun_api': JIANDAOYUN_API_HEADERS,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def get_headers(cls, module_name: str = 'default', **overrides) -> Dict[str, str]:
|
||||
"""
|
||||
获取指定模块的请求头
|
||||
|
||||
Args:
|
||||
module_name: 模块名称
|
||||
**overrides: 覆盖的请求头配置
|
||||
|
||||
Returns:
|
||||
请求头字典
|
||||
"""
|
||||
# 获取基础配置
|
||||
config = cls._module_headers.get(module_name, cls.DEFAULT_HEADERS)
|
||||
|
||||
# 创建配置副本
|
||||
header_config = HeaderConfig(
|
||||
referer=overrides.get('referer', config.referer),
|
||||
user_agent=overrides.get('user_agent', config.user_agent),
|
||||
content_type=overrides.get('content_type', config.content_type),
|
||||
authorization=overrides.get('authorization', config.authorization),
|
||||
custom_headers={**config.custom_headers, **overrides.get('custom_headers', {})}
|
||||
)
|
||||
|
||||
return header_config.to_dict()
|
||||
|
||||
@classmethod
|
||||
def register_module_headers(cls, module_name: str, config: HeaderConfig):
|
||||
"""
|
||||
注册模块的请求头配置
|
||||
|
||||
Args:
|
||||
module_name: 模块名称
|
||||
config: 请求头配置
|
||||
"""
|
||||
cls._module_headers[module_name] = config
|
||||
|
||||
@classmethod
|
||||
def update_module_headers(cls, module_name: str, **updates):
|
||||
"""
|
||||
更新模块的请求头配置
|
||||
|
||||
Args:
|
||||
module_name: 模块名称
|
||||
**updates: 要更新的配置项
|
||||
"""
|
||||
if module_name in cls._module_headers:
|
||||
config = cls._module_headers[module_name]
|
||||
for key, value in updates.items():
|
||||
if hasattr(config, key):
|
||||
setattr(config, key, value)
|
||||
else:
|
||||
config.custom_headers[key] = value
|
||||
|
||||
@@ -1,6 +1,12 @@
|
||||
"""
|
||||
模块注册和路由管理
|
||||
提供统一的模块注册机制，方便添加新功能模块
|
||||
模块注册表模块
|
||||
|
||||
本模块提供统一的模块注册机制，用于管理所有业务模块和操作。
|
||||
使用注册表模式可以：
|
||||
- 统一管理所有操作
|
||||
- 方便添加新功能模块
|
||||
- 支持动态路由和操作查找
|
||||
- 提供操作元数据管理
|
||||
"""
|
||||
from typing import Dict, Callable, Optional, Any
|
||||
from dataclasses import dataclass
|
||||
@@ -8,7 +14,18 @@ from dataclasses import dataclass
|
||||
|
||||
@dataclass
|
||||
class ActionConfig:
|
||||
"""操作配置"""
|
||||
"""
|
||||
操作配置数据类
|
||||
|
||||
存储操作的配置信息，包括处理函数、所属模块、描述等。
|
||||
|
||||
属性:
|
||||
handler: 处理函数，执行具体业务逻辑
|
||||
module_name: 所属模块名称
|
||||
description: 操作描述，用于文档和日志
|
||||
requires_auth: 是否需要认证，默认 True
|
||||
header_module: 使用的请求头模块名称，可选
|
||||
"""
|
||||
handler: Callable # 处理函数
|
||||
module_name: str # 所属模块名称
|
||||
description: Optional[str] = None # 描述
|
||||
@@ -17,9 +34,19 @@ class ActionConfig:
|
||||
|
||||
|
||||
class ModuleRegistry:
|
||||
"""模块注册表"""
|
||||
"""
|
||||
模块注册表类
|
||||
|
||||
统一管理所有业务模块和操作的注册表。
|
||||
支持注册模块实例和操作，并提供查询功能。
|
||||
|
||||
属性:
|
||||
_actions: 操作字典，格式为 {操作名: ActionConfig}
|
||||
_modules: 模块字典，格式为 {模块名: 模块信息}
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
"""初始化模块注册表"""
|
||||
self._actions: Dict[str, ActionConfig] = {}
|
||||
self._modules: Dict[str, Dict[str, Any]] = {}
|
||||
|
||||
@@ -65,7 +92,12 @@ class ModuleRegistry:
|
||||
return self._actions.get(action_name)
|
||||
|
||||
def get_all_actions(self) -> Dict[str, ActionConfig]:
|
||||
"""获取所有注册的操作"""
|
||||
"""
|
||||
获取所有注册的操作
|
||||
|
||||
Returns:
|
||||
Dict[str, ActionConfig]: 所有注册的操作字典的副本
|
||||
"""
|
||||
return self._actions.copy()
|
||||
|
||||
def register_module(self, module_name: str, module_instance: Any, **metadata):
|
||||
@@ -113,4 +145,5 @@ class ModuleRegistry:
|
||||
|
||||
|
||||
# 全局模块注册表实例
|
||||
# 在应用启动时使用此实例来注册所有模块和操作
|
||||
registry = ModuleRegistry()
|
||||
|
||||
Reference in New Issue
Block a user