Natural Language Processing

"""
AI Track Unit 10: Natural Language Processing Foundations
Complete NLP Pipeline: Tokenizer, TF-IDF Vectorizer, Cosine Search & Sentiment Classifier
Author: DevAscent Engineering Team
"""

import math
import re
from collections import Counter
from typing import List, Dict, Tuple, Set

# =====================================================================
# 1. TEXT PREPROCESSING & TOKENIZER
# =====================================================================
STOP_WORDS: Set[str] = {
    'the', 'is', 'at', 'which', 'on', 'a', 'an', 'and', 'or', 'in',
    'this', 'to', 'of', 'for', 'with', 'it', 'as', 'was', 'by'
}

def clean_and_tokenize(text: str, remove_stopwords: bool = True) -> List[str]:
    """Lowercase, strip non-alphanumeric punctuation, and extract word tokens."""
    cleaned = re.sub(r'[^a-zA-Z0-9\s]', ' ', text.lower())
    tokens = cleaned.split()
    if remove_stopwords:
        tokens = [t for t in tokens if t not in STOP_WORDS and len(t) > 1]
    return tokens

# =====================================================================
# 2. CUSTOM TF-IDF VECTORIZER
# =====================================================================
class CustomTFIDFVectorizer:
    def __init__(self):
        self.vocabulary: Dict[str, int] = {}
        self.idf: Dict[str, float] = {}

    def fit(self, corpus: List[str]):
        tokenized_docs = [clean_and_tokenize(doc) for doc in corpus]
        N = len(corpus)

        # Build vocabulary
        unique_words = sorted(list(set(word for doc in tokenized_docs for word in doc)))
        self.vocabulary = {word: idx for idx, word in enumerate(unique_words)}

        # Compute Document Frequency (DF)
        df = Counter()
        for doc in tokenized_docs:
            for word in set(doc):
                df[word] += 1

        # Compute Inverse Document Frequency (IDF) with smoothing (+1)
        for word, count in df.items():
            self.idf[word] = math.log((N + 1) / (count + 1)) + 1.0

    def transform(self, corpus: List[str]) -> List[List[float]]:
        vectors = []
        vocab_size = len(self.vocabulary)

        for doc in corpus:
            tokens = clean_and_tokenize(doc)
            tf_counter = Counter(tokens)
            doc_len = len(tokens) if len(tokens) > 0 else 1.0

            vec = [0.0] * vocab_size
            for word, count in tf_counter.items():
                if word in self.vocabulary:
                    idx = self.vocabulary[word]
                    tf = count / doc_len
                    vec[idx] = tf * self.idf[word]

            # L2 Normalization (Cosine scale)
            norm = math.sqrt(sum(v ** 2 for v in vec))
            if norm > 0:
                vec = [v / norm for v in vec]
            vectors.append(vec)

        return vectors

    def fit_transform(self, corpus: List[str]) -> List[List[float]]:
        self.fit(corpus)
        return self.transform(corpus)

# =====================================================================
# 3. SEMANTIC SEARCH (COSINE SIMILARITY)
# =====================================================================
def cosine_similarity(v1: List[float], v2: List[float]) -> float:
    return sum(a * b for a, b in zip(v1, v2))

# =====================================================================
# 4. NAIVE BAYES SENTIMENT CLASSIFIER WITH LAPLACE SMOOTHING
# =====================================================================
class NaiveBayesSentimentClassifier:
    def __init__(self):
        self.class_priors: Dict[str, float] = {}
        self.word_counts: Dict[str, Counter] = {'pos': Counter(), 'neg': Counter()}
        self.total_words: Dict[str, int] = {'pos': 0, 'neg': 0}
        self.vocab: Set[str] = set()

    def fit(self, documents: List[str], labels: List[str]):
        n = len(documents)
        label_counts = Counter(labels)
        for label in ['pos', 'neg']:
            self.class_priors[label] = label_counts[label] / n

        for doc, label in zip(documents, labels):
            tokens = clean_and_tokenize(doc, remove_stopwords=False)
            for token in tokens:
                self.word_counts[label][token] += 1
                self.total_words[label] += 1
                self.vocab.add(token)

    def predict(self, text: str) -> Tuple[str, Dict[str, float]]:
        tokens = clean_and_tokenize(text, remove_stopwords=False)
        vocab_size = len(self.vocab)

        # Log-probabilities to prevent underflow
        log_probs = {}
        for label in ['pos', 'neg']:
            log_prob = math.log(self.class_priors[label])
            for token in tokens:
                # Laplace Smoothing (+1)
                count = self.word_counts[label][token]
                prob = (count + 1.0) / (self.total_words[label] + vocab_size)
                log_prob += math.log(prob)
            log_probs[label] = log_prob

        best_class = 'pos' if log_probs['pos'] > log_probs['neg'] else 'neg'
        return best_class, log_probs

# =====================================================================
# DEMONSTRATION HARNESS
# =====================================================================
if __name__ == "__main__":
    print("=" * 68)
    print("DEMO 1: TF-IDF VECTORIZATION & SEMANTIC RETRIEVAL SEARCH")
    print("=" * 68)

    docs = [
        "Artificial Intelligence and Machine Learning are transforming modern technology.",
        "Deep Learning utilizes neural networks with multiple layers for computer vision.",
        "Natural language processing powers smart voice assistants and language translation.",
        "Planetary rovers explore the red dusty surface of Mars using autonomous robotics."
    ]

    tfidf = CustomTFIDFVectorizer()
    doc_vectors = tfidf.fit_transform(docs)

    query = "computer neural networks"
    query_vector = tfidf.transform([query])[0]

    print(f"User Query: '{query}'\n")
    print("Semantic Similarity Match Rankings:")
    scores = []
    for idx, doc_vec in enumerate(doc_vectors):
        sim = cosine_similarity(query_vector, doc_vec)
        scores.append((sim, docs[idx]))

    scores.sort(key=lambda item: item[0], reverse=True)
    for rank, (score, doc) in enumerate(scores, 1):
        print(f"  Rank {rank} [Score: {score:.4f}]: {doc}")

    print("\n" + "=" * 68)
    print("DEMO 2: NAIVE BAYES SENTIMENT CLASSIFICATION (WITH LAPLACE SMOOTHING)")
    print("=" * 68)

    training_reviews = [
        "This product is amazing exceptional quality and works beautifully",
        "Outstanding performance absolutely love this device great design",
        "Terrible waste of money completely broken and defective hardware",
        "Horrible customer service arrived late damaged and unusable",
        "Fantastic battery life brilliant screen highly recommended"
    ]
    training_labels = ["pos", "pos", "neg", "neg", "pos"]

    clf = NaiveBayesSentimentClassifier()
    clf.fit(training_reviews, training_labels)

    test_reviews = [
        "This device is fantastic with brilliant performance",
        "Defective waste of money completely unusable and broken"
    ]

    for review in test_reviews:
        predicted_sentiment, _ = clf.predict(review)
        print(f"Review: '{review}'")
        print(f"  --> Predicted Sentiment: {predicted_sentiment.upper()}\n")