CardScoreDataServer/app/crud/crud_card.py

from typing import Optional, List, Dict, Any
from datetime import date
from mysql.connector.pooling import PooledMySQLConnection
import json
from datetime import datetime
import copy

from app.core.config import settings
from app.utils.scheme import CardImageResponse, CardType, SortBy, SortOrder, ImageType
from app.utils.card_score_calculate import calculate_scores_from_images, SCORE_SOURCE_IMAGE_TYPES
from app.core.logger import get_logger

logger = get_logger(__name__)

# 定义灰度图固定的 detection_json 结构
EMPTY_DETECTION_JSON = {
    "result": {
        "center_result": {},
        "defect_result": {
            "defects": []
        }
    }
}


def update_card_scores_and_status(db_conn: PooledMySQLConnection, card_id: int):
    """
    更新cards表中的分数和状态。注意：只基于 card_images 表（主4张图）计算。
    """
    with db_conn.cursor(dictionary=True) as cursor:
        # 1. 获取所有关联图片 (主表)
        query_images = f"SELECT * FROM {settings.DB_IMAGE_TABLE_NAME} WHERE card_id = %s"
        cursor.execute(query_images, (card_id,))
        image_records = cursor.fetchall()

        images = [
            CardImageResponse.model_validate(row)
            for row in image_records
            if row.get("image_type") in SCORE_SOURCE_IMAGE_TYPES
        ]

        # 2. 计算分数和状态
        scores_data = calculate_scores_from_images(images)

        # 3. 更新 cards 表
        query_update_card = (
            f"UPDATE {settings.DB_CARD_TABLE_NAME} SET "
            "detection_score = %s, modified_score = %s, is_edited = %s, updated_at = NOW() "
            "WHERE id = %s"
        )
        params = (
            scores_data["detection_score"],
            scores_data["modified_score"],
            scores_data["is_edited"],
            card_id,
        )
        cursor.execute(query_update_card, params)
        db_conn.commit()


def _parse_json_value(value: Any) -> Optional[Dict[str, Any]]:
    if value is None:
        return None
    if isinstance(value, str):
        try:
            return json.loads(value)
        except json.JSONDecodeError:
            return None
    return value if isinstance(value, dict) else None


def _has_center_box_shapes(center_result: Any) -> bool:
    if not isinstance(center_result, dict):
        return False
    box_result = center_result.get("box_result", {})
    if not isinstance(box_result, dict):
        return False
    inner_shapes = box_result.get("inner_box", {}).get("shapes", [])
    outer_shapes = box_result.get("outer_box", {}).get("shapes", [])
    return bool(inner_shapes or outer_shapes)


def _extract_center_result(json_data: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
    if not json_data:
        return None
    center_result = json_data.get("result", {}).get("center_result")
    if _has_center_box_shapes(center_result):
        return copy.deepcopy(center_result)
    return None


def _apply_center_result(json_data: Optional[Dict[str, Any]], center_result: Dict[str, Any]) -> Dict[str, Any]:
    payload = copy.deepcopy(json_data) if isinstance(json_data, dict) else copy.deepcopy(EMPTY_DETECTION_JSON)
    payload.setdefault("result", {})["center_result"] = copy.deepcopy(center_result)
    return payload


def _share_side_center_results(images: List[CardImageResponse]) -> None:
    """同面各图共享 center_result（优先 fusion/ring，其次 stripe）。"""
    side_priority_types = {
        "front": [
            ImageType.front_fusion.value,
            ImageType.front_ring.value,
            ImageType.front_coaxial.value,
            ImageType.front_stripe1.value,
            ImageType.front_stripe2.value,
            ImageType.front_stripe3.value,
            ImageType.front_stripe4.value,
        ],
        "back": [
            ImageType.back_fusion.value,
            ImageType.back_ring.value,
            ImageType.back_coaxial.value,
            ImageType.back_stripe1.value,
            ImageType.back_stripe2.value,
            ImageType.back_stripe3.value,
            ImageType.back_stripe4.value,
        ],
    }

    for side, priority_types in side_priority_types.items():
        side_center = None
        for image_type in priority_types:
            for img in images:
                if img.image_type != image_type:
                    continue
                for json_field in ("modified_json", "detection_json"):
                    source = _extract_center_result(getattr(img, json_field, None))
                    if source:
                        side_center = source
                        break
                if side_center:
                    break
            if side_center:
                break

        if not side_center:
            continue

        prefix = f"{side}_"
        for img in images:
            if not img.image_type.startswith(prefix):
                continue
            if img.image_type.endswith("_gray"):
                continue
            img.detection_json = _apply_center_result(img.detection_json, side_center)
            if img.modified_json:
                img.modified_json = _apply_center_result(img.modified_json, side_center)


def _construct_gray_image_json(gray_type: str, ring_image_data: Optional[Dict[str, Any]]) -> Dict[str, Any]:
    """
    内部辅助：构建辅助图（灰度图/融合图）的 modified_json
    gray_type: front_gray / back_gray / front_fusion / back_fusion
    ring_image_data: 对应的 ring 图的数据库原始字典数据 (包含 detection_json/modified_json 字段)
    """
    if not ring_image_data:
        return None

    # 获取 Ring 图最新的 JSON (优先取 modified, 没有则取 detection)
    source_json = ring_image_data.get('modified_json')
    if not source_json:
        source_json = ring_image_data.get('detection_json')

    # 解析 JSON 字符串
    if isinstance(source_json, str):
        try:
            source_json = json.loads(source_json)
        except:
            return None

    if not source_json:
        return None

    # 开始筛选 defects
    defects = source_json.get("result", {}).get("defect_result", {}).get("defects", [])
    filtered_defects = []

    for defect in defects:
        # 判断当前缺陷是否属于当前这张辅助图
        # 如果是融合图，就找 fusion_id；如果是灰度图，就找 gray_id
        is_fusion = gray_type in ("front_fusion", "back_fusion")
        key_to_check = "fusion_id" if is_fusion else "gray_id"
        
        if key_to_check in defect:
            filtered_defects.append(defect)

    # 即使没有缺陷，也最好返回一个空的结构，不要返回 None
    # 不然前端可能会觉得没这个字段或者解析报错
    gray_modified_json = copy.deepcopy(EMPTY_DETECTION_JSON)
    gray_modified_json["result"]["defect_result"]["defects"] = filtered_defects

    center_result = source_json.get("result", {}).get("center_result")
    if _has_center_box_shapes(center_result):
        gray_modified_json["result"]["center_result"] = copy.deepcopy(center_result)

    # 还可以把 Ring 图的宽高带过来，防止前端报错
    gray_modified_json["result"]["imageHeight"] = source_json.get("result", {}).get("imageHeight", 0)
    gray_modified_json["result"]["imageWidth"] = source_json.get("result", {}).get("imageWidth", 0)

    return gray_modified_json


def get_card_with_details(db_conn: PooledMySQLConnection, card_id: int) -> Optional[Dict[str, Any]]:
    """获取单个卡牌的完整信息，包含主图和灰度图。"""
    with db_conn.cursor(dictionary=True) as cursor:
        # 1. 获取卡牌信息
        query_card = f"SELECT * FROM {settings.DB_CARD_TABLE_NAME} WHERE id = %s"
        cursor.execute(query_card, (card_id,))
        card_data = cursor.fetchone()
        if not card_data:
            return None

        # 2. 获取主图片 (Card Images)
        query_images = f"SELECT * FROM {settings.DB_IMAGE_TABLE_NAME} WHERE card_id = %s"
        cursor.execute(query_images, (card_id,))
        main_image_records = cursor.fetchall()

        # 3. 获取灰度图片 (Gray Images)
        query_gray = f"SELECT * FROM {settings.DB_GRAY_IMAGE_TABLE_NAME} WHERE card_id = %s"
        cursor.execute(query_gray, (card_id,))
        gray_image_records = cursor.fetchall()

        # 4. 寻找 Ring 图的数据，用于辅助构建灰度图 JSON
        # 建立映射: ImageType -> Record Dict
        main_images_map = {rec['image_type']: rec for rec in main_image_records}

        final_images_list = []

        # 处理主图片，补充全路径
        for row in main_image_records:
            row['image_path'] = settings.get_full_url(row.get('image_path'))
            row['detection_image_path'] = settings.get_full_url(row.get('detection_image_path'))
            row['modified_image_path'] = settings.get_full_url(row.get('modified_image_path'))
            final_images_list.append(CardImageResponse.model_validate(row))

        # 处理 card_gray_images：真灰度图 + ring/stripe 等辅助图（导入时可能落在此表）
        for row in gray_image_records:
            g_type = row['image_type']
            # 主表已有同类型时跳过，避免 images 列表重复
            if g_type in main_images_map:
                continue
            fusion_type = None
            if g_type.startswith("front_") and g_type != ImageType.front_gray.value:
                fusion_type = ImageType.front_fusion.value
            elif g_type.startswith("back_") and g_type != ImageType.back_gray.value:
                fusion_type = ImageType.back_fusion.value

            if g_type in (ImageType.front_gray.value, ImageType.back_gray.value):
                target_ring_type = (
                    ImageType.front_ring.value
                    if g_type == ImageType.front_gray.value
                    else ImageType.back_ring.value
                )
                virtual_detection_json = copy.deepcopy(EMPTY_DETECTION_JSON)
                ring_data = main_images_map.get(target_ring_type)
                virtual_modified_json = _construct_gray_image_json(g_type, ring_data)
                ring_detection = _parse_json_value(ring_data.get("detection_json") if ring_data else None)
                center_result = _extract_center_result(ring_detection)
                if center_result:
                    virtual_detection_json = _apply_center_result(virtual_detection_json, center_result)
            elif fusion_type and fusion_type in main_images_map:
                # ring / stripe 等在辅助表时，JSON 与融合图共用
                fusion_row = main_images_map[fusion_type]
                virtual_detection_json = fusion_row.get("detection_json") or copy.deepcopy(EMPTY_DETECTION_JSON)
                if isinstance(virtual_detection_json, str):
                    virtual_detection_json = json.loads(virtual_detection_json)
                virtual_modified_json = fusion_row.get("modified_json")
                if isinstance(virtual_modified_json, str):
                    virtual_modified_json = json.loads(virtual_modified_json)
            else:
                virtual_detection_json = copy.deepcopy(EMPTY_DETECTION_JSON)
                virtual_modified_json = None

            gray_image_dict = {
                "id": row['id'],
                "card_id": row['card_id'],
                "image_type": g_type,
                "image_path": settings.get_full_url(row['image_path']),
                "created_at": row['created_at'],
                "updated_at": row['updated_at'],
                "detection_json": virtual_detection_json,
                "modified_json": virtual_modified_json,
                "image_name": None,
                "detection_image_path": None,
                "modified_image_path": None,
                "is_edited": False,
            }
            final_images_list.append(CardImageResponse.model_validate(gray_image_dict))

        _share_side_center_results(final_images_list)

        # 5. 获取分数详情：fusion/ring/coaxial 参与算分（每面仅用一份 JSON，不重复扣分）
        main_images_objs = [
            img for img in final_images_list
            if img.image_type in SCORE_SOURCE_IMAGE_TYPES
        ]
        score_details = calculate_scores_from_images(main_images_objs)

        # 6. 对图片列表进行自定义排序（14 类：每面 fusion / gray / ring / stripe1-4）
        sort_priority = {
            ImageType.front_fusion.value: 0,
            ImageType.back_fusion.value: 1,
            ImageType.front_gray.value: 2,
            ImageType.back_gray.value: 3,
            ImageType.front_ring.value: 4,
            ImageType.back_ring.value: 5,
            ImageType.front_stripe1.value: 6,
            ImageType.front_stripe2.value: 7,
            ImageType.front_stripe3.value: 8,
            ImageType.front_stripe4.value: 9,
            ImageType.back_stripe1.value: 10,
            ImageType.back_stripe2.value: 11,
            ImageType.back_stripe3.value: 12,
            ImageType.back_stripe4.value: 13,
            ImageType.front_coaxial.value: 14,
            ImageType.back_coaxial.value: 15,
        }
        final_images_list.sort(key=lambda x: sort_priority.get(x.image_type, 999))

        card_data.update({
            "images": final_images_list,  # 包含所有图片
            "detection_score": score_details["detection_score"],
            "modified_score": score_details["modified_score"],
            "detection_score_detail": score_details["detection_score_detail"],
            "modified_score_detail": score_details["modified_score_detail"]
        })

        return card_data


def get_card_list_with_images(
        db_conn: PooledMySQLConnection,
        card_id: Optional[int],
        card_name: Optional[str],
        card_type: Optional[CardType],
        is_edited: Optional[bool],
        min_detect_score: Optional[float],
        max_detect_score: Optional[float],
        min_mod_score: Optional[float],
        max_mod_score: Optional[float],
        created_start: Optional[date],
        created_end: Optional[date],
        updated_start: Optional[date],
        updated_end: Optional[date],
        sort_by: SortBy,
        sort_order: SortOrder,
        skip: int,
        limit: int
) -> List[Dict[str, Any]]:
    # 此函数逻辑主要是列表展示，不需要复杂的 JSON 构造，
    # 只需要把灰度图的路径也带出来即可。

    with db_conn.cursor(dictionary=True) as cursor:
        query = f"SELECT * FROM {settings.DB_CARD_TABLE_NAME}"
        conditions = []
        params = []
        if card_id is not None: conditions.append("id = %s"); params.append(card_id)
        if card_name: conditions.append("card_name LIKE %s"); params.append(f"%{card_name}%")
        if card_type: conditions.append("card_type = %s"); params.append(card_type.value)
        if is_edited is not None: conditions.append("is_edited = %s"); params.append(is_edited)
        if min_detect_score is not None: conditions.append("detection_score >= %s"); params.append(min_detect_score)
        if max_detect_score is not None: conditions.append("detection_score <= %s"); params.append(max_detect_score)
        if min_mod_score is not None: conditions.append("modified_score >= %s"); params.append(min_mod_score)
        if max_mod_score is not None: conditions.append("modified_score <= %s"); params.append(max_mod_score)
        if created_start: conditions.append("DATE(created_at) >= %s"); params.append(created_start)
        if created_end: conditions.append("DATE(created_at) <= %s"); params.append(created_end)
        if updated_start: conditions.append("DATE(updated_at) >= %s"); params.append(updated_start)
        if updated_end: conditions.append("DATE(updated_at) <= %s"); params.append(updated_end)

        if conditions: query += " WHERE " + " AND ".join(conditions)
        query += f" ORDER BY {sort_by.value} {sort_order.value}, id DESC"
        query += " LIMIT %s OFFSET %s"
        params.extend([limit, skip])

        cursor.execute(query, tuple(params))
        cards = cursor.fetchall()

        if not cards:
            return []

        card_ids = [card['id'] for card in cards]
        format_strings = ','.join(['%s'] * len(card_ids))

        # 1. 查询主图片
        image_query = (
            f"SELECT id, card_id, image_type, image_path, detection_image_path, modified_image_path "
            f"FROM {settings.DB_IMAGE_TABLE_NAME} WHERE card_id IN ({format_strings})"
        )
        cursor.execute(image_query, tuple(card_ids))
        images = cursor.fetchall()

        # 2. 查询灰度图片
        gray_query = (
            f"SELECT id, card_id, image_type, image_path "
            f"FROM {settings.DB_GRAY_IMAGE_TABLE_NAME} WHERE card_id IN ({format_strings})"
        )
        cursor.execute(gray_query, tuple(card_ids))
        gray_images = cursor.fetchall()

        # 分组
        images_by_card_id = {}
        for img in images:
            cid = img['card_id']
            if cid not in images_by_card_id: images_by_card_id[cid] = []
            images_by_card_id[cid].append(img)

        for g_img in gray_images:
            cid = g_img['card_id']
            if cid not in images_by_card_id: images_by_card_id[cid] = []
            # 补齐字段结构以便前端统一处理
            g_img['detection_image_path'] = None
            g_img['modified_image_path'] = None
            images_by_card_id[cid].append(g_img)

        # 附加到卡牌
        for card in cards:
            card['image_path_list'] = {}
            card['detection_image_path_list'] = {}
            card['modified_image_path_list'] = {}

            related_images = images_by_card_id.get(card['id'], [])
            for image_data in related_images:
                image_type = image_data['image_type']
                if image_type:
                    card['image_path_list'][image_type] = settings.get_full_url(image_data.get('image_path'))
                    card['detection_image_path_list'][image_type] = settings.get_full_url(
                        image_data.get('detection_image_path'))
                    card['modified_image_path_list'][image_type] = settings.get_full_url(
                        image_data.get('modified_image_path'))

        return cards


def get_card_list_and_count(
        db_conn: PooledMySQLConnection,
        card_id: Optional[int],
        cardNo: Optional[str],
        card_name: Optional[str],
        card_type: Optional[CardType],
        is_edited: Optional[bool],
        review_state: Optional[int],
        min_detect_score: Optional[float],
        max_detect_score: Optional[float],
        min_mod_score: Optional[float],
        max_mod_score: Optional[float],
        created_start: Optional[date],
        created_end: Optional[date],
        updated_start: Optional[date],
        updated_end: Optional[date],
        sort_by: SortBy,
        sort_order: SortOrder,
        skip: int,
        limit: int,
        permission_user_id: Optional[str] = None,
        bound_user_id: Optional[str] = None
) -> Dict[str, Any]:
    with db_conn.cursor(dictionary=True) as cursor:
        conditions = []
        params = []

        if card_id is not None: conditions.append("id = %s"); params.append(card_id)
        if cardNo: conditions.append("cardNo LIKE %s"); params.append(f"%{cardNo}%")
        if card_name: conditions.append("card_name LIKE %s"); params.append(f"%{card_name}%")
        if card_type: conditions.append("card_type = %s"); params.append(card_type.value)
        if is_edited is not None: conditions.append("is_edited = %s"); params.append(is_edited)
        if review_state is not None: conditions.append("review_state = %s"); params.append(review_state)
        if min_detect_score is not None: conditions.append("detection_score >= %s"); params.append(min_detect_score)
        if max_detect_score is not None: conditions.append("detection_score <= %s"); params.append(max_detect_score)
        if min_mod_score is not None: conditions.append("modified_score >= %s"); params.append(min_mod_score)
        if max_mod_score is not None: conditions.append("modified_score <= %s"); params.append(max_mod_score)
        if created_start: conditions.append("DATE(created_at) >= %s"); params.append(created_start)
        if created_end: conditions.append("DATE(created_at) <= %s"); params.append(created_end)
        if updated_start: conditions.append("DATE(updated_at) >= %s"); params.append(updated_start)
        if updated_end: conditions.append("DATE(updated_at) <= %s"); params.append(updated_end)
        if permission_user_id is not None:
            conditions.append(
                f"id IN (SELECT card_id FROM `{settings.DB_USER_CARD_TABLE_NAME}` WHERE user_id = %s)"
            )
            params.append(permission_user_id)

        where_clause = ""
        if conditions: where_clause = " WHERE " + " AND ".join(conditions)

        # Count
        count_query = f"SELECT COUNT(*) as total FROM {settings.DB_CARD_TABLE_NAME}" + where_clause
        cursor.execute(count_query, tuple(params))
        total_count = cursor.fetchone()['total']

        # List
        data_query = f"SELECT * FROM {settings.DB_CARD_TABLE_NAME}" + where_clause
        data_query += f" ORDER BY {sort_by.value} {sort_order.value}, id DESC"
        data_query += " LIMIT %s OFFSET %s"
        data_params = params.copy()
        data_params.extend([limit, skip])

        cursor.execute(data_query, tuple(data_params))
        cards = cursor.fetchall()

        if cards:
            card_ids = [card['id'] for card in cards]
            format_strings = ','.join(['%s'] * len(card_ids))

            bound_card_ids = set()
            if bound_user_id is not None:
                # 只标记当前页卡片是否绑定到指定外部用户，不影响原有列表筛选逻辑。
                bound_query = (
                    f"SELECT card_id FROM `{settings.DB_USER_CARD_TABLE_NAME}` "
                    f"WHERE user_id = %s AND card_id IN ({format_strings})"
                )
                cursor.execute(bound_query, tuple([bound_user_id] + card_ids))
                bound_card_ids = {row['card_id'] for row in cursor.fetchall()}

            # 主图
            image_query = (
                f"SELECT id, card_id, image_type, image_path, detection_image_path, modified_image_path "
                f"FROM {settings.DB_IMAGE_TABLE_NAME} WHERE card_id IN ({format_strings})"
            )
            cursor.execute(image_query, tuple(card_ids))
            images = cursor.fetchall()

            # [NEW] 灰度图
            gray_query = (
                f"SELECT id, card_id, image_type, image_path "
                f"FROM {settings.DB_GRAY_IMAGE_TABLE_NAME} WHERE card_id IN ({format_strings})"
            )
            cursor.execute(gray_query, tuple(card_ids))
            gray_images = cursor.fetchall()

            images_by_card_id = {}
            for img in images:
                cid = img['card_id']
                if cid not in images_by_card_id: images_by_card_id[cid] = []
                images_by_card_id[cid].append(img)

            # 混入灰度图
            for g_img in gray_images:
                cid = g_img['card_id']
                if cid not in images_by_card_id: images_by_card_id[cid] = []
                g_img['detection_image_path'] = None
                g_img['modified_image_path'] = None
                images_by_card_id[cid].append(g_img)

            for card in cards:
                card['is_bound'] = card['id'] in bound_card_ids
                card['image_path_list'] = {}
                card['detection_image_path_list'] = {}
                card['modified_image_path_list'] = {}
                related_images = images_by_card_id.get(card['id'], [])
                for image_data in related_images:
                    image_type = image_data['image_type']
                    if image_type:
                        card['image_path_list'][image_type] = settings.get_full_url(image_data.get('image_path'))
                        card['detection_image_path_list'][image_type] = settings.get_full_url(
                            image_data.get('detection_image_path'))
                        card['modified_image_path_list'][image_type] = settings.get_full_url(
                            image_data.get('modified_image_path'))

        return {
            "total": total_count,
            "list": cards
        }