File size: 10,154 Bytes
e918eaf
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
"""
AI & Tech News Scraper
Fetches news from popular tech resources and big tech company blogs
"""

import feedparser
import requests
from bs4 import BeautifulSoup
from datetime import datetime, timedelta
from typing import List, Dict
import logging

logger = logging.getLogger(__name__)


class AITechNewsScraper:
    """Scraper for AI and tech news from major sources and company blogs"""

    # AI/Tech News Sources (RSS + Web)
    SOURCES = {
        # Major Tech News
        'TechCrunch AI': {
            'url': 'https://techcrunch.com/category/artificial-intelligence/feed/',
            'type': 'rss',
            'category': 'ai'
        },
        'The Verge AI': {
            'url': 'https://www.theverge.com/ai-artificial-intelligence/rss/index.xml',
            'type': 'rss',
            'category': 'ai'
        },
        'VentureBeat AI': {
            'url': 'https://venturebeat.com/category/ai/feed/',
            'type': 'rss',
            'category': 'ai'
        },
        'MIT Technology Review AI': {
            'url': 'https://www.technologyreview.com/topic/artificial-intelligence/feed',
            'type': 'rss',
            'category': 'ai'
        },
        'Ars Technica AI': {
            'url': 'https://feeds.arstechnica.com/arstechnica/technology-lab',
            'type': 'rss',
            'category': 'tech'
        },
        'Wired AI': {
            'url': 'https://www.wired.com/feed/tag/ai/latest/rss',
            'type': 'rss',
            'category': 'ai'
        },

        # Big Tech Company Blogs
        'OpenAI Blog': {
            'url': 'https://openai.com/blog/rss.xml',
            'type': 'rss',
            'category': 'ai'
        },
        'Google AI Blog': {
            'url': 'https://blog.google/technology/ai/rss/',
            'type': 'rss',
            'category': 'ai'
        },
        'Microsoft AI Blog': {
            'url': 'https://blogs.microsoft.com/ai/feed/',
            'type': 'rss',
            'category': 'ai'
        },
        'Meta AI Blog': {
            'url': 'https://ai.meta.com/blog/rss/',
            'type': 'rss',
            'category': 'ai'
        },
        'DeepMind Blog': {
            'url': 'https://deepmind.google/blog/rss.xml',
            'type': 'rss',
            'category': 'ai'
        },
        'Anthropic News': {
            'url': 'https://www.anthropic.com/news/rss.xml',
            'type': 'rss',
            'category': 'ai'
        },
        'AWS AI Blog': {
            'url': 'https://aws.amazon.com/blogs/machine-learning/feed/',
            'type': 'rss',
            'category': 'ai'
        },
        'NVIDIA AI Blog': {
            'url': 'https://blogs.nvidia.com/feed/',
            'type': 'rss',
            'category': 'ai'
        },

        # Research & Academia
        'Stanford HAI': {
            'url': 'https://hai.stanford.edu/news/rss.xml',
            'type': 'rss',
            'category': 'research'
        },
        'Berkeley AI Research': {
            'url': 'https://bair.berkeley.edu/blog/feed.xml',
            'type': 'rss',
            'category': 'research'
        },
    }

    def __init__(self):
        """Initialize the AI/Tech news scraper"""
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
        })

    def scrape_ai_tech_news(self, max_items: int = 100, hours: int = 48) -> List[Dict]:
        """
        Scrape AI and tech news from all sources

        Args:
            max_items: Maximum number of news items to return
            hours: Only include news from the last N hours

        Returns:
            List of news items with standardized format
        """
        all_news = []
        cutoff_time = datetime.now() - timedelta(hours=hours)

        for source_name, source_config in self.SOURCES.items():
            try:
                if source_config['type'] == 'rss':
                    news_items = self._scrape_rss_feed(
                        source_name,
                        source_config['url'],
                        source_config['category'],
                        cutoff_time
                    )
                    all_news.extend(news_items)
                    logger.info(f"Scraped {len(news_items)} items from {source_name}")

            except Exception as e:
                logger.error(f"Error scraping {source_name}: {e}")
                continue

        # Sort by timestamp (newest first)
        all_news.sort(key=lambda x: x['timestamp'], reverse=True)

        # Limit to max_items
        return all_news[:max_items]

    def _scrape_rss_feed(self, source_name: str, feed_url: str,
                        category: str, cutoff_time: datetime) -> List[Dict]:
        """Scrape a single RSS feed"""
        news_items = []

        try:
            feed = feedparser.parse(feed_url)

            for entry in feed.entries:
                try:
                    # Parse timestamp
                    if hasattr(entry, 'published_parsed') and entry.published_parsed:
                        timestamp = datetime(*entry.published_parsed[:6])
                    elif hasattr(entry, 'updated_parsed') and entry.updated_parsed:
                        timestamp = datetime(*entry.updated_parsed[:6])
                    else:
                        timestamp = datetime.now()

                    # Skip old news
                    if timestamp < cutoff_time:
                        continue

                    # Extract title and summary
                    title = entry.get('title', 'No title')
                    summary = entry.get('summary', entry.get('description', ''))

                    # Clean HTML from summary
                    if summary:
                        soup = BeautifulSoup(summary, 'html.parser')
                        summary = soup.get_text().strip()
                        # Limit summary length
                        if len(summary) > 300:
                            summary = summary[:297] + '...'

                    # Determine impact and sentiment based on keywords
                    impact = self._determine_impact(title, summary)
                    sentiment = self._determine_sentiment(title, summary)

                    news_item = {
                        'title': title,
                        'summary': summary or title,
                        'source': source_name,
                        'url': entry.get('link', ''),
                        'timestamp': timestamp,
                        'category': category,
                        'impact': impact,
                        'sentiment': sentiment,
                        'is_breaking': self._is_breaking_news(title, summary),
                        'likes': 0,  # No engagement data for RSS
                        'retweets': 0,
                        'reddit_score': 0,
                        'reddit_comments': 0
                    }

                    news_items.append(news_item)

                except Exception as e:
                    logger.error(f"Error parsing entry from {source_name}: {e}")
                    continue

        except Exception as e:
            logger.error(f"Error fetching RSS feed {feed_url}: {e}")

        return news_items

    def _determine_impact(self, title: str, summary: str) -> str:
        """Determine impact level based on keywords"""
        text = f"{title} {summary}".lower()

        high_impact_keywords = [
            'breakthrough', 'announce', 'launch', 'release', 'new model',
            'gpt', 'claude', 'gemini', 'llama', 'chatgpt',
            'billion', 'trillion', 'acquisition', 'merger',
            'regulation', 'ban', 'lawsuit', 'security breach',
            'major', 'significant', 'revolutionary', 'first-ever'
        ]

        medium_impact_keywords = [
            'update', 'improve', 'enhance', 'study', 'research',
            'partnership', 'collaboration', 'funding', 'investment',
            'expands', 'grows', 'adopts', 'implements'
        ]

        for keyword in high_impact_keywords:
            if keyword in text:
                return 'high'

        for keyword in medium_impact_keywords:
            if keyword in text:
                return 'medium'

        return 'low'

    def _determine_sentiment(self, title: str, summary: str) -> str:
        """Determine sentiment based on keywords"""
        text = f"{title} {summary}".lower()

        positive_keywords = [
            'breakthrough', 'success', 'achieve', 'improve', 'advance',
            'innovative', 'revolutionary', 'launch', 'release', 'win',
            'growth', 'expand', 'partnership', 'collaboration'
        ]

        negative_keywords = [
            'fail', 'issue', 'problem', 'concern', 'worry', 'risk',
            'ban', 'lawsuit', 'breach', 'hack', 'leak', 'crisis',
            'decline', 'loss', 'shutdown', 'controversy'
        ]

        positive_count = sum(1 for kw in positive_keywords if kw in text)
        negative_count = sum(1 for kw in negative_keywords if kw in text)

        if positive_count > negative_count:
            return 'positive'
        elif negative_count > positive_count:
            return 'negative'
        else:
            return 'neutral'

    def _is_breaking_news(self, title: str, summary: str) -> bool:
        """Determine if news is breaking"""
        text = f"{title} {summary}".lower()

        breaking_indicators = [
            'breaking', 'just announced', 'just released', 'just launched',
            'alert', 'urgent', 'developing', 'live', 'now:'
        ]

        return any(indicator in text for indicator in breaking_indicators)

    def get_statistics(self) -> Dict:
        """Get statistics - returns empty for backward compatibility"""
        return {
            'total': 0,
            'high_impact': 0,
            'breaking': 0,
            'last_update': 'Managed by cache',
            'by_category': {
                'ai': 0,
                'tech': 0,
                'research': 0
            }
        }