63 lines
1.9 KiB
Python
63 lines
1.9 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
|
||
|
||
def register_scheduled_jobs(job_instance):
|
||
"""
|
||
注册所有被装饰的定时任务到Job实例
|
||
|
||
:param job_instance: Job类实例，必须有add_job方法
|
||
:return: 注册的任务数量
|
||
"""
|
||
count = 0
|
||
|
||
# 获取实例的所有方法
|
||
for name, method in inspect.getmembers(job_instance, predicate=inspect.ismethod):
|
||
# 检查原始函数是否在scheduled_tasks中
|
||
original_func = method.__func__
|
||
|
||
for task in scheduled_tasks:
|
||
if task["func"].__name__ == original_func.__name__ and task["enabled"]:
|
||
# 注册任务
|
||
job_instance.add_job(method, task["cron"], task["name"])
|
||
count += 1
|
||
break
|
||
|
||
LOG.info(f"已注册 {count} 个定时任务")
|
||
return count |