Contra-Signal / backend /utils /api_clients.py
GSMSB's picture
chore: optimize AI models, improve PDF extraction latency, and clean dependencies
964462b
Raw
History Blame Contribute Delete
2.04 kB
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)