65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
import functools
|
||
import inspect
|
||
import logging
|
||
from typing import Callable, Optional, Dict, List, Any
|
||
|
||
LOG = logging.getLogger("JobDecorator")
|
||
|
||
# 存储所有被装饰的任务
|
||
scheduled_tasks = []
|
||
|
||
def scheduled_job(cron: str, name: Optional[str] = None, enabled: bool = True):
|
||
"""
|
||
定时任务装饰器，用于标记需要定时执行的方法
|
||
|
||
:param cron: cron表达式，例如 "0 0 * * * *" 表示每小时执行一次
|
||
:param name: 任务名称，默认使用方法名
|
||
:param enabled: 是否启用该任务，默认为True
|
||
"""
|
||
def decorator(func):
|
||
task_name = name or func.__name__
|
||
|
||
# 记录任务信息
|
||
task_info = {
|
||
"func": func,
|
||
"cron": cron,
|
||
"name": task_name,
|
||
"enabled": enabled
|
||
}
|
||
scheduled_tasks.append(task_info)
|
||
|
||
LOG.info(f"注册定时任务: {task_name}, cron: {cron}, enabled: {enabled}")
|
||
|
||
@functools.wraps(func)
|
||
def wrapper(*args, **kwargs):
|
||
return func(*args, **kwargs)
|
||
|
||
return wrapper
|
||
|
||
return decorator
|
||
|
||
# 修改 register_scheduled_jobs 函数
|
||
|
||
def register_scheduled_jobs(job_instance):
|
||
"""注册所有标记了 @scheduled_job 装饰器的方法
|
||
|
||
Args:
|
||
job_instance: 包含定时任务的实例
|
||
"""
|
||
if not hasattr(job_instance, 'scheduled_jobs'):
|
||
return
|
||
|
||
for task in job_instance.scheduled_jobs:
|
||
method = getattr(job_instance, task["method_name"])
|
||
# 使用正确的方法添加任务
|
||
if hasattr(job_instance, 'add_job'):
|
||
job_instance.add_job(method, task["cron"], task["name"])
|
||
elif hasattr(job_instance, 'add_scheduled_job'):
|
||
job_instance.add_scheduled_job(method, task["cron"], task["name"])
|
||
else:
|
||
# 如果没有合适的方法，记录错误
|
||
import logging
|
||
logging.getLogger(__name__).error(
|
||
f"无法为 {job_instance.__class__.__name__} 添加定时任务，"
|
||
f"缺少 add_job 或 add_scheduled_job 方法"
|
||
) |