mtsd/backend/app/tasks/delivery_warning.py

"""配送时效预警定时任务

依据订单分配时写入的配送截止时间（biz_delivery_task.effective_time）及区域表 sys_region.delivery_warn_advance_minutes（缺省 30 分钟）生成运营端预警：

- 进入「截止时间前 N 分钟」窗口且仍未完成配送：生成一条 advance_deadline 预警；
- 已超过截止时间仍未完成：生成 timeout 预警；
- 配送员确认送达（任务状态 delivering）或任务取消/完成时，由业务侧自动结案预警。

建议：每 1～5 分钟执行一次 check_delivery_warnings(db)。
"""
import logging
from datetime import datetime
from typing import Optional

from sqlalchemy.orm import Session

from app.models.models import BizDeliveryTask, BizDeliveryWarning, SysRegion
from app.core.utils import generate_id
from app.crud.delivery_warning_resolve import auto_resolve_open_warnings_for_task

logger = logging.getLogger(__name__)

DEFAULT_ADVANCE_MINUTES = 30


class DeliveryWarningTask:
    """配送时效预警任务"""

    def __init__(self, db: Session):
        self.db = db

    def check_delivery_warnings(self) -> dict:
        """检查配送任务并生成预警"""
        result = {
            "checked": 0,
            "warnings_created": 0,
            "notifications_sent": 0,
            "errors": []
        }

        try:
            tasks = (
                self.db.query(BizDeliveryTask)
                .filter(BizDeliveryTask.is_del == 0)
                .all()
            )
            result["checked"] = len(tasks)

            for task in tasks:
                try:
                    if task.status in ("delivering", "completed", "cancelled"):
                        auto_resolve_open_warnings_for_task(self.db, task.id)
                        continue

                    if task.status not in ("pending", "assigned", "picking"):
                        continue

                    warnings = self._check_task_warnings(task)
                    result["warnings_created"] += len(warnings)

                    for warning in warnings:
                        self._send_warning_notification(warning)
                        result["notifications_sent"] += 1

                except Exception as e:
                    logger.error(f"处理任务 {task.id} 时出错: {str(e)}")
                    result["errors"].append({
                        "task_id": task.id,
                        "error": str(e)
                    })

            self.db.commit()

        except Exception as e:
            logger.error(f"配送时效预警任务执行失败: {str(e)}")
            self.db.rollback()
            result["errors"].append({"task_id": "all", "error": str(e)})

        return result

    def _region_advance_minutes(self, region_id: str) -> int:
        reg = self.db.query(SysRegion).filter(SysRegion.id == region_id, SysRegion.is_del == 0).first()
        if not reg:
            return DEFAULT_ADVANCE_MINUTES
        m = getattr(reg, "delivery_warn_advance_minutes", None)
        if m is None or int(m) < 1:
            return DEFAULT_ADVANCE_MINUTES
        return int(m)

    def _check_task_warnings(self, task: BizDeliveryTask) -> list:
        warnings = []

        if not task.effective_time:
            return warnings

        now = datetime.now()
        remaining_minutes = int((task.effective_time - now).total_seconds() // 60)
        advance = self._region_advance_minutes(task.region_id)

        warning_type = None
        urgency_level = None
        message = None
        if remaining_minutes < 0:
            warning_type = "timeout"
            urgency_level = "已超时"
            message = f"已超过配送时效 {abs(remaining_minutes)} 分钟，仍未完成配送"
        elif remaining_minutes <= advance:
            warning_type = "advance_deadline"
            urgency_level = "时效预警"
            message = f"距配送截止不足 {advance} 分钟（剩余约 {remaining_minutes} 分钟），请督促配送"
        else:
            return warnings

        w = self._upsert_task_warning(
            task=task,
            warning_type=warning_type,
            urgency_level=urgency_level,
            remaining_minutes=remaining_minutes,
            message=message,
        )
        if w:
            warnings.append(w)

        return warnings

    def _upsert_task_warning(
        self,
        task: BizDeliveryTask,
        warning_type: str,
        urgency_level: str,
        remaining_minutes: int,
        message: str
    ) -> Optional[BizDeliveryWarning]:
        """单任务单预警：存在则更新，不存在则创建"""
        try:
            warning = self._get_latest_warning_by_task(task.id)
            now = datetime.now()
            if warning:
                warning.warning_type = warning_type
                warning.urgency_level = urgency_level
                warning.remaining_minutes = remaining_minutes
                warning.message = message
                warning.order_id = task.order_id
                warning.is_sent = 0
                # 未完成前统一保持未结案（若历史上被手工结案，则重新激活）
                warning.is_resolved = 0
                warning.resolved_time = None
                warning.resolved_by = None
                warning.update_at = now
                warning.update_by = "system"
                self.db.flush()
                return warning

            warning = BizDeliveryWarning(
                id=generate_id(),
                task_id=task.id,
                order_id=task.order_id,
                warning_type=warning_type,
                urgency_level=urgency_level,
                remaining_minutes=remaining_minutes,
                message=message,
                is_sent=0,
                is_resolved=0,
                create_by="system",
                create_name="系统",
            )
            self.db.add(warning)
            self.db.flush()
            return warning
        except Exception as e:
            logger.error(f"创建预警记录失败: {str(e)}")
            return None

    def _get_latest_warning_by_task(self, task_id: str) -> Optional[BizDeliveryWarning]:
        return self.db.query(BizDeliveryWarning).filter(
            BizDeliveryWarning.task_id == task_id,
            BizDeliveryWarning.is_del == 0,
        ).order_by(BizDeliveryWarning.update_at.desc(), BizDeliveryWarning.create_at.desc()).first()

    def _send_warning_notification(self, warning: BizDeliveryWarning) -> bool:
        """发送预警通知（可通过微信模板消息等）"""
        try:
            # 获取配送员信息
            task = self.db.query(BizDeliveryTask).filter(
                BizDeliveryTask.id == warning.task_id
            ).first()

            if not task or not task.deliver_id:
                logger.warning(f"任务 {warning.task_id} 未分配配送员，跳过通知")
                return False

            # TODO: 调用微信模板消息接口发送通知
            # from app.core.wechat_pay import send_delivery_warning_message
            # send_delivery_warning_message(
            #     deliver_id=task.deliver_id,
            #     task_no=task.task_no,
            #     message=warning.message
            # )

            logger.info(f"已发送预警通知: {task.task_no} - {warning.message}")
            return True

        except Exception as e:
            logger.error(f"发送预警通知失败: {str(e)}")
            return False


def check_delivery_warnings(db: Session) -> dict:
    """执行配送时效预警检查（供外部调用）"""
    task = DeliveryWarningTask(db)
    return task.check_delivery_warnings()


def cleanup_expired_warnings(db: Session, days: int = 7) -> int:
    """清理过期的预警记录
    
    Args:
        db: 数据库会话
        days: 保留天数，默认7天
    
    Returns:
        int: 删除的记录数
    """
    from datetime import timedelta

    cutoff_date = datetime.now() - timedelta(days=days)

    count = db.query(BizDeliveryWarning).filter(
        BizDeliveryWarning.create_at < cutoff_date,
        BizDeliveryWarning.is_resolved == 1,
        BizDeliveryWarning.is_del == 0
    ).update({"is_del": 1})

    db.commit()
    logger.info(f"已清理 {count} 条过期预警记录")
    return count