Enabled for local deployment.

.gitignore

@@ -0,0 +1,10 @@
+.env
+.env.local
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+.Python
+env/
+venv/
+*.log

README.md

@@ -4,7 +4,7 @@ emoji: 🏥
 colorFrom: blue
 colorTo: green
 sdk: streamlit
-sdk_version: "1.38.0"
+sdk_version: "1.44.1"
 app_file: app.py
 pinned: false
 ---

app.py

@@ -1,9 +1,13 @@
+from dotenv import load_dotenv
+import os
 import streamlit as st
 from groq import Groq
-import os
 from database import get_symptoms, get_disease_info
 from user_management import show_user_management, save_history_if_logged_in
 
+# Load environment variables from .env.local
+load_dotenv(dotenv_path=".env.local")
+
 # Initialize Groq client
 groq_client = Groq(api_key=os.getenv("GROQ_API_KEY"))
 
@@ -63,7 +67,7 @@ if st.button("Ask"):
                 max_tokens=500
             )
             st.write("**Response**:")
-            st.write(response.choices[0].message.content)
+            st.markdown(response.choices[0].message.content)
             
             # Save to history if user is logged in
             if st.session_state.get("user_id"):
@@ -75,4 +79,4 @@ if st.button("Ask"):
 
 # Footer
 st.markdown("---")
-st.write("Developed by xAI. Powered by Groq and Neon.")
+st.write("Developed by Pacman. Powered by AI.")

database.py

@@ -1,67 +1,90 @@
+from dotenv import load_dotenv
 import os
 import psycopg2
 from contextlib import contextmanager
 from typing import List, Tuple, Optional
 
+# Load environment variables from .env.local
+load_dotenv(dotenv_path=".env.local")
+
 # Database connection string from environment variable
 DB_CONNECT = os.getenv("DB_CONNECT")
 
+if not DB_CONNECT:
+    raise ValueError("DB_CONNECT environment variable not set. Please check .env.local.")
+
 @contextmanager
 def get_db_connection():
     """Context manager for database connections."""
-    conn = psycopg2.connect(DB_CONNECT)
+    conn = None
     try:
+        conn = psycopg2.connect(DB_CONNECT)
         yield conn
+    except psycopg2.Error as e:
+        raise Exception(f"Database connection failed: {str(e)}")
     finally:
-        conn.close()
+        if conn:
+            conn.close()
 
 def get_symptoms() -> List[str]:
     """Fetch all symptom names from the database."""
-    with get_db_connection() as conn:
-        with conn.cursor() as cur:
-            cur.execute("SELECT symptom_name FROM mb_symptoms ORDER BY symptom_name")
-            return [row[0] for row in cur.fetchall()]
+    try:
+        with get_db_connection() as conn:
+            with conn.cursor() as cur:
+                cur.execute("SELECT symptom_name FROM mb_symptoms ORDER BY symptom_name")
+                return [row[0] for row in cur.fetchall()]
+    except Exception as e:
+        raise Exception(f"Error fetching symptoms: {str(e)}")
 
 def get_disease_info(symptoms: List[str]) -> List[Tuple[str, str, List[str]]]:
     """Get possible diseases and precautions based on symptoms."""
-    with get_db_connection() as conn:
-        with conn.cursor() as cur:
-            # Calculate severity score and match diseases
-            placeholders = ','.join(['%s'] * len(symptoms))
-            query = """
-                SELECT d.disease_name, d.description, 
-                       ARRAY[COALESCE(d.precaution_1, ''), COALESCE(d.precaution_2, ''), 
-                             COALESCE(d.precaution_3, ''), COALESCE(d.precaution_4, '')] as precautions
-                FROM mb_diseases d
-                JOIN mb_disease_symptoms ds ON d.disease_id = ds.disease_id
-                JOIN mb_symptoms s ON ds.symptom_id = s.symptom_id
-                WHERE s.symptom_name IN (%s)
-                GROUP BY d.disease_id, d.disease_name, d.description, d.precaution_1, d.precaution_2, d.precaution_3, d.precaution_4
-                ORDER BY COUNT(*) DESC
-                LIMIT 5
-            """ % placeholders
-            cur.execute(query, symptoms)
-            return cur.fetchall()
+    try:
+        with get_db_connection() as conn:
+            with conn.cursor() as cur:
+                # Calculate severity score and match diseases
+                placeholders = ','.join(['%s'] * len(symptoms))
+                query = """
+                    SELECT d.disease_name, d.description, 
+                           ARRAY[COALESCE(d.precaution_1, ''), COALESCE(d.precaution_2, ''), 
+                                 COALESCE(d.precaution_3, ''), COALESCE(d.precaution_4, '')] as precautions
+                    FROM mb_diseases d
+                    JOIN mb_disease_symptoms ds ON d.disease_id = ds.disease_id
+                    JOIN mb_symptoms s ON ds.symptom_id = s.symptom_id
+                    WHERE s.symptom_name IN (%s)
+                    GROUP BY d.disease_id, d.disease_name, d.description, d.precaution_1, d.precaution_2, d.precaution_3, d.precaution_4
+                    ORDER BY COUNT(*) DESC
+                    LIMIT 5
+                """ % placeholders
+                cur.execute(query, symptoms)
+                return cur.fetchall()
+    except Exception as e:
+        raise Exception(f"Error fetching disease info: {str(e)}")
 
 def save_user_history(user_id: str, symptoms: str, predicted_diseases: str) -> None:
     """Save user symptom history to database."""
-    with get_db_connection() as conn:
-        with conn.cursor() as cur:
-            cur.execute(
-                "INSERT INTO mb_history (user_id, symptoms, predicted_diseases) VALUES (%s, %s, %s)",
-                (user_id, symptoms, predicted_diseases)
-            )
-            conn.commit()
+    try:
+        with get_db_connection() as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "INSERT INTO mb_history (user_id, symptoms, predicted_diseases) VALUES (%s, %s, %s)",
+                    (user_id, symptoms, predicted_diseases)
+                )
+                conn.commit()
+    except Exception as e:
+        raise Exception(f"Error saving user history: {str(e)}")
 
 def get_user_history(user_id: str) -> List[Tuple[int, str, str, str]]:
     """Retrieve user history from database."""
-    with get_db_connection() as conn:
-        with conn.cursor() as cur:
-            cur.execute(
-                "SELECT history_id, symptoms, predicted_diseases, query_timestamp FROM mb_history WHERE user_id = %s ORDER BY query_timestamp DESC",
-                (user_id,)
-            )
-            return cur.fetchall()
+    try:
+        with get_db_connection() as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "SELECT history_id, symptoms, predicted_diseases, query_timestamp FROM mb_history WHERE user_id = %s ORDER BY query_timestamp DESC",
+                    (user_id,)
+                )
+                return cur.fetchall()
+    except Exception as e:
+        raise Exception(f"Error fetching user history: {str(e)}")
 
 def register_user(user_id: str, dob: str, email: Optional[str]) -> bool:
     """Register a new user."""
@@ -70,16 +93,21 @@ def register_user(user_id: str, dob: str, email: Optional[str]) -> bool:
             with conn.cursor() as cur:
                 cur.execute(
                     "INSERT INTO mb_users (user_id, date_of_birth, email) VALUES (%s, %s, %s)",
-                    (user_id, dob, email)
+                    (user_id, dob, email or None)
                 )
                 conn.commit()
                 return True
     except psycopg2.IntegrityError:
         return False
+    except Exception as e:
+        raise Exception(f"Error registering user: {str(e)}")
 
 def user_exists(user_id: str) -> bool:
     """Check if user_id already exists."""
-    with get_db_connection() as conn:
-        with conn.cursor() as cur:
-            cur.execute("SELECT 1 FROM mb_users WHERE user_id = %s", (user_id,))
-            return cur.fetchone() is not None
+    try:
+        with get_db_connection() as conn:
+            with conn.cursor() as cur:
+                cur.execute("SELECT 1 FROM mb_users WHERE user_id = %s", (user_id,))
+                return cur.fetchone() is not None
+    except Exception as e:
+        raise Exception(f"Error checking user existence: {str(e)}")

requirements.txt

@@ -1,3 +1,4 @@
-streamlit==1.38.0
-groq==0.11.0
-psycopg2-binary==2.9.9
+streamlit==1.44.1
+groq
+psycopg2-binary
+python-dotenv

user_management.py

@@ -36,22 +36,22 @@ def show_user_management():
             if not user_id:
                 st.sidebar.error("User ID is required.")
             elif user_exists(user_id):
-                st.sidebar.error("User ID already taken. Please choose another.")
+                st.sidebar.error("User ID already exists. Please choose another.")
             else:
-                if register_user(user_id, dob, email or None):
-                    st.session_state.user_id = user_id
-                    st.sidebar.success("Registered successfully!")
-                else:
-                    st.sidebar.error("Registration failed. Please try again.")
-        
+                if register_user(user_id, dob, email st.session_state.user_id = user_id
+                st.sidebar.success("Registered successfully!")
+                
         if st.sidebar.button("Login"):
-            if user_exists(user_id):
+            if user_exists(user_id,):
                 st.session_state.user_id = user_id
                 st.sidebar.success("Logged in successfully!")
             else:
-                st.sidebar.error("User ID not found. Please register or try another ID.")
+                st.sidebar.error("User ID not found. Please register or try again.")
 
-def save_history_if_logged_in(symptoms: str, predicted_diseases: str):
-    """Save history if user is logged in."""
-    if st.session_state.get("user_id"):
-        save_user_history(st.session_state.user_id, symptoms, predicted_diseases)
+def save_history_if_logged_in(user_id: str, symptoms: str, predicted_diseases: str):
+    """Save user history if user is logged in."""
+    try:
+        if user_id.get("user_id"):
+            save_user_history(st.session_state.user_id, symptoms, predicted_diseases)
+    except Exception as e:
+        raise Exception(f"Error saving user history: {str(e)}")