71 lines
3.0 KiB
Python
71 lines
3.0 KiB
Python
import os
|
|
from pathlib import Path
|
|
from fastapi import APIRouter, UploadFile, File, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from Crud import FileCrud
|
|
from Schemas.UserSchemas import TokenData
|
|
from Utils.CommonUtils import file_md5
|
|
from Utils.SqlAlchemyUtils import get_db
|
|
from Schemas import FileSechemas
|
|
from Utils.AuthUtils import token_data_depend
|
|
|
|
router = APIRouter(prefix="/api/daily/file", tags=["文件"])
|
|
allow_file_type = {
|
|
'image',
|
|
'audio',
|
|
'video',
|
|
"application"
|
|
}
|
|
|
|
|
|
@router.post('/file_upload', response_model=FileSechemas.UserFileInfo, summary='文件上次')
|
|
def file_upload(file: UploadFile = File(...),
|
|
db: Session = Depends(get_db), token_data: TokenData = Depends(token_data_depend)):
|
|
mime_type = file.content_type
|
|
common_type, file_type = file.content_type.split('/')
|
|
if common_type not in allow_file_type:
|
|
raise Exception('不允许上传该类型文件')
|
|
content_start = file.file.read(1024)
|
|
file_md = file_md5(content_start)
|
|
save_path = Path(f"static_data/upload_files/{common_type}/{file_type}")
|
|
file_save_path = save_path / file_md
|
|
file_url = f'/static_data/upload_files/{common_type}/{file_type}/{file_md}'
|
|
if not os.path.exists(save_path):
|
|
os.makedirs(save_path)
|
|
user_file = FileCrud.get_file_by_md(db, file_md)
|
|
if not user_file:
|
|
if not os.path.exists(file_save_path):
|
|
with open(file_save_path, 'wb') as f:
|
|
f.write(content_start)
|
|
while True:
|
|
content = file.file.read(10240)
|
|
if content:
|
|
f.write(content)
|
|
else:
|
|
break
|
|
user_file = FileCrud.add_file(db,
|
|
FileSechemas.UserFileCreate(md=file_md, file_name=file.filename,
|
|
mime_type=mime_type,
|
|
user_id=token_data.email, common_type=common_type,
|
|
file_url=file_url))
|
|
return user_file.to_dict()
|
|
|
|
|
|
@router.post("/query", response_model=FileSechemas.UserFileQueryRes, summary='文件查询')
|
|
def query(req: FileSechemas.UserFileQuery, token_data: TokenData = Depends(token_data_depend),
|
|
db: Session = Depends(get_db)):
|
|
# if req.user_id and req.user_id != token_data.email:
|
|
# raise HTTPException(detail='无权查询其他用户文件', status_code=403)
|
|
count, items = FileCrud.file_query(db, req)
|
|
item_list = [item.to_dict() for item in items]
|
|
return FileSechemas.UserFileQueryRes(count=count, item_list=item_list)
|
|
|
|
|
|
@router.post("/get", response_model=FileSechemas.UserFileInfo, summary='文件获取')
|
|
def query(req: FileSechemas.UserFileGet,
|
|
db: Session = Depends(get_db)):
|
|
file = FileCrud.file_get(db, req.md)
|
|
if not file:
|
|
raise HTTPException(detail="未找到文件", status_code=404)
|
|
return file.to_dict()
|