shanxia2.0/backend/app/tasks/__init__.py

"""定时任务模块"""
import logging

from apscheduler.schedulers.background import BackgroundScheduler

logger = logging.getLogger(__name__)
scheduler = BackgroundScheduler()


def init_scheduled_tasks(get_db_session):
    """初始化定时任务"""
    if scheduler.running:
        return
    scheduler.start()
    logger.info("定时任务调度器已启动")

    from app.tasks.customer_pool import (
        check_customer_protect_time,
        pool_recovery_job,
        remind_customer_recycle,
        warn_customer_recycle,
    )
    from app.tasks.db_auto_plan import auto_generate_monthly_plans_job
    from app.tasks.pl_cert_expiry import scan_pl_cert_expiry_job
    from app.tasks.pl_personnel_age import refresh_personnel_ages_job

    scheduler.add_job(
        check_customer_protect_time,
        "cron", hour=1, minute=0,
        id="check_customer_protect_time",
        replace_existing=True,
    )
    scheduler.add_job(
        refresh_personnel_ages_job,
        "cron", hour=1, minute=0,
        id="refresh_personnel_ages",
        replace_existing=True,
    )
    scheduler.add_job(
        auto_generate_monthly_plans_job,
        "cron", day=1, hour=1, minute=30,
        id="db_auto_generate_monthly_plans",
        replace_existing=True,
    )
    scheduler.add_job(
        remind_customer_recycle,
        "cron", hour=9, minute=0,
        id="remind_customer_recycle",
        replace_existing=True,
    )
    scheduler.add_job(
        pool_recovery_job,
        "cron", hour=2, minute=0,
        id="pool_recovery_job",
        replace_existing=True,
    )
    scheduler.add_job(
        warn_customer_recycle,
        "cron", hour=10, minute=0,
        id="warn_customer_recycle",
        replace_existing=True,
    )
    scheduler.add_job(
        scan_pl_cert_expiry_job,
        "cron", hour=8, minute=0,
        id="scan_pl_cert_expiry",
        replace_existing=True,
    )
    logger.info(
        "已注册定时任务: check_customer_protect_time(01:00), "
        "refresh_personnel_ages(01:00), "
        "db_auto_generate_monthly_plans(每月1日01:30), "
        "pool_recovery_job(02:00), remind_customer_recycle(09:00), "
        "warn_customer_recycle(10:00), scan_pl_cert_expiry(08:00)"
    )


def shutdown_scheduler():
    if scheduler.running:
        scheduler.shutdown(wait=False)
        logger.info("定时任务调度器已关闭")


__all__ = [
    "scheduler",
    "init_scheduled_tasks",
    "shutdown_scheduler",
]