Update app.py

app.py

@@ -5,9 +5,10 @@ import time
 import os
 import sys
 import psutil
-from urllib.parse import urlparse, quote
+from urllib.parse import quote
 import threading
 import re
+from pathlib import Path
 
 # Configure logging
 logging.basicConfig(
@@ -18,74 +19,66 @@ logging.basicConfig(
 logger = logging.getLogger("SemanticSearchApp")
 
 # Security validation functions
-def is_valid_url(url: str) -> bool:
-    """Validate URL format and safety."""
+def is_valid_url(url):
+    """Validate URL format and safety"""
     try:
         result = urlparse(url)
         return all([result.scheme, result.netloc])
-    except Exception:
+    except:
         return False
 
-def sanitize_query(query: str) -> str:
-    """Sanitize user input to prevent injection attacks."""
+def sanitize_query(query):
+    """Sanitize user input to prevent injection attacks"""
     try:
-        # Remove non-alphanumeric characters except spaces and hyphens.
-        clean_query = re.sub(r'[^\w\s-]', '', query)
-        return clean_query[:256]  # Truncate to prevent overly long queries.
+        return re.sub(r'[^\w\s-]', '', query)[:256]
     except Exception as e:
         logger.error(f"Query sanitization failed: {str(e)}")
-        return query[:256]  # Fallback truncation.
-
-# Diagnostics integration
-try:
-    from diagnostics import diagnose_parquet_files
-    diagnostics_available = True
-except ImportError:
-    diagnostics_available = False
-    logger.warning("Diagnostics module not available")
-
-def add_diagnostics_ui(search_system: OptimizedSemanticSearch):
-    """Enhanced diagnostics UI with proper directory checks."""
+        return query[:256]
+
+def add_diagnostics_ui(search_system):
+    """Enhanced diagnostics with accurate path handling"""
     with st.sidebar.expander("🔧 Diagnostics", expanded=False):
-        if st.button("Run Full System Check"):
-            with st.spinner("Performing comprehensive system check..."):
-                col1, col2 = st.columns(2)
-                
-                # Get actual paths from the search system.
-                metadata_dir = search_system.metadata_mgr.metadata_dir
-                faiss_dir = Path("compressed_shards")  # Assuming FAISS index dir, if applicable.
-
-                with col1:
-                    st.subheader("📂 Metadata Validation")
-                    if metadata_dir.exists():
-                        parquet_files = list(metadata_dir.glob("*.parquet"))
-                        st.write(f"Directory: `{metadata_dir}`")
-                        st.write(f"Parquet Files Found: {'✅' if parquet_files else '❌'}")
-                        if diagnostics_available and diagnose_parquet_files(str(metadata_dir)):
-                            st.success("✅ Metadata shards valid")
-                        else:
-                            st.error("❌ Metadata issues detected")
-                    else:
-                        st.error("Metadata directory not found")
-
-                with col2:
-                    st.subheader("📚 FAISS Validation")
-                    # For the FAISS index, we rely on our search system.
-                    if hasattr(search_system, 'index') and search_system.index is not None:
-                        st.success(f"✅ FAISS index loaded with {search_system.index.ntotal} vectors")
-                    else:
-                        st.error("❌ No FAISS index loaded")
-
-                st.subheader("💻 System Resources")
-                col_res1, col_res2 = st.columns(2)
-                with col_res1:
-                    st.metric("Memory Usage", 
-                              f"{psutil.Process().memory_info().rss // 1024 ** 2} MB",
-                              help="Current process memory usage")
-                with col_res2:
-                    st.metric("CPU Utilization", 
-                              f"{psutil.cpu_percent()}%",
-                              help="Total system CPU usage")
+        col1, col2 = st.columns(2)
+        
+        # Metadata validation
+        with col1:
+            st.subheader("📂 Metadata Validation")
+            metadata_dir = search_system.metadata_mgr.metadata_dir
+            if metadata_dir.exists():
+                parquet_files = list(metadata_dir.glob("*.parquet"))
+                status = len(parquet_files) > 0
+                st.write(f"Directory: `{metadata_dir}`")
+                st.write(f"Parquet Files: {len(parquet_files)}")
+                st.success("✅ Valid metadata" if status else "❌ No parquet files found")
+            else:
+                st.error("Metadata directory not found")
+
+        # FAISS validation
+        with col2:
+            st.subheader("📚 FAISS Validation")
+            faiss_path = Path("combined_index.faiss")
+            if faiss_path.exists():
+                st.write(f"Index Path: `{faiss_path}`")
+                if hasattr(search_system, 'index') and search_system.index:
+                    st.success(f"✅ Index loaded")
+                    st.write(f"Vectors: {search_system.index.ntotal:,}")
+                else:
+                    st.error("❌ Index not loaded")
+            else:
+                st.error("FAISS index not found")
+
+        # System resources
+        st.subheader("💻 System Resources")
+        col_res1, col_res2 = st.columns(2)
+        with col_res1:
+            mem_usage = psutil.Process().memory_info().rss // 1024 ** 2
+            st.metric("Memory Usage", f"{mem_usage} MB")
+
+        with col_res2:
+            cpu_usage = psutil.cpu_percent()
+            status_color = "#ff0000" if cpu_usage > 80 else "#00ff00"
+            st.markdown(f"<span style='color:{status_color}'>CPU: {cpu_usage}%</span>", 
+                       unsafe_allow_html=True)
 
 def main():
     st.set_page_config(
@@ -94,146 +87,108 @@ def main():
         layout="wide"
     )
 
-    # Initialize the optimized search system with caching.
-    @st.cache_resource(ttl=3600, show_spinner="Initializing search engine...")
-    def init_search_system():
-        try:
-            system = OptimizedSemanticSearch()
-            logger.info("Search system initialized successfully")
-            return system
-        except Exception as e:
-            logger.error(f"System initialization failed: {str(e)}")
-            st.error("Critical system initialization error. Check logs.")
-            st.stop()
-
-    # Custom CSS for enhanced visual design.
+    # Custom CSS styling
     st.markdown("""
     <style>
-    div[data-testid="stExpander"] div[role="button"] p {
-        font-size: 1.2rem;
-        font-weight: bold;
-        color: #1e88e5;
-    }
-    a.source-link {
-        color: #1a73e8 !important;
-        text-decoration: none !important;
-        border-bottom: 2px solid transparent;
-        transition: all 0.3s ease;
-    }
-    a.source-link:hover {
-        border-bottom-color: #1a73e8;
-        opacity: 0.9;
+    .metric-box {
+        padding: 15px;
+        border-radius: 8px;
+        background: #f8f9fa;
+        margin: 10px 0;
+        box-shadow: 0 2px 4px rgba(0,0,0,0.1);
     }
-    .similarity-badge {
-        padding: 0.2em 0.5em;
-        border-radius: 4px;
-        background: #e3f2fd;
-        color: #1e88e5;
-        font-weight: 500;
+    .result-card {
+        padding: 15px;
+        border-left: 4px solid #1e88e5;
+        margin: 10px 0;
+        background: #fff;
     }
     </style>
     """, unsafe_allow_html=True)
 
+    # Initialize search system
+    @st.cache_resource(ttl=3600)
+    def init_search_system():
+        try:
+            system = OptimizedSemanticSearch()
+            system.initialize_system()
+            logger.info("Search system initialized")
+            return system
+        except Exception as e:
+            logger.error(f"Initialization failed: {str(e)}")
+            st.error("System initialization error")
+            st.stop()
+
     try:
         search_system = init_search_system()
     except Exception as e:
-        st.error(f"Failed to initialize search system: {str(e)}")
+        st.error(f"Critical error: {str(e)}")
         st.stop()
 
+    # Main UI components
     st.title("🔍 Semantic Search Engine")
-    
-    # Search input with sanitization.
-    query = st.text_input("Enter your search query:", 
-                          placeholder="Search documents...",
-                          max_chars=200)
-    
+    query = st.text_input("Search knowledge base:", placeholder="Enter your query...")
+
     if query:
-        try:
-            clean_query = sanitize_query(query)
-            if not clean_query:
-                st.warning("Please enter a valid search query")
-                st.stop()
-                
-            with st.spinner("🔍 Searching through documents..."):
-                start_time = time.time()
-                results = search_system.search(clean_query, top_k=5)
+        clean_query = sanitize_query(query)
+        if not clean_query:
+            st.warning("Invalid query format")
+            st.stop()
+
+        with st.spinner("Analyzing documents..."):
+            start_time = time.time()
+            try:
+                results = search_system.search(clean_query, 5)
                 search_duration = time.time() - start_time
                 
-                if results:
-                    st.subheader(f"Top Results (Search completed in {search_duration:.2f}s)")
-                    
-                    for res in results:
-                        with st.expander(f"{res.get('title', 'Untitled')}"):
-                            col1, col2 = st.columns([3, 1])
-                            with col1:
-                                st.markdown(f"**Summary:** {res.get('summary', '')}")
-                            with col2:
-                                st.markdown(
-                                    f"<div class='similarity-badge'>"
-                                    f"Confidence: {res.get('similarity', 0):.1%}"
-                                    f"</div>", 
-                                    unsafe_allow_html=True
-                                )
-                                st.progress(float(res.get("similarity", 0)))
-                            if res.get("source"):
-                                # Validate URL before displaying.
-                                if is_valid_url(res["source"]):
-                                    st.markdown(f"<a class='source-link' href='{res['source']}' target='_blank'>Source Link</a>", unsafe_allow_html=True)
-                                else:
-                                    st.warning("Invalid source URL")
+                if not results:
+                    st.warning("No matches found")
+                    st.info("Try refining your search terms")
                 else:
-                    st.warning("No matching documents found")
-                    st.info("Try these tips:")
-                    st.markdown("""
-                    - Use more specific keywords
-                    - Check your spelling
-                    - Avoid special characters
-                    """)
-        except Exception as e:
-            logger.error(f"Search failed: {str(e)}")
-            st.error("Search operation failed. Please try again.")
+                    st.subheader(f"Top Results ({search_duration:.2f}s)")
+                    for res in results:
+                        with st.expander(res.get('title', 'Untitled')):
+                            st.markdown(f"**Summary**: {res.get('summary', '')}")
+                            similarity = res.get('similarity', 0)
+                            st.progress(similarity)
+                            st.markdown(f"**Confidence**: {similarity:.1%}")
+                            source = res.get('source', '')
+                            if source and is_valid_url(source):
+                                st.markdown(f"[View Source]({source})")
+                            elif res.get('title'):
+                                st.markdown(f"[Google Scholar Search](https://scholar.google.com/scholar?q={quote(res['title'])})")
 
-    # Sidebar: System monitoring and diagnostics.
+            except Exception as e:
+                logger.error(f"Search error: {str(e)}")
+                st.error("Search operation failed")
+
+    # System status sidebar
     with st.sidebar:
-        st.subheader("📊 System Status")
+        st.subheader("📊 System Health")
+        
         col1, col2 = st.columns(2)
         with col1:
-            st.metric("Total Documents", 
-                      f"{search_system.metadata_mgr.total_docs:,}",
-                      help="Total indexed documents in the system")
+            st.metric("Documents", f"{search_system.metadata_mgr.total_docs:,}")
+        
         with col2:
-            if hasattr(search_system, 'index'):
-                st.metric("FAISS Vectors", 
-                          f"{search_system.index.ntotal:,}",
-                          help="Number of vectors in the FAISS index")
-        st.metric("Active Memory", 
-                  f"{psutil.Process().memory_info().rss // 1024 ** 2} MB",
-                  help="Current memory usage by the application")
-
-        if diagnostics_available:
+            vectors = search_system.index.ntotal if hasattr(search_system, 'index') else 0
+            st.metric("Vectors", f"{vectors:,}")
+
+        # Diagnostics section
+        if st.checkbox("Show advanced diagnostics"):
             add_diagnostics_ui(search_system)
-        else:
-            st.warning("Diagnostics module not available")
 
-        if st.button("🩺 Run Health Check"):
-            try:
-                system_stats = {
-                    "shards_loaded": 1 if hasattr(search_system, 'index') else 0,
-                    "metadata_records": search_system.metadata_mgr.total_docs,
-                    "memory_usage": f"{psutil.Process().memory_info().rss // 1024 ** 2} MB",
-                    "active_threads": threading.active_count(),
-                    "system_load": f"{os.getloadavg()[0]:.2f}"
-                }
-                st.json(system_stats)
-            except Exception as e:
-                st.error(f"Health check failed: {str(e)}")
+        # System monitoring
+        st.subheader("⚙️ Monitoring")
+        with st.expander("Performance"):
+            mem = psutil.virtual_memory()
+            st.write(f"Memory: {mem.percent}% used")
+            st.write(f"CPU Cores: {psutil.cpu_count()}")
+            st.write(f"Active threads: {threading.active_count()}")
 
-        if st.button("♻️ Clear Cache"):
-            try:
-                st.cache_resource.clear()
-                st.experimental_rerun()
-            except Exception as e:
-                st.error(f"Cache clearance failed: {str(e)}")
+        if st.button("🔄 Refresh System"):
+            st.cache_resource.clear()
+            st.rerun()
 
 if __name__ == "__main__":
-    main()
+    main()