198 lines
7.8 KiB
Python
198 lines
7.8 KiB
Python
# _*_ coding : UTF-8 _*_
|
|
# @Time : 2025/01/25 22:16
|
|
# @UpdateTime : 2025/01/25 22:16
|
|
# @Author : sonder
|
|
# @File : file.py
|
|
# @Software : PyCharm
|
|
# @Comment : 本程序
|
|
import os
|
|
from datetime import datetime
|
|
from fastapi import APIRouter, UploadFile, File, Path, Depends, Request, Query
|
|
from fastapi.responses import FileResponse, JSONResponse
|
|
|
|
from annotation.log import Log
|
|
from config.constant import BusinessType
|
|
from config.env import UploadConfig
|
|
from controller.login import LoginController
|
|
from exceptions.exception import ModelValidatorException, ServiceException
|
|
from models import File as FileModel
|
|
from schemas.common import BaseResponse
|
|
from schemas.file import UploadFileResponse, GetFileInfoResponse, GetFileListResponse
|
|
from utils.response import Response
|
|
from utils.upload import Upload
|
|
|
|
fileAPI = APIRouter(
|
|
prefix="/file",
|
|
)
|
|
|
|
|
|
@fileAPI.post("/upload", response_model=UploadFileResponse, response_class=JSONResponse, summary="上传文件")
|
|
@Log(title="上传文件", business_type=BusinessType.INSERT)
|
|
async def upload_file(
|
|
request: Request,
|
|
file: UploadFile = File(..., description="上传的文件"),
|
|
current_user: dict = Depends(LoginController.get_current_user),
|
|
):
|
|
# 1. 检查文件扩展名是否允许
|
|
file_extension = os.path.splitext(file.filename)[1][1:].lower() # 获取文件扩展名并转换为小写
|
|
if file_extension not in UploadConfig.DEFAULT_ALLOWED_EXTENSION:
|
|
raise ModelValidatorException(message="文件类型不支持")
|
|
|
|
# 2. 生成唯一的文件名
|
|
timestamp = datetime.now().strftime("%Y%m%d%H%M%S%f")
|
|
unique_filename = f"{current_user.get('id')}_{timestamp}.{file_extension}"
|
|
|
|
# 3. 保存文件到服务器
|
|
file_path = os.path.join(UploadConfig.UPLOAD_PATH, unique_filename)
|
|
with open(file_path, "wb") as buffer:
|
|
buffer.write(await file.read())
|
|
|
|
# 4. 构建文件的相对路径和绝对路径
|
|
relative_path = os.path.join(UploadConfig.UPLOAD_PREFIX, unique_filename) # 相对路径
|
|
absolute_path = os.path.abspath(file_path) # 绝对路径
|
|
|
|
# 5. 将文件信息保存到数据库
|
|
file_record = await FileModel.create(
|
|
name=file.filename,
|
|
size=os.path.getsize(file_path),
|
|
file_type=file.content_type,
|
|
absolute_path=absolute_path,
|
|
relative_path=relative_path,
|
|
uploader_id=current_user.get("id"),
|
|
)
|
|
result = await FileModel.get_or_none(id=file_record.id).values(
|
|
id="id",
|
|
name="name",
|
|
size="size",
|
|
file_type="file_type",
|
|
relative_path="relative_path",
|
|
absolute_path="absolute_path",
|
|
uploader_id="uploader__id",
|
|
uploader_username="uploader__username",
|
|
uploader_nickname="uploader__nickname",
|
|
uploader_department_id="uploader__department__id",
|
|
uploader_department_name="uploader__department__name",
|
|
create_time="create_time",
|
|
update_time="update_time",
|
|
)
|
|
return Response.success(data=result)
|
|
|
|
|
|
@fileAPI.get("/{id}", summary="下载文件")
|
|
@Log(title="获取文件", business_type=BusinessType.SELECT)
|
|
async def download_file(
|
|
request: Request,
|
|
id: str = Path(..., description="文件ID"),
|
|
):
|
|
# 1. 查询文件记录
|
|
file_record = await FileModel.get_or_none(id=id, del_flag=1)
|
|
if not file_record:
|
|
raise ServiceException(message="文件不存在!")
|
|
|
|
# 2. 检查文件是否存在
|
|
if not os.path.exists(file_record.absolute_path):
|
|
raise ServiceException(message="文件不存在!")
|
|
|
|
# 3. 返回文件内容
|
|
return FileResponse(
|
|
path=file_record.absolute_path,
|
|
filename=file_record.name,
|
|
media_type=file_record.file_type,
|
|
)
|
|
|
|
|
|
@fileAPI.get("/info/{id}", response_class=JSONResponse, response_model=GetFileInfoResponse, summary="获取文件信息")
|
|
@Log(title="获取文件信息", business_type=BusinessType.SELECT)
|
|
async def get_file_info(
|
|
request: Request,
|
|
id: str = Path(..., description="文件ID"),
|
|
current_user: dict = Depends(LoginController.get_current_user),
|
|
):
|
|
# 1. 查询文件记录
|
|
file_record = await FileModel.get_or_none(id=id, del_flag=1)
|
|
if not file_record:
|
|
raise ServiceException(message="文件不存在!")
|
|
result = await FileModel.get_or_none(id=id, del_flag=1).values(
|
|
id="id",
|
|
name="name",
|
|
size="size",
|
|
file_type="file_type",
|
|
relative_path="relative_path",
|
|
absolute_path="absolute_path",
|
|
uploader_id="uploader__id",
|
|
uploader_username="uploader__username",
|
|
uploader_nickname="uploader__nickname",
|
|
uploader_department_id="uploader__department__id",
|
|
uploader_department_name="uploader__department__name",
|
|
create_time="create_time",
|
|
update_time="update_time",
|
|
)
|
|
return Response.success(data=result)
|
|
|
|
|
|
@fileAPI.delete("/delete/{id}", response_class=JSONResponse, response_model=BaseResponse, summary="删除文件")
|
|
@fileAPI.post("/delete/{id}", response_class=JSONResponse, response_model=BaseResponse, summary="删除文件")
|
|
@Log(title="删除文件", business_type=BusinessType.DELETE)
|
|
async def delete_file(
|
|
request: Request,
|
|
id: str = Path(..., description="文件ID"),
|
|
current_user: dict = Depends(LoginController.get_current_user), ):
|
|
# 1. 查询文件记录
|
|
file_record = await FileModel.get_or_none(id=id, del_flag=1)
|
|
if not file_record:
|
|
raise ServiceException(message="文件不存在!")
|
|
if Upload.check_file_exists(file_record.absolute_path):
|
|
Upload.delete_file(file_record.absolute_path)
|
|
await file_record.delete()
|
|
return Response.success()
|
|
|
|
|
|
@fileAPI.get("/list", response_class=JSONResponse, response_model=GetFileListResponse, summary="获取文件列表")
|
|
@Log(title="获取文件列表", business_type=BusinessType.SELECT)
|
|
async def get_file_list(
|
|
request: Request,
|
|
page: int = Query(default=1, description="页码"),
|
|
pageSize: int = Query(default=10, description="每页数量"),
|
|
name: str = Query(default=None, description="文件名"),
|
|
file_type: str = Query(default=None, description="文件类型"),
|
|
uploader_id: str = Query(default=None, description="上传者ID"),
|
|
uploader_username: str = Query(default=None, description="上传者用户名"),
|
|
uploader_nickname: str = Query(default=None, description="上传者昵称"),
|
|
department_id: str = Query(default=None, description="上传者部门ID"),
|
|
department_name: str = Query(default=None, description="上传者部门名称"),
|
|
current_user: dict = Depends(LoginController.get_current_user), ):
|
|
# 1. 查询文件记录
|
|
filterArgs = {
|
|
f'{k}__icontains': v for k, v in {
|
|
'name': name,
|
|
'file_type': file_type,
|
|
'uploader__id': uploader_id,
|
|
'uploader__username': uploader_username,
|
|
'uploader__nickname': uploader_nickname,
|
|
'uploader__department__id': department_id,
|
|
'uploader__department__name': department_name
|
|
}.items() if v
|
|
}
|
|
total = await FileModel.filter(**filterArgs).count()
|
|
result = await FileModel.filter(**filterArgs).order_by('-create_time').offset((page - 1) * pageSize).limit(
|
|
pageSize).values(
|
|
id="id",
|
|
name="name",
|
|
size="size",
|
|
file_type="file_type",
|
|
relative_path="relative_path",
|
|
absolute_path="absolute_path",
|
|
uploader_id="uploader__id",
|
|
uploader_username="uploader__username",
|
|
uploader_nickname="uploader__nickname",
|
|
uploader_department_id="uploader__department__id",
|
|
uploader_department_name="uploader__department__name",
|
|
create_time="create_time",
|
|
update_time="update_time",
|
|
)
|
|
return Response.success(data={
|
|
"total": total,
|
|
"result": result,
|
|
"page": page,
|
|
})
|