shanxia/crm/py-backend/app/scheduler.py

import logging
from datetime import datetime, timedelta

from apscheduler.schedulers.background import BackgroundScheduler
from sqlalchemy import text

from app.database import get_db
from app.models.crm_customer import CrmCustomer, CrmCustomerPool
from app.models.crm_payment import CrmPaymentPlan
from app.models.crm_visit import CrmVisit
from app.utils import next_id

logger = logging.getLogger(__name__)

PROTECT_MIN_FOLLOW_COUNT = 3
PROTECT_REMIND_DAYS = 3
PAYMENT_REMIND_DAYS = 7


def _get_tenant_ids() -> list[str]:
    """Get all active tenant IDs."""
    db = next(get_db())
    try:
        rows = db.execute(text("SELECT id FROM sys_tenant WHERE status = '0' AND del_flag = '0'")).fetchall()
        return [str(r[0]) for r in rows]
    finally:
        db.close()


# ──────────────────────────────────────────────────────────────────────
# Customer Pool Tasks
# ──────────────────────────────────────────────────────────────────────

def check_protect_period_expired():
    """Check customers whose protect period has expired. Runs daily at 1am."""
    logger.info("======== Start: check protect period expired ========")
    db = next(get_db())
    try:
        now = datetime.now()
        tenants = _get_tenant_ids()
        recycle_count = 0
        qualified_count = 0

        for tenant_id in tenants:
            customers = db.query(CrmCustomer).filter(
                CrmCustomer.del_flag == 0,
                CrmCustomer.tenant_id == tenant_id,
                CrmCustomer.protect_time.isnot(None),
                CrmCustomer.protect_time <= now,
                CrmCustomer.status == 'normal',
                CrmCustomer.owner_user_id.isnot(None),
            ).all()

            for c in customers:
                follow_count = c.follow_count or 0
                if follow_count < PROTECT_MIN_FOLLOW_COUNT:
                    _recycle_to_pool(db, c, f"保护期到期未达标：跟进{follow_count}次，要求{PROTECT_MIN_FOLLOW_COUNT}次")
                    recycle_count += 1
                else:
                    c.protect_time = None
                    db.commit()
                    qualified_count += 1

        logger.info(f"======== Protect period check done: total={recycle_count + qualified_count}, qualified={qualified_count}, recycled={recycle_count} ========")
    except Exception as e:
        db.rollback()
        logger.error(f"Protect period check failed: {e}")
    finally:
        db.close()


def remind_protect_period():
    """Remind about customers whose protect period expires in 3 days. Runs daily at 9am."""
    logger.info("======== Start: remind protect period ========")
    db = next(get_db())
    try:
        now = datetime.now()
        deadline = now + timedelta(days=PROTECT_REMIND_DAYS)
        tenants = _get_tenant_ids()
        remind_count = 0

        for tenant_id in tenants:
            customers = db.query(CrmCustomer).filter(
                CrmCustomer.del_flag == 0,
                CrmCustomer.tenant_id == tenant_id,
                CrmCustomer.protect_time.isnot(None),
                CrmCustomer.protect_time.between(now, deadline),
                CrmCustomer.status == 'normal',
                CrmCustomer.owner_user_id.isnot(None),
            ).all()

            for c in customers:
                follow_count = c.follow_count or 0
                if follow_count < PROTECT_MIN_FOLLOW_COUNT:
                    logger.info(f"Remind owner [{c.owner_user_name}]: customer [{c.customer_name}] protect period expiring in {PROTECT_REMIND_DAYS} days, follow {follow_count}/{PROTECT_MIN_FOLLOW_COUNT}")
                    remind_count += 1

        logger.info(f"======== Protect period remind done: {remind_count} reminders ========")
    except Exception as e:
        logger.error(f"Protect period remind failed: {e}")
    finally:
        db.close()


def auto_recycle_customers():
    """Auto-recycle long-unfollowed customers to pool. Runs daily at 2am."""
    logger.info("======== Start: auto recycle customers ========")
    db = next(get_db())
    try:
        now = datetime.now()
        tenants = _get_tenant_ids()
        recycle_count = 0

        for tenant_id in tenants:
            # Rule 1: New customers created >15 days ago with 0 follows
            new_deadline = now - timedelta(days=15)
            new_customers = db.query(CrmCustomer).filter(
                CrmCustomer.del_flag == 0,
                CrmCustomer.tenant_id == tenant_id,
                CrmCustomer.create_time <= new_deadline,
                CrmCustomer.status == 'normal',
                CrmCustomer.owner_user_id.isnot(None),
                CrmCustomer.level != 'key',
            ).filter(
                (CrmCustomer.last_time.is_(None)) | (CrmCustomer.follow_count == 0)
            ).all()

            for c in new_customers:
                _recycle_to_pool(db, c, "自动回收：创建后15天未跟进")
                recycle_count += 1

            # Rule 2: Old customers last followed >30 days ago
            old_deadline = now - timedelta(days=30)
            old_customers = db.query(CrmCustomer).filter(
                CrmCustomer.del_flag == 0,
                CrmCustomer.tenant_id == tenant_id,
                CrmCustomer.last_time.isnot(None),
                CrmCustomer.last_time <= old_deadline,
                CrmCustomer.status == 'normal',
                CrmCustomer.owner_user_id.isnot(None),
                CrmCustomer.level != 'key',
            ).filter(
                (CrmCustomer.opportunity_count == 0) | (CrmCustomer.opportunity_count.is_(None))
            ).all()

            for c in old_customers:
                days = (now - c.last_time).days if c.last_time else 0
                _recycle_to_pool(db, c, f"自动回收：最后跟进后{days}天未再跟进")
                recycle_count += 1

        logger.info(f"======== Auto recycle done: {recycle_count} customers recycled ========")
    except Exception as e:
        db.rollback()
        logger.error(f"Auto recycle failed: {e}")
    finally:
        db.close()


def remind_auto_recycle():
    """Warn about customers approaching auto-recycle. Runs daily at 8am."""
    logger.info("======== Start: remind auto recycle ========")
    db = next(get_db())
    try:
        now = datetime.now()
        tenants = _get_tenant_ids()
        remind_count = 0

        for tenant_id in tenants:
            # Rule 1: New customers created >=12 days with 0 follows
            warn_deadline = now - timedelta(days=12)
            new_customers = db.query(CrmCustomer).filter(
                CrmCustomer.del_flag == 0,
                CrmCustomer.tenant_id == tenant_id,
                CrmCustomer.create_time <= warn_deadline,
                CrmCustomer.create_time >= now - timedelta(days=15),
                CrmCustomer.status == 'normal',
                CrmCustomer.owner_user_id.isnot(None),
                CrmCustomer.level != 'key',
            ).filter(
                (CrmCustomer.last_time.is_(None)) | (CrmCustomer.follow_count == 0)
            ).all()

            for c in new_customers:
                days = (now - c.create_time).days if c.create_time else 0
                remaining = 15 - days
                if 0 < remaining <= 3:
                    logger.info(f"Remind [{c.owner_user_name}]: customer [{c.customer_name}] created {days}d ago, {remaining}d until auto-recycle")
                    remind_count += 1

            # Rule 2: Old customers last followed 27-30 days ago
            old_customers = db.query(CrmCustomer).filter(
                CrmCustomer.del_flag == 0,
                CrmCustomer.tenant_id == tenant_id,
                CrmCustomer.last_time.isnot(None),
                CrmCustomer.last_time <= now - timedelta(days=27),
                CrmCustomer.last_time >= now - timedelta(days=30),
                CrmCustomer.status == 'normal',
                CrmCustomer.owner_user_id.isnot(None),
                CrmCustomer.level != 'key',
            ).all()

            for c in old_customers:
                days = (now - c.last_time).days if c.last_time else 0
                remaining = 30 - days
                if 0 < remaining <= 3:
                    logger.info(f"Remind [{c.owner_user_name}]: customer [{c.customer_name}] last follow {days}d ago, {remaining}d until auto-recycle")
                    remind_count += 1

        logger.info(f"======== Auto recycle remind done: {remind_count} reminders ========")
    except Exception as e:
        logger.error(f"Auto recycle remind failed: {e}")
    finally:
        db.close()


def _recycle_to_pool(db, customer: CrmCustomer, reason: str):
    """Move customer back to public pool."""
    original_owner_id = customer.owner_user_id
    original_owner_name = customer.owner_user_name

    customer.status = 'pool'
    customer.owner_user_id = None
    customer.owner_user_name = None
    customer.into_pool_time = datetime.now()
    customer.into_pool_by = 0
    customer.into_pool_reason = reason
    customer.protect_time = None
    customer.claim_time = None

    pool_record = CrmCustomerPool(
        id=next_id(),
        customer_id=customer.id,
        customer_name=customer.customer_name,
        operation_type='recycle',
        from_user_id=original_owner_id,
        from_user_name=original_owner_name,
        reason=reason,
        create_by=0,
        create_by_name='系统',
        create_time=datetime.now(),
        tenant_id=customer.tenant_id,
    )
    db.add(pool_record)
    db.commit()
    logger.info(f"Customer [{customer.customer_name}] recycled to pool, previous owner: {original_owner_name}, reason: {reason}")


# ──────────────────────────────────────────────────────────────────────
# Payment Plan Tasks
# ──────────────────────────────────────────────────────────────────────

def check_overdue_payment_plans():
    """Mark payment plans as overdue. Runs daily at 1am."""
    logger.info("======== Start: check overdue payment plans ========")
    db = next(get_db())
    try:
        today = datetime.now().date()
        tenants = _get_tenant_ids()
        updated = 0

        for tenant_id in tenants:
            plans = db.query(CrmPaymentPlan).filter(
                CrmPaymentPlan.del_flag == 0,
                CrmPaymentPlan.tenant_id == tenant_id,
                CrmPaymentPlan.plan_date < today,
                CrmPaymentPlan.status.in_(['pending', 'partial']),
                CrmPaymentPlan.is_overdue == 0,
            ).all()

            for p in plans:
                overdue_days = (today - p.plan_date).days
                p.is_overdue = 1
                p.overdue_days = overdue_days
                updated += 1
                level = "轻微" if overdue_days <= 7 else "一般" if overdue_days <= 15 else "严重" if overdue_days <= 30 else "恶意"
                logger.info(f"Overdue [{level}]: plan {p.plan_no}, customer {p.customer_name}, {overdue_days}d overdue")

        db.commit()
        logger.info(f"======== Overdue check done: {updated} plans updated ========")
    except Exception as e:
        db.rollback()
        logger.error(f"Overdue check failed: {e}")
    finally:
        db.close()


def send_payment_reminder():
    """Send payment reminders for plans due within 7 days. Runs daily at 9am."""
    logger.info("======== Start: send payment reminders ========")
    db = next(get_db())
    try:
        today = datetime.now().date()
        deadline = today + timedelta(days=PAYMENT_REMIND_DAYS)
        tenants = _get_tenant_ids()
        remind_count = 0

        for tenant_id in tenants:
            plans = db.query(CrmPaymentPlan).filter(
                CrmPaymentPlan.del_flag == 0,
                CrmPaymentPlan.tenant_id == tenant_id,
                CrmPaymentPlan.plan_date >= today,
                CrmPaymentPlan.plan_date <= deadline,
                CrmPaymentPlan.status.in_(['pending', 'partial']),
                CrmPaymentPlan.is_overdue == 0,
            ).all()

            for p in plans:
                days_left = (p.plan_date - today).days
                if p.remind_status != 'sent' or days_left == 0:
                    logger.info(f"Remind [{p.owner_user_name}]: plan {p.plan_no}, customer {p.customer_name}, {days_left}d until due, amount {p.plan_amount}")
                    p.remind_status = 'sent'
                    p.last_remind_time = datetime.now()
                    remind_count += 1

        db.commit()
        logger.info(f"======== Payment remind done: {remind_count} reminders ========")
    except Exception as e:
        db.rollback()
        logger.error(f"Payment remind failed: {e}")
    finally:
        db.close()


def reset_remind_status():
    """Reset remind status for expired plans. Runs daily at 2am."""
    logger.info("======== Start: reset remind status ========")
    db = next(get_db())
    try:
        today = datetime.now().date()
        tenants = _get_tenant_ids()
        reset_count = 0

        for tenant_id in tenants:
            plans = db.query(CrmPaymentPlan).filter(
                CrmPaymentPlan.del_flag == 0,
                CrmPaymentPlan.tenant_id == tenant_id,
                CrmPaymentPlan.plan_date < today,
                CrmPaymentPlan.remind_status == 'sent',
                CrmPaymentPlan.is_overdue == 0,
            ).all()

            for p in plans:
                p.remind_status = 'expired'
                reset_count += 1

        db.commit()
        logger.info(f"======== Remind status reset done: {reset_count} records ========")
    except Exception as e:
        db.rollback()
        logger.error(f"Remind status reset failed: {e}")
    finally:
        db.close()


# ──────────────────────────────────────────────────────────────────────
# Visit Reminder Task
# ──────────────────────────────────────────────────────────────────────

def send_visit_reminder():
    """Send reminders for upcoming visits. Runs daily at 9am."""
    logger.info("======== Start: send visit reminders ========")
    db = next(get_db())
    try:
        now = datetime.now()
        deadline = now + timedelta(days=7)
        tenants = _get_tenant_ids()
        remind_count = 0

        for tenant_id in tenants:
            visits = db.query(CrmVisit).filter(
                CrmVisit.del_flag == 0,
                CrmVisit.tenant_id == tenant_id,
                CrmVisit.status.in_(['0', 'pending']),
                CrmVisit.plan_visit_date.isnot(None),
                CrmVisit.plan_visit_date <= deadline,
            ).all()

            for v in visits:
                days_left = (v.plan_visit_date - now).days if v.plan_visit_date else 0
                urgency = "今日" if days_left == 0 else "明日" if days_left == 1 else f"{days_left}天后"
                logger.info(f"Remind [{v.owner_user_name}]: visit {v.visit_no}, customer {v.customer_name}, {urgency} ({v.visit_type})")
                remind_count += 1

        logger.info(f"======== Visit remind done: {remind_count} reminders ========")
    except Exception as e:
        logger.error(f"Visit remind failed: {e}")
    finally:
        db.close()


# ──────────────────────────────────────────────────────────────────────
# Scheduler Setup
# ──────────────────────────────────────────────────────────────────────

_scheduler: BackgroundScheduler | None = None


def start_scheduler():
    global _scheduler
    if _scheduler is not None:
        return

    _scheduler = BackgroundScheduler(timezone="Asia/Shanghai")

    # 1am: Protect period check + Overdue payment check
    _scheduler.add_job(check_protect_period_expired, 'cron', hour=1, minute=0, id='check_protect_period')
    _scheduler.add_job(check_overdue_payment_plans, 'cron', hour=1, minute=5, id='check_overdue_payment')

    # 2am: Auto recycle + Reset remind status
    _scheduler.add_job(auto_recycle_customers, 'cron', hour=2, minute=0, id='auto_recycle_customers')
    _scheduler.add_job(reset_remind_status, 'cron', hour=2, minute=5, id='reset_remind_status')

    # 8am: Auto recycle warning
    _scheduler.add_job(remind_auto_recycle, 'cron', hour=8, minute=0, id='remind_auto_recycle')

    # 9am: Protect period remind + Payment remind + Visit remind
    _scheduler.add_job(remind_protect_period, 'cron', hour=9, minute=0, id='remind_protect_period')
    _scheduler.add_job(send_payment_reminder, 'cron', hour=9, minute=5, id='send_payment_reminder')
    _scheduler.add_job(send_visit_reminder, 'cron', hour=9, minute=10, id='send_visit_reminder')

    _scheduler.start()
    logger.info("Scheduler started with 8 jobs")