daily/Router/FileRouter.py

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)
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_save_path = save_path / file_md
    file_url = f'/static_data/upload_files/{common_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))
    else:
        raise HTTPException(detail='文件已经上传', status_code=303)
    return user_file.to_dict()


@router.post("/query", response_model=FileSechemas.UserFileQueryRes)
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)