# 藏品路由 - 使用字段编码 import os import uuid import re from typing import Optional, List from fastapi import APIRouter, Depends, HTTPException, status, Query, UploadFile, File from sqlalchemy import or_, func, text from sqlalchemy.orm import Session from app.core.database import get_db from app.core.auth import get_current_user from app.core.logging_config import logger from app.models.models import User, Collection, CollectionImage, Operation from app.schemas.schemas import ( CollectionCreate, CollectionUpdate, CollectionResponse, CollectionListResponse, CollectionImageResponse ) from app.services.oss import upload_to_oss, get_oss_path, delete_from_oss, get_public_url router = APIRouter(prefix="/api/collections", tags=["藏品"]) def to_camel_case(data: dict) -> dict: """将字段编码转换为 camelCase 格式""" if not data: return data mapping = { 'f99_90_id': 'id', 'f99_91_user_id': 'userId', 'f99_92_created_at': 'createdAt', 'f99_93_updated_at': 'updatedAt', 'f01_01_name': 'name', 'f01_02_code': 'code', 'f01_03_category': 'category', 'f01_04_status': 'status', 'f01_05_remark': 'remark', 'f02_10_prefix_serial': 'prefixSerial', 'f02_11_version': 'version', 'f02_12_packaging': 'packaging', 'f02_13_rarity': 'rarity', 'f03_20_is_graded': 'isGraded', 'f03_21_grading_company': 'gradingCompany', 'f03_22_grading_score': 'gradingScore', 'f03_23_three_star': 'threeStar', 'f04_30_special_mark': 'specialMark', 'f04_31_serial_feature': 'serialFeature', 'f04_32_issuer': 'issuer', 'f04_33_issue_year': 'issueYear', 'f04_34_material': 'material', 'f04_35_denomination': 'denomination', 'f04_36_issue_quantity': 'issueQuantity', 'f05_40_cost_price': 'costPrice', 'f05_41_target_price': 'targetPrice', 'f05_42_goal_price': 'goalPrice', 'f05_43_repair_fee': 'repairFee', 'f05_44_grading_fee': 'gradingFee', 'f06_50_purpose': 'purpose', } return {mapping.get(k, k): v for k, v in data.items()} # 编码生成函数 def generate_code(version: str, user_id: str, db: Session) -> str: """自动生成藏品编号 - 按用户独立编码""" import re # 查询当前用户的非空编码（不与其他用户混算） user_codes = db.query(Collection.f01_02_code).filter( Collection.f01_02_code.isnot(None), Collection.f99_91_user_id == user_id ).all() max_num = 0 for (code,) in user_codes: # 只处理纯数字或 4 位数字编码（忽略 TEST001 等特殊编码） if re.match(r'^\d{4}$', code): try: num = int(code) if num > max_num: max_num = num except (ValueError, TypeError): pass # 当前用户最大号 +1 next_num = max_num + 1 return str(next_num).zfill(4) @router.get("/next-code") def get_next_code( current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """获取下一个藏品编号""" next_code = generate_code("2024 龙", current_user.f99_90_id, db) return {"code": 200, "data": {"nextCode": next_code}} @router.get("") def get_collections( category: Optional[str] = None, status: Optional[str] = None, user_id: Optional[str] = None, search: Optional[str] = None, page: int = Query(1, ge=1), limit: int = Query(20, ge=1, le=100), current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """获取藏品列表""" # admin 用户可以看到所有藏品，普通用户只能看到自己的 from sqlalchemy.orm import joinedload if current_user.role == "admin": # 联表查询获取用户名 query = db.query(Collection, User.f01_01_name.label('owner_name')).join( User, Collection.f99_91_user_id == User.f99_90_id, isouter=True ) else: query = db.query(Collection).filter(or_(Collection.f99_91_user_id == current_user.f99_90_id, current_user.role == "admin")) if category: query = query.filter(Collection.f01_03_category == category) if user_id: query = query.filter(Collection.f99_91_user_id == user_id) if status: query = query.filter(Collection.f01_04_status == status) if search: query = query.filter( (Collection.f01_01_name.contains(search)) | (Collection.f01_05_remark.contains(search)) ) # 总数（分开查询避免join影响count） if current_user.role == "admin": total = db.query(Collection).count() else: total = query.count() # 分页 data = query.order_by(Collection.f99_92_created_at.desc()) \ .offset((page - 1) * limit) \ .limit(limit) \ .all() # 转换为字典列表并转为 camelCase data_list = [] for item in data: # 处理联表查询结果 if current_user.role == "admin": collection_item, owner_name = item else: collection_item = item owner_name = None item_dict = { 'f99_90_id': collection_item.f99_90_id, 'f99_91_user_id': collection_item.f99_91_user_id, 'owner_name': owner_name, # 所属用户名（仅管理员可见） 'f01_01_name': collection_item.f01_01_name, 'f01_02_code': collection_item.f01_02_code, 'f01_03_category': collection_item.f01_03_category, 'f01_04_status': collection_item.f01_04_status, 'f01_05_remark': collection_item.f01_05_remark, 'f02_10_prefix_serial': collection_item.f02_10_prefix_serial, 'f02_11_version': collection_item.f02_11_version, 'f02_12_packaging': collection_item.f02_12_packaging, 'f02_13_rarity': collection_item.f02_13_rarity, 'f03_20_is_graded': collection_item.f03_20_is_graded, 'f03_21_grading_company': collection_item.f03_21_grading_company, 'f03_22_grading_score': collection_item.f03_22_grading_score, 'f03_23_three_star': collection_item.f03_23_three_star, 'f04_30_special_mark': collection_item.f04_30_special_mark, 'f04_31_serial_feature': collection_item.f04_31_serial_feature, 'f04_32_issuer': collection_item.f04_32_issuer, 'f04_33_issue_year': collection_item.f04_33_issue_year, 'f04_34_material': collection_item.f04_34_material, 'f04_35_denomination': collection_item.f04_35_denomination, 'f04_36_issue_quantity': collection_item.f04_36_issue_quantity, 'f05_40_cost_price': float(collection_item.f05_40_cost_price) if collection_item.f05_40_cost_price else None, 'f05_41_target_price': float(collection_item.f05_41_target_price) if collection_item.f05_41_target_price else None, 'f05_42_goal_price': float(collection_item.f05_42_goal_price) if collection_item.f05_42_goal_price else None, 'f05_43_repair_fee': float(collection_item.f05_43_repair_fee) if collection_item.f05_43_repair_fee else None, 'f05_44_grading_fee': float(collection_item.f05_44_grading_fee) if collection_item.f05_44_grading_fee else None, 'f06_50_purpose': collection_item.f06_50_purpose, 'f99_92_created_at': collection_item.f99_92_created_at.isoformat() if collection_item.f99_92_created_at else None, 'images': [] } # 加载图片数据 from app.models.models import CollectionImage images = db.query(CollectionImage).filter( CollectionImage.collection_id == collection_item.f99_90_id ).all() for img in images: item_dict['images'].append({ 'id': img.id, 'filename': img.filename, 'original_name': img.original_name, 'path': img.path, 'created_at': img.created_at.isoformat() if img.created_at else None }) data_list.append(to_camel_case(item_dict)) return { "data": data_list, "pagination": { "page": page, "limit": limit, "total": total, "pages": (total + limit - 1) // limit } } @router.get("/stats") def get_stats( current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """获取藏品统计""" # 获取所有藏品 if current_user.role == "admin": all_collections = db.query(Collection).all() else: all_collections = db.query(Collection).filter( or_(Collection.f99_91_user_id == current_user.f99_90_id, current_user.role == "admin") ).all() # 总数 total_count = len(all_collections) # 按分类统计 from collections import Counter by_category = Counter(c.f01_03_category for c in all_collections).items() # 按状态统计 by_status = Counter(c.f01_04_status for c in all_collections).items() # 按是否评级统计 by_graded = Counter(c.f03_20_is_graded for c in all_collections).items() # 新增：8 个分布统计 by_packaging = Counter(c.f02_12_packaging for c in all_collections if c.f02_12_packaging).items() by_rarity = Counter(c.f02_13_rarity for c in all_collections if c.f02_13_rarity).items() by_version = Counter(c.f02_11_version for c in all_collections if c.f02_11_version).items() by_grading_company = Counter(c.f03_21_grading_company for c in all_collections if c.f03_20_is_graded and c.f03_21_grading_company).items() by_grading_score = Counter(c.f03_22_grading_score for c in all_collections if c.f03_20_is_graded and c.f03_22_grading_score).items() by_special_mark = Counter(c.f04_30_special_mark for c in all_collections if c.f04_30_special_mark).items() # 总成本： SUM(cost_price + repair_fee + grading_fee) for ALL collections total_cost = sum( (c.f05_40_cost_price or 0) + (c.f05_43_repair_fee or 0) + (c.f05_44_grading_fee or 0) for c in all_collections ) # 预期利润： SUM(target_price - cost_price) for collections with target_price > 0 expected_profit = sum( (c.f05_41_target_price or 0) - (c.f05_40_cost_price or 0) for c in all_collections if c.f05_41_target_price and c.f05_41_target_price > 0 ) # 已售商品：状态为 sold 且出售价 > 0 sold_collections = [ c for c in all_collections if c.f01_04_status == 'sold' and c.f05_42_goal_price and c.f05_42_goal_price > 0 ] # 总收入：SUM(出售价) for 已售商品（售价>0） total_revenue = sum( c.f05_42_goal_price or 0 for c in sold_collections ) # 总利润（已实现利润）：SUM(出售价 - 成本价 - 修复费 - 评级费) for 已售商品 # 单藏品总成本 = 成本价 + 修复费 + 评级费 total_profit = sum( (c.f05_42_goal_price or 0) - (c.f05_40_cost_price or 0) - (c.f05_43_repair_fee or 0) - (c.f05_44_grading_fee or 0) for c in sold_collections ) return { "totalCount": total_count, "byCategory": [{"category": c, "count": n} for c, n in by_category], "byStatus": [{"status": s, "count": n} for s, n in by_status], "byGrading": [{"isGraded": g, "count": n} for g, n in by_graded], "byPackaging": [{"packaging": p, "count": n} for p, n in by_packaging], "byRarity": [{"rarity": r, "count": n} for r, n in by_rarity], "byVersion": [{"version": v, "count": n} for v, n in by_version], "byGradingCompany": [{"company": c, "count": n} for c, n in by_grading_company], "byGradingScore": [{"score": s, "count": n} for s, n in by_grading_score], "bySpecialMark": [{"mark": m, "count": n} for m, n in by_special_mark], # 盈亏统计（只统计已售且有价格的藏品） "byProfitLoss": [ {"type": "profit", "label": "盈利", "count": sum(1 for c in sold_collections if c.f05_42_goal_price and c.f05_40_cost_price and c.f05_42_goal_price > c.f05_40_cost_price)}, {"type": "loss", "label": "亏损", "count": sum(1 for c in sold_collections if c.f05_42_goal_price and c.f05_40_cost_price and c.f05_42_goal_price <= c.f05_40_cost_price)} ], "totalCost": total_cost, "totalTarget": sum(c.f05_41_target_price or 0 for c in all_collections), "expectedProfit": expected_profit, "totalRevenue": total_revenue, "totalProfit": total_profit } @router.get("/by-user/{user_id}") def get_collections_by_user( user_id: str, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """获取指定用户的所有藏品（管理员专用）""" if current_user.role != "admin": raise HTTPException(status_code=403, detail="E00050: 仅管理员可访问") collections = db.query(Collection).filter( Collection.f99_91_user_id == user_id ).all() result = [] for c in collections: result.append({ "f99_90_id": c.f99_90_id, "name": c.f01_01_name, "code": c.f01_02_code, "prefixSerial": c.f02_10_prefix_serial, "version": c.f02_11_version, "category": c.f01_03_category, "status": c.f01_04_status, "packaging": c.f02_12_packaging, "isGraded": c.f03_20_is_graded, "costPrice": c.f05_40_cost_price, "f01_01_name": c.f01_01_name, "f02_10_prefix_serial": c.f02_10_prefix_serial, "f02_11_version": c.f02_11_version, "f01_04_status": c.f01_04_status, "userId": c.f99_91_user_id, "f05_40_cost_price": c.f05_40_cost_price, "f99_92_created_at": c.f99_92_created_at.isoformat() if c.f99_92_created_at else None }) return result @router.get("/{collection_id}") def get_collection( collection_id: str, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """获取单个藏品详情""" result = db.execute( text("SELECT * FROM collections WHERE f99_90_id = :id"), {"id": collection_id} ).fetchone() if not result: raise HTTPException(status_code=404, detail="E00033: 藏品不存在") collection = dict(result._mapping) # 非管理员只能查看自己的藏品 if current_user.role != "admin" and collection.get('f99_91_user_id') != current_user.f99_90_id: raise HTTPException(status_code=403, detail="无权访问") result_dict = { 'f99_90_id': collection.get('f99_90_id'), 'f99_91_user_id': collection.get('f99_91_user_id'), 'f01_01_name': collection.get('f01_01_name'), 'f01_02_code': collection.get('f01_02_code'), 'f01_03_category': collection.get('f01_03_category'), 'f01_04_status': collection.get('f01_04_status'), 'f01_05_remark': collection.get('f01_05_remark'), 'f02_10_prefix_serial': collection.get('f02_10_prefix_serial'), 'f02_11_version': collection.get('f02_11_version'), 'f02_12_packaging': collection.get('f02_12_packaging'), 'f02_13_rarity': collection.get('f02_13_rarity'), 'f03_20_is_graded': collection.get('f03_20_is_graded'), 'f03_21_grading_company': collection.get('f03_21_grading_company'), 'f03_22_grading_score': collection.get('f03_22_grading_score'), 'f03_23_three_star': collection.get('f03_23_three_star'), 'f04_30_special_mark': collection.get('f04_30_special_mark'), 'f04_31_serial_feature': collection.get('f04_31_serial_feature'), 'f04_32_issuer': collection.get('f04_32_issuer'), 'f04_33_issue_year': collection.get('f04_33_issue_year'), 'f04_34_material': collection.get('f04_34_material'), 'f04_35_denomination': collection.get('f04_35_denomination'), 'f04_36_issue_quantity': collection.get('f04_36_issue_quantity'), 'f05_40_cost_price': float(collection.get('f05_40_cost_price')) if collection.get('f05_40_cost_price') else None, 'f05_41_target_price': float(collection.get('f05_41_target_price')) if collection.get('f05_41_target_price') else None, 'f05_42_goal_price': float(collection.get('f05_42_goal_price')) if collection.get('f05_42_goal_price') else None, 'f05_43_repair_fee': float(collection.get('f05_43_repair_fee')) if collection.get('f05_43_repair_fee') else None, 'f05_44_grading_fee': float(collection.get('f05_44_grading_fee')) if collection.get('f05_44_grading_fee') else None, 'f06_50_purpose': collection.get('f06_50_purpose'), 'f99_92_created_at': collection.get('f99_92_created_at').isoformat() if collection.get('f99_92_created_at') else None, 'images': [] } # 加载图片数据 images = db.query(CollectionImage).filter( CollectionImage.collection_id == collection_id ).all() for img in images: result_dict['images'].append({ 'id': img.id, 'filename': img.filename, 'original_name': img.original_name, 'path': img.path, 'created_at': img.created_at.isoformat() if img.created_at else None }) return to_camel_case(result_dict) @router.post("") def create_collection( collection_data: CollectionCreate, force: bool = False, # 是否强制保存（忽略重复警告） current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """创建藏品 - 支持冠字号查重""" from app.core.logging_config import logger # 自动生成编码 final_code = collection_data.f01_02_code or generate_code( collection_data.f02_11_version or '2024 龙', current_user.f99_90_id, db ) # 编号查重（如果提供了编号且不是强制保存） if not force and final_code: existing_code = db.query(Collection).filter( Collection.f01_02_code == final_code, or_(Collection.f99_91_user_id == current_user.f99_90_id, current_user.role == "admin") ).first() if existing_code: logger.warning(f"发现重复编号：{final_code}, 已存在藏品 ID: {existing_code.f99_90_id}") return { "error": { "code": "DUPLICATE_CODE", "message": f"藏品编号 {final_code} 已存在，请使用其他编号" } } # 冠字号查重（如果提供了冠字号且不是强制保存） if not force and collection_data.f02_10_prefix_serial: # 查询当前用户是否有相同冠字号的藏品 existing = db.query(Collection).filter( Collection.f02_10_prefix_serial == collection_data.f02_10_prefix_serial, or_(Collection.f99_91_user_id == current_user.f99_90_id, current_user.role == "admin") ).first() if existing: logger.warning(f"发现重复冠字号：{collection_data.f02_10_prefix_serial}, 已存在藏品 ID: {existing.f99_90_id}") # 返回警告信息，让前端询问用户是否继续 return { "warning": { "code": "DUPLICATE_SERIAL", "message": f"发现重复冠字号：{collection_data.f02_10_prefix_serial}", "existing_collection": { "id": existing.f99_90_id, "name": existing.f01_01_name, "code": existing.f01_02_code, "prefix_serial": existing.f02_10_prefix_serial } }, "data": { "ask_continue": True } } collection = Collection( f99_91_user_id=current_user.f99_90_id, f01_01_name=collection_data.f01_01_name, f01_02_code=final_code, f01_03_category=collection_data.f01_03_category, f01_04_status=collection_data.f01_04_status or "in_collection", f01_05_remark=collection_data.f01_05_remark, f02_10_prefix_serial=collection_data.f02_10_prefix_serial, f02_11_version=collection_data.f02_11_version, f02_12_packaging=collection_data.f02_12_packaging, f02_13_rarity=collection_data.f02_13_rarity, f03_20_is_graded=collection_data.f03_20_is_graded or False, f03_21_grading_company=collection_data.f03_21_grading_company, f03_22_grading_score=collection_data.f03_22_grading_score, f03_23_three_star=collection_data.f03_23_three_star or False, f04_30_special_mark=collection_data.f04_30_special_mark, f04_31_serial_feature=collection_data.f04_31_serial_feature, f04_32_issuer=collection_data.f04_32_issuer, f04_33_issue_year=collection_data.f04_33_issue_year, f04_34_material=collection_data.f04_34_material, f04_35_denomination=collection_data.f04_35_denomination, f04_36_issue_quantity=collection_data.f04_36_issue_quantity, f05_40_cost_price=collection_data.f05_40_cost_price, f05_41_target_price=collection_data.f05_41_target_price, f05_42_goal_price=collection_data.f05_42_goal_price, f05_43_repair_fee=collection_data.f05_43_repair_fee, f05_44_grading_fee=collection_data.f05_44_grading_fee, f06_50_purpose=collection_data.f06_50_purpose ) db.add(collection) db.commit() db.refresh(collection) return { 'f99_90_id': collection.f99_90_id, 'f99_91_user_id': collection.f99_91_user_id, 'f01_01_name': collection.f01_01_name, 'f01_02_code': collection.f01_02_code, 'f01_03_category': collection.f01_03_category, 'f01_04_status': collection.f01_04_status, 'f01_05_remark': collection.f01_05_remark, 'f99_92_created_at': collection.f99_92_created_at.isoformat() if collection.f99_92_created_at else None, 'message': '创建成功' } @router.put("/{collection_id}") def update_collection( collection_id: str, collection_data: CollectionUpdate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """更新藏品""" collection = db.query(Collection).filter( Collection.f99_90_id == collection_id, or_(Collection.f99_91_user_id == current_user.f99_90_id, current_user.role == "admin") ).first() if not collection: raise HTTPException(status_code=404, detail="E00033: 藏品不存在") # 更新字段 - 使用 model_fields_set 检查哪些字段被设置 for field_name in collection_data.model_fields_set: value = getattr(collection_data, field_name) if value is not None: setattr(collection, field_name, value) db.commit() db.refresh(collection) return { "f99_90_id": collection.f99_90_id, "message": "更新成功" } @router.delete("/{collection_id}") def delete_collection( collection_id: str, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """删除藏品""" # 验证权限并检查是否存在 collection = db.query(Collection).filter( Collection.f99_90_id == collection_id, or_(Collection.f99_91_user_id == current_user.f99_90_id, current_user.role == "admin") ).first() if not collection: raise HTTPException(status_code=404, detail="E00033: 藏品不存在") # 使用原生 SQL 删除（避免 ORM 级联查询字段不匹配问题） from sqlalchemy import text # 1. 删除关联的 operations（f99_91_user_id 关联到 collections.f99_90_id） db.execute(text("DELETE FROM operations WHERE f99_91_user_id = :id"), {"id": collection_id}) # 2. 删除关联的图片 db.execute(text("DELETE FROM collection_images WHERE collection_id = :id"), {"id": collection_id}) # 3. 删除藏品本身 db.execute(text("DELETE FROM collections WHERE f99_90_id = :id"), {"id": collection_id}) db.commit() return {"message": "删除成功"} @router.post("/upload-image") async def upload_image( collection_id: str = None, file: UploadFile = File(...), current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """上传藏品图片 - 文件名格式：用户名 - 藏品编号 - 冠字号""" try: # 验证藏品是否存在 collection = db.query(Collection).filter( Collection.f99_90_id == collection_id ).first() if not collection: raise HTTPException(status_code=404, detail="E00033: 藏品不存在") # 获取用户信息（用于文件名） owner = db.query(User).filter(User.f99_90_id == collection.f99_91_user_id).first() username = owner.f01_01_name if owner else "unknown" # 获取藏品信息（用于文件名） code = collection.f01_02_code or "0000" prefix_serial = collection.f02_10_prefix_serial or "" # 检查文件类型 if not file.content_type.startswith('image/'): raise HTTPException(status_code=400, detail="E00038: 只能上传图片文件") # 检查文件大小（限制 10MB） file_size = 0 content = await file.read() file_size = len(content) if file_size > 10 * 1024 * 1024: # 10MB raise HTTPException(status_code=400, detail=f"E00039: 图片大小不能超过 10MB（当前{file_size // 1024 // 1024}MB）") # 生成OSS存储路径 user_id = collection.f99_91_user_id file_extension = file.filename.split('.')[-1] if '.' in file.filename else 'jpg' # 清理特殊字符 clean_username = re.sub(r'[^\w\u4e00-\u9fff\-]', '', username) clean_serial = re.sub(r'[^\w\u4e00-\u9fff\-]', '', prefix_serial) # 文件名格式：用户名-藏品编号-冠字号.jpg if clean_serial: filename = f"{clean_username}-{code}-{clean_serial}.{file_extension}" else: filename = f"{clean_username}-{code}.{file_extension}" # 生成OSS key oss_key, unique_name = get_oss_path("collections", user_id=user_id, filename=filename) # 上传到OSS image_url = upload_to_oss(content, oss_key) # 创建图片记录（保存OSS URL） image = CollectionImage( id=str(uuid.uuid4()), collection_id=collection_id, filename=unique_name, original_name=file.filename, path=image_url # 保存OSS URL ) db.add(image) db.commit() db.refresh(image) logger.info(f"图片上传成功：{image_url}, collection_id={collection_id}") return { "message": "上传成功", "image_id": image.id, "filename": unique_name, "url": image_url } except HTTPException: raise except Exception as e: logger.error(f"图片上传失败：{str(e)}") raise HTTPException(status_code=500, detail="上传失败") @router.delete("/images/{image_id}") async def delete_image( image_id: str, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """删除藏品图片""" try: # 查找图片记录 image = db.query(CollectionImage).filter( CollectionImage.id == image_id ).first() if not image: raise HTTPException(status_code=404, detail="E00033: 图片不存在") # 检查权限 collection = db.query(Collection).filter( Collection.f99_90_id == image.collection_id ).first() if collection and current_user.role != "admin" and collection.f99_91_user_id != current_user.f99_90_id: raise HTTPException(status_code=403, detail="E00014: 无权删除此图片") # 删除OSS文件（如果path是OSS URL） if image.path and image.path.startswith("https://"): # 从OSS URL提取key try: oss_key = image.path.replace("https://jiachenlong-oss.oss-cn-hangzhou.aliyuncs.com/", "") delete_from_oss(oss_key) except Exception as e: logger.warning(f"OSS文件删除失败: {e}") elif image.path and os.path.exists(image.path): # 兼容旧的本地上传 os.remove(image.path) # 删除数据库记录 db.delete(image) db.commit() return {"message": "删除成功"} except HTTPException: raise except Exception as e: logger.error(f"图片删除失败：{str(e)}") raise HTTPException(status_code=500, detail="删除失败")