CardScoreDataServer/app/crud/crud_card.py

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

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


def update_card_scores_and_status(db_conn: PooledMySQLConnection, card_id: int):
    """
    根据卡牌关联的图片数量和内容，更新cards表中的分数和状态。
    - 如果图片满4张，计算并更新 detection_score, modified_score, is_edited。
    - 如果图片不足4张，将分数置为NULL，is_edited置为False。
    - 自动更新 updated_at 字段。
    """
    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()

        # 将数据库行转换为Pydantic模型，便于处理JSON
        images = [CardImageResponse.model_validate(row) for row in image_records]

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

        # 3. 更新 cards 表
        # 注意: updated_at 会由数据库自动更新
        query_update_card = (
            f"UPDATE {settings.DB_CARD_TABLE_NAME} SET "
            "detection_score = %s, modified_score = %s, is_edited = %s, updated_at = %s "
            "WHERE id = %s"
        )
        params = (
            scores_data["detection_score"],
            scores_data["modified_score"],
            scores_data["is_edited"],
            datetime.now(),  # 手动更新时间戳以确保触发
            card_id,
        )
        cursor.execute(query_update_card, params)
        db_conn.commit()


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. 获取所有关联的图片信息
        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]

        # 3. 获取分数详情 (如果需要)
        # 只有当图片满4张时，分数详情才有意义
        score_details = calculate_scores_from_images(images)
        card_data.update({
            "images": images,
            "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_name: Optional[str],
        card_type: Optional[CardType],
        sort_by: SortBy,
        sort_order: SortOrder,
        skip: int,
        limit: int
) -> List[Dict[str, Any]]:
    """获取带筛选和排序功能的卡牌列表，并附带其关联的图片信息。"""
    with db_conn.cursor(dictionary=True) as cursor:
        # 1. 构建动态查询
        query = f"SELECT * FROM {settings.DB_CARD_TABLE_NAME}"
        conditions = []
        params = []

        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 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 []

        # 2. 一次性获取所有相关卡牌的图片 (避免 N+1 查询)
        card_ids = [card['id'] for card in cards]

        # 使用 IN 子句和占位符
        format_strings = ','.join(['%s'] * len(card_ids))
        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()

        # 3. 将图片按 card_id 分组
        images_by_card_id = {}
        for image in images:
            cid = image['card_id']
            if cid not in images_by_card_id:
                images_by_card_id[cid] = []
            images_by_card_id[cid].append(image)

        # 4. 将图片附加到对应的卡牌上
        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'], [])

            # 遍历图片记录，用 image_type 作为 key 来填充三个字典
            for image_data in related_images:
                image_type = image_data['image_type']  # e.g., 'front_face'
                if image_type:  # 确保 image_type 不为空
                    card['image_path_list'][image_type] = image_data.get('image_path')
                    card['detection_image_path_list'][image_type] = image_data.get('detection_image_path')
                    card['modified_image_path_list'][image_type] = image_data.get('modified_image_path')

        return cards