"""附件管理 API""" from typing import Optional, List from fastapi import APIRouter, Depends, Query, UploadFile, File, Form from sqlalchemy.orm import Session from app.core.database import get_db from app.core.auth import get_current_user from app.core.response import success_response, error_response from app.crud.attachment import attachment_crud from app.core.storage import storage router = APIRouter(prefix="/api/web/admin/attachment", tags=["附件管理"]) # 业务类型常量 BUSINESS_TYPES = { "delivery_pickup": "配送取货凭证", "delivery_deliver": "配送送达凭证", "delivery_prepare": "备货完成凭证", "settlement": "结算文件", "order": "订单附件", "inquiry": "询价附件", "quotation": "报价附件", "other": "其他" } @router.get("/list", response_model=dict) def list_attachments( business_type: Optional[str] = Query(None, description="业务类型"), business_id: Optional[str] = Query(None, description="业务ID"), page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(20, ge=1, le=100, description="每页条数"), db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """获取附件列表""" from app.models.models import BizAttachment from sqlalchemy import func query = db.query(BizAttachment).filter(BizAttachment.is_del == 0) if business_type: query = query.filter(BizAttachment.business_type == business_type) if business_id: query = query.filter(BizAttachment.business_id == business_id) total = query.count() skip = (page - 1) * pageSize attachments = query.order_by(BizAttachment.create_at.desc()).offset(skip).limit(pageSize).all() return success_response({ "list": [{ "id": att.id, "businessType": att.business_type, "businessTypeText": BUSINESS_TYPES.get(att.business_type, att.business_type), "businessId": att.business_id, "fileName": att.file_name, "fileUrl": att.file_url, "fileSize": att.file_size, "fileType": att.file_type, "uploadBy": att.upload_by, "uploadByName": att.upload_by_name, "createAt": str(att.create_at) if att.create_at else None } for att in attachments], "total": total, "page": page, "pageSize": pageSize }) @router.get("/business/{business_type}/{business_id}", response_model=dict) def get_business_attachments( business_type: str, business_id: str, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """获取指定业务的附件列表""" attachments = attachment_crud.get_by_business(db, business_type, business_id) return success_response([{ "id": att.id, "fileName": att.file_name, "fileUrl": att.file_url, "fileSize": att.file_size, "fileType": att.file_type, "uploadByName": att.upload_by_name, "createAt": str(att.create_at) if att.create_at else None } for att in attachments]) @router.post("/upload", response_model=dict) async def upload_attachment( business_type: str = Form(..., description="业务类型"), business_id: Optional[str] = Form(None, description="业务ID"), file: UploadFile = File(..., description="上传文件"), db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """上传附件""" try: allowed_image = frozenset({ "image/jpeg", "image/png", "image/gif", "image/webp", "image/heic", "image/heif", }) allowed_video = frozenset({ "video/mp4", "video/quicktime", "video/webm", "video/x-msvideo", }) allowed_doc = frozenset({ "application/pdf", "application/msword", "application/vnd.openxmlformats-officedocument.wordprocessingml.document", }) allowed_types = allowed_image | allowed_video | allowed_doc ct = (file.content_type or "").strip().lower() or "application/octet-stream" if ct not in allowed_types: return error_response(400, f"不支持的文件类型: {file.content_type}") # 读取文件内容 content = await file.read() def _max_bytes_for(content_type: str) -> int: if content_type.startswith("video/"): return 100 * 1024 * 1024 if content_type.startswith("image/"): return 10 * 1024 * 1024 return 20 * 1024 * 1024 max_bytes = _max_bytes_for(ct) if len(content) > max_bytes: label = "视频" if ct.startswith("video/") else ("图片" if ct.startswith("image/") else "文件") mb = max_bytes // (1024 * 1024) return error_response(400, f"{label}大小不能超过 {mb}MB") # 上传到存储 folder = f"attachments/{business_type}" file_url = storage.upload_file(content, file.filename or "file", folder) # 创建附件记录 from app.core.utils import get_file_size_display attachment = attachment_crud.create_attachment( db=db, business_type=business_type, business_id=business_id, file_name=file.filename or "file", file_url=file_url, file_size=len(content), file_type=file.content_type, upload_by=current_user.get("user_id"), upload_by_name=current_user.get("real_name", current_user.get("username", "")) ) return success_response({ "id": attachment.id, "fileName": attachment.file_name, "fileUrl": attachment.file_url, "fileSize": attachment.file_size, "fileSizeDisplay": get_file_size_display(attachment.file_size), "fileType": attachment.file_type }) except Exception as e: return error_response(500, f"上传失败: {str(e)}") @router.delete("/{attachment_id}", response_model=dict) def delete_attachment( attachment_id: str, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """删除附件""" attachment = attachment_crud.get(db, attachment_id) if not attachment: return error_response(404, "附件不存在") if attachment.is_del == 1: return error_response(404, "附件已删除") # 软删除 attachment_crud.soft_delete(db, attachment_id) return success_response({"message": "删除成功"}) @router.get("/business-types", response_model=dict) def get_business_types( current_user: dict = Depends(get_current_user) ): """获取业务类型列表""" return success_response([{ "value": key, "label": value } for key, value in BUSINESS_TYPES.items()])