Initial commit

wangzhibo

2025-07-27 e0fb1b468f7d607ff78232d250271c46f603c3a9

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
import threading
import time
from datetime import datetime
from functools import partial
 
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger
from auto_media_publisher.tasks.task_class_config import TASKS_CONFIG
from auto_media_publisher.utils.logger import get_logger
from auto_media_publisher.utils.utils_database import _init_database
from datetime import datetime
 
class auto_media_publisherTaskSchedulerThread(threading.Thread):
    """
    核心任务调度类。通过读取 TASKS_CONFIG 中的配置的任务信息启动。
    使用：
    scheduler_thread = auto_media_publisherTaskSchedulerThread()
    scheduler_thread.daemon = True
    scheduler_thread.start()
    直接在一个 main.py 中启动， 也可用在 flask等 app 的类中启动。
    """
    def __init__(self):
        super().__init__()
        self.scheduler = BackgroundScheduler()
        self.tasks_config = TASKS_CONFIG
        self.task_instances = {}
        self.logger = get_logger(self.__class__.__name__,"admin")
        self._running = True
        _init_database()
 
    def _run_task(self, task_id, task):
        try:
            print(f"{datetime.now()} ▶️ 开始执行 {task_id}")
            task.run()
            print(f"{datetime.now()} ✅ 执行完毕 {task_id}")
            self.logger.info(f"任务执行完毕: {task_id}")
        except Exception as e:
            print(f"❌ {datetime.now()} 任务 {task_id} 执行异常: {e}")
            self.logger.exception(f"任务 {task_id} 异常: {e}")
            task.last_log += f"\n运行异常: {e}"
 
 
    def _setup_tasks(self):
        for item in TASKS_CONFIG:
            task_id = item.get("id")
            if not item.get("enabled"):
                self.logger.info(f"任务 {task_id} enabled 为 False,掠过")
                continue
            taskClass = item.get("task_class")
            handler = taskClass()
            self.task_instances[task_id] = handler
            trigger_type = item.get("trigger", "interval")
            trigger_args = item.get("trigger_args", {})
 
            job_func = partial(self._run_task, task_id, handler)
 
            if trigger_type == "interval":
                trigger = IntervalTrigger(**trigger_args)
                self.scheduler.add_job(
                    job_func,
                    trigger,
                    id=task_id,
                    next_run_time=datetime.now(),  # 立即执行一次
                    max_instances=1     # 如果上次 任务还未执行完毕，就不再创建新的Job了。
                )
            elif trigger_type == "cron":
                trigger = CronTrigger(**trigger_args)
                self.scheduler.add_job(job_func, trigger, id=task_id,max_instances=1)
            else:
                self.logger.error(f"不支持的 trigger 类型: {trigger_type}")
                raise ValueError(f"Invalid trigger type: {trigger_type}")
 
            self.logger.info(f"已添加任务 {task_id} 到调度器")
 
    def run(self):
        try:
            self.logger.info("auto_media_publisherTaskSchedulerThread 启动")
            self._setup_tasks()
            self.scheduler.start()
            while self._running:
                time.sleep(1)  # 保持线程活跃
        except Exception as e:
            self.logger.exception(f"任务  run setup_tasks 异常: {e}")
 
 
    def shutdown(self,wait=False):
        self._running = False
        self.scheduler.shutdown(wait)
        self.logger.info("auto_media_publisherTaskSchedulerThread 已停止")
 
    def get_task_info(self):
        print(self.task_instances)
        return {
            tid: {
                "last_run": str(t.last_run) if t.last_run else "Never",
                "last_log": t.last_log
            } for tid, t in self.task_instances.items()
        }