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

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_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]]:
    """获取带筛选和排序功能的卡牌列表，并附带其关联的图片信息。"""
    with db_conn.cursor(dictionary=True) as cursor:
        # 1. 构建动态查询
        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)

        # --- 分数范围 ---
        # detection_score
        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)

        # modified_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)

        # --- 新增筛选 3: 日期范围 ---
        # 使用 DATE() 函数将数据库的时间戳转换为日期进行比较，
        # 这样 '2023-01-01' 可以匹配 '2023-01-01 12:34:56'

        # Created At
        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)

        # Updated At
        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)

        # --- 拼接 WHERE 子句 ---
        if conditions:
            query += " WHERE " + " AND ".join(conditions)

        # --- 排序和分页 ---
        # 注意: 这里的 user input (sort_by) 是 Enum，比较安全，但在拼接 SQL 时仍需注意
        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. 一次性获取所有相关卡牌的图片 (逻辑保持不变)
        card_ids = [card['id'] for card in cards]
        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'], [])
            for image_data in related_images:
                image_type = image_data['image_type']
                if 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