CardScoreDataServer/app/utils/rating_report_utils.py

import io
import json
import cv2
import numpy as np
from fastapi import HTTPException
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 PIL import Image
from app.utils.scheme import ImageType
from mysql.connector.pooling import PooledMySQLConnection
from app.core.logger import get_logger

logger = get_logger(__name__)


def parse_json_value(raw_value: Any) -> Dict[str, Any]:
    if raw_value is None:
        return {}
    if isinstance(raw_value, dict):
        return raw_value
    if isinstance(raw_value, str):
        try:
            return json.loads(raw_value)
        except json.JSONDecodeError:
            return {}
    return {}


def format_datetime(dt: Any) -> str:
    if isinstance(dt, datetime):
        return dt.strftime("%Y-%m-%d %H:%M:%S")
    return ""


def save_rating_report_history(
        db_conn: PooledMySQLConnection,
        card_id: int,
        card_no: str,
        report_name: str,
        report_json: Dict[str, Any]
) -> int:
    try:
        with db_conn.cursor() as cursor:
            insert_sql = (
                f"INSERT INTO {settings.RATING_REPORT_HISTORY_TABLE_NAME} "
                "(card_id, cardNo, report_name, report_json) "
                "VALUES (%s, %s, %s, %s)"
            )
            cursor.execute(
                insert_sql,
                (card_id, card_no, report_name, json.dumps(report_json, ensure_ascii=False))
            )
            db_conn.commit()
            return cursor.lastrowid
    except Exception as e:
        db_conn.rollback()
        logger.error(f"保存评级报告历史失败: {e}")
        raise HTTPException(status_code=500, detail="保存评级报告历史失败。")


def get_active_json(image_data: Any) -> Optional[Dict]:
    """获取有效的json数据，优先 modified_json"""
    if not image_data:
        return None

    # image_data 可能是 Pydantic 对象或 字典，做兼容处理
    if hasattr(image_data, "modified_json"):
        mj = image_data.modified_json
        dj = image_data.detection_json
    else:
        mj = image_data.get("modified_json")
        dj = image_data.get("detection_json")

    # 注意：根据 schema.py，这里读出来已经是 dict 了，不需要 json.loads
    # 如果数据库里存的是 null，读出来是 None
    if mj:
        return mj
    return dj


def crop_defect_image(original_image_path_str: str, min_rect: List, output_filename: str) -> str:
    """
    通过 MinIO 切割缺陷图片为正方形并保存
    """
    try:
        # ★ 将进来的全路径 URL 剥离为相对路径 (如 /Data/xxx.jpg) 供 MinIO 读取
        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}")
            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
            center_x = int(center_x)
            center_y = int(center_y)
            rect_w = int(rect_w)
            rect_h = int(rect_h)

            resize_scale = 1.5 - abs(abs(angle % 90) - 45) / 90
            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)

            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)

            out_rel_path = f"/DefectImage/{output_filename}"
            out_object_name = f"{settings.MINIO_BASE_PREFIX}{out_rel_path}"

            minio_client.put_object(
                settings.MINIO_BUCKET,
                out_object_name,
                out_bytes,
                len(out_bytes.getvalue()),
                content_type="image/jpeg"
            )

            return settings.get_full_url(out_rel_path)

    except Exception as e:
        logger.error(f"切割并上传图片失败: {e}")
        return ""