""" Bot interaction feature extraction for user clustering. """ import pandas as pd import numpy as np from typing import Dict, List, Optional, Tuple, Any import gc class BotFeatureExtractor: """Extract features from bot interaction data.""" def __init__(self): self.default_bot_features = { "total_bot_actions": 0, "bot_action_rate": 0, "api_usage_tier": 0, "download_audio_count": 0, "download_video_count": 0, "download_audio_wav_count": 0, "share_count": 0, "total_downloads": 0, "download_diversity": 0, "audio_preference_ratio": 0, "creator_collector_score": 0, "creator_promoter_score": 0, } def extract_features( self, bots_action_df: pd.DataFrame, total_clip_df: pd.DataFrame, user_ids: pd.Series, features_df: pd.DataFrame, verbose: bool = True, ) -> Tuple[pd.DataFrame, Dict[str, Any]]: """ Extract bot interaction features for specified users. Args: bots_action_df: DataFrame containing bot action data total_clip_df: DataFrame containing clip data for mapping user_ids: Series of user IDs to process features_df: Main features dataframe (for total_clips_created) verbose: Whether to print progress messages Returns: Tuple of (DataFrame with bot interaction features, summary dict) """ if verbose: print(" Checking clip ID mapping...") # Determine clip ID columns clip_id_col = self._get_clip_id_column(total_clip_df) bot_action_clip_col = self._get_bot_action_clip_column(bots_action_df) if not clip_id_col or not bot_action_clip_col or len(bots_action_df) == 0: if verbose: print(" ⚠️ Cannot map bot actions to users (missing required columns)") return self._create_default_features(user_ids), {"bot_users": 0} try: # Create clip to user mapping clip_map = self._create_clip_user_mapping( total_clip_df, user_ids, clip_id_col ) # Map bot actions to users user_bot_actions = self._map_bot_actions_to_users( bots_action_df, bot_action_clip_col, clip_map ) if len(user_bot_actions) == 0: if verbose: print(" ⚠️ No bot actions found for filtered users") return self._create_default_features(user_ids), {"bot_users": 0} # Extract bot statistics bot_stats = self._extract_bot_statistics(user_bot_actions, verbose) # Add creator statistics for bot action rate if "total_clips_created" in features_df.columns: creator_stats = features_df[features_df["total_clips_created"] > 0][ ["user_id", "total_clips_created"] ] bot_stats = bot_stats.merge(creator_stats, on="user_id", how="left") # Fill NaN values immediately after merge bot_stats["total_clips_created"] = bot_stats[ "total_clips_created" ].fillna(0) # Use np.where to avoid division by zero and NaN bot_stats["bot_action_rate"] = np.where( bot_stats["total_clips_created"] > 0, bot_stats["total_bot_actions"] / bot_stats["total_clips_created"], 0, ).round(2) # Add API usage tiers bot_stats = self._add_api_usage_tiers(bot_stats) # Add creator behavior scores bot_stats = self._add_creator_behavior_scores(bot_stats) # Create summary before printing summary = self._create_summary(user_bot_actions, bot_stats) if verbose: self._print_summary(user_bot_actions, bot_stats) return bot_stats, summary except Exception as e: if verbose: print(f" ⚠️ Error processing bot actions: {e}") return self._create_default_features(user_ids), { "bot_users": 0, "error": str(e), } def _get_clip_id_column(self, total_clip_df: pd.DataFrame) -> Optional[str]: """Determine the clip ID column name.""" if "id" in total_clip_df.columns: return "id" elif "clip_id" in total_clip_df.columns: return "clip_id" return None def _get_bot_action_clip_column( self, bots_action_df: pd.DataFrame ) -> Optional[str]: """Determine the bot action clip ID column name.""" if "clip_id" in bots_action_df.columns: return "clip_id" elif "id" in bots_action_df.columns: return "id" return None def _create_clip_user_mapping( self, total_clip_df: pd.DataFrame, user_ids: pd.Series, clip_id_col: str ) -> pd.DataFrame: """Create mapping from clip IDs to user IDs.""" clip_map = total_clip_df[total_clip_df["user_id"].isin(user_ids)][ [clip_id_col, "user_id"] ].copy() if clip_id_col != "clip_id": clip_map.rename(columns={clip_id_col: "clip_id"}, inplace=True) return clip_map def _map_bot_actions_to_users( self, bots_action_df: pd.DataFrame, bot_action_clip_col: str, clip_map: pd.DataFrame, ) -> pd.DataFrame: """Map bot actions to users via clip IDs.""" if bot_action_clip_col != "clip_id": bots_action_df_renamed = bots_action_df.rename( columns={bot_action_clip_col: "clip_id"} ) else: bots_action_df_renamed = bots_action_df return bots_action_df_renamed.merge(clip_map, on="clip_id", how="inner") def _extract_bot_statistics( self, user_bot_actions: pd.DataFrame, verbose: bool ) -> pd.DataFrame: """Extract bot action statistics per user.""" # Check available columns available_cols = set(user_bot_actions.columns) if verbose: print( f" Available bot action columns: {sorted(list(available_cols))[:10]}..." ) # Build aggregation dictionary agg_dict = {"clip_id": "count"} # Total actions # Numeric columns to aggregate potential_cols = [ "download_audio_count", "download_video_count", "download_audio_wav_count", "share_count", "download_count", "play_count", "like_count", ] for col in potential_cols: if col in available_cols: agg_dict[col] = ["sum", "mean"] # Perform aggregation bot_stats = user_bot_actions.groupby("user_id", as_index=False).agg(agg_dict) # Flatten column names bot_stats = self._flatten_column_names(bot_stats) # Process aggregated columns bot_stats = self._process_aggregated_columns(bot_stats, potential_cols) # Calculate derived features bot_stats = self._calculate_derived_features(bot_stats) return bot_stats def _flatten_column_names(self, df: pd.DataFrame) -> pd.DataFrame: """Flatten multi-level column names.""" new_cols = [] for col in df.columns: if isinstance(col, tuple) and col[1]: new_cols.append(f"{col[0]}_{col[1]}") elif isinstance(col, tuple): new_cols.append(col[0]) else: new_cols.append(col) df.columns = new_cols df.rename(columns={"clip_id_count": "total_bot_actions"}, inplace=True) return df def _process_aggregated_columns( self, bot_stats: pd.DataFrame, potential_cols: List[str] ) -> pd.DataFrame: """Process aggregated columns, keeping only sum values.""" for col in potential_cols: sum_col = f"{col}_sum" mean_col = f"{col}_mean" if sum_col in bot_stats.columns: bot_stats[col] = bot_stats[sum_col].copy() # Drop intermediate columns bot_stats.drop([sum_col, mean_col], axis=1, errors="ignore") return bot_stats def _calculate_derived_features(self, bot_stats: pd.DataFrame) -> pd.DataFrame: """Calculate derived bot interaction features.""" # Total downloads download_cols = [ "download_audio_count", "download_video_count", "download_audio_wav_count", ] existing_download_cols = [ col for col in download_cols if col in bot_stats.columns ] if existing_download_cols: bot_stats["total_downloads"] = bot_stats[existing_download_cols].sum(axis=1) # Download diversity bot_stats["download_diversity"] = ( bot_stats[existing_download_cols] > 0 ).sum(axis=1) # Audio preference ratio - use np.where to avoid NaN audio_cols = ["download_audio_count", "download_audio_wav_count"] audio_cols = [col for col in audio_cols if col in bot_stats.columns] if audio_cols and "total_downloads" in bot_stats.columns: audio_downloads = bot_stats[audio_cols].sum(axis=1) bot_stats["audio_preference_ratio"] = np.where( bot_stats["total_downloads"] > 0, audio_downloads / bot_stats["total_downloads"], 0, ).round(3) # Share engagement ratio - use np.where to avoid NaN if "share_count" in bot_stats.columns: bot_stats["share_engagement_ratio"] = np.where( bot_stats["total_bot_actions"] > 0, bot_stats["share_count"] / bot_stats["total_bot_actions"], 0, ).round(3) return bot_stats def _add_api_usage_tiers(self, bot_stats: pd.DataFrame) -> pd.DataFrame: """Add API usage tier categorization.""" bot_stats["api_usage_tier"] = pd.cut( bot_stats["total_bot_actions"], bins=[0, 5, 20, 50, 100, float("inf")], labels=[0, 1, 2, 3, 4], # 0=minimal to 4=extreme include_lowest=True, ).astype(int) return bot_stats def _add_creator_behavior_scores(self, bot_stats: pd.DataFrame) -> pd.DataFrame: """Add creator behavior scores.""" # Creator collector score - use np.where to avoid NaN if "total_downloads" in bot_stats.columns: bot_stats["creator_collector_score"] = np.where( bot_stats["total_bot_actions"] > 0, bot_stats["total_downloads"] / bot_stats["total_bot_actions"], 0, ).round(3) else: bot_stats["creator_collector_score"] = 0 # Creator promoter score - use np.where to avoid NaN if "share_count" in bot_stats.columns: bot_stats["creator_promoter_score"] = np.where( bot_stats["total_bot_actions"] > 0, bot_stats["share_count"] / bot_stats["total_bot_actions"], 0, ).round(3) else: bot_stats["creator_promoter_score"] = 0 return bot_stats def _create_default_features(self, user_ids: pd.Series) -> pd.DataFrame: """Create default bot features for users without bot actions.""" default_df = pd.DataFrame({"user_id": user_ids}) for col, default_val in self.default_bot_features.items(): default_df[col] = default_val return default_df def _print_summary( self, user_bot_actions: pd.DataFrame, bot_stats: pd.DataFrame ) -> None: """Print summary statistics.""" bot_users = len(bot_stats) print( f"📊 Found: {len(user_bot_actions):,} bot actions from {bot_users:,} users" ) if "total_downloads" in bot_stats.columns: download_users = (bot_stats["total_downloads"] > 0).sum() avg_downloads = ( bot_stats[bot_stats["total_downloads"] > 0]["total_downloads"].mean() if download_users > 0 else 0 ) print( f"📊 Downloads: {download_users:,} users | Avg: {avg_downloads:.1f}/user" ) if "share_count" in bot_stats.columns: share_users = (bot_stats["share_count"] > 0).sum() avg_shares = ( bot_stats[bot_stats["share_count"] > 0]["share_count"].mean() if share_users > 0 else 0 ) print(f"📊 Shares: {share_users:,} users | Avg: {avg_shares:.1f}/user") def merge_features( self, features_df: pd.DataFrame, bot_features: pd.DataFrame ) -> pd.DataFrame: """Merge bot features into main features dataframe.""" # Get features to merge (exclude user_id and already existing) bot_features_to_use = [] for col in self.default_bot_features.keys(): if col in bot_features.columns and col not in features_df.columns: bot_features_to_use.append(col) if not bot_features_to_use: print(" ⚠️ Bot features already exist in dataframe, skipping merge") return features_df # Merge with main features features_df = features_df.merge( bot_features[["user_id"] + bot_features_to_use], on="user_id", how="left" ) # Fill missing values for all bot features for col in bot_features_to_use: if col in features_df.columns: features_df[col] = features_df[col].fillna( self.default_bot_features.get(col, 0) ) return features_df def _create_summary( self, user_bot_actions: pd.DataFrame, bot_stats: pd.DataFrame ) -> Dict[str, Any]: """Create summary statistics for bot features.""" summary = { "bot_users": len(bot_stats), "total_bot_actions": len(user_bot_actions), "avg_bot_actions_per_user": bot_stats["total_bot_actions"].mean() if len(bot_stats) > 0 else 0, } # Add download statistics if "total_downloads" in bot_stats.columns: summary["download_users"] = (bot_stats["total_downloads"] > 0).sum() summary["avg_downloads_per_user"] = ( bot_stats[bot_stats["total_downloads"] > 0]["total_downloads"].mean() if summary["download_users"] > 0 else 0 ) # Add share statistics if "share_count" in bot_stats.columns: summary["share_users"] = (bot_stats["share_count"] > 0).sum() summary["avg_shares_per_user"] = ( bot_stats[bot_stats["share_count"] > 0]["share_count"].mean() if summary["share_users"] > 0 else 0 ) # Add API usage tier distribution if "api_usage_tier" in bot_stats.columns: tier_counts = bot_stats["api_usage_tier"].value_counts().to_dict() summary["api_tier_distribution"] = tier_counts summary["power_users"] = bot_stats[bot_stats["api_usage_tier"] >= 3].shape[ 0 ] return summary