去掉数据库

app/api/image_data.py

@@ -1,242 +0,0 @@
-import os
-import uuid
-import json
-from typing import Optional, Dict, Any, List
-from fastapi import APIRouter, File, UploadFile, Depends, HTTPException, Form, Query
-from fastapi.responses import JSONResponse
-from fastapi.concurrency import run_in_threadpool
-from pydantic import BaseModel, field_validator
-import mysql.connector
-
-from app.core.config import settings
-from app.core.logger import get_logger
-from app.core.database_loader import get_db_connection
-from app.services.score_service import ScoreService
-from app.api.score_inference import ScoreType
-
-logger = get_logger(__name__)
-router = APIRouter()
-
-
-# --- Pydantic 模型定义 ---
-class ImageDataUpdate(BaseModel):
-    img_result_json: Dict[str, Any]
-
-
-class ImageDataResponse(BaseModel):
-    img_id: int
-    img_name: Optional[str] = None
-    img_path: str
-    img_result_json: Dict[str, Any]
-
-    # 添加一个验证器来自动处理从数据库来的字符串
-    @field_validator('img_result_json', mode='before')
-    @classmethod
-    def parse_json_string(cls, value):
-        if isinstance(value, str):
-            try:
-                return json.loads(value)
-            except json.JSONDecodeError:
-                raise ValueError("Invalid JSON string provided")
-        return value
-
-
-# --- API 端点实现 ---
-
-@router.post("/", response_model=ImageDataResponse, summary="创建新的图片记录")
-async def create_image_data(
-        score_type: ScoreType = Form(...),
-        is_reflect_card: bool = Form(False),
-        img_name: Optional[str] = Form(None),
-        file: UploadFile = File(...),
-        db_conn: mysql.connector.connection.MySQLConnection = Depends(get_db_connection)
-):
-    # 1. 保存图片到本地
-    file_extension = os.path.splitext(file.filename)[1]
-    unique_filename = f"{uuid.uuid4()}{file_extension}"
-    img_path = settings.DATA_DIR / unique_filename
-
-    try:
-        image_bytes = await file.read()
-        with open(img_path, "wb") as f:
-            f.write(image_bytes)
-    except Exception as e:
-        logger.error(f"保存图片失败: {e}")
-        raise HTTPException(status_code=500, detail="无法保存图片文件")
-
-    # 2. 调用 ScoreService 生成 JSON 数据
-    try:
-        service = ScoreService()
-        json_result = await run_in_threadpool(
-            service.score_inference,
-            score_type=score_type.value,
-            is_reflect_card=is_reflect_card,
-            image_bytes=image_bytes
-        )
-    except Exception as e:
-        # 如果推理失败，删除已保存的图片
-        os.remove(img_path)
-        logger.error(f"分数推理失败: {e}")
-        raise HTTPException(status_code=500, detail=f"分数推理时发生错误: {e}")
-
-    # 3. 存入数据库
-    cursor = None
-    try:
-        cursor = db_conn.cursor(dictionary=True)
-        insert_query = (
-            f"INSERT INTO {settings.DB_TABLE_NAME} (img_name, img_path, img_result_json) "
-            "VALUES (%s, %s, %s)"
-        )
-        json_string = json.dumps(json_result, ensure_ascii=False)
-        cursor.execute(insert_query, (img_name, str(img_path), json_string))
-        new_id = cursor.lastrowid
-        db_conn.commit()
-
-        logger.info(f"成功创建记录, ID: {new_id}")
-        return {
-            "img_id": new_id,
-            "img_name": img_name,
-            "img_path": str(img_path),
-            "img_result_json": json_result
-        }
-
-    except mysql.connector.Error as err:
-        db_conn.rollback()
-        os.remove(img_path)
-        logger.error(f"数据库插入失败: {err}")
-        raise HTTPException(status_code=500, detail="数据库操作失败")
-    finally:
-        if cursor:
-            cursor.close()
-
-
-@router.get("/{img_id}", response_model=ImageDataResponse, summary="根据ID查询记录")
-def get_image_data_by_id(
-        img_id: int,
-        db_conn: mysql.connector.connection.MySQLConnection = Depends(get_db_connection)
-):
-    cursor = None
-    try:
-        cursor = db_conn.cursor(dictionary=True)
-        query = f"SELECT * FROM {settings.DB_TABLE_NAME} WHERE img_id = %s"
-        cursor.execute(query, (img_id,))
-        record = cursor.fetchone()
-
-        if record is None:
-            raise HTTPException(status_code=404, detail="记录未找到")
-
-        # Pydantic 验证器会自动处理转换，这里不再需要手动转换
-        return record
-
-    except mysql.connector.Error as err:
-        logger.error(f"数据库查询失败: {err}")
-        raise HTTPException(status_code=500, detail="数据库查询失败")
-    finally:
-        if cursor:
-            cursor.close()
-
-
-@router.get("/", response_model=List[ImageDataResponse], summary="根据名称查询记录")
-def get_image_data_by_name(
-        img_name: str,
-        db_conn: mysql.connector.connection.MySQLConnection = Depends(get_db_connection)
-):
-    cursor = None
-    try:
-        cursor = db_conn.cursor(dictionary=True)
-        query = f"SELECT * FROM {settings.DB_TABLE_NAME} WHERE img_name = %s"
-        cursor.execute(query, (img_name,))
-        records = cursor.fetchall()
-
-        # Pydantic 验证器会自动处理转换，这里不再需要手动转换
-        return records
-
-    except mysql.connector.Error as err:
-        logger.error(f"数据库查询失败: {err}")
-        raise HTTPException(status_code=500, detail="数据库查询失败")
-    finally:
-        if cursor:
-            cursor.close()
-
-
-@router.put("/{img_id}", response_model=ImageDataResponse, summary="更新记录的JSON数据")
-def update_image_data_json(
-        img_id: int,
-        data: ImageDataUpdate,
-        db_conn: mysql.connector.connection.MySQLConnection = Depends(get_db_connection)
-):
-    # ... (这个函数逻辑保持不变)
-    cursor = None
-    try:
-        cursor = db_conn.cursor(dictionary=True)
-        check_query = f"SELECT img_path FROM {settings.DB_TABLE_NAME} WHERE img_id = %s"
-        cursor.execute(check_query, (img_id,))
-        record = cursor.fetchone()
-        if not record:
-            raise HTTPException(status_code=404, detail="记录未找到")
-
-        update_query = (
-            f"UPDATE {settings.DB_TABLE_NAME} SET img_result_json = %s WHERE img_id = %s"
-        )
-        json_string = json.dumps(data.img_result_json, ensure_ascii=False)
-        cursor.execute(update_query, (json_string, img_id))
-        db_conn.commit()
-
-        if cursor.rowcount == 0:
-            raise HTTPException(status_code=404, detail="记录未找到或数据未改变")
-
-        logger.info(f"成功更新记录, ID: {img_id}")
-        return get_image_data_by_id(img_id, db_conn)
-
-    except mysql.connector.Error as err:
-        db_conn.rollback()
-        logger.error(f"数据库更新失败: {err}")
-        raise HTTPException(status_code=500, detail="数据库更新失败")
-    finally:
-        if cursor:
-            cursor.close()
-
-
-@router.delete("/{img_id}", summary="删除记录和对应的图片文件")
-def delete_image_data(
-        img_id: int,
-        db_conn: mysql.connector.connection.MySQLConnection = Depends(get_db_connection)
-):
-    # ... (这个函数无需修改)
-    cursor = None
-    try:
-        cursor = db_conn.cursor(dictionary=True)
-
-        query = f"SELECT img_path FROM {settings.DB_TABLE_NAME} WHERE img_id = %s"
-        cursor.execute(query, (img_id,))
-        record = cursor.fetchone()
-
-        if not record:
-            raise HTTPException(status_code=404, detail="记录未找到")
-
-        img_path = record['img_path']
-
-        delete_query = f"DELETE FROM {settings.DB_TABLE_NAME} WHERE img_id = %s"
-        cursor.execute(delete_query, (img_id,))
-        db_conn.commit()
-
-        if cursor.rowcount == 0:
-            raise HTTPException(status_code=404, detail="记录删除失败，可能已被删除")
-
-        try:
-            if os.path.exists(img_path):
-                os.remove(img_path)
-                logger.info(f"成功删除图片文件: {img_path}")
-        except Exception as e:
-            logger.error(f"删除图片文件失败: {img_path}, 错误: {e}")
-
-        logger.info(f"成功删除记录, ID: {img_id}")
-        return JSONResponse(content={"message": f"记录 {img_id} 已成功删除"}, status_code=200)
-
-    except mysql.connector.Error as err:
-        db_conn.rollback()
-        logger.error(f"数据库删除失败: {err}")
-        raise HTTPException(status_code=500, detail="数据库删除失败")
-    finally:
-        if cursor:
-            cursor.close()

app/core/config.py

@@ -15,31 +15,16 @@ class CardModelConfig:
 class Settings:
     API_Inference_prefix: str = "/api/card_inference"
     API_Score_prefix: str = "/api/card_score"
-    API_ImageData_prefix: str = "/api/image_data"
 
     BASE_PATH = Path(__file__).parent.parent.parent.absolute()
 
     TEMP_WORK_DIR = BASE_PATH / "_temp_work"
-    DATA_DIR = BASE_PATH / "Data"
     SCORE_CONFIG_PATH = BASE_PATH / "app/core/scoring_config.json"
 
     # 图片像素与真实图片缩放比例
     PIXEL_RESOLUTION = 24.54
     CORNER_SIZE_MM = 3.0
 
-    # --- 数据库配置 ---
-    DB_NAME = 'card_score_database'
-    DB_TABLE_NAME = 'image_records'
-    DATABASE_CONFIG: Dict[str, str] = {
-        'user': 'root',
-        'password': '123456',
-        'host': '127.0.0.1',
-    }
-    # 连接到指定数据库的配置
-    DATABASE_CONFIG_WITH_DB: Dict[str, str] = {
-        **DATABASE_CONFIG,
-        'database': DB_NAME
-    }
 
     # 使用一个字典来管理所有卡片检测模型
     # key (如 'outer_box') 将成为 API 路径中的 {inference_type}

app/core/database_loader.py

@@ -1,100 +0,0 @@
-import mysql.connector
-from mysql.connector import errorcode
-from .config import settings
-from app.core.logger import get_logger
-
-logger = get_logger(__name__)
-
-# 全局的连接池
-db_connection_pool = None
-
-
-def init_database():
-    """
-    初始化数据库：如果数据库或表不存在，则创建它们。
-    """
-    logger.info("--- 开始初始化数据库 ---")
-
-    # 1. 尝试连接MySQL服务器（不指定数据库）
-    try:
-        cnx = mysql.connector.connect(**settings.DATABASE_CONFIG)
-        cursor = cnx.cursor()
-
-        # 2. 创建数据库（如果不存在）
-        cursor.execute(f"CREATE DATABASE IF NOT EXISTS {settings.DB_NAME} DEFAULT CHARACTER SET 'utf8mb4'")
-        logger.info(f"数据库 '{settings.DB_NAME}' 已准备就绪。")
-
-        # 3. 切换到目标数据库
-        cnx.database = settings.DB_NAME
-
-        # 4. 创建表（如果不存在）
-        table_description = (
-            f"CREATE TABLE IF NOT EXISTS `{settings.DB_TABLE_NAME}` ("
-            "  `img_id` INT AUTO_INCREMENT PRIMARY KEY,"
-            "  `img_name` VARCHAR(255) NULL,"
-            "  `img_path` VARCHAR(512) NOT NULL,"
-            "  `img_result_json` JSON NOT NULL,"
-            "  `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP"
-            ") ENGINE=InnoDB"
-        )
-        cursor.execute(table_description)
-        logger.info(f"数据表 '{settings.DB_TABLE_NAME}' 已准备就绪。")
-
-    except mysql.connector.Error as err:
-        logger.error(f"数据库初始化失败: {err}")
-        exit(1)  # 初始化失败直接退出程序
-    finally:
-        if 'cursor' in locals() and cursor:
-            cursor.close()
-        if 'cnx' in locals() and cnx.is_connected():
-            cnx.close()
-
-    logger.info("--- 数据库初始化完成 ---")
-
-
-def load_database_pool():
-    """
-    在应用启动时创建数据库连接池。
-    """
-    global db_connection_pool
-    if db_connection_pool is None:
-        logger.info("--- 创建数据库连接池 ---")
-        try:
-            db_connection_pool = mysql.connector.pooling.MySQLConnectionPool(
-                pool_name="mypool",
-                pool_size=5,  # 池中保持的连接数
-                **settings.DATABASE_CONFIG_WITH_DB
-            )
-            logger.info("--- 数据库连接池创建成功 ---")
-        except mysql.connector.Error as err:
-            logger.error(f"创建数据库连接池失败: {err}")
-            exit(1)
-
-
-def close_database_pool():
-    """
-    在应用关闭时，不需要手动关闭连接池，连接器会自动处理。
-    这个函数留作备用。
-    """
-    logger.info("--- 数据库连接池将自动关闭 ---")
-
-
-# --- FastAPI 依赖注入 ---
-def get_db_connection():
-    """
-    一个FastAPI依赖项，用于从池中获取数据库连接。
-    它确保连接在使用后返回到池中。
-    """
-    if db_connection_pool is None:
-        raise RuntimeError("数据库连接池未初始化")
-
-    db_conn = None
-    try:
-        db_conn = db_connection_pool.get_connection()
-        yield db_conn
-    except mysql.connector.Error as err:
-        logger.error(f"获取数据库连接失败: {err}")
-        # 这里可以根据需要抛出HTTPException
-    finally:
-        if db_conn and db_conn.is_connected():
-            db_conn.close()

app/main.py

@@ -1,11 +1,9 @@
 from fastapi import FastAPI
 from contextlib import asynccontextmanager
 
-from .core.database_loader import init_database, load_database_pool, close_database_pool
 from .core.model_loader import load_models, unload_models
 from app.api.card_inference import router as card_inference_router
 from app.api.score_inference import router as score_inference_router
-from app.api.image_data import router as image_data_router
 import os
 
 from .core.config import settings
@@ -18,21 +16,15 @@ async def lifespan(main_app: FastAPI):
     os.makedirs(settings.TEMP_WORK_DIR, exist_ok=True)
     os.makedirs(settings.DATA_DIR, exist_ok=True)
 
-    # --- 数据库初始化 ---
-    init_database()
-    load_database_pool()
-
     # --- 模型加载 ---
     load_models()
     yield
 
     print("--- 应用关闭 ---")
-    close_database_pool()
     unload_models()
 
 
 app = FastAPI(title="卡片框和缺陷检测服务", lifespan=lifespan)
 
 app.include_router(card_inference_router, prefix=settings.API_Inference_prefix)
-app.include_router(score_inference_router, prefix=settings.API_Score_prefix)
-app.include_router(image_data_router, prefix=settings.API_ImageData_prefix, tags=["Image Data"])
+app.include_router(score_inference_router, prefix=settings.API_Score_prefix)