Update app.py

app.py

@@ -1,17 +1,12 @@
 import gradio as gr
 import pandas as pd
 import numpy as np
-import torch
-from transformers import AutoTokenizer, AutoModelForSequenceClassification
 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
 
-# Configure logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
@@ -23,15 +18,13 @@ class RecommendationWeights:
 
 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', 'Timestamp'}  # 添加时间戳列
+            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
@@ -39,43 +32,14 @@ class TweetPreprocessor:
             logger.error(f"Error loading data: {e}")
             raise
 
-    def _clean_text(self, text: str) -> str:
-        """清理文本内容，移除无意义的内容"""
-        if pd.isna(text) or len(str(text).strip()) < 10:  # 排除过短或空的文本
-            return ""
-        
-        # 移除URL
-        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 sentiment and popularity metrics."""
-        # 清理文本
-        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['Sentiment'] = self.data['Text'].apply(self._get_sentiment)
         self.data['Popularity'] = self._normalize_popularity()
-        
-        # 添加时间衰减因子
-        self.data['Time_Weight'] = self._calculate_time_weight()
+        self.data['Credibility'] = np.random.choice([0, 1], size=len(self.data), p=[0.3, 0.7])
         return self.data
     
-    def _calculate_time_weight(self) -> pd.Series:
-        """计算时间权重，越新的内容权重越高"""
-        current_time = datetime.now()
-        self.data['Timestamp'] = pd.to_datetime(self.data['Timestamp'])
-        time_diff = (current_time - self.data['Timestamp']).dt.total_seconds()
-        return np.exp(-time_diff / (7 * 24 * 3600))  # 7天的衰减周期
-
     @staticmethod
     def _get_sentiment(text: str) -> float:
-        """Calculate sentiment polarity for a text."""
         try:
             return TextBlob(str(text)).sentiment.polarity
         except Exception as e:
@@ -83,63 +47,19 @@ class TweetPreprocessor:
             return 0.0
 
     def _normalize_popularity(self) -> pd.Series:
-        """Normalize popularity scores using min-max scaling."""
         popularity = self.data['Retweets'] + self.data['Likes']
         return (popularity - popularity.min()) / (popularity.max() - popularity.min() + 1e-6)
-class FakeNewsClassifier:
-    def __init__(self, model_name: str):
-        """Initialize the fake news classifier."""
-        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-        self.model_name = model_name
-        self.model, self.tokenizer = self._load_model()
-        
-    def _load_model(self) -> Tuple[AutoModelForSequenceClassification, AutoTokenizer]:
-        """Load the model and tokenizer."""
-        try:
-            tokenizer = AutoTokenizer.from_pretrained(self.model_name)
-            model = AutoModelForSequenceClassification.from_pretrained(self.model_name).to(self.device)
-            return model, tokenizer
-        except Exception as e:
-            logger.error(f"Error loading model: {e}")
-            raise
-
-    @torch.no_grad()
-    def predict_batch(self, texts: List[str], batch_size: int = 32) -> np.ndarray:
-        """Predict fake news probability for a batch of texts."""
-        predictions = []
-        
-        for i in range(0, len(texts), batch_size):
-            batch_texts = texts[i:i + batch_size]
-            inputs = self.tokenizer(
-                batch_texts,
-                return_tensors="pt",
-                padding=True,
-                truncation=True,
-                max_length=128
-            ).to(self.device)
-            
-            outputs = self.model(**inputs)
-            batch_predictions = outputs.logits.argmax(dim=1).cpu().numpy()
-            predictions.extend(batch_predictions)
-            
-        return np.array(predictions)     
 
 class RecommendationSystem:
-    def __init__(self, data_path: Path, model_name: str):
-        """Initialize the recommendation system."""
+    def __init__(self, data_path: Path):
         self.preprocessor = TweetPreprocessor(data_path)
-        self.classifier = FakeNewsClassifier(model_name)
         self.data = None
         self.setup_system()
 
     def setup_system(self):
-        """Set up the recommendation system."""
         self.data = self.preprocessor.calculate_metrics()
-        predictions = self.classifier.predict_batch(self.data['Clean_Text'].tolist())
-        self.data['Credibility'] = [1 if pred == 1 else -1 for pred in predictions]
 
     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"}
 
@@ -149,17 +69,15 @@ class RecommendationSystem:
             self.data['Credibility'] * normalized_weights.visibility +
             self.data['Sentiment'] * normalized_weights.sentiment +
             self.data['Popularity'] * normalized_weights.popularity
-        ) * self.data['Time_Weight']  # 考虑时间因素
+        )
 
         top_recommendations = (
-            self.data.nlargest(100, 'Final_Score')
-            .sample(num_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 = {
@@ -171,9 +89,8 @@ class RecommendationSystem:
             }
             
             formatted_results.append({
-                "text": row['Clean_Text'],
-                "scores": score_details,
-                "timestamp": row['Timestamp'].strftime("%Y-%m-%d %H:%M")
+                "text": row['Text'],
+                "scores": score_details
             })
             
         return {
@@ -183,34 +100,40 @@ class RecommendationSystem:
 
     @staticmethod
     def _get_sentiment_label(sentiment_score: float) -> str:
-        """Convert sentiment score to human-readable label."""
         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]:
-        """Provide explanation for different score components."""
         return {
-            "可信度": "基于机器学习模型对内容可信度的评估",
-            "情感倾向": "文本的情感倾向分析结果",
-            "热度": "根据点赞和转发数量计算的归一化热度分数",
-            "时间权重": "考虑内容时效性的权重因子"
+            "可信度": "内容可信度评估",
+            "情感倾向": "文本的情感分析结果",
+            "热度": "基于点赞和转发的热度分数"
         }
 
 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("""
         # 推文推荐系统
-        
-        这个系统通过多个维度来为您推荐高质量的推文:
-        - **可信度**: 评估内容的可靠性
-        - **情感倾向**: 分析文本的情感色彩
-        - **热度**: 考虑内容的受欢迎程度
-        - **时效性**: 优先推荐较新的内容
+        调整下方的权重来获取个性化推荐：
         """)
         
         with gr.Row():
@@ -246,11 +169,9 @@ def create_gradio_interface(recommendation_system: RecommendationSystem) -> gr.I
     return interface
 
 def main():
-    """Main function to run the application."""
     try:
         recommendation_system = RecommendationSystem(
-            data_path=Path('twitter_dataset.csv'),
-            model_name="hamzab/roberta-fake-news-classification"
+            data_path=Path('twitter_dataset.csv')
         )
         interface = create_gradio_interface(recommendation_system)
         interface.launch()
@@ -259,4 +180,4 @@ def main():
         raise
 
 if __name__ == "__main__":
-    main() 
+    main()