87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
from http.client import HTTPException
|
|
import os
|
|
from typing import Annotated, List, Optional
|
|
from fastapi import APIRouter, Body, File, UploadFile
|
|
|
|
from constants.documents import UPLOAD_ACCEPTED_FILE_TYPES
|
|
from models.decomposed_file_info import DecomposedFileInfo
|
|
from services import TEMP_FILE_SERVICE
|
|
from services.documents_loader import DocumentsLoader
|
|
from utils.randomizers import get_random_uuid
|
|
from utils.validators import validate_files
|
|
|
|
FILES_ROUTER = APIRouter(prefix="/files", tags=["Files"])
|
|
|
|
|
|
@FILES_ROUTER.post("/upload", response_model=List[str])
|
|
async def upload_files(files: Optional[List[UploadFile]]):
|
|
if not files:
|
|
raise HTTPException(400, "Documents are required")
|
|
|
|
temp_dir = TEMP_FILE_SERVICE.create_temp_dir(get_random_uuid())
|
|
|
|
validate_files(files, True, True, 50, UPLOAD_ACCEPTED_FILE_TYPES)
|
|
|
|
temp_files: List[str] = []
|
|
if files:
|
|
for each_file in files:
|
|
temp_path = TEMP_FILE_SERVICE.create_temp_file_path(
|
|
each_file.filename, temp_dir
|
|
)
|
|
with open(temp_path, "wb") as f:
|
|
content = await each_file.read()
|
|
f.write(content)
|
|
|
|
temp_files.append(temp_path)
|
|
|
|
return temp_files
|
|
|
|
|
|
@FILES_ROUTER.post("/decompose", response_model=List[DecomposedFileInfo])
|
|
async def decompose_files(file_paths: Annotated[List[str], Body(embed=True)]):
|
|
temp_dir = TEMP_FILE_SERVICE.create_temp_dir(get_random_uuid())
|
|
|
|
txt_files = []
|
|
other_files = []
|
|
for file_path in file_paths:
|
|
if file_path.endswith(".txt"):
|
|
txt_files.append(file_path)
|
|
else:
|
|
other_files.append(file_path)
|
|
|
|
documents_loader = DocumentsLoader(file_paths=other_files)
|
|
await documents_loader.load_documents(temp_dir)
|
|
parsed_documents = documents_loader.documents
|
|
|
|
response = []
|
|
for index, parsed_doc in enumerate(parsed_documents):
|
|
file_path = TEMP_FILE_SERVICE.create_temp_file_path(
|
|
f"{get_random_uuid()}.txt", temp_dir
|
|
)
|
|
parsed_doc = parsed_doc.replace("<br>", "\n")
|
|
with open(file_path, "w") as text_file:
|
|
text_file.write(parsed_doc)
|
|
response.append(
|
|
DecomposedFileInfo(
|
|
name=os.path.basename(other_files[index]), file_path=file_path
|
|
)
|
|
)
|
|
|
|
# Return the txt documents as it is
|
|
for each_file in txt_files:
|
|
response.append(
|
|
DecomposedFileInfo(name=os.path.basename(each_file), file_path=each_file)
|
|
)
|
|
|
|
return response
|
|
|
|
|
|
@FILES_ROUTER.post("/update")
|
|
async def update_files(
|
|
file_path: Annotated[str, Body()],
|
|
file: Annotated[UploadFile, File()],
|
|
):
|
|
with open(file_path, "wb") as f:
|
|
f.write(await file.read())
|
|
|
|
return {"message": "File updated successfully"}
|