""" Spark TikTok Analytics POC - Dashboard Module Handles UI components, visualizations, and dashboard layouts Optimized for Container Runtime with responsive design """ import streamlit as st import pandas as pd from datetime import datetime, timedelta from typing import Dict, List, Optional, Any import logging # Note: Using Streamlit native charts instead of plotly for Container Runtime compatibility # plotly.express as px # plotly.graph_objects as go # from plotly.subplots import make_subplots # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class DashboardModule: """ Dashboard and visualization module for TikTok analytics Handles all UI components and data visualizations """ def __init__(self): self.colors = { 'primary': '#FF6B6B', 'secondary': '#4ECDC4', 'tertiary': '#45B7D1', 'success': '#96CEB4', 'warning': '#FFEAA7', 'danger': '#DDA0DD', 'dark': '#2F3542', 'light': '#F1F2F6' } self.chart_config = { 'displayModeBar': False, 'responsive': True } # ================================================================= # MAIN DASHBOARD COMPONENTS # ================================================================= def render_main_header(self, title: str = "Spark TikTok Analytics POC"): """Render main application header""" st.set_page_config( page_title=title, page_icon="🎵", layout="wide", initial_sidebar_state="expanded" ) st.markdown("""

🎵 Spark TikTok Analytics POC

Professional TikTok Sound Analytics Platform

""", unsafe_allow_html=True) def render_sidebar_navigation(self) -> str: """Render sidebar navigation and return selected page""" st.sidebar.markdown("### 🎵 Navigation") pages = { "🏠 Dashboard": "dashboard", "🔍 Song Search": "search", "📊 Analytics": "analytics", "⚙️ Settings": "settings", "📈 Database Stats": "stats" } selected_page = st.sidebar.selectbox( "Select Page", options=list(pages.keys()), index=0 ) return pages[selected_page] def render_metrics_row(self, metrics: Dict[str, Any]): """Render key metrics in a row layout""" col1, col2, col3, col4 = st.columns(4) with col1: self._render_metric_card( "Total Videos", metrics.get('total_videos', 0), "🎬" ) with col2: self._render_metric_card( "Total Likes", metrics.get('total_likes', 0), "❤️" ) with col3: self._render_metric_card( "Unique Creators", metrics.get('unique_creators', 0), "👥" ) with col4: self._render_metric_card( "Viral Videos", metrics.get('viral_videos_count', 0), "🔥" ) def render_song_tracking_table(self, songs_df: pd.DataFrame): """Render active song tracking table""" if songs_df.empty: st.info("No songs currently being tracked. Add a song to get started!") return st.markdown("### 🎵 Active Song Tracking") # Format the dataframe for display display_df = songs_df.copy() # Format columns for better display if 'LAST_SCRAPED_AT' in display_df.columns: display_df['Last Scraped'] = pd.to_datetime(display_df['LAST_SCRAPED_AT']).dt.strftime('%Y-%m-%d %H:%M') if 'SCRAPE_STATUS' in display_df.columns: display_df['Status'] = display_df['SCRAPE_STATUS'].apply(self._format_status) # Select columns to display display_columns = ['ARTIST', 'TRACK', 'Status', 'Last Scraped', 'MAX_RESULTS_PER_SCRAPE'] available_columns = [col for col in display_columns if col in display_df.columns] if available_columns: st.dataframe( display_df[available_columns], use_container_width=True, hide_index=True ) else: st.dataframe(display_df, use_container_width=True, hide_index=True) # ================================================================= # VISUALIZATION COMPONENTS # ================================================================= def render_engagement_overview_chart(self, analytics: Dict[str, Any]): """Render engagement overview chart""" if not analytics or analytics.get('total_videos', 0) == 0: st.info("No video data available for engagement overview") return st.markdown("### 📊 Engagement Overview") # Create engagement metrics chart metrics = ['Likes', 'Comments', 'Shares', 'Plays'] values = [ analytics.get('total_likes', 0), analytics.get('total_comments', 0), analytics.get('total_shares', 0), analytics.get('total_plays', 0) ] fig = go.Figure(data=[ go.Bar( x=metrics, y=values, marker=dict( color=[self.colors['primary'], self.colors['secondary'], self.colors['tertiary'], self.colors['success']] ), text=[f"{v:,}" for v in values], textposition='auto' ) ]) fig.update_layout( title="Total Engagement Metrics", yaxis_title="Count", template="plotly_white", height=400 ) st.plotly_chart(fig, use_container_width=True, config=self.chart_config) def render_viral_distribution_chart(self, videos_df: pd.DataFrame): """Render viral vs non-viral video distribution""" if videos_df.empty: st.info("No video data available for viral distribution") return st.markdown("### 🔥 Viral Content Distribution") col1, col2 = st.columns(2) with col1: # Viral percentage pie chart viral_counts = videos_df['is_viral'].value_counts() labels = ['Non-Viral', 'Viral'] values = [viral_counts.get(False, 0), viral_counts.get(True, 0)] fig_pie = go.Figure(data=[ go.Pie( labels=labels, values=values, colors=[self.colors['light'], self.colors['danger']], hole=.3 ) ]) fig_pie.update_layout( title="Viral vs Non-Viral Videos", height=350 ) st.plotly_chart(fig_pie, use_container_width=True, config=self.chart_config) with col2: # Engagement score distribution if 'engagement_score' in videos_df.columns: fig_hist = px.histogram( videos_df, x='engagement_score', nbins=20, title="Engagement Score Distribution", color_discrete_sequence=[self.colors['tertiary']] ) fig_hist.update_layout( xaxis_title="Engagement Score", yaxis_title="Number of Videos", height=350 ) st.plotly_chart(fig_hist, use_container_width=True, config=self.chart_config) def render_top_creators_chart(self, top_creators: List[Dict[str, Any]]): """Render top creators chart""" if not top_creators: st.info("No creator data available") return st.markdown("### 👑 Top Performing Creators") creators_df = pd.DataFrame(top_creators) fig = go.Figure(data=[ go.Bar( x=creators_df['username'], y=creators_df['total_likes'], marker=dict(color=self.colors['secondary']), text=[f"{v:,}" for v in creators_df['total_likes']], textposition='auto', hovertemplate="%{x}
" + "Total Likes: %{y:,}
" + "Videos: " + creators_df['video_count'].astype(str) + "" ) ]) fig.update_layout( title="Top Creators by Total Likes", xaxis_title="Creator", yaxis_title="Total Likes", template="plotly_white", height=400, xaxis_tickangle=-45 ) st.plotly_chart(fig, use_container_width=True, config=self.chart_config) def render_content_category_chart(self, categories: Dict[str, int]): """Render content category distribution""" if not categories: st.info("No content category data available") return st.markdown("### 🎭 Content Categories") # Create horizontal bar chart for categories categories_df = pd.DataFrame(list(categories.items()), columns=['Category', 'Count']) categories_df = categories_df.sort_values('Count', ascending=True) fig = go.Figure(data=[ go.Bar( y=categories_df['Category'], x=categories_df['Count'], orientation='h', marker=dict(color=self.colors['warning']), text=[f"{v}" for v in categories_df['Count']], textposition='auto' ) ]) fig.update_layout( title="Video Content Categories", xaxis_title="Number of Videos", template="plotly_white", height=max(300, len(categories) * 40) ) st.plotly_chart(fig, use_container_width=True, config=self.chart_config) def render_trending_hashtags(self, hashtags: List[Dict[str, Any]]): """Render trending hashtags visualization""" if not hashtags: st.info("No hashtag data available") return st.markdown("### #️⃣ Trending Hashtags") # Display hashtags in columns cols = st.columns(3) for idx, hashtag_data in enumerate(hashtags[:9]): # Show top 9 with cols[idx % 3]: self._render_hashtag_card(hashtag_data) def render_momentum_gauge(self, momentum_score: float): """Render momentum score as gauge chart""" st.markdown("### 🚀 Momentum Score") fig = go.Figure(go.Indicator( mode="gauge+number+delta", value=momentum_score, domain={'x': [0, 1], 'y': [0, 1]}, title={'text': "Song Momentum"}, delta={'reference': 0.5}, gauge={ 'axis': {'range': [None, 1]}, 'bar': {'color': self.colors['primary']}, 'steps': [ {'range': [0, 0.3], 'color': self.colors['light']}, {'range': [0.3, 0.7], 'color': self.colors['warning']}, {'range': [0.7, 1], 'color': self.colors['success']} ], 'threshold': { 'line': {'color': "red", 'width': 4}, 'thickness': 0.75, 'value': 0.8 } } )) fig.update_layout(height=300) st.plotly_chart(fig, use_container_width=True, config=self.chart_config) # ================================================================= # INTERACTIVE COMPONENTS # ================================================================= def render_song_search_form(self) -> Dict[str, Any]: """Render song search form and return search parameters""" st.markdown("### 🔍 Search Chartmetric Songs") with st.form("song_search_form"): col1, col2 = st.columns(2) with col1: artist = st.text_input("Artist Name", placeholder="e.g., Taylor Swift") with col2: track = st.text_input("Track Name", placeholder="e.g., Shake It Off") search_submitted = st.form_submit_button("🔍 Search Songs", use_container_width=True) return { 'artist': artist.strip() if artist else None, 'track': track.strip() if track else None, 'submitted': search_submitted } def render_song_selection_table(self, songs_df: pd.DataFrame) -> Optional[Dict[str, Any]]: """Render song selection table with action buttons""" if songs_df.empty: return None st.markdown("### 🎵 Search Results") # Format dataframe for display display_df = songs_df.copy() display_df['Select'] = False display_df = display_df[['Select', 'ARTIST', 'TRACK', 'POSTS_LATEST', 'ACTIVE']] # Create interactive table edited_df = st.data_editor( display_df, use_container_width=True, hide_index=True, column_config={ "Select": st.column_config.CheckboxColumn("Select", default=False), "POSTS_LATEST": st.column_config.NumberColumn("Posts", format="%d"), "ACTIVE": st.column_config.CheckboxColumn("Active") } ) # Get selected songs selected_songs = edited_df[edited_df['Select'] == True] if not selected_songs.empty: st.markdown(f"**Selected {len(selected_songs)} songs:**") col1, col2 = st.columns(2) with col1: max_results = st.slider("Max Videos per Song", 10, 500, 100, step=10) with col2: include_media = st.checkbox("Include Thumbnails", value=False) if st.button("🚀 Start Tracking Selected Songs", type="primary", use_container_width=True): return { 'songs': selected_songs, 'max_results': max_results, 'include_thumbnails': include_media } return None def render_scraping_progress(self, current_step: str, progress: float = None): """Render scraping progress indicator""" st.markdown("### ⏳ Scraping Progress") steps = [ "🔍 Searching songs", "🎵 Validating music URLs", "🌐 Starting API scraping", "📥 Collecting video data", "💾 Storing to database", "📊 Calculating analytics", "✅ Complete" ] current_idx = next((i for i, step in enumerate(steps) if current_step.lower() in step.lower()), 0) if progress is not None: st.progress(progress) for i, step in enumerate(steps): if i < current_idx: st.success(f"✅ {step}") elif i == current_idx: st.info(f"⏳ {step}") else: st.empty() # ================================================================= # INSIGHTS AND ALERTS COMPONENTS # ================================================================= def render_insights_panel(self, insights: List[Dict[str, Any]]): """Render insights and recommendations panel""" if not insights: st.info("No insights available yet. Add some tracked songs to see insights!") return st.markdown("### 💡 Key Insights") # Group insights by importance high_importance = [i for i in insights if i.get('importance') == 'high'] medium_importance = [i for i in insights if i.get('importance') == 'medium'] low_importance = [i for i in insights if i.get('importance') == 'low'] # Render high importance insights first for insight in high_importance: self._render_insight_card(insight, 'error') # Red for high importance for insight in medium_importance: self._render_insight_card(insight, 'warning') # Yellow for medium for insight in low_importance: self._render_insight_card(insight, 'info') # Blue for low def render_alerts_panel(self, analytics: Dict[str, Any]): """Render system alerts and notifications""" st.markdown("### 🚨 System Alerts") alerts = [] # Check for data freshness if 'calculated_at' in analytics: calc_time = datetime.fromisoformat(analytics['calculated_at'].replace('Z', '+00:00')) hours_old = (datetime.now() - calc_time).total_seconds() / 3600 if hours_old > 24: alerts.append({ 'type': 'warning', 'title': 'Stale Data', 'message': f'Analytics data is {hours_old:.1f} hours old. Consider refreshing.' }) # Check for low video counts if analytics.get('total_videos', 0) < 5: alerts.append({ 'type': 'info', 'title': 'Low Video Count', 'message': 'Very few videos found. Consider expanding search criteria or waiting for more content.' }) # Check for high viral rate (good news!) if analytics.get('viral_percentage', 0) > 25: alerts.append({ 'type': 'success', 'title': 'High Viral Rate', 'message': f"{analytics['viral_percentage']:.1f}% viral rate detected! This song is trending." }) if not alerts: st.success("✅ All systems normal") else: for alert in alerts: getattr(st, alert['type'])(f"**{alert['title']}**: {alert['message']}") # ================================================================= # SETTINGS AND CONFIGURATION # ================================================================= def render_settings_panel(self) -> Dict[str, Any]: """Render application settings panel""" st.markdown("### ⚙️ Application Settings") with st.expander("🔑 API Configuration"): apify_token = st.text_input( "Apify API Token", type="password", help="Your Apify token for TikTok scraping", key="apify_token" ) test_api = st.button("Test API Connection") if test_api and apify_token: from api_module import api_client result = api_client.test_api_connection(apify_token) if result.get('success'): st.success(f"✅ {result['message']}") if 'monthly_usage' in result: usage = result['monthly_usage'] st.info(f"Monthly usage: {usage.get('requests', 0)} requests") else: st.error(f"❌ {result.get('error', 'Connection failed')}") with st.expander("📊 Analytics Settings"): viral_threshold = st.number_input( "Viral Threshold (Likes)", min_value=1000, max_value=100000, value=10000, step=1000, help="Minimum likes to consider a video viral" ) refresh_interval = st.selectbox( "Data Refresh Interval", options=["1 hour", "6 hours", "12 hours", "24 hours"], index=3 ) cache_duration = st.slider( "Cache Duration (minutes)", min_value=5, max_value=60, value=15, help="How long to cache database queries" ) with st.expander("🎨 Display Settings"): theme = st.selectbox( "Color Theme", options=["Default", "Dark", "Light", "Custom"], index=0 ) charts_per_row = st.slider( "Charts per Row", min_value=1, max_value=3, value=2 ) show_raw_data = st.checkbox( "Show Raw Data Tables", value=False, help="Display raw data tables for debugging" ) return { 'apify_token': apify_token, 'viral_threshold': viral_threshold, 'refresh_interval': refresh_interval, 'cache_duration': cache_duration, 'theme': theme, 'charts_per_row': charts_per_row, 'show_raw_data': show_raw_data } def render_database_stats_panel(self, stats: Dict[str, Any]): """Render database statistics panel""" st.markdown("### 📈 Database Statistics") if not stats: st.warning("Unable to load database statistics") return col1, col2 = st.columns(2) with col1: st.markdown("#### 📊 Chartmetric Data") chartmetric = stats.get('chartmetric', {}) st.metric("Total Tracks", chartmetric.get('TOTAL_TRACKS', 0)) st.metric("Unique Artists", chartmetric.get('UNIQUE_ARTISTS', 0)) st.metric("Active Tracks", chartmetric.get('ACTIVE_TRACKS', 0)) with col2: st.markdown("#### 🎵 POC Data") st.metric("Tracked Songs", stats.get('tracked_songs', 0)) videos = stats.get('videos', {}) st.metric("Total Videos", videos.get('TOTAL_VIDEOS', 0)) st.metric("Songs with Videos", videos.get('SONGS_WITH_VIDEOS', 0)) analytics = stats.get('analytics', {}) st.metric("Analytics Records", analytics.get('TOTAL_ANALYTICS', 0)) # ================================================================= # UTILITY METHODS # ================================================================= def _render_metric_card(self, title: str, value: Any, icon: str): """Render a metric card""" formatted_value = f"{value:,}" if isinstance(value, (int, float)) else str(value) st.markdown(f"""
{icon}
{formatted_value}
{title}
""", unsafe_allow_html=True) def _render_hashtag_card(self, hashtag_data: Dict[str, Any]): """Render a hashtag card""" hashtag = hashtag_data.get('hashtag', '') count = hashtag_data.get('count', 0) percentage = hashtag_data.get('percentage', 0) st.markdown(f"""
#{hashtag}
{count} videos ({percentage}%)
""", unsafe_allow_html=True) def _render_insight_card(self, insight: Dict[str, Any], alert_type: str): """Render an insight card""" title = insight.get('title', 'Insight') message = insight.get('message', '') getattr(st, alert_type)(f"**{title}**: {message}") def _format_status(self, status: str) -> str: """Format scrape status for display""" status_map = { 'NEVER_SCRAPED': '🆕 Never Scraped', 'DUE_FOR_SCRAPE': '⏰ Due for Scrape', 'UP_TO_DATE': '✅ Up to Date' } return status_map.get(status, status) # Global instance for use across the application dashboard = DashboardModule()