TG-5: Streamlit Application Base

app.py

@@ -0,0 +1,170 @@
+import streamlit as st
+import pymysql
+import myloginpath
+import ollama
+import bcrypt
+
+# -------------------------------------------------------------------
+# Database Connections (PoLP & SoD)
+# -------------------------------------------------------------------
+def get_db_connection(login_path):
+    """Dynamically connect using myloginpath to preserve Segregation of Duties."""
+    try:
+        conf = myloginpath.parse(login_path)
+        return pymysql.connect(
+            host=conf.get('host', '127.0.0.1'),
+            user=conf.get('user'),
+            password=conf.get('password'),
+            database='food_db',
+            cursorclass=pymysql.cursors.DictCursor
+        )
+    except Exception as e:
+        st.error(f"Failed to connect using login-path '{login_path}'. Did you run mysql_config_editor?")
+        st.sidebar.error(f"Connection Error: {e}")
+        return None
+
+# -------------------------------------------------------------------
+# Authentication Logic
+# -------------------------------------------------------------------
+def verify_login(username, password):
+    conn = get_db_connection('app_auth')
+    if not conn: return False
+    
+    with conn.cursor() as cursor:
+        cursor.execute("SELECT password_hash FROM users WHERE username = %s LIMIT 1", (username,))
+        result = cursor.fetchone()
+        conn.close()
+        
+        if result:
+            # Check the hash
+            return bcrypt.checkpw(password.encode('utf-8'), result['password_hash'].encode('utf-8'))
+    return False
+
+def register_user(username, password):
+    conn = get_db_connection('app_auth')
+    if not conn: return False
+    
+    hashed = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
+    try:
+        with conn.cursor() as cursor:
+            cursor.execute("INSERT INTO users (username, password_hash) VALUES (%s, %s)", (username, hashed))
+            conn.commit()
+        conn.close()
+        return True
+    except pymysql.err.IntegrityError:
+        return False  # Username exists
+
+# -------------------------------------------------------------------
+# UI Flow
+# -------------------------------------------------------------------
+st.set_page_config(page_title="Food AI Explorer", page_icon="🍔", layout="wide")
+
+if "authenticated_user" not in st.session_state:
+    st.session_state["authenticated_user"] = None
+
+# Sidebar Authentication
+with st.sidebar:
+    st.title("User Portal 🔐")
+    if st.session_state["authenticated_user"]:
+        st.success(f"Logged in as: {st.session_state['authenticated_user']}")
+        if st.button("Logout"):
+            st.session_state["authenticated_user"] = None
+            st.rerun()
+    else:
+        tab1, tab2 = st.tabs(["Login", "Register"])
+        with tab1:
+            l_user = st.text_input("Username", key="l_user")
+            l_pass = st.text_input("Password", type="password", key="l_pass")
+            if st.button("Login"):
+                if verify_login(l_user, l_pass):
+                    st.session_state["authenticated_user"] = l_user
+                    st.success("Logged in successfully!")
+                    st.rerun()
+                else:
+                    st.error("Invalid username or password.")
+        with tab2:
+            r_user = st.text_input("Username", key="r_user")
+            r_pass = st.text_input("Password", type="password", key="r_pass")
+            if st.button("Register"):
+                if len(r_pass) < 6:
+                    st.error("Password too short.")
+                elif register_user(r_user, r_pass):
+                    st.success("Registered successfully! Please log in.")
+                else:
+                    st.error("Username already exists.")
+
+# Main Application Logic
+if not st.session_state["authenticated_user"]:
+    st.title("🍔 Food AI Local Explorer")
+    st.info("Please login or register on the sidebar to interact with the LLM.")
+    st.stop()  # Halt execution here, keeping it secure.
+
+# --- Authenticated App ---
+st.title("🍔 Food AI Local Explorer")
+st.markdown("Interrogate your database leveraging your private secure stack.")
+
+# Checking products via Reader Login path
+conn_reader = get_db_connection('app_reader')
+if conn_reader:
+    with conn_reader.cursor() as cursor:
+        cursor.execute("SELECT COUNT(*) as total FROM products;")
+        total_products = cursor.fetchone()['total']
+        st.sidebar.info(f"Database Scope: {total_products} products.")
+
+tab_chat, tab_explore = st.tabs(["💬 AI Chat", "🔍 Food Search & Details"])
+
+with tab_chat:
+    st.subheader("Chat with the Context")
+    if "messages" not in st.session_state:
+        st.session_state["messages"] = [{"role": "assistant", "content": "How can I help you analyze the food data today?"}]
+
+    for msg in st.session_state.messages:
+        st.chat_message(msg["role"]).write(msg["content"])
+
+    if prompt := st.chat_input("Ask about the food items..."):
+        st.session_state.messages.append({"role": "user", "content": prompt})
+        st.chat_message("user").write(prompt)
+        sys_prompt = "You are a helpful data analyst AI. Answer strictly using local data contexts."
+        
+        with st.spinner("Analyzing locally..."):
+            try:
+                response = ollama.chat(model='llama3', messages=[
+                    {'role': 'system', 'content': sys_prompt},
+                    {'role': 'user', 'content': prompt}
+                ])
+                ai_reply = response['message']['content']
+            except Exception as e:
+                ai_reply = f"Hold on! Could not reach Ollama Engine. Error: {e}"
+
+        st.session_state.messages.append({"role": "assistant", "content": ai_reply})
+        st.chat_message("assistant").write(ai_reply)
+
+with tab_explore:
+    st.subheader("Raw Data Search")
+    search_query = st.text_input("Search Product Name or Ingredient (e.g. 'Nutella' or 'Sugar')")
+    
+    if st.button("Search Database") and search_query and conn_reader:
+        with st.spinner("Querying MySQL..."):
+            try:
+                with conn_reader.cursor() as cursor:
+                    # Leverage the FULLTEXT INDEX built in init.sql
+                    query = """
+                        SELECT code, product_name, generic_name, brands, ingredients_text 
+                        FROM products 
+                        WHERE MATCH(product_name, ingredients_text) AGAINST(%s IN NATURAL LANGUAGE MODE)
+                        LIMIT 50
+                    """
+                    cursor.execute(query, (search_query,))
+                    results = cursor.fetchall()
+            except Exception as e:
+                st.error(f"SQL Error: {e}")
+                results = []
+                
+        if results:
+            st.success(f"Found {len(results)} matching records!")
+            st.dataframe(results, use_container_width=True)
+        else:
+            st.warning("No products found matching those terms.")
+
+if conn_reader:
+    conn_reader.close()