diff --git a/job_decorators.py b/job_decorators.py index e61bccd..514ac21 100644 --- a/job_decorators.py +++ b/job_decorators.py @@ -6,10 +6,86 @@ from typing import Callable, Optional, Dict, List, Any LOG = logging.getLogger("JobDecorator") -# 存储所有被装饰的任务 -scheduled_tasks = [] -# 存储所有实例的弱引用 -instances = [] +# 任务注册表类,管理所有任务和实例 +class TaskRegistry: + _instance = None + + @classmethod + def get_instance(cls): + if cls._instance is None: + cls._instance = TaskRegistry() + return cls._instance + + def __init__(self): + if TaskRegistry._instance is not None: + raise RuntimeError("请使用 TaskRegistry.get_instance() 获取实例") + + self.tasks = [] # 存储所有被装饰的任务 + self.instances = [] # 存储所有实例的弱引用 + + def register_task(self, task_info): + """注册任务信息""" + self.tasks.append(task_info) + + def register_instance(self, instance): + """注册实例到实例列表""" + # 检查实例是否已注册 + for ref in self.instances: + if ref() is instance: + return + + # 添加实例的弱引用 + self.instances.append(weakref.ref(instance)) + + # 清理已失效的弱引用 + self.instances = [ref for ref in self.instances if ref() is not None] + + # 尝试为这个实例注册任务 + self.register_tasks_for_instance(instance) + + def register_tasks_for_instance(self, instance): + """为指定实例注册任务""" + class_name = instance.__class__.__name__ + + for task in self.tasks: + # 检查任务是否属于这个类 + if task["class_name"] == class_name and hasattr(instance, task["method_name"]): + method = getattr(instance, task["method_name"]) + + # 注册任务 + if hasattr(instance, 'add_job'): + LOG.info(f"为 {class_name} 添加定时任务: {task['method_name']}, cron: {task['cron']}") + instance.add_job(method, task["cron"], task["name"]) + elif hasattr(instance, 'onEveryTime'): + # 将cron表达式转换为时间字符串 (简化处理,仅支持每天固定时间) + cron_parts = task["cron"].split() + if len(cron_parts) >= 2: + minute, hour = cron_parts[0], cron_parts[1] + # 移除可能的前导0 + if minute.startswith('0') and len(minute) > 1: + minute = minute[1:] + if hour.startswith('0') and len(hour) > 1: + hour = hour[1:] + time_str = f"{hour}:{minute}" + LOG.info(f"为 {class_name} 添加定时任务: {task['method_name']}, 时间: {time_str}") + instance.onEveryTime(time_str, method) + + def initialize(self): + """初始化任务系统,扫描所有已创建的实例并注册任务""" + LOG.info("初始化任务系统") + + # 清理已失效的弱引用 + self.instances = [ref for ref in self.instances if ref() is not None] + + # 为所有实例注册任务 + for ref in self.instances: + instance = ref() + if instance: + self.register_tasks_for_instance(instance) + + +# 获取任务注册表实例 +registry = TaskRegistry.get_instance() def scheduled_job(cron: str, name: Optional[str] = None, enabled: bool = True): @@ -33,7 +109,7 @@ def scheduled_job(cron: str, name: Optional[str] = None, enabled: bool = True): "method_name": func.__name__, "class_name": func.__qualname__.split('.')[0] if '.' in func.__qualname__ else None } - scheduled_tasks.append(task_info) + registry.register_task(task_info) LOG.info(f"注册定时任务: {task_name}, cron: {cron}, enabled: {enabled}") @@ -41,7 +117,7 @@ def scheduled_job(cron: str, name: Optional[str] = None, enabled: bool = True): def wrapper(*args, **kwargs): # 如果是实例方法的第一次调用,注册实例 if args and not isinstance(args[0], type) and hasattr(args[0], '__class__'): - register_instance(args[0]) + registry.register_instance(args[0]) return func(*args, **kwargs) return wrapper @@ -49,62 +125,6 @@ def scheduled_job(cron: str, name: Optional[str] = None, enabled: bool = True): return decorator -def register_instance(instance): - """注册实例到全局实例列表""" - # 检查实例是否已注册 - for ref in instances: - if ref() is instance: - return - - # 添加实例的弱引用 - instances.append(weakref.ref(instance)) - - # 清理已失效的弱引用 - global instances - instances = [ref for ref in instances if ref() is not None] - - # 尝试为这个实例注册任务 - register_tasks_for_instance(instance) - - -def register_tasks_for_instance(instance): - """为指定实例注册任务""" - class_name = instance.__class__.__name__ - - for task in scheduled_tasks: - # 检查任务是否属于这个类 - if task["class_name"] == class_name and hasattr(instance, task["method_name"]): - method = getattr(instance, task["method_name"]) - - # 注册任务 - if hasattr(instance, 'add_job'): - LOG.info(f"为 {class_name} 添加定时任务: {task['method_name']}, cron: {task['cron']}") - instance.add_job(method, task["cron"], task["name"]) - elif hasattr(instance, 'onEveryTime'): - # 将cron表达式转换为时间字符串 (简化处理,仅支持每天固定时间) - cron_parts = task["cron"].split() - if len(cron_parts) >= 2: - minute, hour = cron_parts[0], cron_parts[1] - # 移除可能的前导0 - if minute.startswith('0') and len(minute) > 1: - minute = minute[1:] - if hour.startswith('0') and len(hour) > 1: - hour = hour[1:] - time_str = f"{hour}:{minute}" - LOG.info(f"为 {class_name} 添加定时任务: {task['method_name']}, 时间: {time_str}") - instance.onEveryTime(time_str, method) - - def initialize_job_system(): """初始化任务系统,扫描所有已创建的实例并注册任务""" - LOG.info("初始化任务系统") - - # 清理已失效的弱引用 - global instances - instances = [ref for ref in instances if ref() is not None] - - # 为所有实例注册任务 - for ref in instances: - instance = ref() - if instance: - register_tasks_for_instance(instance) + registry.initialize() \ No newline at end of file