"""配送时效预警 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)