评级报告对比, 和不保存重复数据

app/api/rating_report.py

@@ -1,19 +1,41 @@
-from fastapi import APIRouter, HTTPException, Depends, Query
-from mysql.connector.pooling import PooledMySQLConnection
-
 from datetime import datetime
 
-from app.core.logger import get_logger
+from fastapi import APIRouter, Depends, HTTPException, Query
+from mysql.connector.pooling import PooledMySQLConnection
+
 from app.core.config import settings
 from app.core.database_loader import get_db_connection
+from app.core.logger import get_logger
 from app.crud import crud_card
-from app.utils.rating_report_utils import (get_active_json, save_rating_report_history,
-                                           format_datetime, parse_json_value, crop_defect_image)
+from app.utils.rating_report_utils import (
+    build_defect_compare_key,
+    crop_defect_image,
+    format_datetime,
+    get_active_json,
+    parse_json_value,
+    remove_common_defects,
+    save_rating_report_history,
+)
 
 logger = get_logger(__name__)
 router = APIRouter()
 
 
+def _build_history_detail_response(row: dict, report_data: dict) -> dict:
+    """
+    统一历史详情返回结构，避免不同接口拼装不一致。
+    """
+    return {
+        "ratingId": row.get("rating_id"),
+        "cardId": row.get("card_id"),
+        "cardNo": row.get("cardNo"),
+        "reportName": row.get("report_name"),
+        "createdAt": format_datetime(row.get("created_at")),
+        "updatedAt": format_datetime(row.get("updated_at")),
+        "reportData": report_data,
+    }
+
+
 @router.get("/generate", status_code=200, summary="生成评级报告数据")
 def generate_rating_report(
         cardNo: str,
@@ -22,35 +44,27 @@ def generate_rating_report(
     if not cardNo or not cardNo.strip():
         raise HTTPException(status_code=400, detail="cardNo 不能为空")
 
-    # 根据cardNo 查询id
     try:
         with db_conn.cursor(buffered=True) as cursor:
             query_sql = f"SELECT id FROM {settings.DB_CARD_TABLE_NAME} WHERE cardNo = %s LIMIT 1"
             cursor.execute(query_sql, (cardNo,))
             row = cursor.fetchone()
     except Exception as e:
-        logger.error(f"创建卡牌失败: {e}")
-        raise HTTPException(status_code=500, detail="数据库查询失败。")
+        logger.error(f"根据 cardNo 查询卡牌失败: cardNo={cardNo}, error={e}")
+        raise HTTPException(status_code=500, detail="数据库查询失败")
 
     if not row:
-        raise HTTPException(
-            status_code=404,
-            detail=f"未找到卡牌编号为 {cardNo} 的相关记录"
-        )
+        raise HTTPException(status_code=404, detail=f"未找到 cardNo={cardNo} 的卡牌记录")
 
     card_id = row[0]
     rating_time_now = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+    logger.info(f"开始生成评级报告: cardNo={cardNo}, card_id={card_id}")
 
-    # top_n_defects = 3
-    """
-    根据 Card ID 生成评级报告 JSON
-    """
-    # 1. 获取卡片详情 (复用 Crud 逻辑，确保能拿到所有图片)
+    # 复用卡牌详情查询逻辑，直接拿到当前用于生成报告的图片和缺陷数据。
     card_data = crud_card.get_card_with_details(db_conn, card_id)
     if not card_data:
-        raise HTTPException(status_code=404, detail="未找到该卡片信息")
+        raise HTTPException(status_code=404, detail="未找到该卡牌信息")
 
-    # 初始化返回结构
     response_data = {
         "backImageUrl": "",
         "frontImageUrl": "",
@@ -70,13 +84,10 @@ def generate_rating_report(
         "defectDetailList": []
     }
 
-    # 临时列表用于收集所有缺陷，最后排序取 Top N
+    # 先统一收集缺陷，再按面积排序
     all_defects_collected = []
-
-    # 遍历图片寻找 Front Ring 和 Back Ring
     images = card_data.get("images", [])
 
-    # 辅助字典：defect_type 到 统计字段 的映射
     defect_map_keys = {
         "front_ring": {
             "corner": "cornerFrontNum",
@@ -92,63 +103,44 @@ def generate_rating_report(
 
     for img in images:
         img_type = img.image_type
-
-        # 只处理环光图
         if img_type not in ["front_ring", "back_ring"]:
             continue
 
-        # 设置主图 URL
         if img_type == "front_ring":
             response_data["frontImageUrl"] = img.image_path
         elif img_type == "back_ring":
             response_data["backImageUrl"] = img.image_path
 
-        # 获取有效 JSON
         json_data = get_active_json(img)
         if not json_data or "result" not in json_data:
             continue
 
         result_node = json_data["result"]
 
-        # 1. 处理居中 (Center)
         center_inf = result_node.get("center_result", {}).get("box_result", {}).get("center_inference", {})
         if center_inf:
-            # 格式: L/R=47/53, T/B=51/49 (取整)
-            # center_inference 包含 center_left, center_right, center_top, center_bottom
-            c_str = (
+            center_text = (
                 f"L/R={int(round(center_inf.get('center_left', 0)))}/{int(round(center_inf.get('center_right', 0)))}, "
                 f"T/B={int(round(center_inf.get('center_top', 0)))}/{int(round(center_inf.get('center_bottom', 0)))}"
             )
             if img_type == "front_ring":
-                response_data["centerFront"] = c_str
-                # 2. 处理尺寸 (仅从正面取，或者只要有就取) - mm 转 cm，除以 10，保留2位
-                rw_mm = center_inf.get("real_width_mm", 0)
-                rh_mm = center_inf.get("real_height_mm", 0)
-                response_data["measureWidth"] = round(rw_mm / 10.0, 2)
-                response_data["measureLength"] = round(rh_mm / 10.0, 2)
+                response_data["centerFront"] = center_text
+                response_data["measureWidth"] = round(center_inf.get("real_width_mm", 0) / 10.0, 2)
+                response_data["measureLength"] = round(center_inf.get("real_height_mm", 0) / 10.0, 2)
             else:
-                response_data["centerBack"] = c_str
+                response_data["centerBack"] = center_text
 
-        # 2. 处理缺陷 (Defects)
         defects = result_node.get("defect_result", {}).get("defects", [])
-
         for defect in defects:
-            # 过滤 edit_type == 'del'
             if defect.get("edit_type") == "del":
                 continue
 
-            d_type = defect.get("defect_type", "")  # corner, edge, face
-            d_label = defect.get("label", "")  # scratch, wear, etc.
-
-            # 统计数量
-            count_key = defect_map_keys.get(img_type, {}).get(d_type)
+            defect_type = defect.get("defect_type", "")
+            count_key = defect_map_keys.get(img_type, {}).get(defect_type)
             if count_key:
                 response_data[count_key] += 1
 
-            # 收集详细信息用于 Top N 列表
-            # 需要保存：缺陷对象本身，图片路径，正反面标识
             side_str = "FRONT" if img_type == "front_ring" else "BACK"
-
             all_defects_collected.append({
                 "defect_data": defect,
                 "image_path": img.image_path,
@@ -156,52 +148,38 @@ def generate_rating_report(
                 "area": defect.get("actual_area", 0)
             })
 
-    # 3. 处理 defectDetailList (Top N 切图)
-    # 按实际面积从大到小排序
-    all_defects_collected.sort(key=lambda x: x["area"], reverse=True)
-
-    # top_defects = all_defects_collected[:top_n_defects]
-    top_defects = all_defects_collected
+    all_defects_collected.sort(key=lambda item: item["area"], reverse=True)
 
     final_defect_list = []
-    for idx, item in enumerate(top_defects, start=1):
+    for idx, item in enumerate(all_defects_collected, start=1):
         defect = item["defect_data"]
         side = item["side"]
         original_img_path = item["image_path"]
+        defect_id = idx
+        filename = f"{card_id}_{defect_id}_{rating_time_now}.jpg"
 
-        # 构造 ID
-        d_id = idx  # 1, 2, 3
-
-        # 构造文件名: {card_id}_{seq_id}.jpg
-        filename = f"{card_id}_{d_id}_{rating_time_now}.jpg"
-
-        # 执行切图
         min_rect = defect.get("min_rect")
         defect_img_url = ""
         location_str = ""
 
         if min_rect and len(min_rect) == 3:
-            # 切图并保存
             defect_img_url = crop_defect_image(original_img_path, min_rect, filename)
+            center_x, center_y = min_rect[0]
+            location_str = f"{int(center_x)},{int(center_y)}"
+
+        raw_type = f"{defect.get('defect_type', '').upper()}".strip()
+        type_str_map = {
+            "CORNER": "CORNER",
+            "EDGE": "SIDE",
+            "FACE": "SURFACE"
+        }
+        type_str = type_str_map.get(raw_type)
 
-            # 计算 Location (中心坐标)
-            # min_rect[0] 是 [x, y]
-            cx, cy = min_rect[0]
-            location_str = f"{int(cx)},{int(cy)}"
-
-        # 构造 Type 字符串: defect_type + label (大写)
-        # 例如: defect_type="edge", label="wear" -> "EDGE WEAR"
-        d_type_raw = defect.get("defect_type", "")
-        # d_label_raw = defect.get("label", "")
-
-        type_str = f"{d_type_raw.upper()}".strip()
-        type_str_map = {"CORNER": "CORNER",
-                        "EDGE": "SIDE",
-                        "FACE": "SURFACE"}
-        type_str = type_str_map.get(type_str)
-
+        # compareKey 使用面积、标签、正反面和 location。
+        compare_key = build_defect_compare_key(defect, side, location_str)
         final_defect_list.append({
-            "id": d_id,
+            "id": defect_id,
+            "compareKey": compare_key,
             "side": side,
             "location": location_str,
             "type": type_str,
@@ -210,9 +188,22 @@ def generate_rating_report(
 
     response_data["defectDetailList"] = final_defect_list
     report_name = f"{cardNo}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
-    rating_id = save_rating_report_history(db_conn, card_id, cardNo, report_name, response_data)
-    response_data["ratingId"] = rating_id
-    response_data["reportName"] = report_name
+    history_result = save_rating_report_history(db_conn, card_id, cardNo, report_name, response_data)
+
+    response_data["ratingId"] = history_result["rating_id"]
+    response_data["reportName"] = history_result["report_name"]
+    response_data["historySaved"] = history_result["created"]
+
+    if history_result["created"]:
+        logger.info(
+            f"评级报告生成并保存成功: cardNo={cardNo}, rating_id={history_result['rating_id']}, "
+            f"defect_count={len(final_defect_list)}"
+        )
+    else:
+        logger.info(
+            f"评级报告生成成功，但内容与上一条历史一致，未重复保存: cardNo={cardNo}, "
+            f"rating_id={history_result['rating_id']}, defect_count={len(final_defect_list)}"
+        )
 
     return response_data
 
@@ -252,8 +243,8 @@ def get_rating_report_history_list(
             cursor.execute(query_sql, (cardNo, limit, skip))
             rows = cursor.fetchall()
     except Exception as e:
-        logger.error(f"查询评级报告历史列表失败: {e}")
-        raise HTTPException(status_code=500, detail="查询评级报告历史列表失败。")
+        logger.error(f"查询评级报告历史列表失败: cardNo={cardNo}, error={e}")
+        raise HTTPException(status_code=500, detail="查询评级报告历史列表失败")
 
     history_list = []
     for row in rows:
@@ -274,6 +265,58 @@ def get_rating_report_history_list(
     }
 
 
+@router.get("/history/compare", status_code=200, summary="根据两个 rating_id 对比历史缺陷差异")
+def compare_rating_report_history(
+        rating_id1: int = Query(..., description="第一个历史记录 ID"),
+        rating_id2: int = Query(..., description="第二个历史记录 ID"),
+        db_conn: PooledMySQLConnection = Depends(get_db_connection)
+):
+    try:
+        with db_conn.cursor(dictionary=True) as cursor:
+            query_sql = (
+                f"SELECT rating_id, card_id, cardNo, report_name, report_json, created_at, updated_at "
+                f"FROM {settings.RATING_REPORT_HISTORY_TABLE_NAME} "
+                "WHERE rating_id IN (%s, %s)"
+            )
+            cursor.execute(query_sql, (rating_id1, rating_id2))
+            rows = cursor.fetchall()
+    except Exception as e:
+        logger.error(
+            f"查询评级报告历史对比数据失败: rating_id1={rating_id1}, rating_id2={rating_id2}, error={e}"
+        )
+        raise HTTPException(status_code=500, detail="查询评级报告历史对比数据失败")
+
+    row_map = {row.get("rating_id"): row for row in rows}
+    missing_ids = [rating_id for rating_id in [rating_id1, rating_id2] if rating_id not in row_map]
+    if missing_ids:
+        missing_id_text = ",".join(str(item) for item in missing_ids)
+        raise HTTPException(status_code=404, detail=f"未找到 rating_id={missing_id_text} 的历史记录")
+
+    left_row = row_map[rating_id1]
+    right_row = row_map[rating_id2]
+    left_report, right_report, common_count = remove_common_defects(
+        left_row.get("report_json"),
+        right_row.get("report_json")
+    )
+
+    left_only_count = len(left_report.get("defectDetailList", []))
+    right_only_count = len(right_report.get("defectDetailList", []))
+    logger.info(
+        f"评级报告历史对比完成: rating_id1={rating_id1}, rating_id2={rating_id2}, "
+        f"same_count={common_count}, left_only_count={left_only_count}, right_only_count={right_only_count}"
+    )
+
+    return {
+        "comparisonSummary": {
+            "sameDefectCount": common_count,
+            "leftOnlyCount": left_only_count,
+            "rightOnlyCount": right_only_count
+        },
+        "left": _build_history_detail_response(left_row, left_report),
+        "right": _build_history_detail_response(right_row, right_report)
+    }
+
+
 @router.get("/history/{rating_id}", status_code=200, summary="根据 rating_id 查询单个评级报告历史")
 def get_rating_report_history_detail(
         rating_id: int,
@@ -289,18 +332,10 @@ def get_rating_report_history_detail(
             cursor.execute(query_sql, (rating_id,))
             row = cursor.fetchone()
     except Exception as e:
-        logger.error(f"查询评级报告历史详情失败: {e}")
-        raise HTTPException(status_code=500, detail="查询评级报告历史详情失败。")
+        logger.error(f"查询评级报告历史详情失败: rating_id={rating_id}, error={e}")
+        raise HTTPException(status_code=500, detail="查询评级报告历史详情失败")
 
     if not row:
         raise HTTPException(status_code=404, detail=f"未找到 rating_id={rating_id} 的历史记录")
 
-    return {
-        "ratingId": row.get("rating_id"),
-        "cardId": row.get("card_id"),
-        "cardNo": row.get("cardNo"),
-        "reportName": row.get("report_name"),
-        "createdAt": format_datetime(row.get("created_at")),
-        "updatedAt": format_datetime(row.get("updated_at")),
-        "reportData": parse_json_value(row.get("report_json"))
-    }
+    return _build_history_detail_response(row, parse_json_value(row.get("report_json")))

app/utils/rating_report_utils.py

@@ -1,16 +1,17 @@
+import copy
 import io
 import json
-import cv2
-import numpy as np
-from fastapi import HTTPException
+from collections import Counter
 from datetime import datetime
-from app.core.minio_client import minio_client
-from app.core.config import settings
-from typing import List, Dict, Any, Optional
+from typing import Any, Dict, List, Optional, Tuple
+
+from fastapi import HTTPException
 from PIL import Image
-from app.utils.scheme import ImageType
 from mysql.connector.pooling import PooledMySQLConnection
+
+from app.core.config import settings
 from app.core.logger import get_logger
+from app.core.minio_client import minio_client
 
 logger = get_logger(__name__)
 
@@ -34,15 +35,171 @@ def format_datetime(dt: Any) -> str:
     return ""
 
 
+def _normalize_compare_value(value: Any) -> Any:
+    """
+    归一化对比数据，避免浮点精度和容器类型差异造成误判。
+    """
+    if isinstance(value, float):
+        return round(value, 4)
+    if isinstance(value, tuple):
+        return [_normalize_compare_value(item) for item in value]
+    if isinstance(value, list):
+        return [_normalize_compare_value(item) for item in value]
+    if isinstance(value, dict):
+        return {key: _normalize_compare_value(val) for key, val in value.items()}
+    return value
+
+
+def build_defect_compare_key(defect: Dict[str, Any], side: str = "", location: str = "") -> str:
+    """
+    为原始缺陷构造稳定的对比键。
+    按用户需求，简化为 actual_area + label + side + location 的组合。
+    """
+    actual_area = _normalize_compare_value(defect.get("actual_area"))
+    label = defect.get("label") or ""
+    final_side = side or defect.get("side") or ""
+    final_location = location or defect.get("location") or ""
+    return f"{final_side}|{final_location}|{label}|{actual_area}"
+
+
+def _build_report_defect_compare_key(defect_item: Dict[str, Any]) -> str:
+    """
+    为历史报告里的 defectDetailList 项构造对比键。
+    新数据优先复用 compareKey，老数据回退到 side/location/label/actual_area 组合。
+    """
+    compare_key = defect_item.get("compareKey")
+    if compare_key:
+        return compare_key
+
+    actual_area = defect_item.get("actualArea")
+    if actual_area is None:
+        actual_area = defect_item.get("actual_area")
+    actual_area = _normalize_compare_value(actual_area)
+
+    label = defect_item.get("label")
+    if label is None:
+        label = defect_item.get("type") or ""
+
+    side = defect_item.get("side") or ""
+    location = defect_item.get("location") or ""
+    return f"{side}|{location}|{label}|{actual_area}"
+
+
+def enrich_report_for_compare(report_json: Any) -> Dict[str, Any]:
+    """
+    给历史报告补齐 compareKey，兼容旧数据。
+    """
+    report_data = copy.deepcopy(parse_json_value(report_json))
+    defect_list = report_data.get("defectDetailList")
+
+    if not isinstance(defect_list, list):
+        report_data["defectDetailList"] = []
+        return report_data
+
+    for defect_item in defect_list:
+        if not isinstance(defect_item, dict):
+            continue
+        defect_item["compareKey"] = _build_report_defect_compare_key(defect_item)
+
+    return report_data
+
+
+def build_report_compare_snapshot(report_json: Any) -> Dict[str, Any]:
+    """
+    生成用于“历史去重”的归一化快照。
+    这里会忽略 defectImgUrl 等波动字段，只保留真正反映缺陷是否变化的部分。
+    """
+    report_data = enrich_report_for_compare(report_json)
+    compare_snapshot = copy.deepcopy(report_data)
+
+    normalized_defects = []
+    for defect_item in report_data.get("defectDetailList", []):
+        if not isinstance(defect_item, dict):
+            continue
+        normalized_defects.append({
+            "side": defect_item.get("side") or "",
+            "type": defect_item.get("type") or "",
+            "location": defect_item.get("location") or "",
+            "compareKey": defect_item.get("compareKey") or "",
+        })
+
+    normalized_defects.sort(key=lambda item: item["compareKey"])
+    compare_snapshot["defectDetailList"] = normalized_defects
+
+    return _normalize_compare_value(compare_snapshot)
+
+
+def _filter_common_defects(defect_list: List[Dict[str, Any]], common_counter: Counter) -> List[Dict[str, Any]]:
+    """
+    逐项扣减重复计数，只移除两边真正重合的那一部分。
+    """
+    filtered_list = []
+    for defect_item in defect_list:
+        compare_key = defect_item.get("compareKey") or _build_report_defect_compare_key(defect_item)
+        if common_counter[compare_key] > 0:
+            common_counter[compare_key] -= 1
+            continue
+        filtered_list.append(defect_item)
+    return filtered_list
+
+
+def remove_common_defects(
+        left_report_json: Any,
+        right_report_json: Any
+) -> Tuple[Dict[str, Any], Dict[str, Any], int]:
+    """
+    移除两份历史报告中相同的 defectDetailList 项，只保留差异项。
+    """
+    left_report = enrich_report_for_compare(left_report_json)
+    right_report = enrich_report_for_compare(right_report_json)
+
+    left_defects = [item for item in left_report.get("defectDetailList", []) if isinstance(item, dict)]
+    right_defects = [item for item in right_report.get("defectDetailList", []) if isinstance(item, dict)]
+
+    left_counter = Counter(item.get("compareKey") or _build_report_defect_compare_key(item) for item in left_defects)
+    right_counter = Counter(item.get("compareKey") or _build_report_defect_compare_key(item) for item in right_defects)
+    common_counter = left_counter & right_counter
+    common_count = sum(common_counter.values())
+
+    left_report["defectDetailList"] = _filter_common_defects(left_defects, common_counter.copy())
+    right_report["defectDetailList"] = _filter_common_defects(right_defects, common_counter.copy())
+
+    return left_report, right_report, common_count
+
+
 def save_rating_report_history(
         db_conn: PooledMySQLConnection,
         card_id: int,
         card_no: str,
         report_name: str,
         report_json: Dict[str, Any]
-) -> int:
+) -> Dict[str, Any]:
     try:
-        with db_conn.cursor() as cursor:
+        current_snapshot = build_report_compare_snapshot(report_json)
+
+        with db_conn.cursor(dictionary=True) as cursor:
+            # 只和上一条历史比较，满足“当前和上一次一样就不存”的需求。
+            latest_sql = (
+                f"SELECT rating_id, report_name, report_json "
+                f"FROM {settings.RATING_REPORT_HISTORY_TABLE_NAME} "
+                "WHERE card_id = %s "
+                "ORDER BY rating_id DESC LIMIT 1"
+            )
+            cursor.execute(latest_sql, (card_id,))
+            latest_row = cursor.fetchone()
+
+            if latest_row:
+                latest_snapshot = build_report_compare_snapshot(latest_row.get("report_json"))
+                if latest_snapshot == current_snapshot:
+                    logger.info(
+                        f"评级报告历史未发生变化，跳过保存: cardNo={card_no}, rating_id={latest_row.get('rating_id')}"
+                    )
+                    return {
+                        "rating_id": latest_row.get("rating_id"),
+                        "report_name": latest_row.get("report_name"),
+                        "created": False
+                    }
+
             insert_sql = (
                 f"INSERT INTO {settings.RATING_REPORT_HISTORY_TABLE_NAME} "
                 "(card_id, cardNo, report_name, report_json) "
@@ -53,54 +210,56 @@ def save_rating_report_history(
                 (card_id, card_no, report_name, json.dumps(report_json, ensure_ascii=False))
             )
             db_conn.commit()
-            return cursor.lastrowid
+
+            logger.info(f"评级报告历史保存成功: cardNo={card_no}, rating_id={cursor.lastrowid}")
+            return {
+                "rating_id": cursor.lastrowid,
+                "report_name": report_name,
+                "created": True
+            }
     except Exception as e:
         db_conn.rollback()
         logger.error(f"保存评级报告历史失败: {e}")
-        raise HTTPException(status_code=500, detail="保存评级报告历史失败。")
+        raise HTTPException(status_code=500, detail="保存评级报告历史失败")
 
 
 def get_active_json(image_data: Any) -> Optional[Dict]:
-    """获取有效的json数据，优先 modified_json"""
+    """获取有效的 json 数据，优先使用 modified_json。"""
     if not image_data:
         return None
 
-    # image_data 可能是 Pydantic 对象或 字典，做兼容处理
+    # image_data 可能是 Pydantic 对象或者字典，这里做兼容。
     if hasattr(image_data, "modified_json"):
-        mj = image_data.modified_json
-        dj = image_data.detection_json
+        modified_json = image_data.modified_json
+        detection_json = image_data.detection_json
     else:
-        mj = image_data.get("modified_json")
-        dj = image_data.get("detection_json")
+        modified_json = image_data.get("modified_json")
+        detection_json = image_data.get("detection_json")
 
-    # 注意：根据 schema.py，这里读出来已经是 dict 了，不需要 json.loads
-    # 如果数据库里存的是 null，读出来是 None
-    if mj:
-        return mj
-    return dj
+    if modified_json:
+        return modified_json
+    return detection_json
 
 
 def crop_defect_image(original_image_path_str: str, min_rect: List, output_filename: str) -> str:
     """
-    通过 MinIO 切割缺陷图片为正方形并保存
+    通过 MinIO 裁剪缺陷图，生成正方形缺陷图并回传最终 URL。
     """
     try:
-        # ★ 将进来的全路径 URL 剥离为相对路径 (如 /Data/xxx.jpg) 供 MinIO 读取
+        # 把完整 URL 转成 MinIO 对象路径，例如 /Data/xxx.jpg。
         rel_path = original_image_path_str.replace(settings.DATA_HOST_URL, "")
         rel_path = "/" + rel_path.lstrip('/\\')
         object_name = f"{settings.MINIO_BASE_PREFIX}{rel_path}"
 
-        # 1. 从 MinIO 获取原图字节
         try:
             response = minio_client.get_object(settings.MINIO_BUCKET, object_name)
             image_bytes = response.read()
             response.close()
             response.release_conn()
         except Exception as e:
-            logger.warning(f"从MinIO获取原图失败: {object_name} -> {e}")
+            logger.warning(f"从 MinIO 获取原图失败: {object_name} -> {e}")
             return ""
 
-        # 2. 在内存中用 PIL 切图
         with Image.open(io.BytesIO(image_bytes)) as img:
             img_w, img_h = img.size
             (center_x, center_y), (rect_w, rect_h), angle = min_rect
@@ -113,12 +272,13 @@ def crop_defect_image(original_image_path_str: str, min_rect: List, output_filen
             side_length = max(max(rect_w, rect_h) * resize_scale, 100)
             half_side = side_length / 2
 
-            left, top = max(0, center_x - half_side), max(0, center_y - half_side)
-            right, bottom = min(img_w, center_x + half_side), min(img_h, center_y + half_side)
+            left = max(0, center_x - half_side)
+            top = max(0, center_y - half_side)
+            right = min(img_w, center_x + half_side)
+            bottom = min(img_h, center_y + half_side)
 
             cropped_img = img.crop((left, top, right, bottom))
 
-            # 3. 将切割后的图存入内存流，并上传到 MinIO
             out_bytes = io.BytesIO()
             cropped_img.save(out_bytes, format="JPEG", quality=95)
             out_bytes.seek(0)
@@ -135,7 +295,6 @@ def crop_defect_image(original_image_path_str: str, min_rect: List, output_filen
             )
 
             return settings.get_full_url(out_rel_path)
-
     except Exception as e:
-        logger.error(f"切割并上传图片失败: {e}")
+        logger.error(f"裁剪并上传缺陷图失败: {e}")
         return ""