"""
Event Processor Service - Orchestrates the complete news processing pipeline
"""
from sqlalchemy.orm import Session
from typing import List, Dict, Any, Optional
from datetime import datetime, timezone
import logging
from .. import models, schemas
from ..config import settings
from ..utils.logger import setup_logger
from .news_api_fetcher import NewsAPIFetcher
from .rss_feed_fetcher import RSSFeedFetcher
from .sentiment_analyzer import SentimentAnalyzer
from .nlp_processor import extract_locations
from .geocoder import geocode_location
logger = setup_logger(__name__)
class EventProcessor:
"""
Orchestrates the complete pipeline:
1. Fetch news from NewsAPI
2. Extract locations using spaCy
3. Geocode locations to coordinates
4. Analyze sentiment
5. Save to database
"""
def __init__(self, db: Session):
"""
Initialize the event processor.
Args:
db: Database session
"""
self.db = db
self.news_fetcher = NewsAPIFetcher()
self.rss_fetcher = RSSFeedFetcher()
self.sentiment_analyzer = SentimentAnalyzer()
def fetch_and_process_news(
self,
countries: Optional[List[str]] = None,
category: Optional[str] = None,
page_size: int = 50
) -> List[models.Article]:
"""
Fetch news and process through complete pipeline.
Args:
countries: List of country codes
category: News category
page_size: Number of articles to fetch
Returns:
List of processed Article models
"""
logger.info("Starting news fetch and processing pipeline...")
# Step 1a: Fetch news from NewsAPI
newsapi_articles = self.news_fetcher.fetch_top_headlines(
countries=countries,
category=category,
page_size=page_size
)
logger.info(f"Fetched {len(newsapi_articles)} articles from NewsAPI")
# Step 1b: Fetch from RSS feeds for global coverage
rss_articles = self.rss_fetcher.fetch_all_feeds(
max_articles_per_feed=max(20, page_size // 4)
)
logger.info(f"Fetched {len(rss_articles)} articles from RSS feeds")
# Step 1c: Merge NewsAPI and RSS articles
articles = self._merge_articles(newsapi_articles, rss_articles)
logger.info(f"Using {len(articles)} total articles (NewsAPI + RSS)")
if not articles:
logger.warning("No articles fetched from any source")
return []
# Step 2: Add sentiment analysis
articles = self.sentiment_analyzer.analyze_batch(articles)
# Step 3: Process each article
processed_articles = []
for article_data in articles:
try:
processed = self.process_article(article_data)
if processed:
processed_articles.append(processed)
except Exception as e:
logger.error(f"Error processing article '{article_data.get('title', 'Unknown')}': {e}")
continue
logger.info(f"Successfully processed {len(processed_articles)} articles")
return processed_articles
def process_article(self, article_data: Dict[str, Any]) -> Optional[models.Article]:
"""
Process a single article through the pipeline.
Args:
article_data: Article dictionary from NewsAPI
Returns:
Article model if successful, None otherwise
"""
# Check if article already exists
existing = self.db.query(models.Article).filter(
models.Article.link == article_data['link']
).first()
if existing:
logger.debug(f"Article already exists: {article_data['title'][:50]}...")
return existing
# Create article model
article = models.Article(
title=article_data['title'],
link=article_data['link'],
summary=article_data.get('summary', ''),
published_date=article_data['published_date'],
source_name=article_data['source_name'],
priority=self._map_priority(article_data.get('priority', 'medium')),
sentiment_score=article_data.get('sentiment_score'),
sentiment_label=article_data.get('sentiment_label'),
sentiment_analyzer=article_data.get('sentiment_analyzer', 'hybrid'),
analyzed_at=datetime.now(timezone.utc)
)
# Extract and geocode locations
text_to_analyze = f"{article_data['title']} {article_data.get('summary', '')}"
locations = extract_locations(text_to_analyze, lang='en')
if locations:
logger.info(f"Found {len(locations)} locations: {locations}")
# Geocode the first location (most relevant)
primary_location = locations[0]
geocoded = geocode_location(primary_location)
if geocoded:
# Create location model
location = models.LocationBase(
entity_name=primary_location,
entity_type='GPE',
latitude=geocoded['latitude'],
longitude=geocoded['longitude'],
geocoder_source='Nominatim',
location_type='city'
)
article.locations.append(location)
article.is_processed = True
logger.info(f"Geocoded '{primary_location}' to ({geocoded['latitude']}, {geocoded['longitude']})")
else:
logger.warning(f"Failed to geocode location: {primary_location}")
article.processing_error = f"Failed to geocode: {primary_location}"
else:
logger.warning(f"No locations found in article: {article_data['title'][:50]}...")
article.processing_error = "No locations found"
# Save to database
try:
self.db.add(article)
self.db.commit()
self.db.refresh(article)
logger.info(f"Saved article: {article.title[:50]}...")
return article
except Exception as e:
self.db.rollback()
logger.error(f"Error saving article to database: {e}")
return None
def _map_priority(self, priority_str: str) -> models.Priority:
"""
Map priority string to Priority enum.
Args:
priority_str: Priority as string
Returns:
Priority enum value
"""
priority_map = {
'low': models.Priority.LOW,
'medium': models.Priority.MEDIUM,
'high': models.Priority.HIGH,
'critical': models.Priority.CRITICAL
}
return priority_map.get(priority_str.lower(), models.Priority.MEDIUM)
def _merge_articles(
self,
newsapi_articles: List[Dict[str, Any]],
rss_articles: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
"""
Merge articles from NewsAPI and RSS, removing duplicates.
Args:
newsapi_articles: Articles from NewsAPI
rss_articles: Articles from RSS feeds
Returns:
Merged list of unique articles
"""
merged = list(newsapi_articles) # Start with NewsAPI articles
for rss_article in rss_articles:
# Check if this RSS article is a duplicate
is_duplicate = False
for existing in merged:
if self._is_duplicate(rss_article, existing):
is_duplicate = True
break
if not is_duplicate:
merged.append(rss_article)
return merged
def _is_duplicate(
self,
article1: Dict[str, Any],
article2: Dict[str, Any],
similarity_threshold: float = 0.8
) -> bool:
"""
Check if two articles are duplicates based on title similarity.
Args:
article1: First article
article2: Second article
similarity_threshold: Minimum similarity to consider duplicate
Returns:
True if articles are duplicates
"""
title1 = article1.get('title', '').lower()
title2 = article2.get('title', '').lower()
if not title1 or not title2:
return False
# Simple word-based similarity
words1 = set(title1.split())
words2 = set(title2.split())
if not words1 or not words2:
return False
intersection = words1.intersection(words2)
union = words1.union(words2)
similarity = len(intersection) / len(union)
return similarity >= similarity_threshold
def get_processed_events(
self,
limit: int = 100,
min_lat: Optional[float] = None,
max_lat: Optional[float] = None,
min_lng: Optional[float] = None,
max_lng: Optional[float] = None
) -> List[schemas.Event]:
"""
Get processed articles as Event schemas for the frontend.
Args:
limit: Maximum number of events to return
min_lat, max_lat, min_lng, max_lng: Bounding box filters
Returns:
List of Event schemas
"""
query = self.db.query(models.Article).filter(
models.Article.is_processed == True
).join(models.LocationBase)
# Apply bounding box filter if provided
if all([min_lat, max_lat, min_lng, max_lng]):
query = query.filter(
models.LocationBase.latitude >= min_lat,
models.LocationBase.latitude <= max_lat,
models.LocationBase.longitude >= min_lng,
models.LocationBase.longitude <= max_lng
)
articles = query.order_by(models.Article.published_date.desc()).limit(limit).all()
# Convert to Event schemas
events = []
for article in articles:
if article.locations:
location = article.locations[0] # Use first location
event = schemas.Event(
id=article.id,
title=article.title,
link=article.link,
summary=article.summary,
published_date=article.published_date,
source_name=article.source_name,
priority=article.priority.value,
sentiment_score=article.sentiment_score,
sentiment_label=article.sentiment_label,
location=schemas.EventLocation(
latitude=location.latitude,
longitude=location.longitude,
entity_name=location.entity_name
)
)
events.append(event)
return events
# Example usage
if __name__ == "__main__":
from ..database import SessionLocal
db = SessionLocal()
processor = EventProcessor(db)
# Fetch and process news
articles = processor.fetch_and_process_news(page_size=10)
print(f"Processed {len(articles)} articles")
# Get events
events = processor.get_processed_events(limit=10)
print(f"Retrieved {len(events)} events")
db.close()