File size: 9,811 Bytes
f5b7e31
 
 
 
 
 
 
 
eabeab3
f5b7e31
 
 
0f2373c
 
f5b7e31
 
 
 
 
 
 
 
 
 
 
 
 
 
86fdee3
f5b7e31
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
86fdee3
 
 
f5b7e31
86fdee3
f5b7e31
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0f2373c
f5b7e31
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0f2373c
 
 
 
 
 
 
 
 
 
 
 
 
 
f5b7e31
0f2373c
f5b7e31
 
 
 
0f2373c
f5b7e31
 
0f2373c
f5b7e31
 
 
 
 
0f2373c
 
f5b7e31
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0f2373c
 
f5b7e31
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
86fdee3
0f2373c
 
f5b7e31
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0f2373c
f5b7e31
0f2373c
f5b7e31
eabeab3
 
 
 
 
f5b7e31
 
 
eabeab3
f5b7e31
 
 
 
 
eabeab3
f5b7e31
 
 
 
 
0f2373c
f5b7e31
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
"""
AI News API Handler
Fetches AI-related news from NewsAPI and performs sentiment analysis
"""
import requests
import pandas as pd
from datetime import datetime, timedelta
import os
import json
from dotenv import load_dotenv
from textblob import TextBlob
from typing import List, Dict, Optional
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer as SIA


# Load environment variables
load_dotenv()

class AINewsAnalyzer:
    def __init__(self):
        self.api_key = os.getenv('NEWSAPI_KEY')
        self.base_url = "https://newsapi.org/v2/everything"
        
        if not self.api_key:
            raise ValueError("NewsAPI key not found. Please set NEWSAPI_KEY in your .env file")
    
    def fetch_ai_news(self, 
                      query: str = "artificial intelligence", 
                      days: tuple[int] = (7,14), 
                      language: str = "en",
                      sources: Optional[str] = None,
                      page_size: int = 100) -> List[Dict]:
        """
        Fetch AI-related news from NewsAPI
        
        Args:
            query: Search query for news articles
            days: Number of days to look back
            language: Language code (default: "en")
            sources: Comma-separated string of news sources
            page_size: Number of articles to fetch (max 100)
            
        Returns:
            List of news articles with metadata
        """
        # Calculate date range
        today = datetime.now()
        from_date = today - timedelta(days=days[0]) # 7 
        to_date = today - timedelta(days=days[1]) # 14
        
        print(from_date, to_date)
        # Prepare API parameters
        params = {
            'q': query,
            'from': from_date.strftime('%Y-%m-%d'),
            'to': to_date.strftime('%Y-%m-%d'),
            'language': language,
            'sortBy': 'publishedAt',
            'pageSize': page_size,
            'apiKey': self.api_key
        }
        
        # Add sources if specified
        if sources:
            params['sources'] = sources
        
        try:
            # Make API request
            response = requests.get(self.base_url, params=params)
            response.raise_for_status()
            
            data = response.json()
            
            if data['status'] == 'ok':
                return data['articles']
            else:
                print(f"API Error: {data.get('message', 'Unknown error')}")
                return []
                
        except requests.exceptions.RequestException as e:
            print(f"Request failed: {e}")
            return []
    
    def analyze_sentiment(self, text: str, model: str) -> Dict:
        """
        Analyze sentiment of given text using TextBlob
        
        Args:
            text: Text to analyze
            
        Returns:
            Dictionary with sentiment metrics
        """
        if not text:
            return {
                'polarity': 0.0,
                'subjectivity': 0.0,
                'label': 'neutral',
                'confidence': 0.0
            }
        blob = TextBlob(text)
        subjectivity = blob.sentiment.subjectivity

        # implement Vader Analysis for polarity scores
        if model == "Vader":
            vader = SIA()
            fullpolarity = vader.polarity_scores(text)
            polarity=fullpolarity['compound']
            polarity_thresh = 0.05
        # otherwise 
        else:
            polarity = blob.sentiment.polarity
            polarity_thresh = 0.1

        # Determine sentiment label through polarity threshold
        if polarity > polarity_thresh:
            label = 'positive'
        elif polarity < -polarity_thresh:
            label = 'negative'
        else:
            label = 'neutral'
        
        
        # Calculate confidence (distance from neutral)
        confidence = abs(polarity)
        res = {
            'polarity': polarity,
            'subjectivity': subjectivity,
            'label': label,
            'confidence': confidence
        }
        return res
    def process_news_articles(self, articles: List[Dict], model: str) -> pd.DataFrame:
        """
        Process news articles and add sentiment analysis
        
        Args:
            articles: List of news articles from API
            
        Returns:
            DataFrame with processed articles and sentiment data
        """
        processed_articles = []
        
        for article in articles:
            # Skip articles with missing essential data
            if not article.get('title') or not article.get('publishedAt'):
                continue
            
            # Analyze sentiment of title and description
            title_sentiment = self.analyze_sentiment(article['title'], model=model)
            description_sentiment = self.analyze_sentiment(article['description'], model=model)
            
            # Combine title and description sentiment (weighted toward title)
            combined_polarity = (title_sentiment['polarity'] * 0.7 + 
                               description_sentiment['polarity'] * 0.3)
            combined_subjectivity = (title_sentiment['subjectivity'] * 0.7 + 
                                   description_sentiment['subjectivity'] * 0.3)
            # Determine overall sentiment
            if combined_polarity > 0.1:
                overall_sentiment = 'positive'
            elif combined_polarity < -0.1:
                overall_sentiment = 'negative'
            else:
                overall_sentiment = 'neutral'
            
            processed_article = {
                'title': article['title'],
                'description': article.get('description', ''),
                'url': article['url'],
                'source': article['source']['name'],
                'published_at': article['publishedAt'],
                'author': article.get('author', 'Unknown'),
                'sentiment_label': overall_sentiment,
                'sentiment_polarity': combined_polarity,
                'sentiment_subjectivity': combined_subjectivity,
                'title_sentiment': title_sentiment['label'],
                'title_polarity': title_sentiment['polarity'],
                'description_sentiment': description_sentiment['label'],
                'description_polarity': description_sentiment['polarity']
            }
            
            processed_articles.append(processed_article)
        
        # Convert to DataFrame
        df = pd.DataFrame(processed_articles)
        
        # Convert published_at to datetime
        if not df.empty:
            df['published_at'] = pd.to_datetime(df['published_at'])
            df = df.sort_values('published_at', ascending=False)
        
        return df
    
    def get_ai_news_with_sentiment(self, 
                                   query: str = "artificial intelligence",
                                   days: tuple[int] = (7,14),
                                   sources: Optional[str] = None,
                                   model: str = "Textblob") -> pd.DataFrame:
        """
        Complete pipeline: fetch news and analyze sentiment
        
        Args:
            query: Search query for news articles
            days: Number of days to look back
            sources: Comma-separated string of news sources
            
        Returns:
            DataFrame with news articles and sentiment analysis
        """
        print(f"Fetching {query} news from the last {days} days...")
        
        # Fetch articles
        articles = self.fetch_ai_news(query=query, days=days, sources=sources)
        
        if not articles:
            print("No articles found.")
            return pd.DataFrame()
        
        print(f"Found {len(articles)} articles. Analyzing sentiment...")
        
        # Process and analyze
        df = self.process_news_articles(articles, model=model)
        
        print(f"Processed {len(df)} articles with sentiment analysis. \nUsed {model} for polarity analysis and Textblob for sentiment analysis.")
        return df
def load_config():
    """Load configuration from config.json"""
    with open('config.json', 'r') as f:
        return json.load(f)

if __name__ == "__main__":
    # Test the API when run directly
    analyzer = AINewsAnalyzer()
    config = load_config()
    
    print("Testing AI News Sentiment Analyzer...")
    print("=" * 50)
    
    # Test sentiment analysis
    test_texts = config["test_texts"]
    
    print("\nSentiment Analysis Examples:")
    for text in test_texts:
        sentiment = analyzer.analyze_sentiment(text)
        print(f"Text: {text}")
        print(f"Sentiment: {sentiment['label']} (polarity: {sentiment['polarity']:.2f}\n")
    
    # Test news fetching
    print("Fetching recent AI news...")
    df = analyzer.get_ai_news_with_sentiment(days=3)
    
    if not df.empty:
        print(f"\nFound {len(df)} articles")
        print("\nSentiment Distribution:")
        print(df['sentiment_label'].value_counts())
        
        print("\nTop 3 Most Positive Headlines:")
        positive_articles = df[df['sentiment_label'] == 'positive'].nlargest(3, 'sentiment_polarity')
        for _, article in positive_articles.iterrows():
            print(f"📈 {article['title']} (Score: {article['sentiment_polarity']:.2f})")
        
        print("\nTop 3 Most Negative Headlines:")
        negative_articles = df[df['sentiment_label'] == 'negative'].nsmallest(3, 'sentiment_polarity')
        for _, article in negative_articles.iterrows():
            print(f"📉 {article['title']} (Score: {article['sentiment_polarity']:.2f})")
    else:
        print("No articles found. Check your API key and internet connection.")