|
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 |
|
|
|
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): |
|
self.data = self._load_data(data_path) |
|
|
|
@staticmethod |
|
def _load_data(data_path: Path) -> pd.DataFrame: |
|
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 calculate_metrics(self) -> pd.DataFrame: |
|
self.data['Sentiment'] = self.data['Text'].apply(self._get_sentiment) |
|
self.data['Popularity'] = self._normalize_popularity() |
|
self.data['Credibility'] = np.random.choice([0, 1], size=len(self.data), p=[0.3, 0.7]) |
|
return self.data |
|
|
|
@staticmethod |
|
def _get_sentiment(text: str) -> float: |
|
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: |
|
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): |
|
self.data = self.preprocessor.calculate_metrics() |
|
|
|
def get_recommendations(self, weights: RecommendationWeights, num_recommendations: int = 10) -> Dict: |
|
if not self._validate_weights(weights): |
|
return {"error": "Invalid weights provided"} |
|
|
|
normalized_weights = self._normalize_weights(weights) |
|
|
|
self.data['Final_Score'] = ( |
|
self.data['Credibility'] * normalized_weights.visibility + |
|
self.data['Sentiment'] * normalized_weights.sentiment + |
|
self.data['Popularity'] * normalized_weights.popularity |
|
) |
|
|
|
top_recommendations = ( |
|
self.data.nlargest(num_recommendations, 'Final_Score') |
|
) |
|
|
|
return self._format_recommendations(top_recommendations) |
|
|
|
def _format_recommendations(self, recommendations: pd.DataFrame) -> Dict: |
|
formatted_results = [] |
|
for _, row in recommendations.iterrows(): |
|
score_details = { |
|
"总分": f"{row['Final_Score']:.2f}", |
|
"可信度": "可信" if row['Credibility'] > 0 else "存疑", |
|
"情感倾向": self._get_sentiment_label(row['Sentiment']), |
|
"热度": f"{row['Popularity']:.2f}", |
|
"互动数": f"点赞 {row['Likes']} · 转发 {row['Retweets']}" |
|
} |
|
|
|
formatted_results.append({ |
|
"text": row['Text'], |
|
"scores": score_details |
|
}) |
|
|
|
return { |
|
"recommendations": formatted_results, |
|
"score_explanation": self._get_score_explanation() |
|
} |
|
|
|
@staticmethod |
|
def _get_sentiment_label(sentiment_score: float) -> str: |
|
if sentiment_score > 0.3: |
|
return "积极" |
|
elif sentiment_score < -0.3: |
|
return "消极" |
|
return "中性" |
|
|
|
@staticmethod |
|
def _validate_weights(weights: RecommendationWeights) -> bool: |
|
return all(getattr(weights, field) >= 0 for field in weights.__dataclass_fields__) |
|
|
|
@staticmethod |
|
def _normalize_weights(weights: RecommendationWeights) -> RecommendationWeights: |
|
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]: |
|
return { |
|
"可信度": "内容可信度评估", |
|
"情感倾向": "文本的情感分析结果", |
|
"热度": "基于点赞和转发的热度分数" |
|
} |
|
|
|
|
|
def create_gradio_interface(recommendation_system: RecommendationSystem) -> gr.Interface: |
|
with gr.Blocks(theme=gr.themes.Soft()) as interface: |
|
gr.Markdown(""" |
|
# 推文推荐系统 |
|
调整下方的权重来获取个性化推荐： |
|
""") |
|
|
|
with gr.Row(): |
|
with gr.Column(scale=1): |
|
visibility_weight = gr.Slider( |
|
0, 1, 0.5, |
|
label="可信度权重", |
|
info="调整对内容可信度的重视程度" |
|
) |
|
sentiment_weight = gr.Slider( |
|
0, 1, 0.3, |
|
label="情感倾向权重", |
|
info="调整对情感倾向的重视程度" |
|
) |
|
popularity_weight = gr.Slider( |
|
0, 1, 0.2, |
|
label="热度权重", |
|
info="调整对内容热度的重视程度" |
|
) |
|
submit_btn = gr.Button("获取推荐", variant="primary") |
|
|
|
with gr.Column(scale=2): |
|
results = gr.Dataframe( |
|
headers=["推文内容", "评分详情"], |
|
label="推荐结果" |
|
) |
|
|
|
def format_for_display(recommendations): |
|
rows = [] |
|
for rec in recommendations["recommendations"]: |
|
scores = rec["scores"] |
|
score_text = ( |
|
f"总分: {scores['总分']}\n" |
|
f"可信度: {scores['可信度']}\n" |
|
f"情感倾向: {scores['情感倾向']}\n" |
|
f"热度: {scores['热度']}\n" |
|
f"互动: {scores['互动数']}" |
|
) |
|
rows.append([rec["text"], score_text]) |
|
return rows |
|
|
|
submit_btn.click( |
|
fn=lambda v, s, p: format_for_display( |
|
recommendation_system.get_recommendations(RecommendationWeights(v, s, p)) |
|
), |
|
inputs=[visibility_weight, sentiment_weight, popularity_weight], |
|
outputs=results |
|
) |
|
|
|
return interface |
|
|
|
def main(): |
|
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() |
