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")