shudao-main/shudao-chat-py/routers/hazard.py

"""
Hazard detection routes.
"""

from typing import Any, Dict, List, Optional
import io
import json
import time
import traceback
from urllib.parse import urlparse, parse_qs

import httpx
from fastapi import APIRouter, Depends, Request
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from PIL import Image, ImageDraw, ImageFont

from database import get_db
from models.scene import RecognitionRecord, Scene, SecondScene, ThirdScene
from services.oss_service import oss_service
from services.yolo_service import yolo_service
from utils.crypto import decrypt_url
from utils.label_translator import translate_labels, translate_scenes_for_query
from utils.logger import logger

router = APIRouter()


class HazardRequest(BaseModel):
    """兼容前端与 Go 版请求结构。"""

    scene_name: str = ""
    scene_type: str = ""
    image: str = ""
    image_url: str = ""
    date: str = ""
    title: str = ""
    description: str = ""
    labels: str = ""
    current_step: Optional[int] = None
    hazard_count: Optional[int] = None
    third_scene: str = ""
    second_scene: str = ""
    recognition_image_url: str = ""
    original_image_url: str = ""
    tag_type: str = ""
    hazard_details: Any = Field(default_factory=dict)
    save_history: bool = True

    @property
    def source_image(self) -> str:
        return self.image or self.image_url or self.original_image_url or ""

    @property
    def source_scene(self) -> str:
        return self.scene_name or self.scene_type or self.tag_type


class SaveStepRequest(BaseModel):
    """Save current step for a recognition record."""

    record_id: int
    current_step: int


SCENE_KEY_ALIASES = {
    "tunnel": "tunnel",
    "隧道": "tunnel",
    "隧道工程": "tunnel",
    "simple_supported_bridge": "simple_supported_bridge",
    "bridge": "simple_supported_bridge",
    "桥梁": "simple_supported_bridge",
    "桥梁工程": "simple_supported_bridge",
    "gas_station": "gas_station",
    "加油站": "gas_station",
    "special_equipment": "special_equipment",
    "特种设备": "special_equipment",
    "operate_highway": "operate_highway",
    "运营高速公路": "operate_highway",
}

SCENE_DISPLAY_NAMES = {
    "tunnel": "隧道工程",
    "simple_supported_bridge": "桥梁工程",
    "gas_station": "加油站",
    "special_equipment": "特种设备",
    "operate_highway": "运营高速公路",
}

DEFAULT_SCENE_KEY = "tunnel"


def _get_user_code(user: Any) -> str:
    return (
        getattr(user, "userCode", None)
        or getattr(user, "user_code", None)
        or getattr(user, "account", "")
    )


def _resolve_scene_key(scene_value: str) -> str:
    if not scene_value:
        return ""
    return SCENE_KEY_ALIASES.get(scene_value.strip(), scene_value.strip())


def _scene_exists(db: Session, scene_key: str) -> bool:
    if not scene_key:
        return False
    return (
        db.query(Scene)
        .filter(Scene.scene_en_name == scene_key, Scene.is_deleted == 0)
        .first()
        is not None
    )


def _normalize_label_for_scene(label: str, scene_name: str) -> str:
    if _should_trim_scene_label_prefix(scene_name):
        return _process_highway_label(label)
    return label


def _process_highway_label(label: str) -> str:
    underscore_index = label.find("_")
    if underscore_index == -1:
        return label
    return label[underscore_index + 1 :]


def _should_trim_scene_label_prefix(scene_name: str) -> bool:
    if not scene_name:
        return False
    return any(
        keyword in scene_name
        for keyword in (
            "运营高速公路",
            "operate_highway",
            "隧道",
            "tunnel",
            "简支梁",
            "桥梁",
            "simple_supported_bridge",
        )
    )


def _remove_duplicate_labels(labels: List[str]) -> str:
    unique_labels: List[str] = []
    seen = set()
    for label in labels:
        label = str(label).strip()
        if label and label not in seen:
            seen.add(label)
            unique_labels.append(label)
    return ", ".join(unique_labels)


def _dedupe_list(items: List[str]) -> List[str]:
    seen = set()
    result: List[str] = []
    for item in items:
        if item and item not in seen:
            seen.add(item)
            result.append(item)
    return result


def _resolve_image_url(source_image_url: str) -> str:
    """Resolve encrypted/proxy/public image URLs to a directly fetchable URL."""
    if not source_image_url:
        return source_image_url

    url_text = source_image_url.strip()
    if not url_text:
        return url_text

    if "/apiv1/oss/parse/" in url_text:
        try:
            parsed = urlparse(url_text)
            encrypted_url = parse_qs(parsed.query).get("url", [""])[0]
            if encrypted_url:
                return decrypt_url(encrypted_url)
        except Exception:
            pass

    try:
        decrypted_url = decrypt_url(url_text)
        if decrypted_url and decrypted_url != url_text:
            return decrypted_url
    except Exception:
        pass

    return url_text


def _unique_ordered(items: List[str]) -> List[str]:
    seen = set()
    ordered = []
    for item in items:
        if not item or item in seen:
            continue
        seen.add(item)
        ordered.append(item)
    return ordered


def _build_scene_relations(
    db: Session, normalized_labels: List[str]
) -> tuple[list[str], dict[str, list[str]]]:
    """
    Build scene relations by mapping sub-secondary scenes to secondary scenes,
    then querying for third scenes (hazards).
    
    Args:
        db: Database session
        normalized_labels: List of sub-secondary scene names (e.g., "加油机", "加油枪")
    
    Returns:
        Tuple of (third_scene_names, element_hazards)
        - third_scene_names: List of all third scene names (hazards)
        - element_hazards: Dict mapping original labels to their hazards
    """
    third_scene_names: List[str] = []
    element_hazards: Dict[str, List[str]] = {}

    for label in normalized_labels:
        # Translate sub-secondary scene to secondary scene for database query
        # e.g., "加油机" -> "加油设施及附属场地"
        query_label = translate_scenes_for_query([label])[0] if label else label
        
        # Query SecondScene table using the translated secondary scene name
        second_scene = (
            db.query(SecondScene)
            .filter(
                SecondScene.second_scene_name == query_label,
                SecondScene.is_deleted == 0,
            )
            .first()
        )
        
        if not second_scene:
            # No matching secondary scene found, store empty hazards list
            element_hazards.setdefault(label, [])
            continue

        # Query ThirdScene table for hazards related to this secondary scene
        third_scenes = (
            db.query(ThirdScene)
            .filter(
                ThirdScene.second_scene_id == second_scene.id,
                ThirdScene.is_deleted == 0,
            )
            .all()
        )
        
        if third_scenes:
            current_element_hazards = [item.third_scene_name for item in third_scenes]
            third_scene_names.extend(current_element_hazards)
            # Store hazards under the ORIGINAL sub-secondary scene label
            element_hazards[label] = _dedupe_list(
                element_hazards.get(label, []) + current_element_hazards
            )
        else:
            element_hazards.setdefault(label, [])

    return _dedupe_list(third_scene_names), element_hazards


@router.post("/hazard")
async def hazard(
    request: Request,
    data: HazardRequest,
    db: Session = Depends(get_db),
):
    """隐患识别接口：严格按 Go 版的流程、字段和返回结构实现。"""

    user = request.state.user
    if not user:
        logger.warning("[hazard] request.state.user is empty")
        return {"statusCode": 401, "msg": "获取用户信息失败"}

    user_id = getattr(user, "id", None) or getattr(user, "user_id", None) or 1
    user_code = _get_user_code(user)
    request_id = getattr(request.state, "request_id", None) or f"hazard-{int(time.time() * 1000)}"

    try:
        logger.info(
            "[hazard][%s] incoming payload: user_id=%r, user_code=%r, scene_name=%r, scene_type=%r, image_present=%s, image_url_present=%s, image_len=%s, image_url_len=%s, date=%r",
            request_id,
            user_id,
            user_code,
            data.scene_name,
            data.scene_type,
            bool(data.image),
            bool(data.image_url),
            len(data.image or ""),
            len(data.image_url or ""),
            data.date,
        )
        logger.info(
            "[hazard][%s] request headers: content_type=%r, content_length=%r, user_agent=%r, path=%r, method=%r",
            request_id,
            request.headers.get("content-type"),
            request.headers.get("content-length"),
            request.headers.get("user-agent"),
            request.url.path,
            request.method,
        )

        source_image_url = _resolve_image_url(data.source_image)
        logger.info(
            "[hazard][%s] resolved image url: raw=%r, resolved=%r",
            request_id,
            data.source_image,
            source_image_url,
        )

        scene_key = _resolve_scene_key(data.source_scene)
        if not scene_key:
            scene_key = DEFAULT_SCENE_KEY
        logger.info(
            "[hazard][%s] resolved scene: source_scene=%r, scene_key=%r",
            request_id,
            data.source_scene,
            scene_key,
        )

        if not source_image_url:
            logger.warning(
                "[hazard][%s] validation failed: scene_name=%r, scene_type=%r, image_present=%s, image_url_present=%s, raw_image_len=%s, raw_image_url_len=%s",
                request_id,
                data.scene_name,
                data.scene_type,
                bool(data.image),
                bool(data.image_url),
                len(data.image or ""),
                len(data.image_url or ""),
            )
            return {"statusCode": 400, "msg": "图片链接不能为空"}

        scene_exists = _scene_exists(db, scene_key)
        logger.info(
            "[hazard][%s] scene lookup result: scene_key=%r, exists=%s",
            request_id,
            scene_key,
            scene_exists,
        )
        if not scene_exists:
            logger.warning("[hazard][%s] scene not found: %r", request_id, scene_key)
            return {"statusCode": 500, "msg": "场景名称不存在"}

        logger.info(
            "[hazard][%s] downloading image: url=%r",
            request_id,
            source_image_url,
        )
        try:
            async with httpx.AsyncClient(timeout=30.0) as client:
                img_response = await client.get(source_image_url)
                logger.info(
                    "[hazard][%s] image download response: status_code=%s, content_type=%r, content_length=%s",
                    request_id,
                    img_response.status_code,
                    img_response.headers.get("content-type"),
                    img_response.headers.get("content-length"),
                )
                img_response.raise_for_status()
                image_bytes = img_response.content
        except httpx.HTTPError as e:
            logger.error(
                "[hazard][%s] 下载图片失败: url=%r, error=%s, traceback=%s",
                request_id,
                source_image_url,
                e,
                traceback.format_exc(),
            )
            error_msg = "图片下载失败"
            if isinstance(e, httpx.TimeoutException):
                error_msg = "图片下载超时，请检查网络连接或稍后重试"
            elif isinstance(e, httpx.ConnectError):
                error_msg = "无法连接到图片服务器，请检查图片链接是否正确"
            elif hasattr(e, 'response') and e.response is not None:
                status_code = e.response.status_code
                if status_code == 404:
                    error_msg = "图片不存在，请检查图片链接是否正确"
                elif status_code == 403:
                    error_msg = "无权访问该图片，请检查图片链接权限"
                elif status_code >= 500:
                    error_msg = "图片服务器暂时不可用，请稍后重试"
                else:
                    error_msg = f"图片下载失败（错误代码：{status_code}）"
            return {"statusCode": 500, "msg": error_msg}

        logger.info(
            "[hazard][%s] image downloaded: bytes=%s",
            request_id,
            len(image_bytes),
        )

        try:
            logger.info("[hazard][%s] calling yolo_service.detect_hazards: scene_key=%r", request_id, scene_key)
            yolo_result = await yolo_service.detect_hazards(image_bytes, scene_key)
            logger.info(
                "[hazard][%s] yolo result summary: keys=%s, labels_count=%s, boxes_count=%s, scores_count=%s, model_type=%r",
                request_id,
                list(yolo_result.keys()) if isinstance(yolo_result, dict) else type(yolo_result).__name__,
                len((yolo_result or {}).get("labels", []) or []) if isinstance(yolo_result, dict) else -1,
                len((yolo_result or {}).get("boxes", []) or []) if isinstance(yolo_result, dict) else -1,
                len((yolo_result or {}).get("scores", []) or []) if isinstance(yolo_result, dict) else -1,
                (yolo_result or {}).get("model_type", "") if isinstance(yolo_result, dict) else "",
            )
            logger.debug("[hazard][%s] yolo result raw: %s", request_id, json.dumps(yolo_result, ensure_ascii=False, default=str))
        except httpx.TimeoutException as e:
            logger.error(
                "[hazard][%s] YOLO检测超时: scene_key=%r, error=%s, traceback=%s",
                request_id,
                scene_key,
                e,
                traceback.format_exc(),
            )
            return {"statusCode": 500, "msg": "AI识别服务响应超时，请稍后重试"}
        except httpx.ConnectError as e:
            logger.error(
                "[hazard][%s] YOLO服务连接失败: scene_key=%r, error=%s, traceback=%s",
                request_id,
                scene_key,
                e,
                traceback.format_exc(),
            )
            return {"statusCode": 500, "msg": "无法连接到AI识别服务，请联系管理员"}
        except httpx.HTTPStatusError as e:
            logger.error(
                "[hazard][%s] YOLO服务返回错误状态: scene_key=%r, status_code=%s, error=%s, traceback=%s",
                request_id,
                scene_key,
                e.response.status_code if hasattr(e, 'response') else 'unknown',
                e,
                traceback.format_exc(),
            )
            status_code = e.response.status_code if hasattr(e, 'response') else 500
            if status_code == 400:
                return {"statusCode": 500, "msg": "图片格式不支持或图片已损坏，请更换图片"}
            elif status_code == 404:
                return {"statusCode": 500, "msg": f"未找到场景'{SCENE_DISPLAY_NAMES.get(scene_key, scene_key)}'的识别模型"}
            elif status_code >= 500:
                return {"statusCode": 500, "msg": "AI识别服务暂时不可用，请稍后重试"}
            else:
                return {"statusCode": 500, "msg": f"AI识别服务返回错误（代码：{status_code}）"}
        except ValueError as e:
            logger.error(
                "[hazard][%s] YOLO返回数据格式错误: scene_key=%r, error=%s, traceback=%s",
                request_id,
                scene_key,
                e,
                traceback.format_exc(),
            )
            return {"statusCode": 500, "msg": "AI识别服务返回数据格式错误，请联系管理员"}
        except Exception as e:
            logger.error(
                "[hazard][%s] YOLO检测失败: scene_key=%r, error=%s, traceback=%s",
                request_id,
                scene_key,
                e,
                traceback.format_exc(),
            )
            error_msg = str(e)
            if "timeout" in error_msg.lower():
                return {"statusCode": 500, "msg": "AI识别服务响应超时，请稍后重试"}
            elif "connection" in error_msg.lower():
                return {"statusCode": 500, "msg": "无法连接到AI识别服务，请联系管理员"}
            else:
                return {"statusCode": 500, "msg": f"AI识别失败：{error_msg}"}

        labels = yolo_result.get("labels", []) or []
        boxes = yolo_result.get("boxes", []) or []
        scores = yolo_result.get("scores", []) or []
        model_type = yolo_result.get("model_type", "")

        if not labels:
            logger.info(
                "[hazard][%s] YOLO返回空结果: scene_key=%r, model_type=%r, image_bytes=%s",
                request_id,
                scene_key,
                model_type,
                len(image_bytes),
            )
            return {
                "statusCode": 200,
                "msg": "未检测到隐患",
                "data": {
                    "scene_name": scene_key,
                    "scene_display_name": SCENE_DISPLAY_NAMES.get(scene_key, scene_key),
                    "tag_type": scene_key,
                    "title": f"{SCENE_DISPLAY_NAMES.get(scene_key, scene_key)}隐患提示",
                    "total_detections": 0,
                    "detections": [],
                    "model_type": model_type,
                    "original_image": source_image_url,
                    "annotated_image": source_image_url,
                    "labels": "",
                    "display_labels": [],
                    "third_scenes": [],
                    "element_hazards": {},
                    "no_hazards_detected": True,
                },
            }

        normalized_labels = [
            _normalize_label_for_scene(str(label), scene_key) for label in labels
        ]
        logger.info(
            "[hazard][%s] normalized labels: raw_labels=%s, normalized_labels=%s",
            request_id,
            labels,
            normalized_labels,
        )
        
        # Translate pinyin/English labels to Chinese
        translated_labels = translate_labels(normalized_labels, fallback_to_original=True)
        logger.info(
            "[hazard][%s] translated labels: before=%s, after=%s",
            request_id,
            normalized_labels,
            translated_labels,
        )
        normalized_labels = translated_labels

        detection_results = []
        for i, label in enumerate(normalized_labels):
            if i < len(scores) and i < len(boxes):
                detection_results.append(
                    {
                        "label": label,
                        "score": scores[i],
                        "box": boxes[i],
                        "percent": f"{scores[i] * 100:.2f}%",
                    }
                )
        logger.info(
            "[hazard][%s] detection results built: total=%s, detail=%s",
            request_id,
            len(detection_results),
            detection_results,
        )

        third_scene_names, element_hazards = _build_scene_relations(
            db, normalized_labels
        )
        logger.info(
            "[hazard][%s] scene relations built: third_scene_names=%s, element_hazards=%s",
            request_id,
            third_scene_names,
            element_hazards,
        )

        scene_name = scene_key
        scene_display_name = SCENE_DISPLAY_NAMES.get(scene_key, scene_name)
        tag_type = scene_key
        title = scene_display_name
        description = " ".join(third_scene_names)
        current_step = 1
        hazard_count = len(detection_results)

        # 不再绘制检测框，直接使用原始图片
        logger.info(
            "[hazard][%s] skipping image annotation, using original image",
            request_id,
        )
        recognition_image_url = source_image_url

        try:
            labels_text = _remove_duplicate_labels([item["label"] for item in detection_results])
            if data.labels.strip():
                labels_text = _remove_duplicate_labels(
                    [item.strip() for item in data.labels.split(",") if item.strip()]
                ) or labels_text

            record_title = data.title.strip() or title
            record_description = data.description.strip() or description
            record_third_scene = data.third_scene.strip() or " ".join(third_scene_names)
            record_second_scene = data.second_scene.strip()
            record_tag_type = data.tag_type.strip() or tag_type
            record_current_step = data.current_step if data.current_step is not None else current_step
            record_hazard_count = (
                data.hazard_count if data.hazard_count is not None else hazard_count
            )

            raw_hazard_details = data.hazard_details
            if isinstance(raw_hazard_details, str):
                try:
                    record_hazard_details = json.loads(raw_hazard_details)
                except Exception:
                    record_hazard_details = raw_hazard_details
            elif raw_hazard_details:
                record_hazard_details = raw_hazard_details
            else:
                record_hazard_details = detection_results

            recognition_image_url = (
                data.recognition_image_url.strip() or recognition_image_url
            )

            now_ts = int(time.time())
            record_payload = {
                "original_image_url": source_image_url,
                "recognition_image_url": recognition_image_url,
                "user_id": user_id,
                "scene_type": scene_name,
                "hazard_count": record_hazard_count,
                "current_step": record_current_step,
                "hazard_details": json.dumps(record_hazard_details, ensure_ascii=False)
                if not isinstance(record_hazard_details, str)
                else record_hazard_details,
                "title": record_title,
                "description": record_description,
                "labels": labels_text,
                "tag_type": record_tag_type,
                "second_scene": record_second_scene,
                "third_scene": record_third_scene,
                "created_at": now_ts,
                "updated_at": now_ts,
                "is_deleted": 0,
            }

            logger.info(
                "[hazard][%s] ===== HISTORY SAVE START ===== user_id=%r, scene_key=%r, labels=%r, title=%r, save_history=%s",
                request_id,
                user_id,
                scene_key,
                labels_text,
                record_title,
                data.save_history,
            )
            logger.info(
                "[hazard][%s] history record payload: %s",
                request_id,
                json.dumps(record_payload, ensure_ascii=False, default=str),
            )

            record = RecognitionRecord(**record_payload)
            db.add(record)
            logger.info("[hazard][%s] record added to session, committing db transaction", request_id)
            db.commit()
            logger.info("[hazard][%s] db commit success, refreshing record", request_id)
            db.refresh(record)
            logger.info(
                "[hazard][%s] record saved successfully: id=%r, current_step=%r, created_at=%r, updated_at=%r",
                request_id,
                getattr(record, "id", None),
                getattr(record, "current_step", None),
                getattr(record, "created_at", None),
                getattr(record, "updated_at", None),
            )
        except Exception as e:
            logger.error(
                "[hazard][%s] 识别失败（保存记录阶段）: error=%s, traceback=%s",
                request_id,
                e,
                traceback.format_exc(),
            )
            db.rollback()
            logger.info("[hazard][%s] db rollback executed after record save failure", request_id)
            
            error_msg = str(e)
            if "connection" in error_msg.lower() or "lost connection" in error_msg.lower():
                return {"statusCode": 500, "msg": "数据库连接失败，请稍后重试"}
            elif "duplicate" in error_msg.lower() or "unique" in error_msg.lower():
                return {"statusCode": 500, "msg": "记录已存在，请勿重复提交"}
            elif "timeout" in error_msg.lower():
                return {"statusCode": 500, "msg": "数据库操作超时，请稍后重试"}
            elif "disk" in error_msg.lower() or "space" in error_msg.lower():
                return {"statusCode": 500, "msg": "存储空间不足，请联系管理员"}
            else:
                return {"statusCode": 500, "msg": f"保存识别记录失败：{error_msg}"}

        display_labels = _dedupe_list([item["label"] for item in detection_results])

        response_data = {
            "scene_name": scene_name,
            "scene_display_name": scene_display_name,
            "tag_type": tag_type,
            "title": record_title,
            "description": record_description,
            "current_step": record_current_step,
            "hazard_count": record_hazard_count,
            "total_detections": len(detection_results),
            "detections": detection_results,
            "model_type": model_type,
            "original_image": source_image_url,
            "annotated_image": recognition_image_url,
            "recognition_image_url": recognition_image_url,
            "labels": labels_text,
            "display_labels": display_labels,
            "third_scenes": third_scene_names,
            "element_hazards": element_hazards,
            "record_id": getattr(record, "id", None),
        }
        logger.info(
            "[hazard][%s] response prepared successfully: %s",
            request_id,
            json.dumps(response_data, ensure_ascii=False, default=str),
        )
        return {
            "statusCode": 200,
            "msg": "识别成功",
            "data": response_data,
        }

    except Exception as e:
        logger.error(
            "[hazard][%s] 处理失败: error=%s, traceback=%s",
            request_id,
            e,
            traceback.format_exc(),
        )
        db.rollback()
        logger.info("[hazard][%s] db rollback executed in outer exception handler", request_id)
        
        error_msg = str(e)
        if "connection" in error_msg.lower():
            return {"statusCode": 500, "msg": "服务连接失败，请稍后重试"}
        elif "timeout" in error_msg.lower():
            return {"statusCode": 500, "msg": "服务响应超时，请稍后重试"}
        elif "memory" in error_msg.lower() or "out of memory" in error_msg.lower():
            return {"statusCode": 500, "msg": "系统资源不足，请稍后重试或联系管理员"}
        elif "permission" in error_msg.lower() or "denied" in error_msg.lower():
            return {"statusCode": 500, "msg": "权限不足，请联系管理员"}
        else:
            return {"statusCode": 500, "msg": f"识别处理失败：{error_msg}"}


@router.post("/save_step")
async def save_step(
    request: Request,
    data: SaveStepRequest,
    db: Session = Depends(get_db),
):
    """Update RecognitionRecord.current_step."""

    user = request.state.user
    if not user:
        logger.warning("[save_step] unauthorized request: request.state.user is empty")
        return {"statusCode": 401, "msg": "未授权"}

    user_id = getattr(user, "user_id", None) or getattr(user, "id", None)
    request_id = getattr(request.state, "request_id", None) or f"save-step-{int(time.time() * 1000)}"

    try:
        logger.info(
            "[save_step][%s] incoming payload: user_id=%r, record_id=%r, current_step=%r, path=%r, method=%r",
            request_id,
            user_id,
            data.record_id,
            data.current_step,
            request.url.path,
            request.method,
        )
        logger.info(
            "[save_step][%s] updating RecognitionRecord: record_id=%r, user_id=%r",
            request_id,
            data.record_id,
            user_id,
        )

        affected = (
            db.query(RecognitionRecord)
            .filter(
                RecognitionRecord.id == data.record_id,
                RecognitionRecord.user_id == user_id,
            )
            .update(
                {
                    "current_step": data.current_step,
                    "updated_at": int(time.time()),
                }
            )
        )
        logger.info("[save_step][%s] update result: affected_rows=%s", request_id, affected)

        if affected == 0:
            logger.warning(
                "[save_step][%s] record not found or no permission: record_id=%r, user_id=%r",
                request_id,
                data.record_id,
                user_id,
            )
            return {"statusCode": 404, "msg": "记录不存在"}

        logger.info("[save_step][%s] committing db transaction", request_id)
        db.commit()

        logger.info(
            "[save_step][%s] save success: record_id=%r, current_step=%r",
            request_id,
            data.record_id,
            data.current_step,
        )
        return {
            "statusCode": 200,
            "msg": "保存成功",
            "data": {
                "record_id": data.record_id,
                "current_step": data.current_step,
            },
        }

    except Exception as e:
        logger.error(
            "[save_step][%s] 异常: error=%s, traceback=%s",
            request_id,
            e,
            traceback.format_exc(),
        )
        db.rollback()
        logger.info("[save_step][%s] db rollback executed after save_step failure", request_id)
        return {"statusCode": 500, "msg": f"保存失败: {str(e)}"}


async def _draw_boxes_and_watermark(
    image_bytes: bytes,
    hazards: List[Dict[str, Any]],
    user_name: str,
    user_account: str,
) -> bytes:
    """Draw detection boxes and a tiled watermark on the image."""

    try:
        image = Image.open(io.BytesIO(image_bytes)).convert("RGBA")
        width, height = image.size

        overlay = Image.new("RGBA", (width, height), (255, 255, 255, 0))
        draw = ImageDraw.Draw(overlay)

        try:
            font = ImageFont.truetype(
                "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 20
            )
            font_small = ImageFont.truetype(
                "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 14
            )
        except Exception:
            try:
                font = ImageFont.truetype("C:/Windows/Fonts/msyh.ttc", 20)
                font_small = ImageFont.truetype("C:/Windows/Fonts/msyh.ttc", 14)
            except Exception:
                font = ImageFont.load_default()
                font_small = ImageFont.load_default()

        for hazard in hazards:
            bbox = hazard.get("bbox", []) or hazard.get("box", [])
            label = hazard.get("label", "")
            confidence = hazard.get("confidence", 0)

            if len(bbox) == 4:
                x1, y1, x2, y2 = bbox
                draw.rectangle([x1, y1, x2, y2], outline=(255, 0, 0, 255), width=3)
                text = f"{label} {confidence:.2f}"
                draw.text(
                    (x1, max(0, y1 - 25)),
                    text,
                    fill=(255, 0, 0, 255),
                    font=font,
                )

        current_date = time.strftime("%Y/%m/%d")
        watermarks = [user_name or "", user_account or "", current_date]
        watermarks = [text for text in watermarks if text]
        if not watermarks:
            watermarks = [current_date]

        text_height_estimate = 50
        text_width_estimate = 150
        angle = 45

        watermark_layer = Image.new(
            "RGBA", (width * 2, height * 2), (255, 255, 255, 0)
        )
        watermark_draw = ImageDraw.Draw(watermark_layer)

        for y in range(-height, height * 2, text_height_estimate):
            for x in range(-width, width * 2, text_width_estimate):
                row_index = int(y / text_height_estimate) % len(watermarks)
                watermark_draw.text(
                    (x, y),
                    watermarks[row_index],
                    fill=(128, 128, 128, 60),
                    font=font_small,
                )

        watermark_layer = watermark_layer.rotate(
            angle, expand=False, fillcolor=(255, 255, 255, 0)
        )

        crop_x = (watermark_layer.width - width) // 2
        crop_y = (watermark_layer.height - height) // 2
        watermark_layer = watermark_layer.crop(
            (crop_x, crop_y, crop_x + width, crop_y + height)
        )

        image = Image.alpha_composite(image, watermark_layer)
        image = Image.alpha_composite(image, overlay)

        final_image = image.convert("RGB")
        output = io.BytesIO()
        final_image.save(output, format="JPEG", quality=95)
        return output.getvalue()

    except Exception as e:
        logger.error(
            "[_draw_boxes_and_watermark] 图片处理失败: %s, traceback=%s",
            e,
            traceback.format_exc(),
        )
        return image_bytes