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

"""CRM 客户公海池定时任务"""
import logging
from datetime import datetime, timedelta

from sqlalchemy.orm import Session

from app.core.database import SessionLocal
from app.core.utils import generate_uuid
from app.models.crm_models import CrmCustomer, CrmCustomerPool, CrmCustomerPoolRule

logger = logging.getLogger(__name__)

PROTECT_MIN_FOLLOW_COUNT = 3
PROTECT_REMIND_DAYS = 3


def check_customer_protect_time():
    """每天凌晨1点：检查保护期到期客户，跟进次数不足则退回公海"""
    db: Session = SessionLocal()
    try:
        logger.info("======== 开始执行保护期到期检查任务 ========")

        now = datetime.now()
        customers = (
            db.query(CrmCustomer)
            .filter(
                CrmCustomer.del_flag == 0,
                CrmCustomer.status == "normal",
                CrmCustomer.protect_time.isnot(None),
                CrmCustomer.protect_time <= now,
                CrmCustomer.owner_user_id.isnot(None),
            )
            .all()
        )

        logger.info("查询到 %d 个保护期到期的客户", len(customers))
        recycle_count = 0
        qualified_count = 0

        for customer in customers:
            try:
                follow_count = customer.follow_count or 0
                if follow_count < PROTECT_MIN_FOLLOW_COUNT:
                    _recycle_to_pool(db, customer, f"保护期到期未达标：跟进次数{follow_count}次，要求{PROTECT_MIN_FOLLOW_COUNT}次")
                    recycle_count += 1
                else:
                    customer.protect_time = None
                    customer.update_time = now
                    db.flush()
                    qualified_count += 1
            except Exception:
                logger.exception("处理客户【%s】保护期到期失败", customer.customer_name)

        db.commit()
        logger.info(
            "======== 保护期到期检查任务完成：共 %d 个，达标 %d 个，退回公海 %d 个 ========",
            len(customers), qualified_count, recycle_count,
        )
    except Exception:
        db.rollback()
        logger.exception("保护期到期检查任务异常")
    finally:
        db.close()


def remind_customer_recycle():
    """每天早上9点：保护期到期提醒（3天内到期且跟进不足）"""
    db: Session = SessionLocal()
    try:
        logger.info("======== 开始执行保护期到期提醒任务 ========")

        now = datetime.now()
        remind_deadline = now + timedelta(days=PROTECT_REMIND_DAYS)

        customers = (
            db.query(CrmCustomer)
            .filter(
                CrmCustomer.del_flag == 0,
                CrmCustomer.status == "normal",
                CrmCustomer.protect_time.isnot(None),
                CrmCustomer.protect_time > now,
                CrmCustomer.protect_time <= remind_deadline,
                CrmCustomer.owner_user_id.isnot(None),
            )
            .all()
        )

        logger.info("查询到 %d 个客户保护期即将到期", len(customers))
        remind_count = 0

        for customer in customers:
            try:
                follow_count = customer.follow_count or 0
                if follow_count < PROTECT_MIN_FOLLOW_COUNT:
                    remaining_days = (customer.protect_time - now).days
                    logger.info(
                        "提醒负责人【%s】：客户【%s】保护期还剩%d天，当前跟进%d次，需达到%d次",
                        customer.owner_user_name, customer.customer_name,
                        remaining_days, follow_count, PROTECT_MIN_FOLLOW_COUNT,
                    )
                    remind_count += 1
            except Exception:
                logger.exception("提醒客户【%s】保护期到期失败", customer.customer_name)

        logger.info("======== 保护期到期提醒任务完成：共提醒 %d 个客户 ========", remind_count)
    except Exception:
        logger.exception("保护期到期提醒任务异常")
    finally:
        db.close()


def pool_recovery_job():
    """每天凌晨2点：执行公海池规则自动回收"""
    db: Session = SessionLocal()
    try:
        logger.info("======== 开始执行公海池规则自动回收任务 ========")

        from app.crud.crm.pool_rule import _execute_single_rule

        tenants = [r[0] for r in db.query(CrmCustomerPoolRule.tenant_no).filter(
            CrmCustomerPoolRule.del_flag == 0,
            CrmCustomerPoolRule.status == "enable",
        ).distinct().all()]

        total = 0
        for tenant_no in tenants:
            rules = (
                db.query(CrmCustomerPoolRule)
                .filter(
                    CrmCustomerPoolRule.tenant_no == tenant_no,
                    CrmCustomerPoolRule.status == "enable",
                    CrmCustomerPoolRule.del_flag == 0,
                )
                .order_by(CrmCustomerPoolRule.sort.asc())
                .all()
            )
            for rule in rules:
                try:
                    total += _execute_single_rule(db, tenant_no, rule)
                except Exception:
                    logger.exception("执行公海规则失败: %s", rule.pool_name)

        db.commit()
        logger.info("======== 公海池规则自动回收任务完成：共回收 %d 个客户 ========", total)
    except Exception:
        db.rollback()
        logger.exception("公海池规则自动回收任务异常")
    finally:
        db.close()


def warn_customer_recycle():
    """每天上午10点：即将被回收客户预警"""
    db: Session = SessionLocal()
    try:
        logger.info("======== 开始执行即将被回收客户预警任务 ========")

        now = datetime.now()
        tenants = [r[0] for r in db.query(CrmCustomerPoolRule.tenant_no).filter(
            CrmCustomerPoolRule.del_flag == 0,
            CrmCustomerPoolRule.status == "enable",
        ).distinct().all()]

        warn_count = 0
        for tenant_no in tenants:
            rules = (
                db.query(CrmCustomerPoolRule)
                .filter(
                    CrmCustomerPoolRule.tenant_no == tenant_no,
                    CrmCustomerPoolRule.status == "enable",
                    CrmCustomerPoolRule.del_flag == 0,
                )
                .all()
            )
            for rule in rules:
                try:
                    warn_days = max(3, rule.timeout_days - 3)
                    warn_date = now + timedelta(days=warn_days)

                    if rule.rule_type == "no_follow":
                        at_risk = (
                            db.query(CrmCustomer)
                            .filter(
                                CrmCustomer.tenant_no == tenant_no,
                                CrmCustomer.del_flag == 0,
                                CrmCustomer.status == "normal",
                                CrmCustomer.deal_status != "dealed",
                                CrmCustomer.last_time.isnot(None),
                                CrmCustomer.last_time <= warn_date,
                            )
                            .all()
                        )
                    elif rule.rule_type == "no_deal":
                        at_risk = (
                            db.query(CrmCustomer)
                            .filter(
                                CrmCustomer.tenant_no == tenant_no,
                                CrmCustomer.del_flag == 0,
                                CrmCustomer.status == "normal",
                                CrmCustomer.deal_status == "not_deal",
                                CrmCustomer.claim_time.isnot(None),
                                CrmCustomer.claim_time <= warn_date,
                            )
                            .all()
                        )
                    else:
                        continue

                    for c in at_risk:
                        logger.info(
                            "预警：客户【%s】即将被规则【%s】回收，负责人【%s】",
                            c.customer_name, rule.pool_name, c.owner_user_name,
                        )
                        warn_count += 1
                except Exception:
                    logger.exception("预警规则【%s】执行失败", rule.pool_name)

        logger.info("======== 即将被回收客户预警任务完成：共 %d 个客户 ========", warn_count)
    except Exception:
        logger.exception("即将被回收客户预警任务异常")
    finally:
        db.close()


def _recycle_to_pool(db: Session, customer: CrmCustomer, reason: str):
    """将客户退回公海池"""
    now = datetime.now()
    original_owner_id = customer.owner_user_id or ""
    original_owner_name = customer.owner_user_name or ""

    customer.status = "pool"
    customer.owner_user_id = None
    customer.owner_user_name = None
    customer.into_pool_time = now
    customer.into_pool_by = None
    customer.into_pool_reason = reason
    customer.protect_time = None
    customer.claim_time = None
    customer.update_time = now

    pool_record = CrmCustomerPool(
        id=generate_uuid(),
        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,
        tenant_no=customer.tenant_no,
        create_by=None,
        create_by_name="系统",
        create_time=now,
        update_time=now,
    )
    db.add(pool_record)
    db.flush()