locate-the-news / backend / app / services / nlp_processor.py
nlp_processor.py
Raw
"""
NLP Processor Service - Extracts location entities from text using spaCy.
"""
import spacy
from typing import List, Optional

from ..utils.logger import setup_logger

logger = setup_logger(__name__)

# Lazy loading of spaCy models
_nlp_en: Optional[spacy.Language] = None
_nlp_tr: Optional[spacy.Language] = None


def _load_nlp_model(lang: str) -> Optional[spacy.Language]:
    """
    Lazy loads spaCy models. Returns None if model is not installed.
    """
    global _nlp_en, _nlp_tr
    
    try:
        if lang == "tr":
            if _nlp_tr is None:
                try:
                    _nlp_tr = spacy.load("tr_core_news_sm")
                except OSError:
                    logger.warning("Turkish spaCy model (tr_core_news_sm) not found. Install with: python -m spacy download tr_core_news_sm")
                    return None
            return _nlp_tr
        else:
            if _nlp_en is None:
                try:
                    _nlp_en = spacy.load("en_core_web_sm")
                except OSError:
                    logger.warning("English spaCy model (en_core_web_sm) not found. Install with: python -m spacy download en_core_web_sm")
                    return None
            return _nlp_en
    except Exception as e:
        logger.error(f"Error loading spaCy model for {lang}: {e}")
        return None


def extract_locations(text: str, lang: str = "en") -> List[str]:
    """
    Extracts location entities (cities, countries, regions) from text.
    
    Args:
        text: The text to analyze
        lang: Language code ("en" or "tr")
        
    Returns:
        List of location names found in the text
    """
    if not text or not text.strip():
        return []
    
    nlp = _load_nlp_model(lang)
    if nlp is None:
        logger.warning(f"spaCy model not available for {lang}, returning empty list")
        return []
    
    try:
        doc = nlp(text)
        places = []
        
        for ent in doc.ents:
            # GPE = Geopolitical Entity (countries, cities, states)
            # LOC = Location (mountains, bodies of water, etc.)
            if ent.label_ in ("GPE", "LOC"):
                places.append(ent.text)
        
        # Remove duplicates while preserving order
        seen = set()
        unique_places = []
        for place in places:
            if place not in seen:
                seen.add(place)
                unique_places.append(place)
        
        return unique_places
    except Exception as e:
        logger.error(f"Error extracting locations: {e}")
        return []