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