Spaces:
Sleeping
Sleeping
Add automatic document detection and collection clearing: Implement file hash tracking to automatically clear all Qdrant collections when a new document is uploaded, ensuring data isolation between different documents. Added manual reset button in Data Management tab for force clearing when reprocessing same file."
Browse files
app.py
CHANGED
|
@@ -1,6 +1,7 @@
|
|
| 1 |
import gradio as gr
|
| 2 |
import logging
|
| 3 |
import os
|
|
|
|
| 4 |
from typing import List, Tuple, Optional, Dict, Any
|
| 5 |
import pandas as pd
|
| 6 |
from qdrant_client import QdrantClient
|
|
@@ -24,6 +25,9 @@ logger = logging.getLogger(__name__)
|
|
| 24 |
document_processor = None
|
| 25 |
embeddings_pipeline = None
|
| 26 |
|
|
|
|
|
|
|
|
|
|
| 27 |
# OpenAI client (lazy init)
|
| 28 |
_openai_client: Optional[OpenAI] = None
|
| 29 |
|
|
@@ -34,6 +38,58 @@ def _get_openai_client() -> OpenAI:
|
|
| 34 |
_openai_client = OpenAI(api_key=api_key)
|
| 35 |
return _openai_client
|
| 36 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 37 |
def initialize_components():
|
| 38 |
"""Initialize the pipeline components with error handling for cloud deployment."""
|
| 39 |
global document_processor, embeddings_pipeline
|
|
@@ -57,6 +113,7 @@ def initialize_components():
|
|
| 57 |
def process_document(file_path: str, chunking_methods: List[str]) -> Tuple[str, pd.DataFrame]:
|
| 58 |
"""
|
| 59 |
Process a document with selected chunking methods.
|
|
|
|
| 60 |
|
| 61 |
Args:
|
| 62 |
file_path: Path to the uploaded file
|
|
@@ -65,6 +122,8 @@ def process_document(file_path: str, chunking_methods: List[str]) -> Tuple[str,
|
|
| 65 |
Returns:
|
| 66 |
Tuple of (status_message, chunks_dataframe)
|
| 67 |
"""
|
|
|
|
|
|
|
| 68 |
try:
|
| 69 |
if document_processor is None:
|
| 70 |
return "β Error: Document processor not initialized. Check system configuration.", pd.DataFrame()
|
|
@@ -78,6 +137,26 @@ def process_document(file_path: str, chunking_methods: List[str]) -> Tuple[str,
|
|
| 78 |
if not chunking_methods:
|
| 79 |
return "β Error: Please select at least one chunking method", pd.DataFrame()
|
| 80 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 81 |
# Process the document
|
| 82 |
logger.info(f"Processing document: {file_path}")
|
| 83 |
chunks_by_method = document_processor.process_document(
|
|
@@ -117,6 +196,10 @@ def process_document(file_path: str, chunking_methods: List[str]) -> Tuple[str,
|
|
| 117 |
except Exception as e:
|
| 118 |
logger.error(f"Error processing document: {str(e)}")
|
| 119 |
return f"β Error: {str(e)}", pd.DataFrame()
|
|
|
|
|
|
|
|
|
|
|
|
|
| 120 |
|
| 121 |
def search_chunks(query: str, top_k: int = None) -> Tuple[str, pd.DataFrame]:
|
| 122 |
"""
|
|
@@ -913,6 +996,12 @@ with gr.Blocks(title="Document Chunking & Search System", theme=gr.themes.Soft()
|
|
| 913 |
gr.Markdown("**Delete All Collections:**")
|
| 914 |
gr.Markdown("β οΈ **Warning**: This will delete ALL collections permanently!")
|
| 915 |
delete_all_btn = gr.Button("ποΈ Delete All Collections", variant="stop")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 916 |
|
| 917 |
delete_output = gr.Textbox(
|
| 918 |
label="Deletion Status",
|
|
@@ -950,6 +1039,11 @@ with gr.Blocks(title="Document Chunking & Search System", theme=gr.themes.Soft()
|
|
| 950 |
delete_all_collections,
|
| 951 |
outputs=delete_output
|
| 952 |
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 953 |
|
| 954 |
if __name__ == "__main__":
|
| 955 |
# Set up environment for deployment
|
|
|
|
| 1 |
import gradio as gr
|
| 2 |
import logging
|
| 3 |
import os
|
| 4 |
+
import hashlib
|
| 5 |
from typing import List, Tuple, Optional, Dict, Any
|
| 6 |
import pandas as pd
|
| 7 |
from qdrant_client import QdrantClient
|
|
|
|
| 25 |
document_processor = None
|
| 26 |
embeddings_pipeline = None
|
| 27 |
|
| 28 |
+
# Global variable to track current document identity
|
| 29 |
+
current_document_hash = None
|
| 30 |
+
|
| 31 |
# OpenAI client (lazy init)
|
| 32 |
_openai_client: Optional[OpenAI] = None
|
| 33 |
|
|
|
|
| 38 |
_openai_client = OpenAI(api_key=api_key)
|
| 39 |
return _openai_client
|
| 40 |
|
| 41 |
+
def get_file_hash(file_path: str) -> str:
|
| 42 |
+
"""
|
| 43 |
+
Calculate MD5 hash of a file to track document identity.
|
| 44 |
+
|
| 45 |
+
Args:
|
| 46 |
+
file_path: Path to the file
|
| 47 |
+
|
| 48 |
+
Returns:
|
| 49 |
+
MD5 hash string of the file content
|
| 50 |
+
"""
|
| 51 |
+
hash_md5 = hashlib.md5()
|
| 52 |
+
try:
|
| 53 |
+
with open(file_path, "rb") as f:
|
| 54 |
+
for chunk in iter(lambda: f.read(4096), b""):
|
| 55 |
+
hash_md5.update(chunk)
|
| 56 |
+
return hash_md5.hexdigest()
|
| 57 |
+
except Exception as e:
|
| 58 |
+
logger.error(f"Error calculating file hash for {file_path}: {str(e)}")
|
| 59 |
+
return str(hash(file_path)) # Fallback to simple hash of path
|
| 60 |
+
|
| 61 |
+
def clear_all_collections_for_new_document() -> bool:
|
| 62 |
+
"""
|
| 63 |
+
Clear all collections in Qdrant when a new document is detected.
|
| 64 |
+
|
| 65 |
+
Returns:
|
| 66 |
+
bool: True if clearing was successful or not needed, False on error
|
| 67 |
+
"""
|
| 68 |
+
try:
|
| 69 |
+
logger.info("ποΈ Nuevo documento detectado: limpiando todas las colecciones existentes...")
|
| 70 |
+
result = delete_all_collections()
|
| 71 |
+
if "β
" in result:
|
| 72 |
+
logger.info("β
Todas las colecciones fueron eliminadas exitosamente")
|
| 73 |
+
return True
|
| 74 |
+
else:
|
| 75 |
+
logger.warning(f"β οΈ Limpieza de colecciones completada con advertencias: {result}")
|
| 76 |
+
return True
|
| 77 |
+
except Exception as e:
|
| 78 |
+
logger.error(f"β Error al limpiar colecciones para nuevo documento: {str(e)}")
|
| 79 |
+
return False
|
| 80 |
+
|
| 81 |
+
def reset_document_tracking() -> str:
|
| 82 |
+
"""
|
| 83 |
+
Reset the document tracking hash to force collection clearing on next upload.
|
| 84 |
+
|
| 85 |
+
Returns:
|
| 86 |
+
Status message
|
| 87 |
+
"""
|
| 88 |
+
global current_document_hash
|
| 89 |
+
current_document_hash = None
|
| 90 |
+
logger.info("π Document tracking reset - next upload will clear all collections")
|
| 91 |
+
return "β
Document tracking reset. Next document upload will clear all existing collections."
|
| 92 |
+
|
| 93 |
def initialize_components():
|
| 94 |
"""Initialize the pipeline components with error handling for cloud deployment."""
|
| 95 |
global document_processor, embeddings_pipeline
|
|
|
|
| 113 |
def process_document(file_path: str, chunking_methods: List[str]) -> Tuple[str, pd.DataFrame]:
|
| 114 |
"""
|
| 115 |
Process a document with selected chunking methods.
|
| 116 |
+
Automatically clears all collections if a different document is detected.
|
| 117 |
|
| 118 |
Args:
|
| 119 |
file_path: Path to the uploaded file
|
|
|
|
| 122 |
Returns:
|
| 123 |
Tuple of (status_message, chunks_dataframe)
|
| 124 |
"""
|
| 125 |
+
global current_document_hash
|
| 126 |
+
|
| 127 |
try:
|
| 128 |
if document_processor is None:
|
| 129 |
return "β Error: Document processor not initialized. Check system configuration.", pd.DataFrame()
|
|
|
|
| 137 |
if not chunking_methods:
|
| 138 |
return "β Error: Please select at least one chunking method", pd.DataFrame()
|
| 139 |
|
| 140 |
+
# Calculate hash of the current document
|
| 141 |
+
new_document_hash = get_file_hash(file_path)
|
| 142 |
+
logger.info(f"Document hash: {new_document_hash}")
|
| 143 |
+
|
| 144 |
+
# Check if this is a different document than the previous one
|
| 145 |
+
if current_document_hash is not None and current_document_hash != new_document_hash:
|
| 146 |
+
logger.info(f"π Documento diferente detectado (hash anterior: {current_document_hash[:8]}..., nuevo: {new_document_hash[:8]}...)")
|
| 147 |
+
|
| 148 |
+
# Clear all collections for the new document
|
| 149 |
+
clear_success = clear_all_collections_for_new_document()
|
| 150 |
+
if not clear_success:
|
| 151 |
+
logger.warning("β οΈ FallΓ³ la limpieza de colecciones, pero continuando con el procesamiento...")
|
| 152 |
+
elif current_document_hash is None:
|
| 153 |
+
logger.info("π Primer documento cargado en esta sesiΓ³n")
|
| 154 |
+
else:
|
| 155 |
+
logger.info("π Mismo documento que el anterior, manteniendo embeddings existentes")
|
| 156 |
+
|
| 157 |
+
# Update the current document hash
|
| 158 |
+
current_document_hash = new_document_hash
|
| 159 |
+
|
| 160 |
# Process the document
|
| 161 |
logger.info(f"Processing document: {file_path}")
|
| 162 |
chunks_by_method = document_processor.process_document(
|
|
|
|
| 196 |
except Exception as e:
|
| 197 |
logger.error(f"Error processing document: {str(e)}")
|
| 198 |
return f"β Error: {str(e)}", pd.DataFrame()
|
| 199 |
+
|
| 200 |
+
except Exception as e:
|
| 201 |
+
logger.error(f"Error processing document: {str(e)}")
|
| 202 |
+
return f"β Error: {str(e)}", pd.DataFrame()
|
| 203 |
|
| 204 |
def search_chunks(query: str, top_k: int = None) -> Tuple[str, pd.DataFrame]:
|
| 205 |
"""
|
|
|
|
| 996 |
gr.Markdown("**Delete All Collections:**")
|
| 997 |
gr.Markdown("β οΈ **Warning**: This will delete ALL collections permanently!")
|
| 998 |
delete_all_btn = gr.Button("ποΈ Delete All Collections", variant="stop")
|
| 999 |
+
|
| 1000 |
+
with gr.Row():
|
| 1001 |
+
with gr.Column():
|
| 1002 |
+
gr.Markdown("**Reset Document Tracking:**")
|
| 1003 |
+
gr.Markdown("π Force collection clearing on next document upload (useful when reprocessing the same file)")
|
| 1004 |
+
reset_tracking_btn = gr.Button("π Reset Document Tracking", variant="secondary")
|
| 1005 |
|
| 1006 |
delete_output = gr.Textbox(
|
| 1007 |
label="Deletion Status",
|
|
|
|
| 1039 |
delete_all_collections,
|
| 1040 |
outputs=delete_output
|
| 1041 |
)
|
| 1042 |
+
|
| 1043 |
+
reset_tracking_btn.click(
|
| 1044 |
+
reset_document_tracking,
|
| 1045 |
+
outputs=delete_output
|
| 1046 |
+
)
|
| 1047 |
|
| 1048 |
if __name__ == "__main__":
|
| 1049 |
# Set up environment for deployment
|