Spaces:
Sleeping
Sleeping
File size: 2,811 Bytes
13ba451 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
from fastapi import APIRouter, Form, File, UploadFile,Query
from typing import List,Optional
from service import FileService
from function import support_function
from fastapi import HTTPException
from response import ResponseFile as res
from request import RequestFile
router = APIRouter()
ALLOWED_EXTENSIONS = {'csv', 'txt', 'doc', 'docx', 'pdf', 'xlsx', 'pptx', 'json', 'html'}
def allowed_file(filename):
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
@router.delete("/delete", tags=["File"])
async def delete_folder(request: RequestFile.RequestDeleteAllFile):
check = support_function.check_value_user_id_controller(request.user_id)
if check is not True:
return check
# request = RequestFile.RequestDeleteAllFile(user_id=user_id)
return await FileService.deleteAllFile(request)
@router.get("/list_name_files", tags=["File"])
async def get_name(user_id: str):
check = support_function.check_value_user_id_controller(user_id)
if check is not True:
return check
request = RequestFile.RequestGetNameFile(user_id=user_id)
return await FileService.listNameFiles(request)
@router.delete("/delete_file", tags=["File"])
async def delete_one_file(request: RequestFile.RequestDeleteFile):
user_id = request.user_id
check = support_function.check_value_user_id_controller(user_id)
if check is not True:
return check
name_file = request.name_file
if name_file is None or name_file.strip() == "":
return res.ReponseError(status=400,
data=res.Message(message="Name file is required."))
return await FileService.deleteFile(request)
@router.post("/chatbot/download_folder", tags=["File"])
async def download_folder_from_dropbox(request: RequestFile.RequestDownLoadFolder):
user_id = request.user_id
check = support_function.check_value_user_id_controller(user_id)
if check is not True:
return check
return await FileService.download_folder(request)
@router.post("/chatbot/download_files", tags=["File"])
async def download_file_by_id(request: RequestFile.RequestDownLoadFile):
user_id = request.user_id
check = support_function.check_value_user_id_controller(user_id)
if check is not True:
return check
return await FileService.download_file(request)
@router.post("/upload_files", tags=["File"])
async def upload_files_dropbox(user_id: str = Form(None), files: Optional[List[UploadFile]] = File(None)):
check = support_function.check_value_user_id_controller(user_id)
if check is not True:
return check
request = RequestFile.RequestUploadFile(files=files, user_id=user_id)
return await FileService.upload_files(request) |