From cb399ef3bbbad410794bc1e5f3197bad2c82af34 Mon Sep 17 00:00:00 2001 From: ZZX9599 <536509593@qq.com> Date: Wed, 3 Sep 2025 20:07:38 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BA=BA=E8=84=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- schema/face_schema.py | 33 +++++ service/face_service.py | 268 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 301 insertions(+) create mode 100644 schema/face_schema.py create mode 100644 service/face_service.py diff --git a/schema/face_schema.py b/schema/face_schema.py new file mode 100644 index 0000000..5bb5a80 --- /dev/null +++ b/schema/face_schema.py @@ -0,0 +1,33 @@ +from datetime import datetime +from pydantic import BaseModel, Field + + +# ------------------------------ +# 请求模型（前端传参校验） +# ------------------------------ +class FaceCreateRequest(BaseModel): + """创建人脸记录请求模型""" + id: int = Field(..., description="主键ID") + name: str = Field(None, max_length=255, description="名称") + eigenvalue: str = Field(None, max_length=255, description="特征") + + +class FaceUpdateRequest(BaseModel): + """更新人脸记录请求模型""" + name: str = Field(None, max_length=255, description="名称") + eigenvalue: str = Field(None, max_length=255, description="特征") + + +# ------------------------------ +# 响应模型（后端返回数据） +# ------------------------------ +class FaceResponse(BaseModel): + """人脸记录响应模型""" + id: int = Field(..., description="主键ID") + name: str = Field(None, description="名称") + eigenvalue: str = Field(None, description="特征") + created_at: datetime = Field(..., description="记录创建时间") + updated_at: datetime = Field(..., description="记录更新时间") + + # 支持从数据库查询结果转换 + model_config = {"from_attributes": True} diff --git a/service/face_service.py b/service/face_service.py new file mode 100644 index 0000000..7d767d7 --- /dev/null +++ b/service/face_service.py @@ -0,0 +1,268 @@ +from fastapi import APIRouter, Depends, HTTPException +from mysql.connector import Error as MySQLError + +from ds.db import db +from schema.face_schema import FaceCreateRequest, FaceUpdateRequest, FaceResponse +from schema.response_schema import APIResponse +from middle.auth_middleware import get_current_user +from schema.user_schema import UserResponse + +# 创建人脸接口路由（前缀 /faces、标签用于 Swagger 分类） +router = APIRouter( + prefix="/faces", + tags=["人脸管理"] +) + + +# ------------------------------ +# 1. 创建人脸记录 +# ------------------------------ +@router.post("", response_model=APIResponse, summary="创建人脸记录") +async def create_face( + face: FaceCreateRequest +): + """ + 创建人脸记录： + - 需登录认证 + - 插入新的人脸记录到数据库 + - 返回创建成功信息 + """ + conn = None + cursor = None + try: + conn = db.get_connection() + cursor = conn.cursor(dictionary=True) + + # 1. 检查ID是否已存在 + check_query = "SELECT id FROM face WHERE id = %s" + cursor.execute(check_query, (face.id,)) + existing_face = cursor.fetchone() + if existing_face: + raise HTTPException( + status_code=400, + detail=f"ID为 {face.id} 的人脸记录已存在" + ) + + # 2. 插入新人脸记录到数据库 + insert_query = """ + INSERT INTO face (id, name, eigenvalue) + VALUES (%s, %s, %s) + """ + cursor.execute(insert_query, (face.id, face.name, face.eigenvalue)) + conn.commit() + + # 3. 查询刚创建的记录并返回 + select_query = "SELECT * FROM face WHERE id = %s" + cursor.execute(select_query, (face.id,)) + created_face = cursor.fetchone() + + return APIResponse( + code=201, # 201 表示资源创建成功 + message="人脸记录创建成功", + data=FaceResponse(**created_face) + ) + except MySQLError as e: + if conn: + conn.rollback() + raise Exception(f"创建人脸记录失败：{str(e)}") from e + finally: + db.close_connection(conn, cursor) + + +# ------------------------------ +# 2. 获取单个人脸记录 +# ------------------------------ +@router.get("/{face_id}", response_model=APIResponse, summary="获取单个人脸记录") +async def get_face( + face_id: int, + current_user: UserResponse = Depends(get_current_user) # 需登录认证 +): + """ + 获取单个人脸记录： + - 需登录认证 + - 根据ID查询人脸记录 + - 返回查询到的人脸信息 + """ + conn = None + cursor = None + try: + conn = db.get_connection() + cursor = conn.cursor(dictionary=True) + + query = "SELECT * FROM face WHERE id = %s" + cursor.execute(query, (face_id,)) + face = cursor.fetchone() + + if not face: + raise HTTPException( + status_code=404, + detail=f"ID为 {face_id} 的人脸记录不存在" + ) + + return APIResponse( + code=200, + message="人脸记录查询成功", + data=FaceResponse(**face) + ) + except MySQLError as e: + raise Exception(f"查询人脸记录失败：{str(e)}") from e + finally: + db.close_connection(conn, cursor) + + +# ------------------------------ +# 3. 获取所有人脸记录 +# ------------------------------ +@router.get("", response_model=APIResponse, summary="获取所有人脸记录") +async def get_all_faces( +): + """ + 获取所有人脸记录： + - 需登录认证 + - 查询所有人脸记录（不需要分页） + - 返回所有人脸信息列表 + """ + conn = None + cursor = None + try: + conn = db.get_connection() + cursor = conn.cursor(dictionary=True) + + query = "SELECT * FROM face ORDER BY id" + cursor.execute(query) + faces = cursor.fetchall() + + return APIResponse( + code=200, + message="所有人脸记录查询成功", + data=[FaceResponse(**face) for face in faces] + ) + except MySQLError as e: + raise Exception(f"查询所有人脸记录失败：{str(e)}") from e + finally: + db.close_connection(conn, cursor) + + +# ------------------------------ +# 4. 更新人脸记录 +# ------------------------------ +@router.put("/{face_id}", response_model=APIResponse, summary="更新人脸记录") +async def update_face( + face_id: int, + face_update: FaceUpdateRequest, + current_user: UserResponse = Depends(get_current_user) # 需登录认证 +): + """ + 更新人脸记录： + - 需登录认证 + - 根据ID更新人脸记录信息 + - 返回更新后的人脸信息 + """ + conn = None + cursor = None + try: + conn = db.get_connection() + cursor = conn.cursor(dictionary=True) + + # 1. 检查记录是否存在 + check_query = "SELECT id FROM face WHERE id = %s" + cursor.execute(check_query, (face_id,)) + existing_face = cursor.fetchone() + if not existing_face: + raise HTTPException( + status_code=404, + detail=f"ID为 {face_id} 的人脸记录不存在" + ) + + # 2. 构建更新语句（只更新提供的字段） + update_fields = [] + params = [] + + if face_update.name is not None: + update_fields.append("name = %s") + params.append(face_update.name) + + if face_update.eigenvalue is not None: + update_fields.append("eigenvalue = %s") + params.append(face_update.eigenvalue) + + if not update_fields: + raise HTTPException( + status_code=400, + detail="至少需要提供一个字段进行更新" + ) + + params.append(face_id) # WHERE条件的参数 + + update_query = f""" + UPDATE face + SET {', '.join(update_fields)}, updated_at = CURRENT_TIMESTAMP + WHERE id = %s + """ + cursor.execute(update_query, params) + conn.commit() + + # 3. 查询更新后的记录并返回 + select_query = "SELECT * FROM face WHERE id = %s" + cursor.execute(select_query, (face_id,)) + updated_face = cursor.fetchone() + + return APIResponse( + code=200, + message="人脸记录更新成功", + data=FaceResponse(**updated_face) + ) + except MySQLError as e: + if conn: + conn.rollback() + raise Exception(f"更新人脸记录失败：{str(e)}") from e + finally: + db.close_connection(conn, cursor) + + +# ------------------------------ +# 5. 删除人脸记录 +# ------------------------------ +@router.delete("/{face_id}", response_model=APIResponse, summary="删除人脸记录") +async def delete_face( + face_id: int, + current_user: UserResponse = Depends(get_current_user) # 需登录认证 +): + """ + 删除人脸记录： + - 需登录认证 + - 根据ID删除人脸记录 + - 返回删除成功信息 + """ + conn = None + cursor = None + try: + conn = db.get_connection() + cursor = conn.cursor(dictionary=True) + + # 1. 检查记录是否存在 + check_query = "SELECT id FROM face WHERE id = %s" + cursor.execute(check_query, (face_id,)) + existing_face = cursor.fetchone() + if not existing_face: + raise HTTPException( + status_code=404, + detail=f"ID为 {face_id} 的人脸记录不存在" + ) + + # 2. 执行删除操作 + delete_query = "DELETE FROM face WHERE id = %s" + cursor.execute(delete_query, (face_id,)) + conn.commit() + + return APIResponse( + code=200, + message=f"ID为 {face_id} 的人脸记录删除成功", + data=None + ) + except MySQLError as e: + if conn: + conn.rollback() + raise Exception(f"删除人脸记录失败：{str(e)}") from e + finally: + db.close_connection(conn, cursor)