from flask import Flask, request, jsonify, render_template, send_from_directory import os import time import json import hashlib import threading import queue from concurrent.futures import ThreadPoolExecutor, as_completed from werkzeug.utils import secure_filename from pathlib import Path import traceback import faiss from main import RAGSystem from config import ( PDF_DIR, PROCESSED_DIR, WEB_HOST, WEB_PORT, NOTES_DIR, DOCUMENTS_DIR, DIGESTS_DIR, ENRICHMENT_DEFAULT_THREADS, ENRICHMENT_MAX_THREADS, ENRICHMENT_STATUS_PATH, ENRICHMENT_LOG_PATH ) from logging_config import get_logger, log_performance, log_error_with_context from exceptions import FileUploadError, ValidationError from mistral_integration import RAGAgent app = Flask(__name__) # Configuration app.config['UPLOAD_FOLDER'] = str(PDF_DIR) app.config['MAX_CONTENT_LENGTH'] = 50 * 1024 * 1024 # 50MB max file size MAX_CONCURRENT_UPLOADS = int(os.environ.get("MAX_CONCURRENT_UPLOADS", 20)) # Set up logging app_logger = get_logger(__name__) # Global RAG system instance rag_system = RAGSystem() thread_local_agent = threading.local() # Processing status tracker (in-memory, file-based for persistence) PROCESSING_STATUS_FILE = Path(PROCESSED_DIR) / "metadata" / "processing_status.json" PROCESSING_STATUS_RETENTION_SECONDS = 3600 # Keep completed entries for 1 hour PROCESSING_STATUS_STALE_THRESHOLD = 300 # Consider stalled if no update for 5 minutes ENRICHMENT_STATUS_FILE = ENRICHMENT_STATUS_PATH processing_status = {} # {filename: {status, progress, message, start_time}} # Processing queue for serialized file processing (prevents concurrent processing failures) processing_queue = queue.Queue() processing_worker_running = False processing_worker_lock = threading.Lock() def start_processing_worker(): """Start the processing worker thread if not already running""" global processing_worker_running with processing_worker_lock: if processing_worker_running: return processing_worker_running = True worker_thread = threading.Thread(target=processing_worker, daemon=True) worker_thread.start() app_logger.info("Processing worker thread started") def processing_worker(): """Worker thread that processes files from the queue one at a time""" global processing_worker_running app_logger.info("Processing worker started, waiting for files...") while True: try: # Wait for a file to process (with timeout to allow graceful shutdown) try: process_func = processing_queue.get(timeout=60) except queue.Empty: # Check if queue is really empty and no more files expected if processing_queue.empty(): app_logger.info("Processing worker idle for 60s, continuing to wait...") continue # Process the file try: app_logger.info(f"Processing worker picked up a file, queue size: {processing_queue.qsize()}") process_func() except Exception as e: app_logger.error(f"Processing worker error: {e}") finally: processing_queue.task_done() except Exception as e: app_logger.error(f"Processing worker unexpected error: {e}") time.sleep(1) # Prevent tight loop on errors enrichment_status = {} # {status: 'idle'|'running', progress: 0-100, ...} status_lock = threading.Lock() enrichment_log_lock = threading.Lock() vector_store_lock = threading.Lock() # Thread safety for vector store operations def load_processing_status(): """Load processing status from file""" global processing_status if PROCESSING_STATUS_FILE.exists(): try: with open(PROCESSING_STATUS_FILE, 'r') as f: processing_status = json.load(f) except (json.JSONDecodeError, IOError, OSError) as e: app_logger.warning(f"Failed to load processing status: {e}") processing_status = {} else: processing_status = {} def save_processing_status(): """Save processing status to file""" try: PROCESSING_STATUS_FILE.parent.mkdir(parents=True, exist_ok=True) with open(PROCESSING_STATUS_FILE, 'w') as f: json.dump(processing_status, f, indent=2) except Exception as e: app_logger.error(f"Failed to save processing status: {e}") def update_processing_status(filename, status, progress=0, message="", extra=None): """Update processing status for a file with persistent metadata.""" now = time.time() safe_filename = secure_filename(filename) with status_lock: entry = processing_status.get(safe_filename, {}).copy() extra = extra or {} # Preserve initial metadata (start time, size, etc.) if 'started_at' not in entry: entry['started_at'] = extra.get('started_at', now) if 'file_size' not in entry and 'file_size' in extra: entry['file_size'] = extra['file_size'] if 'client_ip' not in entry and 'client_ip' in extra: entry['client_ip'] = extra['client_ip'] if 'original_name' not in entry and 'original_name' in extra: entry['original_name'] = extra['original_name'] # Always store current filename reference entry['filename'] = safe_filename entry.update({ 'status': status, # 'processing', 'completed', 'failed' 'progress': progress, # 0-100 'message': message, 'updated': now }) # Merge any additional metadata (e.g., stage, chunk counts) for key, value in extra.items(): if key not in ('started_at', 'file_size', 'client_ip', 'original_name'): entry[key] = value entry['elapsed_seconds'] = round(now - entry.get('started_at', now), 2) processing_status[safe_filename] = entry save_processing_status() def cleanup_processing_status(retention_seconds=PROCESSING_STATUS_RETENTION_SECONDS): """Remove stale completed/failed processing entries to keep the status file small.""" now = time.time() removed = False with status_lock: for filename in list(processing_status.keys()): entry = processing_status.get(filename, {}) status = entry.get('status') updated = entry.get('updated', entry.get('started_at', now)) if status in ('completed', 'failed') and (now - updated) > retention_seconds: processing_status.pop(filename, None) removed = True if removed: save_processing_status() def normalize_processing_entry(filename, entry, now=None): """Prepare a processing status entry for JSON responses.""" now = now or time.time() normalized = entry.copy() normalized['filename'] = filename started = normalized.get('started_at') or normalized.get('updated') or now normalized['started_at'] = started last_update = normalized.get('updated', started) normalized['last_update'] = last_update normalized['elapsed_seconds'] = round(now - started, 2) stale_age = now - last_update normalized['stale_seconds'] = round(stale_age, 2) normalized['stale'] = normalized.get('status') == 'processing' and stale_age > PROCESSING_STATUS_STALE_THRESHOLD return normalized # Load existing status on startup load_processing_status() def load_enrichment_status(): """Load enrichment status from file""" global enrichment_status default_status = { 'status': 'idle', 'progress': 0, 'message': '', 'files_processed': 0, 'total_files': 0, 'thread_count': ENRICHMENT_DEFAULT_THREADS, 'active_files': {}, 'last_activity': time.time() } if ENRICHMENT_STATUS_FILE.exists(): try: with open(ENRICHMENT_STATUS_FILE, 'r') as f: enrichment_status = json.load(f) except Exception: enrichment_status = default_status.copy() else: enrichment_status = default_status.copy() enrichment_status.setdefault('active_files', {}) enrichment_status.setdefault('thread_count', ENRICHMENT_DEFAULT_THREADS) enrichment_status.setdefault('last_activity', time.time()) def save_enrichment_status(): """Save enrichment status to file""" try: ENRICHMENT_STATUS_FILE.parent.mkdir(parents=True, exist_ok=True) with open(ENRICHMENT_STATUS_FILE, 'w') as f: json.dump(enrichment_status, f, indent=2) except Exception as e: app_logger.error(f"Failed to save enrichment status: {e}") def log_enrichment_event(event_type, **payload): """Append enrichment events to log file for auditing.""" entry = { 'timestamp': time.time(), 'event': event_type, **payload } try: ENRICHMENT_LOG_PATH.parent.mkdir(parents=True, exist_ok=True) with enrichment_log_lock: with open(ENRICHMENT_LOG_PATH, 'a') as f: f.write(json.dumps(entry) + "\n") except Exception as exc: app_logger.debug(f"Failed to write enrichment log entry: {exc}") def update_enrichment_status(status, progress=0, message="", files_processed=0, total_files=0, thread_count=None): """Update enrichment status""" if thread_count is None: thread_count = enrichment_status.get('thread_count', ENRICHMENT_DEFAULT_THREADS) with status_lock: enrichment_status.update({ 'status': status, 'progress': progress, 'message': message, 'files_processed': files_processed, 'total_files': total_files, 'updated': time.time(), 'last_activity': time.time(), 'thread_count': thread_count }) save_enrichment_status() def update_enrichment_file_activity(file_name, chunk_index=None, chunk_total=None, message=None): with status_lock: active = enrichment_status.setdefault('active_files', {}) entry = active.setdefault(file_name, {}) if chunk_index is not None: entry['chunk'] = int(chunk_index) if chunk_total is not None: entry['total_chunks'] = int(chunk_total) if message: entry['message'] = message entry['updated'] = time.time() enrichment_status['last_activity'] = time.time() # Ensure status is 'running' if there are active files if active and enrichment_status.get('status') != 'running': enrichment_status['status'] = 'running' save_enrichment_status() def clear_enrichment_file_activity(file_name): with status_lock: active = enrichment_status.setdefault('active_files', {}) if file_name in active: active.pop(file_name, None) enrichment_status['last_activity'] = time.time() save_enrichment_status() def clear_all_enrichment_activity(): with status_lock: enrichment_status['active_files'] = {} enrichment_status['last_activity'] = time.time() save_enrichment_status() # Load existing enrichment status on startup load_enrichment_status() def get_thread_rag_agent(): """Get or create a thread-local RAGAgent for enrichment.""" if not hasattr(thread_local_agent, 'rag_agent'): thread_local_agent.rag_agent = RAGAgent() return thread_local_agent.rag_agent def sanitize_thread_count(value): """Clamp requested thread count to allowed range.""" try: count = int(value) except (TypeError, ValueError): return ENRICHMENT_DEFAULT_THREADS return max(1, min(count, ENRICHMENT_MAX_THREADS)) def allowed_file(filename): """Check if file extension is allowed""" if not filename or '.' not in filename: return False ext = filename.rsplit('.', 1)[1].lower() return ext in ['pdf', 'txt'] def get_active_processing_jobs(): """Return the number of files currently being processed.""" with status_lock: return sum(1 for entry in processing_status.values() if entry.get('status') == 'processing') @app.route('/') def index(): return render_template( 'index.html', enrichment_default_threads=ENRICHMENT_DEFAULT_THREADS, enrichment_max_threads=ENRICHMENT_MAX_THREADS ) @app.route('/upload', methods=['POST']) def upload_file(): start_time = time.time() client_ip = request.remote_addr or "unknown" try: app_logger.info(f"File upload request from {client_ip}") if 'file' not in request.files: app_logger.warning(f"Upload request missing file part from {client_ip}") return jsonify({'error': 'No file part'}), 400 file = request.files['file'] if file.filename == '': app_logger.warning(f"Upload request with empty filename from {client_ip}") return jsonify({'error': 'No selected file'}), 400 # Handle both single file and multiple files (Flask handles this automatically) # If multiple files, process first one (browser sends one at a time) if not file or not allowed_file(file.filename): app_logger.warning(f"Invalid file type attempt: {file.filename} from {client_ip}") return jsonify({'error': 'File type not allowed. Only PDF and text files are accepted.'}), 400 # Enforce concurrent upload throttle to prevent overload cleanup_processing_status() active_jobs = get_active_processing_jobs() if active_jobs >= MAX_CONCURRENT_UPLOADS: message = ( f'The processing queue already has {active_jobs} active file(s). ' f'Please wait for current uploads to finish before adding more.' ) app_logger.warning(f"Upload throttled for {client_ip}: {message}") return jsonify({ 'error': message, 'active_uploads': active_jobs, 'max_uploads': MAX_CONCURRENT_UPLOADS }), 429 # Validate file size file.seek(0, os.SEEK_END) file_size = file.tell() file.seek(0) max_size = app.config['MAX_CONTENT_LENGTH'] if file_size > max_size: app_logger.warning(f"File too large: {file_size} bytes from {client_ip}") return jsonify({'error': f'File too large. Maximum size is {max_size // (1024*1024)}MB'}), 413 filename = secure_filename(file.filename) file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename) # Check for duplicate file (by name and content hash) file_exists = os.path.exists(file_path) if file_exists: # Calculate hash of existing file existing_hash = None try: with open(file_path, 'rb') as f: existing_hash = hashlib.md5(f.read()).hexdigest() except (IOError, OSError) as e: app_logger.debug(f"Could not read existing file for hash: {e}") existing_hash = None # Calculate hash of new file file.seek(0) new_file_content = file.read() new_file_hash = hashlib.md5(new_file_content).hexdigest() file.seek(0) # Reset for saving if existing_hash == new_file_hash: app_logger.info(f"Duplicate file detected: {filename} (same content hash)") # Safely get document count (thread-safe read) doc_count = 0 try: with vector_store_lock: if hasattr(rag_system, 'vector_store') and rag_system.vector_store: if hasattr(rag_system.vector_store, 'documents') and rag_system.vector_store.documents: doc_count = len(rag_system.vector_store.documents) except Exception as e: app_logger.debug(f"Could not get document count: {e}") return jsonify({ 'message': f'File "{filename}" already exists with identical content. Skipping upload.', 'duplicate': True, 'total_documents': doc_count }), 200 else: # Different content - replace the file app_logger.info(f"Replacing existing file: {filename} (different content)") # Remove old processed data from pdf_processor import PDFProcessor processor = PDFProcessor() processed_files = processor.get_processed_files() if filename in processed_files: # Remove from processed files metadata metadata_file = Path(PROCESSED_DIR) / "metadata" / "processed_files.json" if metadata_file.exists(): try: with open(metadata_file, 'r') as f: metadata = json.load(f) if filename in metadata: del metadata[filename] with open(metadata_file, 'w') as f: json.dump(metadata, f, indent=2) except: pass # Remove old chunks file chunk_file = Path(PROCESSED_DIR) / "documents" / f"{Path(filename).stem}_chunks.json" if chunk_file.exists(): chunk_file.unlink() app_logger.info(f"Saving file: {filename} ({file_size} bytes) from {client_ip}") file.save(file_path) # Process file asynchronously in background to avoid proxy timeouts def process_file_async(): """Process uploaded file in background thread""" try: update_processing_status( filename, 'processing', 10, 'Starting file processing...', extra={'stage': 'initializing'} ) # Step 1: Extract and chunk app_logger.info(f"[Background] Step 1: Extracting text from {filename}") update_processing_status( filename, 'processing', 20, 'Extracting text from PDF...', extra={'stage': 'extract_text'} ) from pdf_processor import PDFProcessor processor = PDFProcessor() chunks = processor.process_pdf(Path(file_path)) # Handle case where file is already processed (process_pdf returns None) if chunks is None: # File was already processed, reload existing chunks app_logger.info(f"[Background] File {filename} already processed, loading existing chunks") update_processing_status( filename, 'processing', 30, 'Loading existing chunks...', extra={'stage': 'loading_existing_chunks'} ) chunk_file = Path(PROCESSED_DIR) / "documents" / f"{Path(filename).stem}_chunks.json" if chunk_file.exists(): try: with open(chunk_file, 'r') as f: chunks = json.load(f) app_logger.info(f"[Background] Loaded {len(chunks)} existing chunks for {filename}") except Exception as e: app_logger.error(f"[Background] Failed to load existing chunks: {e}") update_processing_status( filename, 'failed', 0, f'Failed to load existing chunks: {str(e)}', extra={'stage': 'load_existing_chunks_failed', 'error': str(e)} ) return else: app_logger.error(f"[Background] File {filename} was marked as processed but no chunks found") update_processing_status( filename, 'failed', 0, 'File marked as processed but no chunks found', extra={'stage': 'missing_chunks'} ) return if not chunks or len(chunks) == 0: # Try OCR fallback before giving up app_logger.info(f"[Background] Normal extraction failed for {filename}, trying OCR fallback...") update_processing_status( filename, 'processing', 25, 'Text extraction failed, trying OCR...', extra={'stage': 'ocr_fallback'} ) try: import requests # Try DeepSeek-OCR first (GPU accelerated, port 5003) deepseek_url = 'http://localhost:5003/ocr' tesseract_url = 'http://localhost:5002/ocr' ocr_response = None ocr_service_used = None with open(file_path, 'rb') as pdf_file: files = {'file': (filename, pdf_file, 'application/pdf')} try: app_logger.info(f"[Background] Trying DeepSeek-OCR for {filename}...") ocr_response = requests.post(deepseek_url, files=files, timeout=1200) if ocr_response.status_code == 200 and ocr_response.json().get('success'): ocr_service_used = 'DeepSeek' else: app_logger.warning(f"[Background] DeepSeek-OCR failed, trying Tesseract...") ocr_response = None except Exception as ds_error: app_logger.warning(f"[Background] DeepSeek-OCR error: {ds_error}, trying Tesseract...") # Fallback to Tesseract if DeepSeek failed if ocr_response is None: with open(file_path, 'rb') as pdf_file: files = {'file': (filename, pdf_file, 'application/pdf')} ocr_response = requests.post(tesseract_url, files=files, timeout=1200) ocr_service_used = 'Tesseract' if ocr_response.status_code == 200: ocr_data = ocr_response.json() if ocr_data.get('success') and ocr_data.get('text'): ocr_text = ocr_data['text'] app_logger.info(f"[Background] OCR extracted {len(ocr_text)} chars from {filename}") # Create chunks from OCR text chunk_size = 1000 chunks = [] for i in range(0, len(ocr_text), chunk_size): chunk_text = ocr_text[i:i+chunk_size] if chunk_text.strip(): chunks.append({ 'content': chunk_text, 'text': chunk_text, # Keep both for compatibility 'source': filename, 'chunk_id': len(chunks), 'metadata': {'ocr': True, 'ocr_service': ocr_service_used} }) if chunks: app_logger.info(f"[Background] OCR created {len(chunks)} chunks from {filename}") # Save chunks to JSON file for enrichment chunk_file = Path(PROCESSED_DIR) / "documents" / f"{Path(filename).stem}_chunks.json" chunk_file.parent.mkdir(parents=True, exist_ok=True) try: with open(chunk_file, 'w') as f: import json json.dump(chunks, f, indent=2) app_logger.info(f"[Background] Saved {len(chunks)} OCR chunks to {chunk_file}") except Exception as save_error: app_logger.warning(f"[Background] Failed to save chunks to file: {save_error}") update_processing_status( filename, 'processing', 40, f'OCR ({ocr_service_used}) extracted {len(chunks)} chunks. Processing...', extra={'stage': 'ocr_complete', 'chunk_count': len(chunks), 'ocr_service': ocr_service_used} ) else: app_logger.warning(f"[Background] OCR failed for {filename}: {ocr_response.text}") except Exception as ocr_error: app_logger.warning(f"[Background] OCR fallback error for {filename}: {ocr_error}") # If still no chunks after OCR, mark as failed if not chunks or len(chunks) == 0: app_logger.error(f"[Background] Failed to extract text from {filename} even with OCR. File may be corrupted.") update_processing_status( filename, 'failed', 0, 'Failed to extract text. File may be corrupted or empty.', extra={'stage': 'extract_text_failed'} ) return app_logger.info(f"[Background] Step 2: Extracted {len(chunks)} chunks from {filename}") update_processing_status( filename, 'processing', 50, f'Extracted {len(chunks)} chunks. Adding to vector store...', extra={'stage': 'chunks_ready', 'chunk_count': len(chunks)} ) # Step 2: Add raw chunks to vector store (enrichment happens manually via UI button) app_logger.info(f"[Background] Step 3: Adding {len(chunks)} raw chunks to vector store") update_processing_status( filename, 'processing', 60, 'Adding chunks to vector store...', extra={'stage': 'vector_store_prepare'} ) # Ensure vector store is initialized if not hasattr(rag_system, 'vector_store') or not rag_system.vector_store: app_logger.error("[Background] Vector store not initialized") update_processing_status( filename, 'failed', 0, 'Vector store not initialized', extra={'stage': 'vector_store_missing'} ) return rag_system.initialize_model() app_logger.info(f"[Background] Step 4: Adding {len(chunks)} raw chunks to vector store") update_processing_status( filename, 'processing', 85, f'Adding {len(chunks)} chunks to vector store...', extra={'stage': 'vector_store_write', 'chunk_count': len(chunks)} ) try: # Thread-safe vector store access with vector_store_lock: rag_system.vector_store.add_documents(chunks) rag_system.vector_store.save_index() except Exception as e: app_logger.error(f"[Background] Error adding chunks to vector store: {e}") update_processing_status( filename, 'failed', 0, f'Error adding to vector store: {str(e)}', extra={'stage': 'vector_store_error', 'error': str(e)} ) return # Safely get document count doc_count = 0 try: if hasattr(rag_system.vector_store, 'documents') and rag_system.vector_store.documents: doc_count = len(rag_system.vector_store.documents) except Exception as e: app_logger.debug(f"[Background] Could not get document count: {e}") duration = time.time() - start_time log_performance(app_logger, "file_upload_processing", duration, filename=filename, file_size=file_size, total_documents=doc_count, client_ip=client_ip) app_logger.info(f"[Background] Successfully processed uploaded file: {filename} ({len(chunks)} chunks, {round(duration, 2)}s)") update_processing_status( filename, 'completed', 100, f'Successfully processed! Created {len(chunks)} chunks in {round(duration, 2)}s', extra={ 'stage': 'completed', 'chunk_count': len(chunks), 'duration_seconds': round(duration, 2) } ) except Exception as e: error_msg = str(e) log_error_with_context( app_logger, e, {"operation": "file_processing", "filename": filename, "client_ip": client_ip}, f"Error processing uploaded file: {filename}" ) update_processing_status( filename, 'failed', 0, f'Error: {error_msg}', extra={'stage': 'failed', 'error': error_msg} ) # Initialize processing status update_processing_status( filename, 'processing', 0, 'File uploaded, starting processing...', extra={ 'stage': 'queued', 'started_at': start_time, 'file_size': file_size, 'client_ip': client_ip, 'original_name': file.filename } ) # Add to processing queue (single worker thread processes one at a time) processing_queue.put(process_file_async) start_processing_worker() # Ensure worker is running app_logger.info(f"File {filename} added to processing queue, queue size: {processing_queue.qsize()}") # Return immediately to avoid proxy timeout app_logger.info(f"File {filename} saved, processing started in background") return jsonify({ 'message': f'File "{filename}" uploaded successfully. Processing in background...', 'status': 'processing', 'filename': filename, 'file_size': file_size, 'progress': 0 }), 202 # 202 Accepted - request accepted for processing except Exception as e: duration = time.time() - start_time log_error_with_context( app_logger, e, {"operation": "file_upload", "client_ip": client_ip, "duration": duration}, "Unexpected error during file upload" ) return jsonify({'error': f'Upload failed: {str(e)}'}), 500 @app.route('/ask', methods=['POST']) def ask_question(): start_time = time.time() try: data = request.get_json() if not data or 'question' not in data: return jsonify({'error': 'Question is required'}), 400 question = data['question'].strip() if not question: return jsonify({'error': 'Question cannot be empty'}), 400 # Initialize RAG system if needed if not hasattr(rag_system, '_initialized') or not rag_system._initialized: try: print("Initializing RAG system for question...") doc_count = rag_system.setup_system() rag_system._initialized = True print(f"System ready with {doc_count} documents") except Exception as init_e: print(f"Error initializing RAG system: {init_e}") return jsonify({'error': f'System initialization failed: {str(init_e)}'}), 500 # Get answer from RAG system result = rag_system.search_and_answer(question) if isinstance(result, str): return jsonify({'answer': result, 'sources': []}), 200 else: return jsonify(result), 200 except Exception as e: duration = time.time() - start_time log_error_with_context( app_logger, e, {"operation": "question_processing", "client_ip": client_ip, "duration": duration}, "Unexpected error in ask_question endpoint" ) return jsonify({'error': f'Error processing question: {str(e)}'}), 500 @app.route('/status', methods=['GET']) def get_status(): try: # Don't initialize RAG system here - it's too slow and blocks the request # Just return file list and status info doc_count = 0 model_loaded = False # Try to get doc count from vector store if it's already loaded (thread-safe read) try: if hasattr(rag_system, '_initialized') and rag_system._initialized: with vector_store_lock: if hasattr(rag_system, 'vector_store') and rag_system.vector_store: if hasattr(rag_system.vector_store, 'documents') and rag_system.vector_store.documents: doc_count = len(rag_system.vector_store.documents) model_loaded = rag_system.rag_agent is not None else: # Try to load vector store metadata without full initialization try: embeddings_dir = Path(PROCESSED_DIR) / "embeddings" metadata_file = embeddings_dir / "documents_metadata.json" if metadata_file.exists(): with open(metadata_file, 'r') as f: metadata = json.load(f) doc_count = len(metadata) if isinstance(metadata, list) else 0 except: pass except Exception as e: app_logger.debug(f"Could not get doc count: {e}") pass pdf_files = list(PDF_DIR.glob("*.pdf")) txt_files = list(PDF_DIR.glob("*.txt")) all_files = sorted(pdf_files + txt_files, key=lambda x: x.stat().st_mtime, reverse=True) # Get file details with sizes and processing status file_list = [] processed_files = {} metadata_file = Path(PROCESSED_DIR) / "metadata" / "processed_files.json" if metadata_file.exists(): try: with open(metadata_file, 'r') as f: processed_files = json.load(f) except: pass for file_path in all_files: file_info = { 'name': file_path.name, 'size': file_path.stat().st_size, 'size_mb': round(file_path.stat().st_size / (1024 * 1024), 2), 'uploaded': file_path.stat().st_mtime, 'processed': file_path.name in processed_files, 'chunks': processed_files.get(file_path.name, {}).get('chunk_count', 0) if file_path.name in processed_files else 0 } file_list.append(file_info) return jsonify({ 'total_documents': doc_count, 'pdf_files': len(pdf_files), 'txt_files': len(txt_files), 'total_files': len(all_files), 'model_loaded': model_loaded, 'file_list': [f.name for f in all_files], 'files': file_list # Detailed file information }), 200 except Exception as e: app_logger.error(f"Status check failed: {e}") import traceback traceback.print_exc() return jsonify({'error': f'Status check failed: {str(e)}'}), 500 @app.route('/static/') def serve_static(filename): return send_from_directory('static', filename) @app.route('/pdf/') def serve_pdf(filename): """Serve PDF files for viewing""" from werkzeug.utils import secure_filename safe_filename = secure_filename(filename) pdf_path = PDF_DIR / safe_filename if pdf_path.exists() and pdf_path.suffix.lower() == '.pdf': return send_from_directory(str(PDF_DIR), safe_filename) return jsonify({'error': 'PDF not found'}), 404 @app.route('/chunks/') def get_chunks(filename): """Get chunks for a specific PDF file""" from werkzeug.utils import secure_filename import json from pathlib import Path safe_filename = secure_filename(filename) chunk_file = Path(PROCESSED_DIR) / "documents" / f"{Path(safe_filename).stem}_chunks.json" if chunk_file.exists(): try: with open(chunk_file, 'r') as f: chunks = json.load(f) return jsonify({ 'success': True, 'filename': safe_filename, 'total_chunks': len(chunks), 'chunks': chunks }), 200 except Exception as e: return jsonify({'error': f'Error reading chunks: {str(e)}'}), 500 else: return jsonify({'error': 'Chunks not found for this file'}), 404 @app.route('/processing-status', methods=['GET']) def list_processing_statuses(): """Return snapshot of all processing jobs (active + recent).""" load_processing_status() cleanup_processing_status() now = time.time() snapshot = { 'timestamp': now, 'active': {}, 'recent': {} } with status_lock: for filename, entry in processing_status.items(): normalized = normalize_processing_entry(filename, entry, now=now) if normalized.get('status') == 'processing': snapshot['active'][filename] = normalized else: if now - normalized.get('last_update', now) <= PROCESSING_STATUS_RETENTION_SECONDS: snapshot['recent'][filename] = normalized snapshot['active_count'] = len(snapshot['active']) snapshot['recent_count'] = len(snapshot['recent']) return jsonify(snapshot), 200 @app.route('/processing-status/') def get_processing_status(filename): """Get processing status for a file""" safe_filename = secure_filename(filename) # Reload status from file load_processing_status() cleanup_processing_status() entry = processing_status.get(safe_filename) if entry: return jsonify(normalize_processing_entry(safe_filename, entry)), 200 else: # Check if file is already processed metadata_file = Path(PROCESSED_DIR) / "metadata" / "processed_files.json" if metadata_file.exists(): try: with open(metadata_file, 'r') as f: processed_files = json.load(f) if safe_filename in processed_files: return jsonify({ 'status': 'completed', 'progress': 100, 'message': 'File already processed', 'chunks': processed_files[safe_filename].get('chunk_count', 0) }), 200 except: pass return jsonify({ 'status': 'unknown', 'progress': 0, 'message': 'No processing status found' }), 404 @app.route('/enrich', methods=['POST']) def enrich_documents(): """Trigger enrichment for new files only""" try: data = request.get_json(silent=True) or {} requested_threads = data.get('threads') thread_count = sanitize_thread_count(requested_threads) # Check if enrichment is already running load_enrichment_status() if enrichment_status.get('status') == 'running': return jsonify({ 'error': 'Enrichment is already running. Please wait for it to complete.', 'status': 'running', 'progress': enrichment_status.get('progress', 0), 'message': enrichment_status.get('message', ''), 'thread_count': enrichment_status.get('thread_count', ENRICHMENT_DEFAULT_THREADS) }), 409 # Conflict # Start enrichment in background thread enrichment_thread = threading.Thread(target=enrich_documents_async, args=(thread_count,), daemon=True) enrichment_thread.start() return jsonify({ 'message': 'Enrichment started. Processing new files only...', 'status': 'started', 'thread_count': thread_count }), 202 # Accepted except Exception as e: app_logger.error(f"Error starting enrichment: {e}") return jsonify({'error': f'Failed to start enrichment: {str(e)}'}), 500 def enrich_documents_async(thread_count=None): """Enrich documents in background - only process files without cards (threaded).""" if thread_count is None: thread_count = ENRICHMENT_DEFAULT_THREADS thread_count = sanitize_thread_count(thread_count) try: log_enrichment_event('run_started', threads=thread_count) update_enrichment_status('running', 0, 'Checking for files that need enrichment...', 0, 0, thread_count) # Get all PDF files pdf_files = list(PDF_DIR.glob("*.pdf")) total_files = len(pdf_files) if total_files == 0: update_enrichment_status('completed', 100, 'No PDF files found', 0, 0, thread_count) log_enrichment_event('run_completed', message='No files to process') return # Check which files already have cards existing_cards = {} if NOTES_DIR.exists(): for card_file in NOTES_DIR.glob("*_cards.json"): try: source_stem = card_file.stem.replace('_cards', '') existing_cards[source_stem] = True except: pass # Find files that need enrichment files_to_enrich = [] processed_files = {} metadata_file = Path(PROCESSED_DIR) / "metadata" / "processed_files.json" if metadata_file.exists(): try: with open(metadata_file, 'r') as f: processed_files = json.load(f) except: pass for pdf_file in pdf_files: file_stem = pdf_file.stem if pdf_file.name in processed_files and file_stem not in existing_cards: files_to_enrich.append(pdf_file) if not files_to_enrich: update_enrichment_status('completed', 100, 'All files already have enriched cards', total_files, total_files, thread_count) log_enrichment_event('run_completed', message='All files already enriched') return total_files_to_process = len(files_to_enrich) log_enrichment_event('files_queued', files=[f.name for f in files_to_enrich]) update_enrichment_status( 'running', 5, f'Found {total_files_to_process} file(s) to enrich. Starting ({thread_count} thread(s))...', 0, total_files_to_process, thread_count ) # Ensure vector store/model ready for final update step if not hasattr(rag_system, '_initialized') or not rag_system._initialized: try: if not hasattr(rag_system, 'vector_store') or not rag_system.vector_store: app_logger.error("Vector store not available for enrichment") update_enrichment_status('failed', 0, 'Vector store not initialized', 0, 0, thread_count) return # Thread-safe initialization with vector_store_lock: rag_system.vector_store.load_existing_index() rag_system.initialize_model() rag_system._initialized = True except Exception as e: app_logger.error(f"Failed to initialize vector store for enrichment: {e}") update_enrichment_status('failed', 0, f'Vector store initialization failed: {str(e)}', 0, 0, thread_count) return results = [] files_processed = 0 # Process files sequentially (chunks are threaded within each file) # This prevents resource contention and allows better chunk-level parallelization for pdf_file in files_to_enrich: try: result = process_file_for_enrichment(pdf_file, thread_count=thread_count) except Exception as e: app_logger.error(f"Error enriching {pdf_file.name}: {e}") result = { 'file': pdf_file.name, 'success': False, 'message': f'Error enriching {pdf_file.name}: {str(e)}', 'enriched_chunks': None, 'chunk_count': 0 } results.append(result) if not result.get('success'): clear_enrichment_file_activity(result['file']) files_processed += 1 # 5% initial + 80% for enrichment stage progress = int(5 + (files_processed / total_files_to_process) * 80) update_enrichment_status( 'running', progress, result.get('message') or f'Finished {pdf_file.name}', files_processed, total_files_to_process, thread_count ) # Sequential vector store updates (remaining 15% progress) enriched_results = [r for r in results if r.get('success') and r.get('enriched_chunks')] total_chunks_processed = sum(len(r.get('enriched_chunks') or []) for r in enriched_results) enriched_count = len(enriched_results) if enriched_results: update_enrichment_status( 'running', 90, 'Updating vector store with enriched chunks...', files_processed, total_files_to_process, thread_count ) for idx, result in enumerate(enriched_results, start=1): try: update_enrichment_file_activity(result['file'], message='Indexing enriched chunks') apply_enriched_chunks_to_vector_store(result['file'], result['enriched_chunks']) log_enrichment_event('vector_update', file=result['file'], chunks=len(result['enriched_chunks'] or [])) progress = int(90 + (idx / len(enriched_results)) * 10) update_enrichment_status( 'running', progress, f'Indexed enriched chunks for {result["file"]}', files_processed, total_files_to_process, thread_count ) except Exception as e: app_logger.error(f"Vector store update failed for {result['file']}: {e}") log_enrichment_event('vector_update_failed', file=result['file'], error=str(e)) finally: clear_enrichment_file_activity(result['file']) # Release memory result['enriched_chunks'] = None else: update_enrichment_status( 'running', 95, 'No enriched chunks generated. Vector store unchanged.', files_processed, total_files_to_process, thread_count ) # Final status if enriched_count > 0: message = f'Enrichment complete! Processed {enriched_count} file(s) with {total_chunks_processed} total chunks.' else: skipped_count = total_files_to_process - enriched_count if skipped_count == total_files_to_process: message = 'Enrichment completed but no files generated new cards. Check logs for details.' else: message = f'Enrichment completed with issues. {enriched_count} succeeded, {skipped_count} skipped.' update_enrichment_status('completed', 100, message, total_files_to_process, total_files_to_process, thread_count) clear_all_enrichment_activity() log_enrichment_event('run_completed', enriched=enriched_count, total=total_files_to_process, chunks=total_chunks_processed, message=message) app_logger.info(f"Enrichment completed: {enriched_count}/{total_files_to_process} files enriched, {total_chunks_processed} total chunks") except Exception as e: app_logger.error(f"Error in enrichment process: {e}") log_enrichment_event('run_failed', error=str(e)) clear_all_enrichment_activity() update_enrichment_status('failed', 0, f'Enrichment failed: {str(e)}', 0, 0, thread_count) def process_file_for_enrichment(pdf_file, thread_count=1): """Worker function to build enriched chunks for a single PDF. Args: pdf_file: Path object for the PDF file thread_count: Number of threads to use for chunk-level parallel processing """ result = { 'file': pdf_file.name, 'success': False, 'message': '', 'enriched_chunks': None, 'chunk_count': 0 } try: update_enrichment_file_activity(pdf_file.name, message='Loading chunks') chunk_file = DOCUMENTS_DIR / f"{pdf_file.stem}_chunks.json" if not chunk_file.exists(): result['message'] = f'No chunks found for {pdf_file.name}, skipping.' log_enrichment_event('file_missing_chunks', file=pdf_file.name) update_enrichment_file_activity(pdf_file.name, message=result['message']) return result with open(chunk_file, 'r') as f: chunks_data = json.load(f) if not chunks_data or not isinstance(chunks_data, list): result['message'] = f'Invalid or empty chunk data for {pdf_file.name}.' log_enrichment_event('file_invalid_chunks', file=pdf_file.name) update_enrichment_file_activity(pdf_file.name, message=result['message']) return result chunk_count = len(chunks_data) update_enrichment_file_activity(pdf_file.name, chunk_index=0, chunk_total=chunk_count, message=f'Generating knowledge cards (starting, {thread_count} thread(s))') log_enrichment_event('file_started', file=pdf_file.name, chunk_count=chunk_count, thread_count=thread_count) agent = get_thread_rag_agent() stop_event = threading.Event() def heartbeat(): while not stop_event.wait(30): # Heartbeat only updates timestamp if no progress update has happened pass heartbeat_thread = threading.Thread(target=heartbeat, daemon=True) heartbeat_thread.start() def progress_callback(current, total): update_enrichment_file_activity( pdf_file.name, chunk_index=current, chunk_total=total, message=f'Generating knowledge cards ({current}/{total})' ) try: enriched_chunks = agent.build_enriched_chunks(chunks_data, progress_callback=progress_callback, thread_count=thread_count) finally: stop_event.set() if enriched_chunks: result.update({ 'success': True, 'enriched_chunks': enriched_chunks, 'chunk_count': len(enriched_chunks), 'message': f'Finished {pdf_file.name} ({len(enriched_chunks)} enriched chunks)' }) update_enrichment_file_activity( pdf_file.name, chunk_index=len(enriched_chunks), chunk_total=len(enriched_chunks), message='Cards ready for indexing' ) log_enrichment_event('file_completed', file=pdf_file.name, enriched_chunks=len(enriched_chunks)) else: result['message'] = f'No enriched chunks generated for {pdf_file.name}.' update_enrichment_file_activity(pdf_file.name, message=result['message']) log_enrichment_event('file_no_cards', file=pdf_file.name) except Exception as e: result['message'] = f'Error enriching {pdf_file.name}: {str(e)}' result['error'] = str(e) update_enrichment_file_activity(pdf_file.name, message=result['message']) log_enrichment_event('file_error', file=pdf_file.name, error=str(e)) return result def apply_enriched_chunks_to_vector_store(source_name, enriched_chunks): """ Safely replace existing vector store entries for a source with enriched chunks. This method uses replace_source_documents which rebuilds the index to ensure consistency and prevent index corruption (index/document count mismatch). """ if not enriched_chunks: app_logger.warning(f"No enriched chunks provided for {source_name}, skipping vector store update") return try: app_logger.info(f"Updating vector store for {source_name} with {len(enriched_chunks)} enriched chunks") # Thread-safe vector store update # Use the safe replacement method that rebuilds the index # This ensures index and documents stay in sync with vector_store_lock: rag_system.vector_store.replace_source_documents(source_name, enriched_chunks) # Save is included in replace_source_documents, but ensure it's saved rag_system.vector_store.save_index() app_logger.info(f"Successfully updated vector store for {source_name}") except Exception as e: app_logger.error(f"Error updating vector store for {source_name}: {e}") raise @app.route('/enrichment-status', methods=['GET']) def get_enrichment_status(): """Get current enrichment status""" load_enrichment_status() # If status is idle but there are active files, it's actually running if enrichment_status.get('status') == 'idle' and enrichment_status.get('active_files'): active_files = enrichment_status.get('active_files', {}) if active_files: enrichment_status['status'] = 'running' return jsonify(enrichment_status), 200 @app.route('/delete/', methods=['DELETE', 'POST']) def delete_file(filename): """Delete a PDF/TXT file and all associated processed data""" try: safe_filename = secure_filename(filename) file_path = PDF_DIR / safe_filename if not file_path.exists(): return jsonify({'error': 'File not found'}), 404 app_logger.info(f"Deleting file: {safe_filename}") # Remove from vector store with vector_store_lock: try: if hasattr(rag_system, 'vector_store') and rag_system.vector_store: rag_system.vector_store._remove_source_from_index(safe_filename) rag_system.vector_store.save_index() app_logger.info(f"Removed {safe_filename} from vector store") except Exception as e: app_logger.warning(f"Error removing from vector store: {e}") # Delete processed files processed_paths = [ DOCUMENTS_DIR / f"{safe_filename}_chunks.json", NOTES_DIR / f"{safe_filename}_cards.json", DIGESTS_DIR / f"{safe_filename}_digest.json" ] deleted_files = [] for proc_path in processed_paths: if proc_path.exists(): try: proc_path.unlink() deleted_files.append(str(proc_path.name)) app_logger.info(f"Deleted processed file: {proc_path.name}") except Exception as e: app_logger.warning(f"Error deleting {proc_path}: {e}") # Remove from processed files metadata metadata_file = Path(PROCESSED_DIR) / "metadata" / "processed_files.json" if metadata_file.exists(): try: with open(metadata_file, 'r') as f: processed_files = json.load(f) if safe_filename in processed_files: del processed_files[safe_filename] with open(metadata_file, 'w') as f: json.dump(processed_files, f, indent=2) app_logger.info(f"Removed {safe_filename} from processed files metadata") except Exception as e: app_logger.warning(f"Error updating metadata: {e}") # Delete the source file try: file_path.unlink() deleted_files.append(safe_filename) app_logger.info(f"Deleted source file: {safe_filename}") except Exception as e: app_logger.error(f"Error deleting source file: {e}") return jsonify({'error': f'Failed to delete source file: {str(e)}'}), 500 return jsonify({ 'success': True, 'message': f'File {safe_filename} and associated data deleted', 'deleted_files': deleted_files }), 200 except Exception as e: app_logger.error(f"Error deleting file: {e}") return jsonify({'error': f'Failed to delete file: {str(e)}'}), 500 @app.route('/delete-all', methods=['DELETE', 'POST']) def delete_all_files(): """Delete all PDFs/TXTs and rebuild empty vector store""" try: app_logger.info("Deleting all files and rebuilding vector store") # Get all files pdf_files = list(PDF_DIR.glob("*.pdf")) txt_files = list(PDF_DIR.glob("*.txt")) all_files = pdf_files + txt_files deleted_count = 0 deleted_files = [] # Delete each file for file_path in all_files: try: safe_filename = secure_filename(file_path.name) # Remove from vector store with vector_store_lock: try: if hasattr(rag_system, 'vector_store') and rag_system.vector_store: rag_system.vector_store._remove_source_from_index(safe_filename) except Exception as e: app_logger.warning(f"Error removing {safe_filename} from vector store: {e}") # Delete processed files processed_paths = [ DOCUMENTS_DIR / f"{safe_filename}_chunks.json", NOTES_DIR / f"{safe_filename}_cards.json", DIGESTS_DIR / f"{safe_filename}_digest.json" ] for proc_path in processed_paths: if proc_path.exists(): proc_path.unlink() # Delete source file file_path.unlink() deleted_files.append(safe_filename) deleted_count += 1 except Exception as e: app_logger.warning(f"Error deleting {file_path.name}: {e}") # Clear processed files metadata metadata_file = Path(PROCESSED_DIR) / "metadata" / "processed_files.json" if metadata_file.exists(): try: with open(metadata_file, 'w') as f: json.dump({}, f) except Exception as e: app_logger.warning(f"Error clearing metadata: {e}") # Rebuild empty vector store with vector_store_lock: try: if hasattr(rag_system, 'vector_store') and rag_system.vector_store: rag_system.vector_store.documents = [] rag_system.vector_store.searchable_texts_raw = [] rag_system.vector_store.searchable_texts_lower = [] if rag_system.vector_store.index is not None: dimension = rag_system.vector_store.index.d rag_system.vector_store.index = faiss.IndexFlatIP(dimension) rag_system.vector_store.save_index() app_logger.info("Rebuilt empty vector store") except Exception as e: app_logger.warning(f"Error rebuilding vector store: {e}") return jsonify({ 'success': True, 'message': f'Deleted {deleted_count} files and rebuilt empty vector store', 'deleted_files': deleted_files, 'deleted_count': deleted_count }), 200 except Exception as e: app_logger.error(f"Error deleting all files: {e}") return jsonify({'error': f'Failed to delete files: {str(e)}'}), 500 @app.route('/reprocessing-status', methods=['GET']) def get_reprocessing_status(): """Get current reprocessing status""" status_file = Path(PROCESSED_DIR) / "metadata" / "reprocessing_status.json" if status_file.exists(): try: with open(status_file, 'r') as f: status = json.load(f) return jsonify(status), 200 except (json.JSONDecodeError, IOError, OSError) as e: app_logger.debug(f"Failed to load reprocessing status: {e}") return jsonify({ 'status': 'idle', 'progress': 0, 'message': 'No reprocessing in progress', 'current_file': '', 'files_processed': 0, 'total_files': 0, 'chunks_created': 0 }), 200 # Memory System - Entity Extraction Endpoints @app.route('/extract-entities', methods=['POST']) def extract_entities_endpoint(): """ Extract user entities from conversation using simple pattern matching POST body: { "user_message": "I am a project manager...", "assistant_response": "Great! I can help..." (optional) } """ try: data = request.get_json() if not data or 'user_message' not in data: return jsonify({ 'success': False, 'error': 'user_message required' }), 400 import re user_message = data['user_message'] entities = { 'demographics': {}, 'interests': [], 'survey_preferences': {}, 'expertise_level': None } # Simple pattern extraction # Extract profession profession_patterns = [ r'I (?:am|work as) (?:a |an )?([a-z\s]+?)(?:\.|,|$)', r'I\'m (?:a |an )?([a-z\s]+?)(?:\.|,|$)', r'(?:my job|profession|role) is ([a-z\s]+?)(?:\.|,|$)' ] for pattern in profession_patterns: match = re.search(pattern, user_message.lower()) if match: entities['demographics']['profession'] = match.group(1).strip() break # Extract interests interest_keywords = [ 'cannabis', 'business', 'finance', 'marketing', 'compliance', 'wellness', 'health', 'growing', 'retail', 'cultivation' ] found_interests = [] for keyword in interest_keywords: if keyword in user_message.lower(): found_interests.append(keyword) if found_interests: entities['interests'] = found_interests # Extract expertise level expert_phrases = ['expert', 'experienced', 'senior', 'advanced', 'years of'] beginner_phrases = ['new to', 'beginner', 'just started', 'learning'] message_lower = user_message.lower() if any(phrase in message_lower for phrase in expert_phrases): entities['expertise_level'] = 'expert' elif any(phrase in message_lower for phrase in beginner_phrases): entities['expertise_level'] = 'beginner' else: entities['expertise_level'] = 'intermediate' app_logger.info(f"Extracted entities: {entities}") return jsonify({ 'success': True, 'entities': entities, 'method': 'simple_patterns', 'confidence': 0.6 }) except Exception as e: app_logger.error(f"Entity extraction failed: {e}") return jsonify({ 'success': False, 'error': str(e) }), 500 if __name__ == '__main__': print("Starting Flask application...") print("RAG system will be initialized on first request") app.run(host=WEB_HOST, port=WEB_PORT, debug=True)