sqxp/backend/app/api/delivery_warning.py

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