""" Admin Analytics Dashboard (Streamlit) Optional admin view for monitoring system metrics. """ import streamlit as st import requests import pandas as pd import plotly.express as px import plotly.graph_objects as go from datetime import datetime, timedelta import os def check_admin_auth(): """Check if admin is authenticated.""" if "admin_token" not in st.session_state: st.session_state.admin_token = None if not st.session_state.admin_token: st.warning("⚠️ Please enter admin token to access analytics") token = st.text_input("Admin Token", type="password", key="token_input") if st.button("Login"): st.session_state.admin_token = token st.rerun() return False return True def get_admin_headers(): """Get authorization headers for admin endpoints.""" return { "Authorization": f"Bearer {st.session_state.admin_token}" } def fetch_metrics_summary(api_url: str): """Fetch metrics summary from admin endpoint.""" try: response = requests.get( f"{api_url}/admin/metrics/summary", headers=get_admin_headers(), timeout=10 ) if response.status_code == 401: st.error("❌ Invalid admin token. Please check and try again.") st.session_state.admin_token = None return None response.raise_for_status() return response.json()["data"] except Exception as e: st.error(f"Failed to fetch summary: {str(e)}") return None def fetch_events_timeline(api_url: str, days: int = 7): """Fetch events timeline.""" try: response = requests.get( f"{api_url}/admin/metrics/events?days={days}", headers=get_admin_headers(), timeout=10 ) response.raise_for_status() return response.json()["data"] except Exception as e: st.error(f"Failed to fetch timeline: {str(e)}") return None def fetch_funnel_analysis(api_url: str, days: int = 7): """Fetch funnel analysis.""" try: response = requests.get( f"{api_url}/admin/metrics/funnel?days={days}", headers=get_admin_headers(), timeout=10 ) response.raise_for_status() return response.json()["data"] except Exception as e: st.error(f"Failed to fetch funnel: {str(e)}") return None def fetch_rate_limit_stats(api_url: str, days: int = 7): """Fetch rate limit statistics.""" try: response = requests.get( f"{api_url}/admin/metrics/rate-limits?days={days}", headers=get_admin_headers(), timeout=10 ) response.raise_for_status() return response.json()["data"] except Exception as e: st.error(f"Failed to fetch rate limit stats: {str(e)}") return None def render_admin_dashboard(): """Render admin analytics dashboard.""" st.title("🔒 Admin Analytics Dashboard") # Check authentication if not check_admin_auth(): return # Logout button if st.button("Logout"): st.session_state.admin_token = None st.rerun() # API URL configuration api_url = st.sidebar.text_input( "API URL", value=os.getenv("API_URL", "http://localhost:7860"), help="FastAPI backend URL" ) # Time range selector days = st.sidebar.slider( "Days to analyze", min_value=1, max_value=30, value=7, help="Number of days to include in analysis" ) # Refresh button if st.sidebar.button("🔄 Refresh Data"): st.rerun() st.divider() # === METRICS SUMMARY === st.header("📊 Metrics Summary") summary = fetch_metrics_summary(api_url) if summary: col1, col2, col3 = st.columns(3) with col1: st.metric("Unique Devices", summary["unique_devices"]) with col2: st.metric("Unique Users", summary["unique_users"]) with col3: total_events = sum(summary["events_by_type"].values()) st.metric("Total Events", total_events) # Events by type st.subheader("Events by Type") events_df = pd.DataFrame([ {"Event Type": k, "Count": v} for k, v in summary["events_by_type"].items() ]) if not events_df.empty: fig = px.bar( events_df, x="Event Type", y="Count", color="Event Type", title="Event Distribution" ) st.plotly_chart(fig, use_container_width=True) st.divider() # === EVENTS TIMELINE === st.header("📈 Events Timeline") timeline_data = fetch_events_timeline(api_url, days) if timeline_data and timeline_data["timeline"]: timeline_df = pd.DataFrame(timeline_data["timeline"]) fig = px.line( timeline_df, x="date", y="count", color="event_type", title=f"Events Over Last {days} Days", labels={"count": "Event Count", "date": "Date", "event_type": "Event Type"} ) st.plotly_chart(fig, use_container_width=True) else: st.info("No timeline data available") st.divider() # === FUNNEL ANALYSIS === st.header("🔀 Conversion Funnel") funnel_data = fetch_funnel_analysis(api_url, days) if funnel_data: stages = funnel_data["funnel_stages"] conversions = funnel_data["conversion_rates"] # Funnel chart funnel_stages = ["DASHBOARD_VIEW", "ANALYSIS_REQUEST", "TASK_QUEUED", "TASK_COMPLETED"] funnel_values = [stages.get(stage, 0) for stage in funnel_stages] fig = go.Figure(go.Funnel( y=funnel_stages, x=funnel_values, textinfo="value+percent initial" )) fig.update_layout(title=f"User Journey Funnel (Last {days} Days)") st.plotly_chart(fig, use_container_width=True) # Conversion metrics st.subheader("Conversion Rates") col1, col2 = st.columns(2) with col1: st.metric("View → Request", f"{conversions['view_to_request']:.1f}%") st.metric("Request → Queued", f"{conversions['request_to_queued']:.1f}%") with col2: st.metric("Queued → Completed", f"{conversions['queued_to_completed']:.1f}%") st.metric("Overall Completion", f"{conversions['overall_completion']:.1f}%") st.divider() # === RATE LIMIT STATS === st.header("🚦 Rate Limiting Statistics") rate_limit_data = fetch_rate_limit_stats(api_url, days) if rate_limit_data: col1, col2 = st.columns(2) with col1: st.metric("Total Rate Limit Hits", rate_limit_data["total_hits"]) with col2: top_devices = len(rate_limit_data["top_devices"]) st.metric("Unique Devices Hit", top_devices) # Top offenders if rate_limit_data["top_devices"]: st.subheader("Top Rate Limited Devices") devices_df = pd.DataFrame(rate_limit_data["top_devices"]) st.dataframe(devices_df, use_container_width=True) # Timeline if rate_limit_data["timeline"]: st.subheader("Rate Limit Hits Over Time") timeline_df = pd.DataFrame(rate_limit_data["timeline"]) fig = px.bar( timeline_df, x="date", y="count", title="Daily Rate Limit Hits", labels={"count": "Hits", "date": "Date"} ) st.plotly_chart(fig, use_container_width=True) if __name__ == "__main__": st.set_page_config( page_title="Admin Analytics", page_icon="🔒", layout="wide" ) render_admin_dashboard()