Spaces:
Running
Running
File size: 2,036 Bytes
a78209f 964462b a78209f 964462b a78209f 964462b a78209f 964462b a78209f 964462b | 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 | import requests
import logging
from typing import List, Dict
import time
from backend.config import NEWS_API_KEY
logger = logging.getLogger(__name__)
class NewsAPIClient:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://newsapi.org/v2/everything"
def fetch_news(self, company_name: str, query_string: str = None, days: int = 7) -> List[Dict]:
"""
Fetches news for the given company from the last `days`.
"""
search_q = query_string if query_string else f'"{company_name}"'
try:
response = requests.get(
self.base_url,
params={
'q': search_q,
'apiKey': self.api_key,
'language': 'en',
'sortBy': 'publishedAt', # Ensure latest news comes first
'pageSize': 20
}
)
response.raise_for_status()
data = response.json()
articles = data.get('articles', [])
# Normalize format
clean_articles = []
for art in articles:
clean_articles.append({
'title': art.get('title'),
'description': art.get('description'),
'url': art.get('url'),
'published_at': art.get('publishedAt'),
'source': art.get('source', {}).get('name')
})
return clean_articles
except Exception as e:
print(f"!!! [NewsAPI] ERROR: {e}")
logger.error(f"NewsAPI error: {str(e)}")
return []
class NewsAggregator:
def __init__(self):
self.client = NewsAPIClient(api_key=NEWS_API_KEY)
def fetch_news(self, company_name: str, query_string: str = None) -> List[Dict]:
# In a full production app, this would try multiple clients
return self.client.fetch_news(company_name, query_string=query_string)
|