Ver código fonte

改进列表查询, 增加总数

AnlaAnla 1 mês atrás
pai
commit
098c91d318
3 arquivos alterados com 200 adições e 2 exclusões
  1. 63 2
      app/api/cards.py
  2. 128 0
      app/crud/crud_card.py
  3. 9 0
      app/utils/scheme.py

+ 63 - 2
app/api/cards.py

@@ -9,7 +9,8 @@ from app.core.config import settings
 from app.core.logger import get_logger
 from app.core.database_loader import get_db_connection
 from app.utils.scheme import (
-    CardDetailResponse, CardListDetailResponse, CardType, SortBy, SortOrder
+    CardDetailResponse, CardListDetailResponse, CardType, SortBy,
+    SortOrder, CardListResponseWrapper, CardListWithTotal
 )
 from app.crud import crud_card
 
@@ -132,7 +133,7 @@ def list_cards_detailed(
 ):
     """获取卡牌的基础信息列表,支持按名称、类型筛选,以及多字段排序和分页。"""
     if page_num is not None:
-        skip = (page_num-1) * limit
+        skip = (page_num - 1) * limit
     try:
         cards_with_images = crud_card.get_card_list_with_images(
             db_conn,
@@ -152,6 +153,66 @@ def list_cards_detailed(
         raise HTTPException(status_code=500, detail="获取数据列表失败。")
 
 
+@router.get("/card_list_filter", response_model=CardListResponseWrapper, summary="获取卡牌列表和总数")
+def card_list_filter(
+        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=0, le=10),
+        max_detection_score: Optional[float] = Query(None, ge=0, le=10),
+        min_modified_score: Optional[float] = Query(None, ge=0, le=10),
+        max_modified_score: Optional[float] = Query(None, ge=0, le=10),
+
+        created_start: Optional[date] = Query(None),
+        created_end: Optional[date] = Query(None),
+        updated_start: Optional[date] = Query(None),
+        updated_end: Optional[date] = Query(None),
+
+        sort_by: SortBy = Query(SortBy.updated_at),
+        sort_order: SortOrder = Query(SortOrder.desc),
+        skip: int = Query(0, ge=0),
+        page_num: int = Query(None, ge=0),
+        limit: int = Query(100, ge=1, le=1000),
+        db_conn: PooledMySQLConnection = db_dependency
+):
+    """
+    获取卡牌列表,返回格式包含 total 和 list。
+    结构: { "data": { "total": 100, "list": [...] } }
+    """
+    if page_num is not None:
+        skip = page_num * limit
+
+    try:
+        result = crud_card.get_card_list_and_count(
+            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
+        )
+
+        # 组装返回数据,注意这里要进行 model_validate 转换 list 中的每一项
+        # 1. 先把 list 里的字典转成 Pydantic 对象
+        validated_list = [CardListDetailResponse.model_validate(c) for c in result['list']]
+
+        # 2. 构造 data 部分
+        data_content = CardListWithTotal(
+            total=result['total'],
+            list=validated_list
+        )
+
+        # 3. 构造最外层包装
+        return CardListResponseWrapper(data=data_content)
+
+    except Exception as e:
+        logger.error(f"查询卡牌列表(带总数)失败: {e}")
+        raise HTTPException(status_code=500, detail="获取数据列表失败。")
+
+
 @router.delete("/delete/{id}", status_code=200, summary="删除卡牌及其所有关联图片")
 def delete_card(id: int, db_conn: PooledMySQLConnection = db_dependency):
     """

+ 128 - 0
app/crud/crud_card.py

@@ -203,3 +203,131 @@ def get_card_list_with_images(
                     card['modified_image_path_list'][image_type] = image_data.get('modified_image_path')
 
         return cards
+
+
+def get_card_list_and_count(
+        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
+) -> Dict[str, Any]:
+    """
+    同时获取符合条件的记录列表和总条数。
+    """
+    with db_conn.cursor(dictionary=True) as cursor:
+        # --- 第一部分:构建 WHERE 子句 (条件逻辑与之前保持一致) ---
+        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)
+
+        where_clause = ""
+        if conditions:
+            where_clause = " WHERE " + " AND ".join(conditions)
+
+        # --- 第二部分:查询总数 (Total) ---
+        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"
+
+        # 复制一份 params 用于数据查询,避免影响原来的列表
+        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))
+            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()
+
+            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)
+
+            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 {
+            "total": total_count,
+            "list": cards
+        }

+ 9 - 0
app/utils/scheme.py

@@ -151,3 +151,12 @@ class CardListDetailResponse(BaseModel):
 
     class Config:
         from_attributes = True
+
+
+class CardListWithTotal(BaseModel):
+    total: int
+    list: List[CardListDetailResponse]
+
+
+class CardListResponseWrapper(BaseModel):
+    data: CardListWithTotal