locate-the-news / backend / app / services / event_builder.py
event_builder.py
Raw
"""
Event Builder Service - Combines RSS fetching, NLP processing, and geocoding
to build events from news articles.
"""
from typing import List, Dict, Optional
from sqlalchemy.orm import Session

from ..models import Article, LocationBase, City, Country, Continent, Priority
from ..schemas import Event, EventLocation
from .rss_fetcher import RssFetcher
from .nlp_processor import extract_locations
from .geocoder import geocode_place
from ..utils.logger import setup_logger

logger = setup_logger(__name__)


def build_events_from_articles(
    articles: List[Article],
    db: Session
) -> List[Event]:
    """
    Processes articles to extract locations and build events.
    
    Args:
        articles: List of Article objects to process
        db: Database session
        
    Returns:
        List of Event objects ready for display on the map
    """
    events = []
    
    for article in articles:
        try:
            # Extract locations from article text
            text = f"{article.title} {article.summary or ''}"
            places = extract_locations(text, lang="en")  # Default to English
            
            if not places:
                logger.debug(f"No locations found in article: {article.title}")
                continue
            
            # Geocode the first location found
            first_place = places[0]
            coords = geocode_place(first_place)
            
            if not coords:
                logger.debug(f"Could not geocode place: {first_place}")
                continue
            
            # Determine priority based on keywords
            priority = determine_priority(article.title, article.summary or "")
            
            # Create event
            event = Event(
                id=article.id,
                title=article.title,
                link=article.link,
                summary=article.summary,
                published_date=article.published_date,
                source_name=article.source_name,
                priority=priority.value if isinstance(priority, Priority) else priority,
                location=EventLocation(
                    latitude=coords["lat"],
                    longitude=coords["lon"],
                    entity_name=first_place
                )
            )
            events.append(event)
            
        except Exception as e:
            logger.error(f"Error building event from article {article.id}: {e}")
            continue
    
    return events


def determine_priority(title: str, summary: str) -> Priority:
    """
    Determines the priority level of a news article based on keywords.
    
    Critical keywords: earthquake, deprem, katliam, massacre, terror, attack
    High keywords: crisis, kriz, conflict, çatışma
    Medium: default
    Low: entertainment, eğlence, sports, spor
    """
    text = (title + " " + summary).lower()
    
    critical_keywords = [
        "earthquake", "deprem", "katliam", "massacre", "terror", "attack",
        "bomb", "bomba", "shooting", "ölüm", "death", "casualties"
    ]
    
    high_keywords = [
        "crisis", "kriz", "conflict", "çatışma", "war", "savaş",
        "emergency", "acil", "disaster", "felaket"
    ]
    
    low_keywords = [
        "entertainment", "eğlence", "sports", "spor", "game", "oyun",
        "music", "müzik", "movie", "film"
    ]
    
    if any(keyword in text for keyword in critical_keywords):
        return Priority.CRITICAL
    elif any(keyword in text for keyword in high_keywords):
        return Priority.HIGH
    elif any(keyword in text for keyword in low_keywords):
        return Priority.LOW
    else:
        return Priority.MEDIUM