改进列表查询, 增加多个可筛选字段

app/api/cards.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, date
 import os
 from typing import Optional, List
 from fastapi import APIRouter, Depends, HTTPException, Query
@@ -114,19 +114,38 @@ def list_cards_detailed(
         card_id: Optional[int] = Query(None, description="筛选条件：卡牌ID (精确匹配)"),
         card_name: Optional[str] = Query(None, description="筛选条件：卡牌名称 (模糊匹配)"),
         card_type: Optional[CardType] = Query(None, description="筛选条件：卡牌类型"),
+        is_edited: Optional[bool] = Query(None, description="筛选：是否已编辑"),
+        min_detection_score: Optional[float] = Query(None, ge=-100, le=10, description="筛选：最小检测分"),
+        max_detection_score: Optional[float] = Query(None, ge=0, le=10, description="筛选：最大检测分"),
+        min_modified_score: Optional[float] = Query(None, ge=-100, le=10, description="筛选：最小修改分"),
+        max_modified_score: Optional[float] = Query(None, ge=0, le=10, description="筛选：最大修改分"),
+        created_start: Optional[date] = Query(None, description="筛选：创建日期起始 (含)"),
+        created_end: Optional[date] = Query(None, description="筛选：创建日期结束 (含)"),
+        updated_start: Optional[date] = Query(None, description="筛选：更新日期起始 (含)"),
+        updated_end: Optional[date] = Query(None, description="筛选：更新日期结束 (含)"),
         sort_by: SortBy = Query(SortBy.updated_at, description="排序字段"),
         sort_order: SortOrder = Query(SortOrder.desc, description="排序顺序"),
         skip: int = Query(0, ge=0, description="分页：跳过的记录数"),
+        page_num: int = Query(None, ge=0, description="分页：第n页(skip*limit)"),
         limit: int = Query(100, ge=1, le=1000, description="分页：每页的记录数"),
         db_conn: PooledMySQLConnection = db_dependency
 ):
     """获取卡牌的基础信息列表，支持按名称、类型筛选，以及多字段排序和分页。"""
+    if page_num is not None:
+        skip = page_num * limit
     try:
         cards_with_images = crud_card.get_card_list_with_images(
-            db_conn, card_id, card_name, card_type, sort_by, sort_order, skip, limit
+            db_conn,
+            card_id, card_name, card_type,
+            is_edited,
+            min_detection_score, max_detection_score,
+            min_modified_score, max_modified_score,
+            created_start, created_end,
+            updated_start, updated_end,
+            sort_by, sort_order, skip, limit
         )
         card_list = [CardListDetailResponse.model_validate(c) for c in cards_with_images]
-        logger.info(card_list)
+        logger.info(f"查询到 {len(card_list)} 条记录")
         return card_list
     except Exception as e:
         logger.error(f"查询卡牌列表失败: {e}")

app/crud/crud_card.py

@@ -1,4 +1,5 @@
 from typing import Optional, List, Dict, Any
+from datetime import date
 from mysql.connector.pooling import PooledMySQLConnection
 import json
 from datetime import datetime
@@ -78,6 +79,15 @@ def get_card_list_with_images(
         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,
@@ -90,7 +100,8 @@ def get_card_list_with_images(
         conditions = []
         params = []
 
-        if card_id:
+        # --- 基础筛选 ---
+        if card_id is not None:
             conditions.append("id = %s")
             params.append(card_id)
         if card_name:
@@ -100,24 +111,67 @@ def get_card_list_with_images(
             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. 一次性获取所有相关卡牌的图片 (避免 N+1 查询)
+        # 2. 一次性获取所有相关卡牌的图片 (逻辑保持不变)
         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 "
@@ -126,7 +180,7 @@ def get_card_list_with_images(
         cursor.execute(image_query, tuple(card_ids))
         images = cursor.fetchall()
 
-        # 3. 将图片按 card_id 分组
+        # 3. 将图片按 card_id 分组 (逻辑保持不变)
         images_by_card_id = {}
         for image in images:
             cid = image['card_id']
@@ -134,20 +188,16 @@ def get_card_list_with_images(
                 images_by_card_id[cid] = []
             images_by_card_id[cid].append(image)
 
-        # 4. 将图片附加到对应的卡牌上
+        # 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 不为空
+                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')