|
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 = '<div style="font-family: sans-serif;">' |
|
|
|
html += ''' |
|
<div style="margin-bottom: 20px; padding: 15px; background-color: #1a1a1a; color: white; border-radius: 8px;"> |
|
<h3 style="margin-top: 0;">Score Guide</h3> |
|
<ul style="margin: 0;"> |
|
<li><strong>Credibility</strong>: Assessment of content reliability</li> |
|
<li><strong>Sentiment</strong>: Text emotional analysis (Positive/Negative/Neutral)</li> |
|
<li><strong>Popularity</strong>: Normalized score based on likes and retweets</li> |
|
</ul> |
|
</div> |
|
''' |
|
|
|
for i, rec in enumerate(raw_recommendations["recommendations"], 1): |
|
scores = rec["scores"] |
|
html += f''' |
|
<div style="margin-bottom: 15px; padding: 15px; border: 1px solid #ddd; border-radius: 8px;"> |
|
<div style="margin-bottom: 10px; font-size: 1.1em;">{rec["text"]}</div> |
|
<div style="display: flex; flex-wrap: wrap; gap: 10px; font-size: 0.9em;"> |
|
<span style="padding: 3px 8px; background-color: #1976d2; color: white; border-radius: 4px;"> |
|
Score: {scores["score"]} |
|
</span> |
|
<span style="padding: 3px 8px; background-color: #2e7d32; color: white; border-radius: 4px;"> |
|
Credibility: {scores["credibility"]} |
|
</span> |
|
<span style="padding: 3px 8px; background-color: #ed6c02; color: white; border-radius: 4px;"> |
|
Sentiment: {scores["sentiment"]} |
|
</span> |
|
<span style="padding: 3px 8px; background-color: #d32f2f; color: white; border-radius: 4px;"> |
|
Popularity: {scores["popularity"]} |
|
</span> |
|
<span style="padding: 3px 8px; background-color: #7b1fa2; color: white; border-radius: 4px;"> |
|
Engagement: {scores["engagement"]} |
|
</span> |
|
</div> |
|
</div> |
|
''' |
|
html += '</div>' |
|
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() |
