neo4j-labs
diff --git a/‎backend/Dockerfile
+1-1 b/‎backend/Dockerfile
+1-1
diff --git a/‎backend/Performance_test.py
+1 b/‎backend/Performance_test.py
+1
diff --git a/‎backend/score.py
+41-25 b/‎backend/score.py
+41-25
diff --git a/‎backend/src/main.py
+50-68 b/‎backend/src/main.py
+50-68
diff --git a/‎backend/src/make_relationships.py
+1 b/‎backend/src/make_relationships.py
+1
@@ -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"]
@@ -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))
 
@@ -171,30 +171,30 @@ async def extract_knowledge_graph_from_file(
 
         if source_type == 'local file':
             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'
@@ -443,25 +443,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")
 
@@ -635,6 +635,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)
@@ -144,7 +144,6 @@ def create_source_node_graph_web_url(graph, model, source_url, source_type):
     obj_source_node.file_type = 'text'
     obj_source_node.file_source = source_type
     obj_source_node.model = model
-    obj_source_node.total_pages = 1
     obj_source_node.url = urllib.parse.unquote(source_url)
     obj_source_node.created_at = datetime.now()
     obj_source_node.file_name = pages[0].metadata['title']
@@ -228,7 +227,7 @@ def create_source_node_graph_url_wikipedia(graph, model, wiki_query, source_type
       lst_file_name.append({'fileName':obj_source_node.file_name,'fileSize':obj_source_node.file_size,'url':obj_source_node.url, 'language':obj_source_node.language, 'status':'Success'})
     return lst_file_name,success_count,failed_count
 
-def extract_graph_from_file_local_file(uri, userName, password, database, model, merged_file_path, fileName, allowedNodes, allowedRelationship):
+def extract_graph_from_file_local_file(uri, userName, password, database, model, merged_file_path, fileName, allowedNodes, allowedRelationship, retry_condition):
 
   logging.info(f'Process file name :{fileName}')
   if retry_condition is None:
@@ -242,35 +241,7 @@ def extract_graph_from_file_local_file(uri, userName, password, database, model,
       raise Exception(f'File content is not available for file : {file_name}')
     return processing_source(uri, userName, password, database, model, file_name, pages, allowedNodes, allowedRelationship, True, merged_file_path)
   else:
-    file_name, pages, file_extension = get_documents_from_file_by_path(merged_file_path,fileName)
-  if pages==None or len(pages)==0:
-    raise Exception(f'File content is not available for file : {file_name}')
-
-  return processing_source(uri, userName, password, database, model, file_name, pages, allowedNodes, allowedRelationship, True, merged_file_path)
-
-def extract_graph_from_file_s3(uri, userName, password, database, model, source_url, aws_access_key_id, aws_secret_access_key, allowedNodes, allowedRelationship):
-
-  if(aws_access_key_id==None or aws_secret_access_key==None):
-    raise Exception('Please provide AWS access and secret keys')
-  else:
-    logging.info("Insert in S3 Block")
-    file_name, pages = get_documents_from_s3(source_url, aws_access_key_id, aws_secret_access_key)
-
-  if pages==None or len(pages)==0:
-    raise Exception(f'File content is not available for file : {file_name}')
-
-  return processing_source(uri, userName, password, database, model, file_name, pages, allowedNodes, allowedRelationship)
-
-def extract_graph_from_web_page(uri, userName, password, database, model, source_url, allowedNodes, allowedRelationship):
-
-  file_name, pages = get_documents_from_web_page(source_url)
-
-  if pages==None or len(pages)==0:
-    raise Exception(f'Content is not available for given URL : {file_name}')
-
-  return processing_source(uri, userName, password, database, model, file_name, pages, allowedNodes, allowedRelationship)
-
-def extract_graph_from_file_youtube(uri, userName, password, database, model, source_url, allowedNodes, allowedRelationship):
+    return processing_source(uri, userName, password, database, model, fileName, [], allowedNodes, allowedRelationship, True, merged_file_path, retry_condition)
 
 def 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):
   if retry_condition is None:
@@ -290,25 +261,41 @@ def extract_graph_from_web_page(uri, userName, password, database, model, source
   if retry_condition is None:
     file_name, pages = get_documents_from_web_page(source_url)
 
-  return processing_source(uri, userName, password, database, model, file_name, pages, allowedNodes, allowedRelationship)
-
-def extract_graph_from_file_Wikipedia(uri, userName, password, database, model, wiki_query, max_sources, language, allowedNodes, allowedRelationship):
-
-  file_name, pages = get_documents_from_Wikipedia(wiki_query, language)
-  if pages==None or len(pages)==0:
-    raise Exception(f'Wikipedia page is not available for file : {file_name}')
-
-  return processing_source(uri, userName, password, database, model, file_name, pages, allowedNodes, allowedRelationship)
-
-def 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):
-
-  file_name, pages = get_documents_from_gcs(gcs_project_id, gcs_bucket_name, gcs_bucket_folder, gcs_blob_filename, access_token)
-  if pages==None or len(pages)==0:
-    raise Exception(f'File content is not available for file : {file_name}')
+    if pages==None or len(pages)==0:
+      raise Exception(f'Content is not available for given URL : {file_name}')
+    return processing_source(uri, userName, password, database, model, file_name, pages, allowedNodes, allowedRelationship)
+  else:
+    return processing_source(uri, userName, password, database, model, file_name, [], allowedNodes, allowedRelationship, retry_condition=retry_condition)
+  
+def extract_graph_from_file_youtube(uri, userName, password, database, model, source_url, file_name, allowedNodes, allowedRelationship, retry_condition):
+  if retry_condition is None:
+    file_name, pages = get_documents_from_youtube(source_url)
 
-  return processing_source(uri, userName, password, database, model, file_name, pages, allowedNodes, allowedRelationship)
+    if pages==None or len(pages)==0:
+      raise Exception(f'Youtube transcript is not available for file : {file_name}')
+    return processing_source(uri, userName, password, database, model, file_name, pages, allowedNodes, allowedRelationship)
+  else:
+     return processing_source(uri, userName, password, database, model, file_name, [], allowedNodes, allowedRelationship, retry_condition=retry_condition)
+    
+def extract_graph_from_file_Wikipedia(uri, userName, password, database, model, wiki_query, language, file_name, allowedNodes, allowedRelationship, retry_condition):
+  if retry_condition is None:
+    file_name, pages = get_documents_from_Wikipedia(wiki_query, language)
+    if pages==None or len(pages)==0:
+      raise Exception(f'Wikipedia page is not available for file : {file_name}')
+    return processing_source(uri, userName, password, database, model, file_name, pages, allowedNodes, allowedRelationship)
+  else:
+    return processing_source(uri, userName, password, database, model, file_name,[], allowedNodes, allowedRelationship, retry_condition=retry_condition)
 
-def processing_source(uri, userName, password, database, model, file_name, pages, allowedNodes, allowedRelationship, is_uploaded_from_local=None, merged_file_path=None):
+def 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):
+  if retry_condition is None:
+    file_name, pages = get_documents_from_gcs(gcs_project_id, gcs_bucket_name, gcs_bucket_folder, gcs_blob_filename, access_token)
+    if pages==None or len(pages)==0:
+      raise Exception(f'File content is not available for file : {file_name}')
+    return processing_source(uri, userName, password, database, model, file_name, pages, allowedNodes, allowedRelationship)
+  else:
+    return processing_source(uri, userName, password, database, model, file_name, [], allowedNodes, allowedRelationship, retry_condition=retry_condition)
+  
+def processing_source(uri, userName, password, database, model, file_name, pages, allowedNodes, allowedRelationship, is_uploaded_from_local=None, merged_file_path=None, retry_condition=None):
   """
    Extracts a Neo4jGraph from a PDF file based on the model.
    
@@ -331,29 +318,24 @@ def processing_source(uri, userName, password, database, model, file_name, pages
 
   total_chunks, chunkId_chunkDoc_list = get_chunkId_chunkDoc_list(graph, file_name, pages, retry_condition)
   result = graphDb_data_Access.get_current_status_document_node(file_name)
-  logging.info("Break down file into chunks")
-  bad_chars = ['"', "\n", "'"]
-  for i in range(0,len(pages)):
-    text = pages[i].page_content
-    for j in bad_chars:
-      if j == '\n':
-        text = text.replace(j, ' ')
-      else:
-        text = text.replace(j, '')
-    pages[i]=Document(page_content=str(text), metadata=pages[i].metadata)
-  create_chunks_obj = CreateChunksofDocument(pages, graph)
-  chunks = create_chunks_obj.split_file_into_chunks()
-  chunkId_chunkDoc_list = create_relation_between_chunks(graph,file_name,chunks)
-  
+ 
+  select_chunks_with_retry=0
+  node_count = 0
+  rel_count = 0
+      
   if len(result) > 0:
     if result[0]['Status'] != 'Processing':      
       obj_source_node = sourceNode()
       status = "Processing"
       obj_source_node.file_name = file_name
       obj_source_node.status = status
-      obj_source_node.total_chunks = len(chunks)
-      obj_source_node.total_pages = len(pages)
+      obj_source_node.total_chunks = total_chunks
       obj_source_node.model = model
+      if retry_condition == START_FROM_LAST_PROCESSED_POSITION:
+          node_count = result[0]['nodeCount']
+          rel_count = result[0]['relationshipCount']
+          select_chunks_with_retry = result[0]['processed_chunk']
+      obj_source_node.processed_chunk = 0+select_chunks_with_retry
       logging.info(file_name)
       logging.info(obj_source_node)
       graphDb_data_Access.update_source_node(obj_source_node)
@@ -363,21 +345,21 @@ def processing_source(uri, userName, password, database, model, file_name, pages
       # selected_chunks = []
       is_cancelled_status = False
       job_status = "Completed"
-      node_count = 0
-      rel_count = 0
+
       for i in range(0, len(chunkId_chunkDoc_list), update_graph_chunk_processed):
         select_chunks_upto = i+update_graph_chunk_processed
         logging.info(f'Selected Chunks upto: {select_chunks_upto}')
         if len(chunkId_chunkDoc_list) <= select_chunks_upto:
           select_chunks_upto = len(chunkId_chunkDoc_list)
         selected_chunks = chunkId_chunkDoc_list[i:select_chunks_upto]
+        
         result = graphDb_data_Access.get_current_status_document_node(file_name)
         is_cancelled_status = result[0]['is_cancelled']
         logging.info(f"Value of is_cancelled : {result[0]['is_cancelled']}")
         if bool(is_cancelled_status) == True:
           job_status = "Cancelled"
           logging.info('Exit from running loop of processing file')
-          exit
+          break
         else:
           processing_chunks_start_time = time.time()
           node_count,rel_count = processing_chunks(selected_chunks,graph,uri, userName, password, database,file_name,model,allowedNodes,allowedRelationship,node_count, rel_count)
@@ -390,8 +372,8 @@ def processing_source(uri, userName, password, database, model, file_name, pages
           obj_source_node.file_name = file_name
           obj_source_node.updated_at = end_time
           obj_source_node.processing_time = processed_time
+          obj_source_node.processed_chunk = select_chunks_upto+select_chunks_with_retry
           obj_source_node.node_count = node_count
-          obj_source_node.processed_chunk = select_chunks_upto
           obj_source_node.relationship_count = rel_count
           graphDb_data_Access.update_source_node(obj_source_node)
 
 
@@ -88,6 +88,7 @@ def update_embedding_create_vector_index(graph, chunkId_chunkDoc_list, file_name
             #             )
             # logging.info('create vector index on chunk embedding')
             result = graph.query("SHOW INDEXES YIELD * WHERE labelsOrTypes = ['__Chunk__'] and name = 'vector'")
+            vector_index = graph.query("SHOW INDEXES YIELD * WHERE labelsOrTypes = ['Chunk'] and type = 'VECTOR' AND name = 'vector' return options")
             if result:
                 logging.info(f"vector index dropped for 'Chunk'")
                 graph.query("DROP INDEX vector IF EXISTS;")