Staging

backend/Dockerfile

@@ -21,4 +21,4 @@ RUN pip install -r requirements.txt
 # Copy application code
 COPY . /code
 # Set command
-CMD ["gunicorn", "score:app", "--workers", "8","--preload","--threads", "8", "--worker-class", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000", "--timeout", "300"]
+CMD ["gunicorn", "score:app", "--workers", "8","--threads", "8", "--worker-class", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000", "--timeout", "300"]

backend/Performance_test.py

@@ -94,6 +94,7 @@ def performance_main():
         for _ in range(CONCURRENT_REQUESTS):
             futures.append(executor.submit(post_request_chunk))
 
+        #  Chatbot request futures
         #  Chatbot request futures
         # for message in CHATBOT_MESSAGES:
         #     futures.append(executor.submit(chatbot_request, message))

backend/score.py

@@ -46,7 +46,6 @@ def sick():
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
-    allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
@@ -137,7 +136,8 @@ async def extract_knowledge_graph_from_file(
     allowedNodes=Form(None),
     allowedRelationship=Form(None),
     language=Form(None),
-    access_token=Form(None)
+    access_token=Form(None),
+    retry_condition=Form(None)
 ):
     """
     Calls 'extract_graph_from_file' in a new thread to create Neo4jGraph from a
@@ -161,30 +161,30 @@ async def extract_knowledge_graph_from_file(
             merged_file_path = os.path.join(MERGED_DIR,file_name)
             logging.info(f'File path:{merged_file_path}')
             result = await asyncio.to_thread(
-                extract_graph_from_file_local_file, uri, userName, password, database, model, merged_file_path, file_name, allowedNodes, allowedRelationship)
+                extract_graph_from_file_local_file, uri, userName, password, database, model, merged_file_path, file_name, allowedNodes, allowedRelationship, retry_condition)
 
         elif source_type == 's3 bucket' and source_url:
             result = await asyncio.to_thread(
-                extract_graph_from_file_s3, uri, userName, password, database, model, source_url, aws_access_key_id, aws_secret_access_key, allowedNodes, allowedRelationship)
+                extract_graph_from_file_s3, uri, userName, password, database, model, source_url, aws_access_key_id, aws_secret_access_key, file_name, allowedNodes, allowedRelationship, retry_condition)
 
         elif source_type == 'web-url':
             result = await asyncio.to_thread(
-                extract_graph_from_web_page, uri, userName, password, database, model, source_url, allowedNodes, allowedRelationship)
+                extract_graph_from_web_page, uri, userName, password, database, model, source_url, file_name, allowedNodes, allowedRelationship, retry_condition)
 
         elif source_type == 'youtube' and source_url:
             result = await asyncio.to_thread(
-                extract_graph_from_file_youtube, uri, userName, password, database, model, source_url, allowedNodes, allowedRelationship)
+                extract_graph_from_file_youtube, uri, userName, password, database, model, source_url, file_name, allowedNodes, allowedRelationship, retry_condition)
 
         elif source_type == 'Wikipedia' and wiki_query:
             result = await asyncio.to_thread(
-                extract_graph_from_file_Wikipedia, uri, userName, password, database, model, wiki_query, max_sources, language, allowedNodes, allowedRelationship)
+                extract_graph_from_file_Wikipedia, uri, userName, password, database, model, wiki_query, language, file_name, allowedNodes, allowedRelationship, retry_condition)
 
         elif source_type == 'gcs bucket' and gcs_bucket_name:
             result = await asyncio.to_thread(
-                extract_graph_from_file_gcs, uri, userName, password, database, model, gcs_project_id, gcs_bucket_name, gcs_bucket_folder, gcs_blob_filename, access_token, allowedNodes, allowedRelationship)
+                extract_graph_from_file_gcs, uri, userName, password, database, model, gcs_project_id, gcs_bucket_name, gcs_bucket_folder, gcs_blob_filename, access_token, file_name, allowedNodes, allowedRelationship, retry_condition)
         else:
             return create_api_response('Failed',message='source_type is other than accepted source')
-        
+
         if result is not None:
             result['db_url'] = uri
             result['api_name'] = 'extract'
@@ -433,25 +433,25 @@ async def generate():
                     logging.info(" SSE Client disconnected")
                     break
                 # get the current status of document node
-                graph = create_graph_database_connection(uri, userName, decoded_password, database)
-                graphDb_data_Access = graphDBdataAccess(graph)
-                result = graphDb_data_Access.get_current_status_document_node(file_name)
-                if result is not None:
-                    status = json.dumps({'fileName':file_name, 
-                    'status':result[0]['Status'],
-                    'processingTime':result[0]['processingTime'],
-                    'nodeCount':result[0]['nodeCount'],
-                    'relationshipCount':result[0]['relationshipCount'],
-                    'model':result[0]['model'],
-                    'total_chunks':result[0]['total_chunks'],
-                    'total_pages':result[0]['total_pages'],
-                    'fileSize':result[0]['fileSize'],
-                    'processed_chunk':result[0]['processed_chunk'],
-                    'fileSource':result[0]['fileSource']
-                    })
+
                 else:
-                    status = json.dumps({'fileName':file_name, 'status':'Failed'})
-                yield status
+                    graph = create_graph_database_connection(uri, userName, decoded_password, database)
+                    graphDb_data_Access = graphDBdataAccess(graph)
+                    result = graphDb_data_Access.get_current_status_document_node(file_name)
+                    print(f'Result of document status in SSE : {result}')
+                    if len(result) > 0:
+                        status = json.dumps({'fileName':file_name, 
+                        'status':result[0]['Status'],
+                        'processingTime':result[0]['processingTime'],
+                        'nodeCount':result[0]['nodeCount'],
+                        'relationshipCount':result[0]['relationshipCount'],
+                        'model':result[0]['model'],
+                        'total_chunks':result[0]['total_chunks'],
+                        'fileSize':result[0]['fileSize'],
+                        'processed_chunk':result[0]['processed_chunk'],
+                        'fileSource':result[0]['fileSource']
+                        })
+                    yield status
             except asyncio.CancelledError:
                 logging.info("SSE Connection cancelled")
 
@@ -495,21 +495,21 @@ async def get_document_status(file_name, url, userName, password, database):
         graph = create_graph_database_connection(uri, userName, decoded_password, database)
         graphDb_data_Access = graphDBdataAccess(graph)
         result = graphDb_data_Access.get_current_status_document_node(file_name)
-        if result is not None:
+        if len(result) > 0:
             status = {'fileName':file_name, 
                 'status':result[0]['Status'],
                 'processingTime':result[0]['processingTime'],
                 'nodeCount':result[0]['nodeCount'],
                 'relationshipCount':result[0]['relationshipCount'],
                 'model':result[0]['model'],
                 'total_chunks':result[0]['total_chunks'],
-                'total_pages':result[0]['total_pages'],
                 'fileSize':result[0]['fileSize'],
                 'processed_chunk':result[0]['processed_chunk'],
                 'fileSource':result[0]['fileSource']
                 }
         else:
             status = {'fileName':file_name, 'status':'Failed'}
+        print(f'Result of document status in refresh : {result}')
         return create_api_response('Success',message="",file_name=status)
     except Exception as e:
         message=f"Unable to get the document status"
@@ -626,6 +626,22 @@ async def merge_duplicate_nodes(uri=Form(), userName=Form(), password=Form(), da
         return create_api_response(job_status, message=message, error=error_message)
     finally:
         gc.collect()
+
+@app.post("/retry_processing")
+async def retry_processing(uri=Form(), userName=Form(), password=Form(), database=Form(), file_name=Form(), retry_condition=Form()):
+    try:
+        graph = create_graph_database_connection(uri, userName, password, database)
+        await asyncio.to_thread(set_status_retry, graph,file_name,retry_condition)
+        #set_status_retry(graph,file_name,retry_condition)
+        return create_api_response('Success',message=f"Status set to Reprocess for filename : {file_name}")
+    except Exception as e:
+        job_status = "Failed"
+        message="Unable to set status to Retry"
+        error_message = str(e)
+        logging.exception(f'{error_message}')
+        return create_api_response(job_status, message=message, error=error_message)
+    finally:
+        gc.collect()        
 
 if __name__ == "__main__":
     uvicorn.run(app)

backend/src/document_sources/gcs_bucket.py

@@ -122,7 +122,6 @@ def merge_file_gcs(bucket_name, original_file_name: str, folder_name_sha1_hashed
       blob.upload_from_file(file_io)
       # pdf_reader = PdfReader(file_io)
       file_size = len(merged_file)
-      # total_pages = len(pdf_reader.pages)
 
       return file_size
   except Exception as e:

backend/src/document_sources/local_file.py

@@ -56,19 +56,19 @@ def get_pages_with_page_numbers(unstructured_pages):
             if page.metadata['page_number']==page_number:
                 page_content += page.page_content
                 metadata = {'source':page.metadata['source'],'page_number':page_number, 'filename':page.metadata['filename'],
-                        'filetype':page.metadata['filetype'], 'total_pages':unstructured_pages[-1].metadata['page_number']}
+                        'filetype':page.metadata['filetype']}
 
             if page.metadata['page_number']>page_number:
                 page_number+=1
-                if not metadata:
-                    metadata = {'total_pages':unstructured_pages[-1].metadata['page_number']}
-                pages.append(Document(page_content = page_content, metadata=metadata))
+                # if not metadata:
+                #     metadata = {'total_pages':unstructured_pages[-1].metadata['page_number']}
+                pages.append(Document(page_content = page_content))
                 page_content='' 
 
             if page == unstructured_pages[-1]:
-                if not metadata:
-                    metadata = {'total_pages':unstructured_pages[-1].metadata['page_number']}
-                pages.append(Document(page_content = page_content, metadata=metadata))
+                # if not metadata:
+                #     metadata = {'total_pages':unstructured_pages[-1].metadata['page_number']}
+                pages.append(Document(page_content = page_content))
 
         elif page.metadata['category']=='PageBreak' and page!=unstructured_pages[0]:
             page_number+=1
@@ -80,7 +80,7 @@ def get_pages_with_page_numbers(unstructured_pages):
             page_content += page.page_content
             metadata_with_custom_page_number = {'source':page.metadata['source'],
                             'page_number':1, 'filename':page.metadata['filename'],
-                            'filetype':page.metadata['filetype'], 'total_pages':1}
+                            'filetype':page.metadata['filetype']}
             if page == unstructured_pages[-1]:
                     pages.append(Document(page_content = page_content, metadata=metadata_with_custom_page_number))
     return pages                

backend/src/document_sources/wikipedia.py

@@ -4,7 +4,7 @@
 
 def get_documents_from_Wikipedia(wiki_query:str, language:str):
   try:
-    pages = WikipediaLoader(query=wiki_query.strip(), lang=language, load_max_docs=1, load_all_available_meta=False).load()
+    pages = WikipediaLoader(query=wiki_query.strip(), lang=language, load_all_available_meta=False).load()
     file_name = wiki_query.strip()
     logging.info(f"Total Pages from Wikipedia = {len(pages)}") 
     return file_name, pages

backend/src/entities/source_node.py

@@ -18,9 +18,9 @@ class sourceNode:
     updated_at:datetime=None
     processing_time:float=None
     error_message:str=None
-    total_pages:int=None
     total_chunks:int=None
     language:str=None
     is_cancelled:bool=None
     processed_chunk:int=None
     access_token:str=None
+    retry_condition:str=None

backend/src/graphDB_dataAccess.py

@@ -37,14 +37,14 @@ def create_source_node(self, obj_source_node:sourceNode):
                             d.processingTime = $pt, d.errorMessage = $e_message, d.nodeCount= $n_count, 
                             d.relationshipCount = $r_count, d.model= $model, d.gcsBucket=$gcs_bucket, 
                             d.gcsBucketFolder= $gcs_bucket_folder, d.language= $language,d.gcsProjectId= $gcs_project_id,
-                            d.is_cancelled=False, d.total_chunks=0, d.processed_chunk=0, d.total_pages=$total_pages,
+                            d.is_cancelled=False, d.total_chunks=0, d.processed_chunk=0,
                             d.access_token=$access_token""",
                             {"fn":obj_source_node.file_name, "fs":obj_source_node.file_size, "ft":obj_source_node.file_type, "st":job_status, 
                             "url":obj_source_node.url,
                             "awsacc_key_id":obj_source_node.awsAccessKeyId, "f_source":obj_source_node.file_source, "c_at":obj_source_node.created_at,
                             "u_at":obj_source_node.created_at, "pt":0, "e_message":'', "n_count":0, "r_count":0, "model":obj_source_node.model,
                             "gcs_bucket": obj_source_node.gcsBucket, "gcs_bucket_folder": obj_source_node.gcsBucketFolder, 
-                            "language":obj_source_node.language, "gcs_project_id":obj_source_node.gcsProjectId, "total_pages": obj_source_node.total_pages,
+                            "language":obj_source_node.language, "gcs_project_id":obj_source_node.gcsProjectId,
                             "access_token":obj_source_node.access_token})
         except Exception as e:
             error_message = str(e)
@@ -71,26 +71,26 @@ def update_source_node(self, obj_source_node:sourceNode):
             if obj_source_node.processing_time is not None and obj_source_node.processing_time != 0:
                 params['processingTime'] = round(obj_source_node.processing_time.total_seconds(),2)
 
-            if obj_source_node.node_count is not None and obj_source_node.node_count != 0:
+            if obj_source_node.node_count is not None :
                 params['nodeCount'] = obj_source_node.node_count
 
-            if obj_source_node.relationship_count is not None and obj_source_node.relationship_count != 0:
+            if obj_source_node.relationship_count is not None :
                 params['relationshipCount'] = obj_source_node.relationship_count
 
             if obj_source_node.model is not None and obj_source_node.model != '':
                 params['model'] = obj_source_node.model
 
-            if obj_source_node.total_pages is not None and obj_source_node.total_pages != 0:
-                params['total_pages'] = obj_source_node.total_pages
-
             if obj_source_node.total_chunks is not None and obj_source_node.total_chunks != 0:
                 params['total_chunks'] = obj_source_node.total_chunks
 
-            if obj_source_node.is_cancelled is not None and obj_source_node.is_cancelled != False:
+            if obj_source_node.is_cancelled is not None:
                 params['is_cancelled'] = obj_source_node.is_cancelled
 
-            if obj_source_node.processed_chunk is not None and obj_source_node.processed_chunk != 0:
+            if obj_source_node.processed_chunk is not None :
                 params['processed_chunk'] = obj_source_node.processed_chunk
+
+            if obj_source_node.retry_condition is not None :
+                params['retry_condition'] = obj_source_node.retry_condition    
 
             param= {"props":params}
 
@@ -187,7 +187,7 @@ def get_current_status_document_node(self, file_name):
         query = """
                 MATCH(d:Document {fileName : $file_name}) RETURN d.status AS Status , d.processingTime AS processingTime, 
                 d.nodeCount AS nodeCount, d.model as model, d.relationshipCount as relationshipCount,
-                d.total_pages AS total_pages, d.total_chunks AS total_chunks , d.fileSize as fileSize, 
+                d.total_chunks AS total_chunks , d.fileSize as fileSize, 
                 d.is_cancelled as is_cancelled, d.processed_chunk as processed_chunk, d.fileSource as fileSource
                 """
         param = {"file_name" : file_name}