Explorar el Código

用户解绑卡牌, 列表判断userid是否绑定过

AnlaAnla hace 1 semana
padre
commit
20d6ed275a
Se han modificado 4 ficheros con 58 adiciones y 19 borrados
  1. 3 1
      app/api/cards.py
  2. 39 15
      app/api/users.py
  3. 15 3
      app/crud/crud_card.py
  4. 1 0
      app/utils/scheme.py

+ 3 - 1
app/api/cards.py

@@ -125,6 +125,7 @@ def get_next_card_details(card_id: int, db_conn: PooledMySQLConnection = db_depe
 @router.get("/card_list_filter", response_model=CardListResponseWrapper, summary="获取卡牌列表和总数 [用户调用]")
 def card_list_filter(
         card_id: Optional[int] = Query(None, description="筛选:卡牌ID"),
+        user_id: Optional[int] = Query(None, ge=0, description="查询卡牌是否绑定到该外部用户ID"),
         cardNo: Optional[str] = Query(None, description="筛选:卡牌编号"),
         card_name: Optional[str] = Query(None, description="筛选:卡牌名称"),
         card_type: Optional[CardType] = Query(None, description="筛选:卡牌类型"),
@@ -166,7 +167,8 @@ def card_list_filter(
             created_start, created_end,
             updated_start, updated_end,
             sort_by, sort_order, skip, limit,
-            None if current_user.get("is_admin") else current_user["id"]
+            None if current_user.get("is_admin") else current_user["id"],
+            user_id
         )
 
         # 组装返回数据,注意这里要进行 model_validate 转换 list 中的每一项

+ 39 - 15
app/api/users.py

@@ -17,7 +17,8 @@ db_dependency = Depends(get_db_connection)
 
 class BindCardRequest(BaseModel):
     user_id: int = Field(..., ge=0)
-    card_id: List[int] = Field(..., min_length=1)
+    bind_card_id: List[int] = Field(default_factory=list)
+    unbind_card_id: List[int] = Field(default_factory=list)
 
 
 def _auth_exception(detail: str = "用户认证信息无效") -> HTTPException:
@@ -96,33 +97,56 @@ def bind_card_to_user(
         db_conn: PooledMySQLConnection = db_dependency
 ):
     try:
-        # 请求字段保持 card_id,但支持一次绑定多个卡片,并自动去重。
-        card_ids = list(dict.fromkeys(data.card_id))
+        # 先解绑再绑定;如果同一张卡同时出现在两个列表,最终以绑定列表为准。
+        bind_card_ids = list(dict.fromkeys(data.bind_card_id))
+        unbind_card_ids = list(dict.fromkeys(data.unbind_card_id))
+        operation_card_ids = list(dict.fromkeys(unbind_card_ids + bind_card_ids))
+
+        if not operation_card_ids:
+            raise HTTPException(status_code=400, detail="绑定列表和解绑列表不能同时为空")
 
         with db_conn.cursor(dictionary=True) as cursor:
-            format_strings = ",".join(["%s"] * len(card_ids))
+            format_strings = ",".join(["%s"] * len(operation_card_ids))
             cursor.execute(
                 f"SELECT id FROM `{settings.DB_CARD_TABLE_NAME}` WHERE id IN ({format_strings})",
-                tuple(card_ids)
+                tuple(operation_card_ids)
             )
             existing_card_ids = {row["id"] for row in cursor.fetchall()}
-            missing_card_ids = sorted(set(card_ids) - existing_card_ids)
+            missing_card_ids = sorted(set(operation_card_ids) - existing_card_ids)
             if missing_card_ids:
                 raise HTTPException(status_code=404, detail=f"卡片未发现: {missing_card_ids}")
 
-            bind_params = [(data.user_id, card_id) for card_id in card_ids]
-            cursor.executemany(
-                f"INSERT IGNORE INTO `{settings.DB_USER_CARD_TABLE_NAME}` (user_id, card_id) VALUES (%s, %s)",
-                bind_params
-            )
-            inserted_count = cursor.rowcount
+            deleted_count = 0
+            if unbind_card_ids:
+                unbind_format_strings = ",".join(["%s"] * len(unbind_card_ids))
+                cursor.execute(
+                    f"DELETE FROM `{settings.DB_USER_CARD_TABLE_NAME}` "
+                    f"WHERE user_id = %s AND card_id IN ({unbind_format_strings})",
+                    tuple([data.user_id] + unbind_card_ids)
+                )
+                deleted_count = cursor.rowcount
+
+            inserted_count = 0
+            if bind_card_ids:
+                bind_params = [(data.user_id, card_id) for card_id in bind_card_ids]
+                cursor.executemany(
+                    f"INSERT IGNORE INTO `{settings.DB_USER_CARD_TABLE_NAME}` (user_id, card_id) VALUES (%s, %s)",
+                    bind_params
+                )
+                inserted_count = cursor.rowcount
+
             db_conn.commit()
 
-        logger.info(f"Admin {current_user['id']} bound cards {card_ids} to external user {data.user_id}")
+        logger.info(
+            f"Admin {current_user['id']} updated card bindings for external user {data.user_id}: "
+            f"bind={bind_card_ids}, unbind={unbind_card_ids}"
+        )
         return {
-            "message": "卡片绑定成功",
+            "message": "卡片绑定关系更新成功",
             "user_id": data.user_id,
-            "card_id": card_ids,
+            "bind_card_id": bind_card_ids,
+            "unbind_card_id": unbind_card_ids,
+            "deleted_count": deleted_count,
             "inserted_count": inserted_count
         }
 

+ 15 - 3
app/crud/crud_card.py

@@ -323,7 +323,8 @@ def get_card_list_and_count(
         sort_order: SortOrder,
         skip: int,
         limit: int,
-        user_id: Optional[int] = None
+        permission_user_id: Optional[int] = None,
+        bound_user_id: Optional[int] = None
 ) -> Dict[str, Any]:
     with db_conn.cursor(dictionary=True) as cursor:
         conditions = []
@@ -343,11 +344,11 @@ def get_card_list_and_count(
         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 user_id is not None:
+        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(user_id)
+            params.append(permission_user_id)
 
         where_clause = ""
         if conditions: where_clause = " WHERE " + " AND ".join(conditions)
@@ -371,6 +372,16 @@ def get_card_list_and_count(
             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 "
@@ -402,6 +413,7 @@ def get_card_list_and_count(
                 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'] = {}

+ 1 - 0
app/utils/scheme.py

@@ -174,6 +174,7 @@ class CardListDetailResponse(BaseModel):
     image_path_list: Dict[str, Optional[str]] = {}
     detection_image_path_list: Dict[str, Optional[str]] = {}
     modified_image_path_list: Dict[str, Optional[str]] = {}
+    is_bound: bool = False
 
     class Config:
         from_attributes = True