Parse files

fai-rag-app/fai-backend/fai_backend/documents/routes.py

@@ -92,3 +92,18 @@ def upload_handler(
         c.FireEvent(event=e.GoToEvent(url='/documents')),
         _('submit_a_question', 'Create Question'''),
     )
+
+
+@router.get('documents/parse_and_save', response_model=list, response_model_exclude_none=True)
+def parse_documents(
+        src_directory_path: str,
+        dest_directory_path: str,
+        dest_file_name: str,
+        file_service: FileUploadService = Depends(get_file_upload_service),
+) -> list:
+    parsed_files = file_service.parse_files(src_directory_path)
+    stringify_parsed_files = [str(elem) for elem in parsed_files]
+
+    file_service.dump_list_to_json(stringify_parsed_files, dest_directory_path, dest_file_name)
+
+    return []

fai-rag-app/fai-backend/fai_backend/files/service.py

@@ -1,3 +1,4 @@
+import json
 import mimetypes
 import os
 import uuid
@@ -6,6 +7,7 @@
 from fastapi import UploadFile
 from pydantic import ByteSize
 
+from fai_backend.files.file_parser import ParserFactory
 from fai_backend.files.models import FileInfo
 
 
@@ -31,19 +33,10 @@ def save_files(self, project_id: str, files: list[UploadFile]) -> str:
                 file_object.write(file.file.read())
         return upload_path
 
-    def list_files(self, project_id: str) -> list[FileInfo]:
-        project_directories = [d for d in os.listdir(self.upload_dir) if d.startswith(f'project_{project_id}_')]
-        if not project_directories:
-            return []
-
-        latest_directory = sorted(project_directories, key=lambda x: (x.split('_')[2], x.split('_')[3]), reverse=True)[
-            0]
-        latest_directory_path = os.path.join(self.upload_dir, latest_directory)
-        upload_date = datetime.fromtimestamp(os.path.getctime(latest_directory_path))
-
+    def get_file_infos(self, directory_path, upload_date: datetime) -> list[FileInfo]:
         file_infos = []
-        for file_name in os.listdir(latest_directory_path):
-            file_path = os.path.join(latest_directory_path, file_name)
+        for file_name in os.listdir(directory_path):
+            file_path = os.path.join(directory_path, file_name)
             if os.path.isfile(file_path):
                 stat = os.stat(file_path)
                 mime_type, _ = mimetypes.guess_type(file_path)
@@ -59,6 +52,18 @@ def list_files(self, project_id: str) -> list[FileInfo]:
 
         return file_infos
 
+    def list_files(self, project_id: str) -> list[FileInfo]:
+        project_directories = [d for d in os.listdir(self.upload_dir) if d.startswith(f'project_{project_id}_')]
+        if not project_directories:
+            return []
+
+        latest_directory = sorted(project_directories, key=lambda x: (x.split('_')[2], x.split('_')[3]), reverse=True)[
+            0]
+        latest_directory_path = os.path.join(self.upload_dir, latest_directory)
+        upload_date = datetime.fromtimestamp(os.path.getctime(latest_directory_path))
+
+        return self.get_file_infos(latest_directory_path, upload_date)
+
     def get_latest_upload_path(self, project_id: str) -> str | None:
         project_directories = [d for d in os.listdir(self.upload_dir) if d.startswith(f'project_{project_id}_')]
         if not project_directories:
@@ -67,3 +72,38 @@ def get_latest_upload_path(self, project_id: str) -> str | None:
         latest_directory = sorted(project_directories, key=lambda x: (x.split('_')[2], x.split('_')[3]), reverse=True)[
             0]
         return os.path.join(self.upload_dir, latest_directory)
+
+    def parse_files(self, src_directory_path: str):
+        parsed_files = []
+
+        upload_date = datetime.fromtimestamp(os.path.getctime(src_directory_path))
+        files = self.get_file_infos(src_directory_path, upload_date)
+
+        for file in files:
+            parser = ParserFactory.get_parser(file.path)
+            parsed_file = parser.parse(file.path)
+            parsed_files.extend(parsed_file)
+
+        return parsed_files
+
+    def dump_list_to_json(self, parsed_files: list[str], dest_directory_path: str, dest_file_name: str):
+        if not os.path.isabs(dest_directory_path):
+            raise ValueError("Destination path must be absolute")
+
+        if not dest_file_name.endswith('.json'):
+            dest_file_name += '.json'
+
+        file_path = os.path.join(dest_directory_path, dest_file_name + '.json')
+        os.makedirs(dest_directory_path, exist_ok=True)
+
+        if os.path.exists(file_path):
+            raise FileExistsError(f"The file {file_path} already exists.")
+
+        try:
+            stringify_parsed_files = [str(elem) for elem in parsed_files]
+            with open(file_path, 'w') as f:
+                json.dump(stringify_parsed_files, f, indent=4)
+        except (IOError, OSError) as e:
+            raise Exception(f"Failed to write to {file_path}: {str(e)}")
+        except TypeError as e:
+            raise Exception(f"Data serialization error: {str(e)}")