sqxp/backend/app/api/delivery_warning.py

"""配送时效预警 API"""
from typing import Optional, List, Dict
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from sqlalchemy import and_, or_
from pydantic import BaseModel, Field

from app.core.database import get_db
from app.core.auth import get_current_user
from app.core.response import success_response, error_response
from app.core.utils import generate_id
from app.models.models import BizDeliveryTask, BizDeliveryWarning, BizDeliveryWarningHis
from app.crud.base import CRUDBase
from app.crud.user import user_crud
from app.crud.order import order_crud

# 注意：该模块提供两套入口：
# - 新入口：/web/admin/delivery-warning/*
# - 兼容入口：/api/web/admin/delivery-warning/*（仅在新入口前加 /api 前缀）
# 这里不能把 legacy_router 的 prefix 写成 /api/web/admin/delivery-warning，
# 否则 include_router(main_router) 会把 main_router 的 prefix 再拼一次，导致路径重复而 404。
main_router = APIRouter(prefix="/web/admin/delivery-warning", tags=["配送时效预警"])
legacy_router = APIRouter(prefix="/api", tags=["配送时效预警(兼容)"])
router = APIRouter()


class DeliveryWarningCRUD(CRUDBase[BizDeliveryWarning]):
    """配送时效预警CRUD"""

    def __init__(self):
        super().__init__(BizDeliveryWarning)

    def create_warning(
        self,
        db: Session,
        task_id: str,
        task_no: str,
        warning_type: str,
        urgency_level: str,
        remaining_minutes: int,
        message: str,
        create_by: str = None,
        create_name: str = None
    ) -> BizDeliveryWarning:
        """创建预警记录"""
        data = {
            "id": generate_id(),
            "task_id": task_id,
            "warning_type": warning_type,
            "urgency_level": urgency_level,
            "remaining_minutes": remaining_minutes,
            "message": message,
            # 历史表由代码侧维护，这里只落基础预警字段
            "is_sent": 0,
            "is_resolved": 0,
            "create_by": create_by,
            "create_name": create_name
        }
        return self.create(db, data)

    def get_by_task(self, db: Session, task_id: str) -> List[BizDeliveryWarning]:
        """获取配送任务的预警记录"""
        return db.query(self.model).filter(
            self.model.task_id == task_id,
            self.model.is_del == 0
        ).order_by(self.model.create_at.desc()).all()

    def mark_handled(
        self,
        db: Session,
        warning_id: str,
        handle_type: str,
        remark: str = None,
        operator_id: str = None,
        operator_name: str = None
    ) -> BizDeliveryWarning:
        """标记预警已处理"""
        warning = self.get(db, warning_id)
        if not warning:
            return None

        # 历史快照：先把主表更新前数据插入到 biz_delivery_warning_his
        his = BizDeliveryWarningHis(
            id=warning.id,
            task_id=warning.task_id,
            order_id=warning.order_id,
            warning_type=warning.warning_type,
            urgency_level=warning.urgency_level,
            remaining_minutes=warning.remaining_minutes,
            message=warning.message,
            is_sent=warning.is_sent,
            sent_time=warning.sent_time,
            is_resolved=warning.is_resolved,
            resolved_time=warning.resolved_time,
            resolved_by=warning.resolved_by,
            resolved_note=warning.resolved_note,
            create_at=warning.create_at,
            create_by=warning.create_by,
            create_name=warning.create_name,
            update_at=warning.update_at,
            update_by=warning.update_by,
            his_create_at=datetime.now(),
        )
        db.add(his)

        # 更新主表（标记已解决/已处理）
        warning.is_resolved = 1
        warning.resolved_time = datetime.now()
        warning.resolved_by = operator_id
        if remark:
            warning.resolved_note = f"{handle_type}:{remark}"
        else:
            warning.resolved_note = handle_type

        warning.update_at = datetime.now()
        warning.update_by = operator_id

        db.commit()
        db.refresh(warning)
        return warning


from datetime import datetime

delivery_warning_crud = DeliveryWarningCRUD()


@main_router.get("/statistics", response_model=dict)
def get_warning_statistics(
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """获取预警统计"""
    try:
        total = db.query(delivery_warning_crud.model).filter(
            delivery_warning_crud.model.is_del == 0
        ).count()

        critical = db.query(delivery_warning_crud.model).filter(
            and_(
                delivery_warning_crud.model.is_del == 0,
                delivery_warning_crud.model.warning_type == "critical",
                delivery_warning_crud.model.is_resolved == 0
            )
        ).count()

        urgent = db.query(delivery_warning_crud.model).filter(
            and_(
                delivery_warning_crud.model.is_del == 0,
                delivery_warning_crud.model.warning_type == "urgent",
                delivery_warning_crud.model.is_resolved == 0
            )
        ).count()

        warning = db.query(delivery_warning_crud.model).filter(
            and_(
                delivery_warning_crud.model.is_del == 0,
                delivery_warning_crud.model.warning_type == "timeout",
                delivery_warning_crud.model.is_resolved == 0
            )
        ).count()

        return success_response({
            "total": total,
            "critical": critical,
            "urgent": urgent,
            "warning": warning
        })
    except Exception as e:
        return error_response(500, f"获取统计数据失败: {str(e)}")


@main_router.get("/list", response_model=dict)
def get_warning_list(
    page: int = Query(1, ge=1, description="页码"),
    pageSize: int = Query(20, ge=1, le=100, description="每页条数", alias="pageSize"),
    warningType: Optional[str] = Query(None, description="预警类型", alias="warningType"),
    status: Optional[str] = Query(None, description="状态(pending/handled)"),
    isResolved: Optional[bool] = Query(None, description="是否已解决", alias="isResolved"),
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """获取预警列表"""
    try:
        query = db.query(delivery_warning_crud.model).filter(
            delivery_warning_crud.model.is_del == 0
        )

        if warningType:
            query = query.filter(delivery_warning_crud.model.warning_type == warningType)

        # 兼容前端传入的 isResolved 参数
        if isResolved is not None:
            query = query.filter(delivery_warning_crud.model.is_resolved == (1 if isResolved else 0))
        elif status:
            # 前端传入 pending/handled 时做简单映射
            if status in ["pending", "unhandled"]:
                query = query.filter(delivery_warning_crud.model.is_resolved == 0)
            else:
                query = query.filter(delivery_warning_crud.model.is_resolved == 1)
        else:
            query = query.filter(delivery_warning_crud.model.is_resolved == 0)

        total = query.count()
        items = (
            query.order_by(delivery_warning_crud.model.create_at.desc())
            .offset((page - 1) * pageSize)
            .limit(pageSize)
            .all()
        )

        # 批量补齐：配送任务、订单、客户、配送员
        task_ids = [it.task_id for it in items if getattr(it, "task_id", None)]
        tasks = (
            db.query(BizDeliveryTask).filter(and_(BizDeliveryTask.id.in_(task_ids), BizDeliveryTask.is_del == 0)).all()
            if task_ids
            else []
        )
        task_map: Dict[str, BizDeliveryTask] = {t.id: t for t in tasks}

        order_ids = []
        for t in tasks:
            if getattr(t, "order_id", None):
                order_ids.append(t.order_id)
        order_ids = list(dict.fromkeys(order_ids))

        order_map: Dict[str, any] = {}
        if order_ids:
            rows = db.query(order_crud.model).filter(and_(order_crud.model.id.in_(order_ids), order_crud.model.is_del == 0)).all()
            order_map = {o.id: o for o in rows}

        deliver_ids = list(
            dict.fromkeys([t.deliver_id for t in tasks if getattr(t, "deliver_id", None)])
        )
        deliver_map: Dict[str, any] = {}
        if deliver_ids:
            rows = db.query(user_crud.model).filter(and_(user_crud.model.id.in_(deliver_ids), user_crud.model.is_del == 0)).all()
            deliver_map = {u.id: u for u in rows}

        client_ids = list(
            dict.fromkeys([getattr(order_map.get(t.order_id), "client_id", None) for t in tasks if t.order_id in order_map])
        )
        client_ids = [cid for cid in client_ids if cid]
        client_map: Dict[str, any] = {}
        if client_ids:
            rows = db.query(user_crud.model).filter(and_(user_crud.model.id.in_(client_ids), user_crud.model.is_del == 0)).all()
            client_map = {u.id: u for u in rows}

        result = []
        for item in items:
            warning_data = _to_response(item)
            task = task_map.get(item.task_id)
            if task:
                warning_data["deliveryId"] = task.id
                warning_data["taskNo"] = task.task_no
                warning_data["deliverId"] = task.deliver_id
                d = deliver_map.get(task.deliver_id) if task.deliver_id else None
                warning_data["deliverName"] = getattr(d, "real_name", None) if d else None
                warning_data["deliverPhone"] = getattr(d, "phone", None) if d else None
                warning_data["deliveryDeadlineAt"] = str(task.effective_time) if task.effective_time else None
                warning_data["expireTime"] = str(task.effective_time) if task.effective_time else None
                warning_data["dispatchRemark"] = task.dispatch_remark

                order = order_map.get(task.order_id) if task.order_id else None
                if order:
                    warning_data["orderId"] = order.id
                    warning_data["orderNo"] = order.order_no
                    warning_data["clientId"] = order.client_id
                    c = client_map.get(order.client_id) if order.client_id else None
                    warning_data["clientName"] = getattr(c, "real_name", None) if c else None
                    warning_data["contactPerson"] = order.contact_person
                    warning_data["contactPhone"] = order.contact_phone
                    warning_data["deliveryAddress"] = order.delivery_address
            result.append(warning_data)

        return success_response({
            "list": result,
            "total": total,
            "page": page,
            "pageSize": pageSize,
            "totalPages": (total + pageSize - 1) // pageSize
        })
    except Exception as e:
        return error_response(500, f"获取预警列表失败: {str(e)}")


@main_router.get("/detail/{warning_id}", response_model=dict)
def get_warning_detail(
    warning_id: str,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """获取预警详情"""
    try:
        warning = delivery_warning_crud.get(db, warning_id)
        if not warning or warning.is_del == 1:
            return error_response(404, "预警不存在")
        return success_response(_to_response(warning))
    except Exception as e:
        return error_response(500, f"获取预警详情失败: {str(e)}")


@main_router.post("/handle/{warning_id}", response_model=dict)
def handle_warning(
    warning_id: str,
    body: dict = None,
    handle_type: str = None,
    remark: str = None,
    note: str = None,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """处理预警"""
    try:
        # 兼容前端 JSON body 与 query 参数多种传参方式
        payload = body or {}
        handle_type = payload.get("handleType") or payload.get("handle_type") or handle_type
        remark = payload.get("remark") or payload.get("note") or remark or note
        if not handle_type:
            handle_type = "manual"  # 默认手动处理
        if not remark:
            remark = "手动处理"

        warning = delivery_warning_crud.mark_handled(
            db=db,
            warning_id=warning_id,
            handle_type=handle_type,
            remark=remark,
            operator_id=current_user.get("user_id"),
            operator_name=current_user.get("real_name")
        )

        if not warning:
            return error_response(404, "预警不存在")

        return success_response({"message": "处理成功"})
    except Exception as e:
        return error_response(500, f"处理预警失败: {str(e)}")


@main_router.post("/notify/{warning_id}", response_model=dict)
def notify_deliver(
    warning_id: str,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """提醒配送员"""
    try:
        warning = delivery_warning_crud.get(db, warning_id)
        if not warning:
            return error_response(404, "预警不存在")

        task = db.query(BizDeliveryTask).filter(BizDeliveryTask.id == warning.task_id).first()
        if not task or not task.deliver_id:
            return error_response(400, "配送任务未分配配送员")

        # TODO: 发送微信模板消息提醒配送员
        notification_sent = True

        if notification_sent:
            return success_response({"message": "已发送提醒"})
        else:
            return error_response(500, "发送提醒失败")
    except Exception as e:
        return error_response(500, f"提醒配送员失败: {str(e)}")


def _to_response(warning: BizDeliveryWarning) -> dict:
    """转换为响应格式"""
    resolved_note = warning.resolved_note or ""
    handle_type = None
    remark = None
    if ":" in resolved_note:
        handle_type, remark = resolved_note.split(":", 1)
    elif resolved_note:
        remark = resolved_note

    status = "handled" if warning.is_resolved else "pending"

    # 兼容：同时保留 snake_case（历史）与 camelCase（前端列表展示）
    return {
        "id": warning.id,
        "taskId": warning.task_id,
        "orderId": warning.order_id,
        "warningType": warning.warning_type,
        "urgencyLevel": warning.urgency_level,
        "remainingMinutes": warning.remaining_minutes,
        "message": warning.message,
        "status": status,
        "isResolved": True if warning.is_resolved else False,
        "resolveNote": remark,
        "resolveTime": str(warning.resolved_time) if warning.resolved_time else None,
        "createAt": str(warning.create_at) if warning.create_at else None,
        # legacy keys
        "task_id": warning.task_id,
        "order_id": warning.order_id,
        "warning_type": warning.warning_type,
        "urgency_level": warning.urgency_level,
        "remaining_minutes": warning.remaining_minutes,
        "create_at": str(warning.create_at) if warning.create_at else None,
        "handle_type": handle_type,
        "remark": remark,
        "handle_by": warning.resolved_by,
        "handle_time": str(warning.resolved_time) if warning.resolved_time else None,
    }


@main_router.post("/trigger-check", response_model=dict)
def trigger_warning_check(
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """手动触发预警检查（供运维使用）"""
    try:
        from app.tasks.delivery_warning import check_delivery_warnings

        result = check_delivery_warnings(db)
        return success_response({
            "message": "检查完成",
            "result": {
                "checked": result.get("checked", 0),
                "warnings_created": result.get("warnings_created", 0),
                "notifications_sent": result.get("notifications_sent", 0),
                "errors": result.get("errors", [])
            }
        })
    except Exception as e:
        return error_response(500, f"检查失败: {str(e)}")


# 将同一套路由挂到兼容前缀下
legacy_router.include_router(main_router)
router.include_router(main_router)
router.include_router(legacy_router)