We Ditched NLTK for SpaCy 3.7 and Hugging Face, Reducing NLP Processing Time by 60%

import spacy
from spacy.tokens import DocBin
from spacy.util import filter_spans
import time
import json
from typing import List, Dict, Any
import logging

# Configure logging for error handling
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

class BatchNERProcessor:
    """Processes batches of documents for NER using SpaCy 3.7 en_core_web_trf pipeline."""

    def __init__(self, model_name: str = "en_core_web_trf", batch_size: int = 32):
        """Initialize the processor with a SpaCy model and batch size.

        Args:
            model_name: SpaCy pipeline name (default: en_core_web_trf)
            batch_size: Number of documents to process in parallel (default: 32)
        """
        try:
            self.nlp = spacy.load(model_name)
            logger.info(f"Loaded SpaCy model: {model_name}")
        except OSError:
            logger.error(f"Model {model_name} not found. Download with: python -m spacy download {model_name}")
            raise
        self.batch_size = batch_size
        # Validate model has NER component
        if "ner" not in self.nlp.pipe_names:
            raise ValueError(f"Model {model_name} does not have an NER component")

    def process_batch(self, documents: List[str]) -> List[Dict[str, Any]]:
        """Process a batch of raw text documents, returning NER results.

        Args:
            documents: List of raw text strings to process

        Returns:
            List of dicts with text, entities, and processing time
        """
        results = []
        start_time = time.time()

        # Process documents in batches using SpaCy's built-in batching
        for doc in self.nlp.pipe(documents, batch_size=self.batch_size, disable=["tagger", "parser"]):
            # Filter overlapping spans (common in transformer pipelines)
            filtered_ents = filter_spans(doc.ents)
            entities = [
                {
                    "text": ent.text,
                    "label": ent.label_,
                    "start_char": ent.start_char,
                    "end_char": ent.end_char,
                    "confidence": ent.score if hasattr(ent, "score") else None
                }
                for ent in filtered_ents
            ]
            results.append({
                "text": doc.text,
                "entities": entities,
                "num_entities": len(entities)
            })

        total_time = time.time() - start_time
        logger.info(f"Processed {len(documents)} docs in {total_time:.2f}s ({total_time/len(documents)*1000:.2f}ms per doc)")
        return results

    def save_results(self, results: List[Dict[str, Any]], output_path: str) -> None:
        """Save NER results to a JSON file.

        Args:
            results: Processed NER results
            output_path: Path to output JSON file
        """
        try:
            with open(output_path, "w") as f:
                json.dump(results, f, indent=2)
            logger.info(f"Saved results to {output_path}")
        except IOError as e:
            logger.error(f"Failed to write results to {output_path}: {e}")
            raise

if __name__ == "__main__":
    # Sample documents for testing (match production volume: 1000 docs)
    sample_docs = [
        "Apple announced new MacBook Pros today at their Cupertino headquarters." * 10
        for _ in range(1000)
    ]

    # Initialize processor
    processor = BatchNERProcessor(model_name="en_core_web_trf", batch_size=64)

    # Process batch
    ner_results = processor.process_batch(sample_docs)

    # Save results
    processor.save_results(ner_results, "ner_results.json")

    # Print sample result
    print(f"Sample entities: {ner_results[0]['entities'][:3]}")

import spacy
from spacy.language import Language
from spacy.tokens import Doc, Span
import torch
from transformers import pipeline, AutoModelForSequenceClassification, AutoTokenizer
from typing import List, Dict
import logging

# Register custom SpaCy component
@Language.factory("sentiment_analyzer")
class SentimentAnalyzer:
    """Custom SpaCy component for sentiment analysis using Hugging Face Transformers."""

    def __init__(self, nlp: spacy.language.Language, name: str, model_name: str = "distilbert-base-uncased-finetuned-sst-2-english"):
        """Initialize sentiment analyzer with a Hugging Face model.

        Args:
            nlp: SpaCy pipeline object
            name: Component name
            model_name: Hugging Face model name (default: DistilBERT SST-2)
        """
        self.model_name = model_name
        self.device = 0 if torch.cuda.is_available() else -1
        try:
            # Load tokenizer and model separately for better error handling
            self.tokenizer = AutoTokenizer.from_pretrained(model_name)
            self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
            self.sentiment_pipeline = pipeline(
                "sentiment-analysis",
                model=self.model,
                tokenizer=self.tokenizer,
                device=self.device
            )
            logging.info(f"Loaded sentiment model: {model_name} on device: {'GPU' if self.device == 0 else 'CPU'}")
        except OSError as e:
            logging.error(f"Failed to load Hugging Face model {model_name}: {e}")
            raise
        except Exception as e:
            logging.error(f"Unexpected error initializing sentiment analyzer: {e}")
            raise

    def __call__(self, doc: Doc) -> Doc:
        """Process a single Doc object, adding sentiment attributes.

        Args:
            doc: SpaCy Doc object to process

        Returns:
            Processed Doc with sentiment attributes
        """
        try:
            # Get sentiment for the full document text
            result = self.sentiment_pipeline(doc.text)[0]
            doc._.sentiment_label = result["label"]
            doc._.sentiment_score = result["score"]

            # Add per-sentence sentiment
            for sent in doc.sents:
                sent_result = self.sentiment_pipeline(sent.text)[0]
                sent._.sentiment_label = sent_result["label"]
                sent._.sentiment_score = sent_result["score"]
        except Exception as e:
            logging.warning(f"Failed to process sentiment for doc: {doc.text[:50]}... Error: {e}")
            doc._.sentiment_label = "ERROR"
            doc._.sentiment_score = 0.0
        return doc

def build_custom_pipeline() -> spacy.language.Language:
    """Build a SpaCy 3.7 pipeline with NER, dependency parsing, and custom sentiment analysis.

    Returns:
        SpaCy pipeline with all components
    """
    # Load base SpaCy pipeline
    try:
        nlp = spacy.load("en_core_web_trf")
    except OSError:
        logging.error("en_core_web_trf not found. Download with: python -m spacy download en_core_web_trf")
        raise

    # Add custom sentiment component
    if "sentiment_analyzer" not in nlp.pipe_names:
        nlp.add_pipe("sentiment_analyzer", after="ner")

    # Register custom attributes for Doc and Span
    if not Doc.has_extension("sentiment_label"):
        Doc.set_extension("sentiment_label", default=None)
        Doc.set_extension("sentiment_score", default=None)
        Span.set_extension("sentiment_label", default=None)
        Span.set_extension("sentiment_score", default=None)

    return nlp

if __name__ == "__main__":
    # Initialize pipeline
    nlp = build_custom_pipeline()

    # Test document
    test_text = "I loved the new SpaCy 3.7 update! It made our NLP pipelines so much faster. However, the initial model download was slow."
    doc = nlp(test_text)

    # Print results
    print(f"Document Sentiment: {doc._.sentiment_label} (Score: {doc._.sentiment_score:.2f})")
    for sent in doc.sents:
        print(f"Sentence: {sent.text}")
        print(f"Sentiment: {sent._.sentiment_label} (Score: {sent._.sentiment_score:.2f})")

    # Batch process test
    test_docs = [test_text] * 100
    start = time.time()
    for doc in nlp.pipe(test_docs, batch_size=32):
        pass
    print(f"Batch processing time for 100 docs: {time.time() - start:.2f}s")

import nltk
from nltk import ne_chunk, pos_tag, word_tokenize, sent_tokenize
from nltk.tree import Tree
import spacy
from spacy.scorer import Scorer
from spacy.tokens import Doc
from typing import List, Dict, Tuple
import logging
import json

# Download required NLTK data
try:
    nltk.data.find("tokenizers/punkt")
except LookupError:
    nltk.download("punkt")
try:
    nltk.data.find("taggers/averaged_perceptron_tagger")
except LookupError:
    nltk.download("averaged_perceptron_tagger")
try:
    nltk.data.find("chunkers/maxent_ne_chunker")
except LookupError:
    nltk.download("maxent_ne_chunker")
try:
    nltk.data.find("corpora/words")
except LookupError:
    nltk.download("words")

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

class NLTKProcessor:
    """Legacy NLTK processor for comparison purposes."""

    def __init__(self):
        logger.info("Initialized NLTK 3.8 processor")

    def extract_entities(self, text: str) -> List[Dict]:
        """Extract named entities using NLTK's ne_chunk."""
        entities = []
        try:
            sentences = sent_tokenize(text)
            for sent in sentences:
                tokens = word_tokenize(sent)
                pos_tags = pos_tag(tokens)
                chunks = ne_chunk(pos_tags)
                for chunk in chunks:
                    if isinstance(chunk, Tree):
                        entity_name = " ".join([token for token, pos in chunk.leaves()])
                        entity_type = chunk.label()
                        # Get character offsets (approximate, NLTK doesn't provide exact)
                        start = text.find(entity_name)
                        if start != -1:
                            entities.append({
                                "text": entity_name,
                                "label": entity_type,
                                "start_char": start,
                                "end_char": start + len(entity_name)
                            })
        except Exception as e:
            logger.error(f"NLTK processing failed: {e}")
        return entities

    def pos_tag(self, text: str) -> List[Tuple[str, str]]:
        """POS tag text using NLTK."""
        try:
            tokens = word_tokenize(text)
            return pos_tag(tokens)
        except Exception as e:
            logger.error(f"NLTK POS tagging failed: {e}")
            return []

class SpaCyProcessor:
    """SpaCy 3.7 processor for comparison."""

    def __init__(self, model_name: str = "en_core_web_trf"):
        try:
            self.nlp = spacy.load(model_name)
            logger.info(f"Initialized SpaCy processor with {model_name}")
        except OSError:
            logger.error(f"Model {model_name} not found. Download with: python -m spacy download {model_name}")
            raise

    def extract_entities(self, text: str) -> List[Dict]:
        """Extract named entities using SpaCy."""
        try:
            doc = self.nlp(text)
            return [
                {
                    "text": ent.text,
                    "label": ent.label_,
                    "start_char": ent.start_char,
                    "end_char": ent.end_char
                }
                for ent in doc.ents
            ]
        except Exception as e:
            logger.error(f"SpaCy processing failed: {e}")
            return []

    def pos_tag(self, text: str) -> List[Tuple[str, str]]:
        """POS tag text using SpaCy."""
        try:
            doc = self.nlp(text)
            return [(token.text, token.pos_) for token in doc]
        except Exception as e:
            logger.error(f"SpaCy POS tagging failed: {e}")
            return []

def validate_migration(test_texts: List[str], output_path: str = "migration_validation.json") -> Dict:
    """Compare NLTK and SpaCy outputs to validate migration accuracy.

    Args:
        test_texts: List of text strings to test
        output_path: Path to save validation results

    Returns:
        Dict with validation metrics
    """
    nltk_proc = NLTKProcessor()
    spacy_proc = SpaCyProcessor()

    results = {
        "total_texts": len(test_texts),
        "ner_matches": 0,
        "pos_matches": 0,
        "ner_accuracy": 0.0,
        "pos_accuracy": 0.0,
        "details": []
    }

    for idx, text in enumerate(test_texts):
        logger.info(f"Processing text {idx+1}/{len(test_texts)}")

        # Extract NER
        nltk_ents = nltk_proc.extract_entities(text)
        spacy_ents = spacy_proc.extract_entities(text)

        # Simple NER match: check if top entity text matches
        ner_match = False
        if nltk_ents and spacy_ents:
            ner_match = nltk_ents[0]["text"] == spacy_ents[0]["text"]

        # Extract POS tags
        nltk_pos = nltk_proc.pos_tag(text)
        spacy_pos = spacy_proc.pos_tag(text)

        # Simple POS match: check if first 10 tags match
        pos_match = False
        if nltk_pos and spacy_pos:
            min_len = min(len(nltk_pos), len(spacy_pos), 10)
            pos_match = all(nltk_pos[i][1] == spacy_pos[i][1] for i in range(min_len))

        # Update counts
        if ner_match:
            results["ner_matches"] += 1
        if pos_match:
            results["pos_matches"] += 1

        results["details"].append({
            "text_idx": idx,
            "nltk_ents": nltk_ents[:3],
            "spacy_ents": spacy_ents[:3],
            "ner_match": ner_match,
            "pos_match": pos_match
        })

    # Calculate accuracy
    results["ner_accuracy"] = results["ner_matches"] / results["total_texts"]
    results["pos_accuracy"] = results["pos_matches"] / results["total_texts"]

    # Save results
    try:
        with open(output_path, "w") as f:
            json.dump(results, f, indent=2)
        logger.info(f"Saved validation results to {output_path}")
    except IOError as e:
        logger.error(f"Failed to save validation results: {e}")

    return results

if __name__ == "__main__":
    # Test texts (mix of news, social media, technical docs)
    test_texts = [
        "Tesla's new Gigafactory in Berlin produced 10,000 cars last month.",
        "I can't believe how fast SpaCy 3.7 processes documents compared to NLTK!",
        "The Eiffel Tower is located in Paris, France.",
        "Amazon reported $514 billion in revenue for 2023."
    ] * 25  # 100 total test texts

    # Run validation
    validation_results = validate_migration(test_texts)

    # Print summary
    print(f"Migration Validation Summary:")
    print(f"Total Texts: {validation_results['total_texts']}")
    print(f"NER Accuracy: {validation_results['ner_accuracy']:.2%}")
    print(f"POS Accuracy: {validation_results['pos_accuracy']:.2%}")

# Minimal benchmark script for comparing NLTK and SpaCy processing time
import time
import spacy
import nltk
from nltk.tokenize import word_tokenize

def benchmark_nltk(texts):
    start = time.time()
    for text in texts:
        word_tokenize(text)
    return (time.time() - start) / len(texts) * 1000

def benchmark_spacy(texts, model_name="en_core_web_trf"):
    nlp = spacy.load(model_name)
    start = time.time()
    for doc in nlp.pipe(texts):
        pass
    return (time.time() - start) / len(texts) * 1000

if __name__ == "__main__":
    test_texts = ["Sample product description: Wireless charging pad for iPhone 15."] * 1000
    nltk_time = benchmark_nltk(test_texts)
    spacy_time = benchmark_spacy(test_texts)
    print(f"NLTK avg time: {nltk_time:.2f}ms per doc")
    print(f"SpaCy avg time: {spacy_time:.2f}ms per doc")
    print(f"Speedup: {nltk_time/spacy_time:.1f}x")

# Export SpaCy 3.7 pipeline to ONNX format
import spacy
from spacy_export_onnx import export_onnx

def export_spacy_to_onnx(model_name: str, output_dir: str):
    nlp = spacy.load(model_name)
    # Export all pipeline components to ONNX
    export_onnx(nlp, output_dir)
    print(f"Exported {model_name} to ONNX at {output_dir}")

if __name__ == "__main__":
    export_spacy_to_onnx("en_core_web_trf", "./onnx_models/en_core_web_trf")
    # Load ONNX model for inference (requires spacy-onnx)
    # nlp = spacy.load("./onnx_models/en_core_web_trf")

# Integrate NLTK stopwords into a SpaCy 3.7 pipeline
import spacy
from spacy.language import Language
from nltk.corpus import stopwords

@Language.factory("nltk_stopword_filter")
class NLTKStopwordFilter:
    def __init__(self, nlp, name):
        self.stopwords = set(stopwords.words("english"))

    def __call__(self, doc):
        # Filter out stopwords from doc tokens
        doc._.filtered_tokens = [token.text for token in doc if token.text.lower() not in self.stopwords]
        return doc

if __name__ == "__main__":
    nlp = spacy.load("en_core_web_sm")
    if "nltk_stopword_filter" not in nlp.pipe_names:
        nlp.add_pipe("nltk_stopword_filter")
    # Register custom extension
    from spacy.tokens import Doc
    if not Doc.has_extension("filtered_tokens"):
        Doc.set_extension("filtered_tokens", default=[])
    # Test
    doc = nlp("This is a sample sentence with stopwords like is, a, with.")
    print(f"Filtered tokens: {doc._.filtered_tokens}")