import gradio as gr import pandas as pd import numpy as np from textblob import TextBlob from typing import List, Dict, Tuple from dataclasses import dataclass from pathlib import Path import logging import re from datetime import datetime logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class RecommendationWeights: visibility: float sentiment: float popularity: float class TweetPreprocessor: def __init__(self, data_path: Path): """Initialize the preprocessor with data path.""" self.data = self._load_data(data_path) @staticmethod def _load_data(data_path: Path) -> pd.DataFrame: """Load and validate the dataset.""" try: data = pd.read_csv(data_path) required_columns = {'Text', 'Retweets', 'Likes'} if not required_columns.issubset(data.columns): raise ValueError(f"Missing required columns: {required_columns - set(data.columns)}") return data except Exception as e: logger.error(f"Error loading data: {e}") raise def _clean_text(self, text: str) -> str: """Clean text content.""" if pd.isna(text) or len(str(text).strip()) < 10: return "" text = re.sub(r'http\S+|www.\S+', '', str(text)) text = re.sub(r'[^\w\s]', '', text) text = ' '.join(text.split()) return text def calculate_metrics(self) -> pd.DataFrame: """Calculate all metrics for tweets.""" self.data['Clean_Text'] = self.data['Text'].apply(self._clean_text) self.data = self.data[self.data['Clean_Text'].str.len() > 0] self.data['Sentiment'] = self.data['Clean_Text'].apply(self._get_sentiment) self.data['Popularity'] = self._normalize_popularity() return self.data @staticmethod def _get_sentiment(text: str) -> float: """Calculate sentiment polarity for a text.""" try: return TextBlob(str(text)).sentiment.polarity except Exception as e: logger.warning(f"Error calculating sentiment: {e}") return 0.0 def _normalize_popularity(self) -> pd.Series: """Normalize popularity scores.""" popularity = self.data['Retweets'] + self.data['Likes'] return (popularity - popularity.min()) / (popularity.max() - popularity.min() + 1e-6) class RecommendationSystem: def __init__(self, data_path: Path): self.preprocessor = TweetPreprocessor(data_path) self.data = None self.setup_system() def setup_system(self): """Initialize the system with preprocessed data.""" self.data = self.preprocessor.calculate_metrics() def recalculate_scores(self, weights: RecommendationWeights): """Recalculate scores based on new weights.""" normalized_weights = self._normalize_weights(weights) self.data['Credibility'] = np.random.choice([0, 1], size=len(self.data), p=[0.3, 0.7]) self.data['Final_Score'] = ( self.data['Credibility'] * normalized_weights.visibility + self.data['Sentiment'] * normalized_weights.sentiment + self.data['Popularity'] * normalized_weights.popularity ) def get_recommendations(self, weights: RecommendationWeights, num_recommendations: int = 10) -> Dict: """Get tweet recommendations based on weights.""" if not self._validate_weights(weights): return {"error": "Invalid weights provided"} self.recalculate_scores(weights) top_recommendations = ( self.data.nlargest(num_recommendations, 'Final_Score') ) return self._format_recommendations(top_recommendations) def _format_recommendations(self, recommendations: pd.DataFrame) -> Dict: """Format recommendations for display.""" formatted_results = [] for _, row in recommendations.iterrows(): score_details = { "score": f"{row['Final_Score']:.2f}", "credibility": "Reliable" if row['Credibility'] > 0 else "Uncertain", "sentiment": self._get_sentiment_label(row['Sentiment']), "popularity": f"{row['Popularity']:.2f}", "engagement": f"Likes {row['Likes']} ยท Retweets {row['Retweets']}" } formatted_results.append({ "text": row['Clean_Text'], "scores": score_details }) return { "recommendations": formatted_results, "score_explanation": self._get_score_explanation() } @staticmethod def _get_sentiment_label(sentiment_score: float) -> str: """Convert sentiment score to label.""" if sentiment_score > 0.3: return "Positive" elif sentiment_score < -0.3: return "Negative" return "Neutral" @staticmethod def _validate_weights(weights: RecommendationWeights) -> bool: """Validate that weights are non-negative.""" return all(getattr(weights, field) >= 0 for field in weights.__dataclass_fields__) @staticmethod def _normalize_weights(weights: RecommendationWeights) -> RecommendationWeights: """Normalize weights to sum to 1.""" total = weights.visibility + weights.sentiment + weights.popularity if total == 0: return RecommendationWeights(1/3, 1/3, 1/3) return RecommendationWeights( visibility=weights.visibility / total, sentiment=weights.sentiment / total, popularity=weights.popularity / total ) @staticmethod def _get_score_explanation() -> Dict[str, str]: """Provide explanation for different score components.""" return { "Credibility": "Content reliability assessment", "Sentiment": "Text emotional analysis result", "Popularity": "Score based on likes and retweets" } def create_gradio_interface(recommendation_system: RecommendationSystem) -> gr.Interface: """Create and configure the Gradio interface.""" with gr.Blocks(theme=gr.themes.Soft()) as interface: gr.Markdown(""" # Tweet Recommendation System Adjust weights to get personalized recommendations Note: To protect user privacy, some tweet content has been redacted or anonymized. """) with gr.Row(): with gr.Column(scale=1): visibility_weight = gr.Slider(0, 1, 0.5, label="Credibility Weight", info="Adjust importance of content credibility") sentiment_weight = gr.Slider(0, 1, 0.3, label="Sentiment Weight", info="Adjust importance of emotional tone") popularity_weight = gr.Slider(0, 1, 0.2, label="Popularity Weight", info="Adjust importance of engagement metrics") submit_btn = gr.Button("Get Recommendations", variant="primary") with gr.Column(scale=2): output_html = gr.HTML() def format_recommendations(raw_recommendations): html = '
' html += '''

Score Guide

''' for i, rec in enumerate(raw_recommendations["recommendations"], 1): scores = rec["scores"] html += f'''
{rec["text"]}
Score: {scores["score"]} Credibility: {scores["credibility"]} Sentiment: {scores["sentiment"]} Popularity: {scores["popularity"]} Engagement: {scores["engagement"]}
''' html += '
' return html def get_recommendations_with_weights(v, s, p): """Get recommendations with current weights.""" weights = RecommendationWeights(v, s, p) return format_recommendations(recommendation_system.get_recommendations(weights)) submit_btn.click( fn=get_recommendations_with_weights, inputs=[visibility_weight, sentiment_weight, popularity_weight], outputs=output_html ) return interface def main(): """Main function to run the application.""" try: recommendation_system = RecommendationSystem( data_path=Path('twitter_dataset.csv') ) interface = create_gradio_interface(recommendation_system) interface.launch() except Exception as e: logger.error(f"Application failed to start: {e}") raise if __name__ == "__main__": main()