fix : improve error llamaparse

api/function.py

@@ -29,15 +29,9 @@ async def data_ingestion(reference, file: UploadFile) -> Any:
             user_id="admin_book_uploaded",
         )
         
-        # # Upload to AWS
-        file_name = f"{reference['title']}"
-        aws_loader = Loader()
-
-        file_obj = file
-        aws_loader.upload_to_s3(file_obj, file_name)
-
         uploader = Uploader(reference, file)
-
+        nodes_with_metadata, file_stream = await uploader.process_documents()
+        
         nodes_with_metadata = await uploader.process_documents()
         if isinstance(nodes_with_metadata, JSONResponse):
             return nodes_with_metadata  # Return the error response directly
@@ -45,6 +39,12 @@ async def data_ingestion(reference, file: UploadFile) -> Any:
         # Build indexes using IndexManager
         index = IndexManager()
         index.build_indexes(nodes_with_metadata)
+        
+        # Upload AWS
+        file_name = f"{reference['title']}"
+        aws_loader = Loader()
+
+        aws_loader.upload_to_s3(file_stream, file_name)
 
         return json.dumps(
             {"status": "success", "message": "Vector Index loaded successfully."}

core/journal_reading/extractor.py

@@ -1,8 +0,0 @@
-
-
-
-
-class Extractor():
-    def __init__(self):
-        pass
-    

core/journal_reading/upload.py

@@ -1,86 +0,0 @@
-import os
-import nest_asyncio
-
-from llama_parse import LlamaParse
-from llama_index.core.node_parser import SimpleNodeParser
-from dotenv import load_dotenv
-from fastapi import UploadFile, HTTPException, File
-from fastapi.responses import JSONResponse
-import fitz
-
-from script.get_metadata import Metadata
-
-load_dotenv()
-nest_asyncio.apply()
-
-
-async def parse_journal(content: bytes, file_name: str):
-    """Parse the journal using LlamaParse."""
-    try:
-        # Initialize the parser
-        parser = LlamaParse(
-            api_key=os.getenv("LLAMA_PARSE_API_KEY"),
-            result_type="markdown",
-            max_timeout=5000,
-        )
-
-        # Load and process the document
-        llama_parse_documents = parser.load_data(
-            content, extra_info={"file_name": file_name}
-        )
-
-        return llama_parse_documents
-
-    except Exception as e:
-        return JSONResponse(status_code=400, content=f"Error processing file: {e}")
-
-
-async def extract_metadata(content: bytes):
-    """Extract metadata from the PDF content."""
-    try:
-        # Open the binary content with PyMuPDF
-        pdf_document = fitz.open("pdf", content)  # "pdf" specifies the format
-
-        # Extract metadata
-        metadata = pdf_document.metadata
-
-        # Prepare metadata dictionary with default values for missing fields
-        metadata_dict = {
-            "title": metadata.get("title", "N/A"),
-            "author": metadata.get("author", "N/A"),
-            "subject": metadata.get("subject", "N/A"),
-            "keywords": metadata.get("keywords", "N/A"),
-            "creation_date": metadata.get("created", "N/A"),
-            "modification_date": metadata.get("modified", "N/A"),
-        }
-
-        return metadata_dict
-
-    except Exception as e:
-        return JSONResponse(status_code=500, content=f"Error inputting metadata: {e}")
-
-
-async def upload_file(file: UploadFile = File(...)):
-    try:
-        # Read the binary content of the uploaded file once
-        content = await file.read()
-        # Parse the journal
-        parsed_documents = await parse_journal(content, file.filename)
-        # Extract metadata
-        metadata_dict = await extract_metadata(content)
-
-        print("Metadata Dictionary : \n\n", metadata_dict)
-
-        metadata_gen = Metadata(metadata_dict)
-        documents_with_metadata = metadata_gen.add_metadata(
-            parsed_documents, metadata_dict
-        )
-
-        print("Document with Metadata : \n\n", documents_with_metadata)
-        print("Banyak documents : \n", len(documents_with_metadata))
-
-        # Return both parsed documents and metadata
-        return {"status": "SUCCESS"}
-
-    except Exception as e:
-        return JSONResponse(status_code=500, content=f"Error processing file: {e}")

core/summarization/summarizer.py

@@ -1,135 +0,0 @@
-from io import BytesIO
-import os
-import base64
-import fitz
-
-from fastapi.responses import JSONResponse
-from llama_index.core.vector_stores import (
-    MetadataFilter,
-    MetadataFilters,
-    FilterCondition,
-)
-
-from llama_index.core import load_index_from_storage
-from llama_index.core.storage import StorageContext
-from llama_index.llms.openai import OpenAI
-from core.parser import parse_topics_to_dict
-from llama_index.core.llms import ChatMessage
-from core.prompt import (
-    SYSTEM_TOPIC_TEMPLATE,
-    USER_TOPIC_TEMPLATE,
-    REFINED_GET_TOPIC_TEMPLATE,
-)
-
-# from langfuse.openai import openai
-
-
-class SummarizeGenerator:
-    def __init__(self, references):
-
-        self.references = references
-        self.llm = OpenAI(temperature=0, model="gpt-4o-mini", max_tokens=4096)
-
-    def extract_pages(self, content_table):
-        try:
-            content_bytes = content_table.file.read()
-            print(content_bytes)
-            # Open the PDF file
-            content_table = fitz.open(stream=content_bytes, filetype="pdf")
-            print(content_table)
-            # content_table = fitz.open(topics_image)
-        except Exception as e:
-            return JSONResponse(status_code=400, content=f"Error opening PDF file: {e}")
-
-        # Initialize a list to collect base64 encoded images
-        pix_encoded_combined = []
-
-        # Iterate over each page to extract images
-        for page_number in range(len(content_table)):
-            try:
-                page = content_table.load_page(page_number)
-                pix_encoded = self._extract_image_as_base64(page)
-                pix_encoded_combined.append(pix_encoded)
-                # print("pix encoded combined", pix_encoded_combined)
-
-            except Exception as e:
-                print(f"Error processing page {page_number}: {e}")
-                continue  # Skip to the next page if there's an error
-
-        if not pix_encoded_combined:
-            return JSONResponse(status_code=404, content="No images found in the PDF")
-
-        return pix_encoded_combined
-
-    def extract_content_table(self, content_table):
-        try:
-            images = self.extract_pages(content_table)
-
-            image_messages = [
-                {
-                    "type": "image_url",
-                    "image_url": {
-                        "url": f"data:image/jpeg;base64,{image}",
-                    },
-                }
-                for image in images
-            ]
-
-            messages = [
-                ChatMessage(
-                    role="system",
-                    content=[{"type": "text", "text": SYSTEM_TOPIC_TEMPLATE}],
-                ),
-                ChatMessage(
-                    role="user",
-                    content=[
-                        {"type": "text", "text": USER_TOPIC_TEMPLATE},
-                        *image_messages,
-                    ],
-                ),
-            ]
-
-            extractor_output = self.llm.chat(messages)
-            print("extractor output : ", extractor_output)
-            refined_extractor_output = self.llm.complete(
-                REFINED_GET_TOPIC_TEMPLATE.format(topics=str(extractor_output))
-            )
-        
-            print("refined extractor output : ",str(refined_extractor_output))
-
-            extractor_dics = dict(parse_topics_to_dict(str(refined_extractor_output)))
-
-            return str(refined_extractor_output), extractor_dics
-
-        except Exception as e:
-            return JSONResponse(status_code=500, content=f"An error occurred: {e}")
-
-    def _extract_image_as_base64(self, page):
-        try:
-            pix = page.get_pixmap()
-            pix_bytes = pix.tobytes()
-            return base64.b64encode(pix_bytes).decode("utf-8")
-        except Exception as e:
-            return JSONResponse(status_code=500, content=f"Error extracting image: {e}")
-
-    def index_summarizer_engine(self, topic, subtopic, index):
-        filters = MetadataFilters(
-            filters=[
-                MetadataFilter(key="title", value=topic),
-                MetadataFilter(key="category", value=subtopic),
-            ],
-            condition=FilterCondition.AND,
-        )
-        
-        # Create the QueryEngineTool with the index and filters 
-        kwargs = {"similarity_top_k": 5, "filters": filters}
-        
-        query_engine = index.as_query_engine(**kwargs)
-
-        return query_engine
-
-    def get_summarizer_engine(self, topic, subtopic):
-        pass
-
-    def prepare_summaries(self):
-        pass

script/document_uploader.py

@@ -1,8 +1,10 @@
+import logging
+import random
+
 from llama_index.core.ingestion import IngestionPipeline
 from llama_index.embeddings.openai import OpenAIEmbedding
 from config import PINECONE_CONFIG
 from pinecone.grpc import PineconeGRPC as Pinecone
-# from service.reader import Reader
 from script.get_metadata import Metadata
 from fastapi import UploadFile, status
 from fastapi.responses import JSONResponse
@@ -12,32 +14,17 @@ from llama_index.core.node_parser import (
     SemanticSplitterNodeParser,
 )
 from llama_index.core import Settings
-# from service.reader_v3 import upload_file
 from service.reader_v4 import upload_file
 
-# from script.get_topic import extract_topic
 
-import logging
-import random
 
 
 class Uploader:
-    # def __init__(self, reference, file: UploadFile, content_table: UploadFile):
     def __init__(self, reference, file: UploadFile):
         self.file = file
-        # self.content_table = content_table
-        # self.reader = Reader()
         self.reference = reference
         self.metadata = Metadata(reference)
 
-    # async def ingest_documents(self, file: UploadFile):
-    #     """Load documents from the storage path."""
-    #     documents = await self.reader.read_from_uploadfile(file)
-    #     print("Banyak document : ", len(documents))
-    #     print("document successfully ingested")
-
-    #     return documents
-
     def check_existing_metadata(self, pinecone_index, title, random_vector):
         try:
             result = pinecone_index.query(
@@ -56,20 +43,11 @@ class Uploader:
 
     async def process_documents(self):
         # Ingest documents
-        # documents = await self.ingest_documents(self.file)
-
-        # Get metadata
-        # documents_with_metadata = self.metadata.apply_metadata(documents)
-        documents_with_metadata = await upload_file(self.reference, self.file)
+        documents_with_metadata, file_stream = await upload_file(self.reference, self.file)
 
         if isinstance(documents_with_metadata, JSONResponse):
             return documents_with_metadata  # Return the error response directly
 
-        # Get Topic
-        # topic_extractor = extract_topic(self.reference, self.content_table)
-        # document_filtered = self.filter_document(documents_with_metadata)
-
-        # embed_model = OpenAIEmbedding()
         embed_model = OpenAIEmbedding(model="text-embedding-3-large")
         Settings.embed_model = embed_model
         # Set up the ingestion pipeline
@@ -80,20 +58,14 @@ class Uploader:
                     breakpoint_percentile_threshold=95,
                     embed_model=embed_model,
                 ),
-                # topic_extractor,
             ]
         )
 
-        # splitter = SemanticSplitterNodeParser(
-        #     buffer_size=1, breakpoint_percentile_threshold=95, embed_model=embed_model
-        # )
-
         #  Run the pipeline
         try:
             print("Pipeline processing completed with Semantic Spliter.")
             nodes_with_metadata = pipeline.run(documents=documents_with_metadata)
-            # nodes_with_metadata = splitter.get_nodes_from_documents(documents_with_metadata)
-            return nodes_with_metadata
+            return nodes_with_metadata, file_stream
 
         except Exception as e:
             try:
@@ -103,7 +75,7 @@ class Uploader:
                     documents_with_metadata
                 )
                 print("Pipeline processing completed with SentenceSplitter fallback.")
-                return nodes_with_metadata
+                return nodes_with_metadata, file_stream
             except Exception as fallback_error:
                 # Log the second error and return JSONResponse for FastAPI
                 logging.error(f"Error with SentenceSplitter fallback: {fallback_error}")

service/reader_v4.py

@@ -1,7 +1,8 @@
 import os
 import nest_asyncio
-from typing import List
+from io import BytesIO
 
+from typing import List
 from dotenv import load_dotenv
 from fastapi import UploadFile
 import joblib
@@ -85,6 +86,10 @@ async def upload_file(reference, file: UploadFile):
     try:
         # Read the binary content of the uploaded file once
         content = await file.read()
+        
+        # Store the file content in a BytesIO stream for reuse later
+        file_stream = BytesIO(content)
+        
         # Parse the journal
         title = reference["title"]
 
@@ -97,7 +102,7 @@ async def upload_file(reference, file: UploadFile):
         print("Banyak documents : \n", len(documents_with_metadata))
 
         # Return both parsed documents and metadata
-        return documents_with_metadata
+        return documents_with_metadata, file_stream
 
     except Exception as e:
         print("error ", e)