调试定时装饰器,优化代码

This commit is contained in:
liuwei
2025-03-20 15:12:31 +08:00
parent 27220e0357
commit 20b8ba0252

View File

@@ -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()