""" Data processing utilities for Spark POC testing Handles video data processing and analytics calculations """ import pandas as pd from datetime import datetime from collections import defaultdict def process_video_data(raw_videos): """Process raw video data from API into structured format""" try: if not raw_videos: return pd.DataFrame() # Convert to DataFrame if it's a list of dicts if isinstance(raw_videos, list): df = pd.DataFrame(raw_videos) else: df = raw_videos.copy() # Ensure required columns exist with default values required_columns = { 'video_id': 'unknown', 'creator': 'unknown_creator', 'views': 0, 'likes': 0, 'shares': 0, 'hearts': 0, 'url': '' } for col, default_value in required_columns.items(): if col not in df.columns: df[col] = default_value # Clean and validate data df['views'] = pd.to_numeric(df['views'], errors='coerce').fillna(0).astype(int) df['likes'] = pd.to_numeric(df['likes'], errors='coerce').fillna(0).astype(int) df['shares'] = pd.to_numeric(df['shares'], errors='coerce').fillna(0).astype(int) df['hearts'] = pd.to_numeric(df['hearts'], errors='coerce').fillna(0).astype(int) # Add processing timestamp df['processed_at'] = datetime.now().isoformat() return df except Exception as e: print(f"Error processing video data: {e}") return pd.DataFrame() def calculate_analytics(video_df): """Calculate aggregate analytics from video data""" try: if video_df.empty: return { 'total_videos': 0, 'total_views': 0, 'total_likes': 0, 'total_shares': 0, 'total_hearts': 0, 'avg_views': 0, 'avg_likes': 0, 'top_video_views': 0, 'top_creator': None } # Basic aggregations analytics = { 'total_videos': len(video_df), 'total_views': int(video_df['views'].sum()), 'total_likes': int(video_df['likes'].sum()), 'total_shares': int(video_df['shares'].sum()), 'total_hearts': int(video_df['hearts'].sum()), } # Averages analytics['avg_views'] = int(video_df['views'].mean()) if len(video_df) > 0 else 0 analytics['avg_likes'] = int(video_df['likes'].mean()) if len(video_df) > 0 else 0 # Top performers if not video_df.empty: top_video = video_df.loc[video_df['views'].idxmax()] analytics['top_video_views'] = int(top_video['views']) analytics['top_video_id'] = top_video['video_id'] # Top creator by total views creator_views = video_df.groupby('creator')['views'].sum() analytics['top_creator'] = creator_views.idxmax() analytics['top_creator_total_views'] = int(creator_views.max()) return analytics except Exception as e: print(f"Error calculating analytics: {e}") return {'error': str(e)} def format_top_performers(video_df, metric='views', limit=5): """Format top performing videos or creators""" try: if video_df.empty: return pd.DataFrame() # Sort by specified metric if metric not in video_df.columns: metric = 'views' # Default fallback top_videos = video_df.nlargest(limit, metric)[ ['video_id', 'creator', 'views', 'likes', 'shares', 'url'] ].copy() # Add rank top_videos['rank'] = range(1, len(top_videos) + 1) return top_videos except Exception as e: print(f"Error formatting top performers: {e}") return pd.DataFrame() def get_creator_analytics(video_df): """Get analytics grouped by creator""" try: if video_df.empty: return pd.DataFrame() creator_stats = video_df.groupby('creator').agg({ 'video_id': 'count', 'views': ['sum', 'mean', 'max'], 'likes': ['sum', 'mean'], 'shares': ['sum', 'mean'] }).round(0) # Flatten column names creator_stats.columns = [ 'total_videos', 'total_views', 'avg_views', 'max_views', 'total_likes', 'avg_likes', 'total_shares', 'avg_shares' ] # Sort by total views creator_stats = creator_stats.sort_values('total_views', ascending=False) # Reset index to make creator a column creator_stats = creator_stats.reset_index() return creator_stats except Exception as e: print(f"Error calculating creator analytics: {e}") return pd.DataFrame() def detect_viral_videos(video_df, view_threshold=50000, like_ratio_threshold=0.05): """Identify potentially viral videos based on engagement metrics""" try: if video_df.empty: return pd.DataFrame() # Calculate engagement ratios df = video_df.copy() df['like_ratio'] = df['likes'] / df['views'].replace(0, 1) # Avoid division by zero df['share_ratio'] = df['shares'] / df['views'].replace(0, 1) # Viral criteria viral_mask = ( (df['views'] >= view_threshold) & (df['like_ratio'] >= like_ratio_threshold) ) viral_videos = df[viral_mask].copy() viral_videos['viral_score'] = ( viral_videos['like_ratio'] * 0.6 + viral_videos['share_ratio'] * 0.4 ) return viral_videos.sort_values('viral_score', ascending=False) except Exception as e: print(f"Error detecting viral videos: {e}") return pd.DataFrame() def generate_insights(analytics, video_df): """Generate insights from analytics data""" try: insights = [] if analytics.get('total_videos', 0) == 0: insights.append("No video data available for analysis") return insights # Video volume insights total_videos = analytics['total_videos'] if total_videos > 100: insights.append(f"🔥 High engagement: {total_videos} videos found using this song") elif total_videos > 50: insights.append(f"📈 Good traction: {total_videos} videos using this song") else: insights.append(f"📊 Moderate usage: {total_videos} videos found") # View insights total_views = analytics['total_views'] avg_views = analytics['avg_views'] if avg_views > 10000: insights.append(f"⭐ Strong performance: Average {avg_views:,} views per video") elif avg_views > 1000: insights.append(f"👍 Decent performance: Average {avg_views:,} views per video") # Top performer insights if 'top_video_views' in analytics and analytics['top_video_views'] > 0: top_views = analytics['top_video_views'] if top_views > 100000: insights.append(f"🚀 Viral potential: Top video has {top_views:,} views") elif top_views > 10000: insights.append(f"📊 Good reach: Top video has {top_views:,} views") # Creator diversity if not video_df.empty: unique_creators = video_df['creator'].nunique() if unique_creators > 50: insights.append(f"🌍 Broad appeal: {unique_creators} different creators") elif unique_creators > 20: insights.append(f"📢 Good spread: {unique_creators} different creators") return insights except Exception as e: return [f"Error generating insights: {str(e)}"] def export_analytics_summary(analytics, video_df, top_videos, top_creators): """Export comprehensive analytics summary""" try: summary = { 'generated_at': datetime.now().isoformat(), 'overview': analytics, 'insights': generate_insights(analytics, video_df), 'top_videos': top_videos.to_dict('records') if not top_videos.empty else [], 'top_creators': top_creators.to_dict('records') if not top_creators.empty else [], 'data_quality': { 'total_records': len(video_df), 'valid_view_counts': (video_df['views'] > 0).sum() if not video_df.empty else 0, 'unique_creators': video_df['creator'].nunique() if not video_df.empty else 0, 'date_range': { 'earliest': video_df['processed_at'].min() if not video_df.empty else None, 'latest': video_df['processed_at'].max() if not video_df.empty else None } } } return summary except Exception as e: return {'error': f'Failed to export summary: {str(e)}'} def validate_data_quality(video_df): """Validate data quality and identify issues""" try: if video_df.empty: return {'status': 'empty', 'issues': ['No data provided']} issues = [] # Check for missing values missing_cols = video_df.isnull().sum() for col, count in missing_cols[missing_cols > 0].items(): issues.append(f"{col}: {count} missing values") # Check for invalid view counts invalid_views = (video_df['views'] < 0).sum() if invalid_views > 0: issues.append(f"{invalid_views} videos with negative view counts") # Check for duplicate videos duplicates = video_df['video_id'].duplicated().sum() if duplicates > 0: issues.append(f"{duplicates} duplicate video IDs found") # Check for unrealistic engagement ratios if not video_df.empty: high_like_ratio = (video_df['likes'] > video_df['views']).sum() if high_like_ratio > 0: issues.append(f"{high_like_ratio} videos with likes > views (unrealistic)") status = 'clean' if not issues else 'issues_found' return {'status': status, 'issues': issues, 'total_records': len(video_df)} except Exception as e: return {'status': 'error', 'issues': [f'Validation error: {str(e)}']}