from flask import Flask, request, jsonify, render_template, send_from_directory import os import time import json import hashlib import threading from concurrent.futures import ThreadPoolExecutor, as_completed from werkzeug.utils import secure_filename from pathlib import Path import traceback from functools import wraps from airagagent.main import RAGSystem from airagagent.config import ( PDF_DIR, PROCESSED_DIR, WEB_HOST, WEB_PORT, NOTES_DIR, DOCUMENTS_DIR, ENRICHMENT_DEFAULT_THREADS, ENRICHMENT_MAX_THREADS, ENRICHMENT_STATUS_PATH, ENRICHMENT_LOG_PATH ) from airagagent.logging_config import get_logger, log_performance, log_error_with_context from airagagent.exceptions import FileUploadError, ValidationError from airagagent.mistral_integration import RAGAgent app = Flask(__name__) # Authentication configuration FLASK_API_KEY = os.environ.get("FLASK_API_KEY") if not FLASK_API_KEY: raise ValueError("FLASK_API_KEY environment variable is not set") def require_api_key(f): """Decorator to require API key authentication for protected endpoints""" @wraps(f) def decorated_function(*args, **kwargs): api_key = request.headers.get('X-API-Key') or request.headers.get('Authorization') if api_key and api_key.startswith('Bearer '): api_key = api_key[7:] # Remove 'Bearer ' prefix if not api_key or api_key != FLASK_API_KEY: return jsonify({'error': 'Unauthorized: Invalid or missing API key'}), 401 return f(*args, **kwargs) return decorated_function # 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", 5)) # 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}} 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']) @require_api_key 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: app_logger.error(f"[Background] Failed to extract text from {filename}. File may be corrupted or empty.") 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 } ) # Start background processing thread processing_thread = threading.Thread(target=process_file_async, daemon=True) processing_thread.start() # 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']) @require_api_key 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 # Extract enhanced search parameters k = data.get('k', 5) domain_filter = data.get('domain_filter') use_keyword_focused = data.get('use_keyword_focused', False) extract_keywords = data.get('extract_keywords', False) expand_query = data.get('expand_query', False) # Use enhanced search if parameters are provided if use_keyword_focused or extract_keywords or expand_query or domain_filter: if use_keyword_focused: # Use keyword-focused search keywords = data.get('keywords', []) if not keywords and extract_keywords: # Extract keywords from question if not provided from keyword_manager import get_keyword_manager km = get_keyword_manager() extracted = km.extract_keywords(question, max_keywords=10) keywords = [kw['keyword'] for kw in extracted] result = rag_system.vector_store.keyword_focused_search(keywords, k=k, domain=domain_filter) # Convert to expected format result = { 'answer': self._format_keyword_results(result, question), 'sources': self._format_sources_from_results(result) } else: # Use enhanced general search result = rag_system.enhanced_search(question, k=k, domain=domain_filter) # Convert to expected format result = { 'answer': self._format_enhanced_results(result, question), 'sources': self._format_sources_from_results(result) } else: # Use standard search result = rag_system.search_and_answer(question, k=k) 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('/add-insight', methods=['POST']) @require_api_key def add_insight(): """Add successful strategy insights to the RAG vector store for global learning""" try: data = request.get_json() if not data or 'content' not in data: return jsonify({'error': 'Content is required'}), 400 content = data['content'] metadata = data.get('metadata', {}) source = data.get('source', 'api') # Create a document chunk from the strategy insight insight_chunk = { 'id': f"strategy_insight_{int(time.time())}_{hash(content) % 10000}", 'content': content, 'metadata': { **metadata, 'source': source, 'type': 'strategy_insight', 'timestamp': time.time() }, 'score': 1.0 # High confidence for manually added insights } # Add to vector store rag_system.vector_store.add_documents([insight_chunk]) app_logger.info(f"Added strategy insight to RAG vector store: {len(content)} chars") return jsonify({ 'success': True, 'message': 'Strategy insight added to vector store', 'chunk_id': insight_chunk['id'] }) except Exception as e: app_logger.error(f"Error adding insight: {str(e)}", exc_info=True) return jsonify({'error': 'Failed to add insight'}), 500 def _format_keyword_results(results, question): """Format keyword-focused search results into answer format.""" if not results: return f"No relevant information found for: {question}" # Create a summary of keyword matches keyword_matches = [] for result in results[:3]: # Top 3 results content = result.get('content', '') matches = result.get('keyword_matches', []) if matches: keyword_matches.extend([f"{m['keyword']} ({m['matches']} matches)" for m in matches[:2]]) if keyword_matches: return f"Found relevant information for: {', '.join(set(keyword_matches[:5]))}" else: return f"Search completed for: {question}" def _format_enhanced_results(results, question): """Format enhanced search results into answer format.""" if not results: return f"No relevant information found for: {question}" # Create a summary based on keyword analysis total_matches = sum(result.get('keyword_analysis', {}).get('total_matches', 0) for result in results[:3]) domains = set() for result in results[:3]: analysis = result.get('keyword_analysis', {}) for match in analysis.get('matched_keywords', []): domains.add(match.get('domain', 'general')) domain_str = ', '.join(domains) if domains else 'general' return f"Found relevant information in {domain_str} domain ({total_matches} keyword matches)" def _format_sources_from_results(results): """Format search results into source format.""" sources = [] for i, result in enumerate(results[:5]): # Top 5 sources metadata = result.get('metadata', {}) source = { 'source': metadata.get('source', f'Document {i+1}'), 'preview': result.get('content', '')[:200] + '...' if len(result.get('content', '')) > 200 else result.get('content', ''), 'score': result.get('score', 0.0) } # Add keyword analysis if available if 'keyword_analysis' in result: analysis = result['keyword_analysis'] source['keyword_matches'] = len(analysis.get('matched_keywords', [])) source['keyword_score'] = analysis.get('keyword_score', 0.0) sources.append(source) return sources @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('/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}") pass return jsonify({ 'status': 'idle', 'progress': 0, 'message': 'No reprocessing in progress', 'current_file': '', 'files_processed': 0, 'total_files': 0, 'chunks_created': 0 }), 200 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)