shanxia/bx/bx-backend/app/routers/attachment.py

import mimetypes
from io import BytesIO
from typing import List
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, status
from fastapi.responses import StreamingResponse
from sqlalchemy.orm import Session

from app.database import get_db
from app.core.deps import get_current_user
from app.models.user import User
from app.schemas.common import ApiResponse
from app.schemas.attachment import AttachmentOut
from app.services.attachment import AttachmentService
from app.services.minio import MinioService

router = APIRouter(prefix="/api/attachments", tags=["附件管理"])


@router.post("/upload", response_model=ApiResponse[AttachmentOut])
def upload_file(
    file: UploadFile = File(...),
    biz_type: str = "",
    biz_id: int = 0,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    file_data = file.file.read()
    content_type = file.content_type or mimetypes.guess_type(file.filename or "")[0] or "application/octet-stream"
    object_name = MinioService.upload_file(file.filename or "unknown", file_data, content_type)
    data = {
        "biz_type": biz_type,
        "biz_id": biz_id,
        "file_name": file.filename or "unknown",
        "file_path": object_name,
        "file_type": content_type,
        "file_size": len(file_data),
        "create_by": str(current_user.id),
    }
    attachment = AttachmentService.create(db, data)
    return ApiResponse(data=AttachmentOut.model_validate(attachment))


@router.get("/download/{attachment_id}")
def download_file(
    attachment_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    attachment = AttachmentService.get_by_id(db, attachment_id)
    if not attachment:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="附件不存在")
    file_data = MinioService.download_file(attachment.file_path)
    file_stream = BytesIO(file_data)
    headers = {"Content-Disposition": f"attachment; filename={attachment.file_name}"}
    return StreamingResponse(file_stream, media_type=attachment.file_type or "application/octet-stream", headers=headers)


@router.get("/list", response_model=ApiResponse[List[AttachmentOut]])
def list_by_biz(
    biz_type: str = "",
    biz_id: int = 0,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    attachments = AttachmentService.get_by_biz(db, biz_type=biz_type, biz_id=biz_id)
    return ApiResponse(data=[AttachmentOut.model_validate(a) for a in attachments])