"""配送时效预警 API""" import logging from typing import Optional, List, Dict from fastapi import APIRouter, Depends, Query, Request from sqlalchemy.orm import Session from sqlalchemy import and_ from app.core.database import get_db from app.core.auth import get_current_user from app.core.admin_request_json import parse_admin_request_json from app.core.response import success_response, error_response from app.core.utils import generate_id from app.models.models import BizDeliveryTask, BizDeliveryWarning 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() _delivery_warning_log = logging.getLogger("sqxp") def _preview_text(val: Optional[str], limit: int = 160) -> str: if val is None: return "" s = str(val).replace("\r", " ").replace("\n", " ") return s if len(s) <= limit else s[: limit - 3] + "..." 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 now = datetime.now() operator_display = operator_name or operator_id or "未知操作人" handle_display = handle_type or "manual" remark_display = " ".join((remark or "已处理").split()) if not remark_display: remark_display = "已处理" process_line = f"{now.strftime('%Y-%m-%d %H:%M:%S')}|{operator_display}|{handle_display}|{remark_display}" _delivery_warning_log.info( "delivery_warning.mark_handled warning_id=%s remark_len=%s remark_preview=%r process_line_preview=%r", warning_id, len(remark_display), _preview_text(remark_display, 120), _preview_text(process_line, 200), ) # 仅记录过程，不变更是否已解决 old_note = (warning.resolved_note or "").strip() warning.resolved_note = f"{old_note}\n{process_line}".strip() if old_note else process_line warning.update_at = 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"), regionId: Optional[str] = Query(None, description="区域ID", alias="regionId"), 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) if regionId: query = query.join( BizDeliveryTask, and_( BizDeliveryTask.id == delivery_warning_crud.model.task_id, BizDeliveryTask.is_del == 0 ) ).filter(BizDeliveryTask.region_id == regionId) # 兼容前端传入的 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: # 默认展示全部（未处理 + 已处理） pass # 单任务单预警：同一 task 仅保留最新一条（兼容历史重复数据） raw_items = query.order_by( delivery_warning_crud.model.update_at.desc(), delivery_warning_crud.model.create_at.desc() ).all() dedup_map: Dict[str, BizDeliveryWarning] = {} for it in raw_items: key = it.task_id or it.id if key not in dedup_map: dedup_map[key] = it dedup_items = list(dedup_map.values()) total = len(dedup_items) start = (page - 1) * pageSize end = start + pageSize items = dedup_items[start:end] # 批量补齐：配送任务、订单、客户、配送员 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) async def handle_warning( warning_id: str, request: Request, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user), note: Optional[str] = Query(None, description="处理备注（兼容 query，优先请求体）"), remark: Optional[str] = Query(None, description="同 note"), ): """处理预警（与 ApiCryptoMiddleware / 运营端加密协议一致：Request.json + parse_admin_request_json）""" try: uid = current_user.get("user_id") qn = (note or "").strip() qr = (remark or "").strip() client_host = request.client.host if request.client else "" _delivery_warning_log.info( "delivery_warning.handle enter warning_id=%s operator_user_id=%s path=%s query_note_len=%s query_remark_len=%s client=%s", warning_id, uid, request.url.path, len(qn), len(qr), client_host, ) data, err = await parse_admin_request_json( request, current_user, source="delivery_warning.handle" ) _delivery_warning_log.info( "delivery_warning.handle after_parse err=%r data_type=%s", err, type(data).__name__, ) if err: _delivery_warning_log.warning( "delivery_warning.handle parse error warning_id=%s err=%s operator_user_id=%s", warning_id, err, uid, ) return error_response(400, err) payload = data if isinstance(data, dict) else {} if isinstance(payload, dict): nk = payload.get("note") rk = payload.get("remark") _delivery_warning_log.info( "delivery_warning.handle payload keys=%s note_len=%s remark_len=%s note_preview=%r remark_preview=%r", sorted(payload.keys()), len(str(nk) if nk is not None else ""), len(str(rk) if rk is not None else ""), _preview_text(str(nk) if nk is not None else "", 100), _preview_text(str(rk) if rk is not None else "", 100), ) else: _delivery_warning_log.warning( "delivery_warning.handle payload_not_dict warning_id=%s type=%s repr_preview=%r", warning_id, type(payload).__name__, _preview_text(repr(data), 200), ) handle_type = (payload.get("handleType") or payload.get("handle_type") or "").strip() or "manual" body_note = str(payload.get("remark") or payload.get("note") or "").strip() raw_note = body_note or qn or qr if not raw_note: _delivery_warning_log.warning( "delivery_warning.handle empty remark warning_id=%s operator_user_id=%s payload_keys=%s body_note_len=%s", warning_id, uid, sorted(payload.keys()) if isinstance(payload, dict) else [], len(body_note), ) return error_response(400, "请填写处理备注") remark_val = raw_note _delivery_warning_log.info( "delivery_warning.handle resolved handle_type=%s remark_len=%s remark_preview=%r", handle_type, len(remark_val), _preview_text(remark_val, 160), ) warning = delivery_warning_crud.mark_handled( db=db, warning_id=warning_id, handle_type=handle_type, remark=remark_val, operator_id=current_user.get("user_id"), operator_name=current_user.get("real_name") ) if not warning: return error_response(404, "预警不存在") _delivery_warning_log.info( "delivery_warning.handle success warning_id=%s resolved_note_total_len=%s", warning_id, len((warning.resolved_note or "")), ) return success_response({"message": "处理记录已保存"}) except Exception as e: _delivery_warning_log.exception( "delivery_warning.handle failed warning_id=%s operator_user_id=%s err=%s", warning_id, current_user.get("user_id"), str(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 process_logs = [] for line in [s.strip() for s in resolved_note.split("\n") if s.strip()]: parts = line.split("|", 3) if len(parts) == 4: t, operator, h_type, note = parts process_logs.append({ "time": t, "operator": operator, "handleType": h_type, "note": note }) handle_type = h_type remark = note elif ":" in line: # 兼容旧格式 handleType:remark h_type, note = line.split(":", 1) process_logs.append({ "time": None, "operator": None, "handleType": h_type, "note": note }) handle_type = h_type remark = note else: process_logs.append({ "time": None, "operator": None, "handleType": "manual", "note": line }) handle_type = "manual" remark = line 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, "processLogs": process_logs, "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)