Upload 10 files

app.py

@@ -0,0 +1,166 @@
+"""
+Hugging Face Space — Indic Language Identifier
+Supports 12 Indic languages + English.
+Uses best available model: Ensemble > CharNgram > Unicode Heuristic fallback.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+import gradio as gr
+from src.utils.languages import LANGUAGE_REGISTRY, ALL_LANGUAGE_CODES
+from src.data.preprocessor import IndicTextPreprocessor
+from src.models.ensemble_model import EnsembleLIDModel, UnicodeHeuristicModel
+
+# ── Model Loading ─────────────────────────────────────────────────────────────
+
+preprocessor = IndicTextPreprocessor()
+model = None
+
+for mpath in ["./artifacts/models/ensemble.pkl", "./artifacts/models/ngram_model.pkl"]:
+    if Path(mpath).exists():
+        try:
+            import pickle
+            with open(mpath, "rb") as f:
+                model = pickle.load(f)
+            print(f"✓ Loaded: {mpath}")
+            break
+        except Exception as e:
+            print(f"Could not load {mpath}: {e}")
+
+if model is None:
+    model = EnsembleLIDModel()
+    print("⚠ No trained model found — using Unicode heuristic fallback")
+
+
+# ── Inference ─────────────────────────────────────────────────────────────────
+
+def predict(text: str) -> str:
+    """Clean inference function: text in → language name out."""
+    if not text or not text.strip():
+        return "Please enter some text."
+    processed = preprocessor.preprocess(text)
+    if hasattr(model, "predict_with_confidence"):
+        result = model.predict_with_confidence([processed])[0]
+        lang = result.get("predicted_lang", "?")
+        conf = result.get("confidence", 0.0)
+    elif hasattr(model, "predict"):
+        lang = model.predict([processed])[0]
+        conf = 0.9
+    else:
+        lang, conf = "eng", 0.5
+
+    info = LANGUAGE_REGISTRY.get(lang)
+    name = info.name if info else lang
+    native = info.native_name if info else ""
+    script = info.script if info else "?"
+    direction = info.direction if info else "LTR"
+    confidence_icon = "🟢" if conf > 0.8 else "🟡" if conf > 0.6 else "🔴"
+
+    return (
+        f"**{name}** ({native})\n\n"
+        f"Code: `{lang}` | Script: {script} | Direction: {direction}\n\n"
+        f"Confidence: {conf:.2%} {confidence_icon}"
+    )
+
+
+def predict_detailed(text: str) -> tuple:
+    """Returns (result_markdown, top5_table) for the full UI."""
+    if not text or not text.strip():
+        return "Please enter some text.", None
+
+    processed = preprocessor.preprocess(text)
+
+    if hasattr(model, "predict_with_confidence"):
+        result = model.predict_with_confidence([processed])[0]
+        lang = result.get("predicted_lang", "?")
+        conf = result.get("confidence", 0.0)
+        top5 = result.get("top_5", [])
+    elif hasattr(model, "predict"):
+        lang = model.predict([processed])[0]
+        conf = 0.9
+        top5 = [{"lang": lang, "probability": conf}]
+    else:
+        lang, conf = "eng", 0.5
+        top5 = []
+
+    info = LANGUAGE_REGISTRY.get(lang)
+    confidence_icon = "🟢" if conf > 0.8 else "🟡" if conf > 0.6 else "🔴"
+
+    result_md = f"""### 🎯 {info.name if info else lang} ({info.native_name if info else ''})
+
+| | |
+|---|---|
+| Code | `{lang}` |
+| Script | {info.script if info else '?'} |
+| Direction | {info.direction if info else 'LTR'} |
+| Family | {info.family if info else '?'} |
+| Confidence | {conf:.2%} {confidence_icon} |
+"""
+
+    table = None
+    if top5:
+        import pandas as pd
+        rows = []
+        for t in top5:
+            i = LANGUAGE_REGISTRY.get(t["lang"])
+            rows.append({
+                "Language": i.name if i else t["lang"],
+                "Code": t["lang"],
+                "Probability": f"{t['probability']:.2%}",
+            })
+        table = pd.DataFrame(rows)
+
+    return result_md, table
+
+
+# ── Sample Texts ──────────────────────────────────────────────────────────────
+
+SAMPLES = [
+    ["भारत एक विशाल देश है जहाँ अनेक भाषाएँ बोली जाती हैं।"],
+    ["বাংলাদেশ এবং পশ্চিমবঙ্গে বাংলা ভাষা প্রচলিত।"],
+    ["தமிழ் மொழி உலகின் மிகவும் பழமையான மொழிகளில் ஒன்றாகும்."],
+    ["India is a diverse country with hundreds of languages."],
+    ["اردو پاک��تان کی قومی زبان ہے۔"],
+    ["ಕನ್ನಡ ಭಾಷೆ ಕರ್ನಾಟಕ ರಾಜ್ಯದ ಅಧಿಕೃತ ಭಾಷೆ."],
+]
+
+# ── Gradio UI ─────────────────────────────────────────────────────────────────
+
+with gr.Blocks(title="Indic Language Identifier") as demo:
+    gr.Markdown("""
+# 🔤 Indic Language Identifier
+Detects language from text across **12 Indic languages + English**.
+Powered by Unicode heuristics + CharNgram ensemble.
+""")
+
+    with gr.Row():
+        with gr.Column(scale=2):
+            text_input = gr.Textbox(
+                label="Input Text",
+                placeholder="Type or paste text in any supported language...",
+                lines=5,
+            )
+            with gr.Row():
+                clear_btn = gr.Button("Clear", variant="secondary")
+                detect_btn = gr.Button("🔍 Detect Language", variant="primary")
+
+            gr.Examples(examples=SAMPLES, inputs=text_input, label="Try a sample")
+
+        with gr.Column(scale=2):
+            result_md = gr.Markdown(label="Result")
+            top5_table = gr.Dataframe(label="Top Predictions", visible=True)
+
+    detect_btn.click(fn=predict_detailed, inputs=text_input, outputs=[result_md, top5_table])
+    clear_btn.click(fn=lambda: ("", None), outputs=[result_md, top5_table])
+    text_input.submit(fn=predict_detailed, inputs=text_input, outputs=[result_md, top5_table])
+
+    gr.Markdown("""---
+**Supported languages:** Bengali · English · Gujarati · Hindi · Kannada · Malayalam · Marathi · Odia · Punjabi · Tamil · Telugu · Urdu
+""")
+
+if __name__ == "__main__":
+    demo.launch()

ensemble_model.py

@@ -0,0 +1,344 @@
+"""
+Ensemble Language Identification Model
+Combines CharNgram + FastText + Unicode heuristics for maximum accuracy.
+Strategies: weighted soft voting, stacking, confidence-based routing.
+"""
+from __future__ import annotations
+
+import json
+import pickle
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+import numpy as np
+import pandas as pd
+from loguru import logger
+from sklearn.metrics import f1_score, accuracy_score, classification_report
+
+from src.utils.languages import (
+    LANGUAGE_REGISTRY,
+    ALL_LANGUAGE_CODES,
+    CODE_TO_LABEL,
+    LABEL_TO_CODE,
+    CONFUSABLE_PAIRS,
+    RTL_LANGUAGES,
+    DEVANAGARI_LANGUAGES,
+)
+from src.data.preprocessor import get_dominant_script
+
+
+# ─────────────────────────────────────────────
+# Unicode Heuristic Model (Rule-based)
+# ─────────────────────────────────────────────
+
+class UnicodeHeuristicModel:
+    """
+    Fast rule-based language identifier using Unicode script analysis.
+    Used as a strong prior for high-confidence cases.
+    """
+
+    # Unique codepoints per language for disambiguation
+    DISTINCTIVE_CHARS = {
+        # Telugu distinctive vowel signs
+        "tel": set(range(0x0C00, 0x0C7F)),
+        # Kannada
+        "kan": set(range(0x0C80, 0x0CFF)),
+        # Malayalam
+        "mal": set(range(0x0D00, 0x0D7F)),
+        # Tamil — note: no voiced stops
+        "tam": set(range(0x0B80, 0x0BFF)),
+        # Odia
+        "ori": set(range(0x0B00, 0x0B7F)),
+        # Gurmukhi (Punjabi)
+        "pan": set(range(0x0A00, 0x0A7F)),
+        # Gujarati
+        "guj": set(range(0x0A80, 0x0AFF)),
+        # Bengali
+        "ben": set(range(0x0980, 0x09FF)),
+        # Arabic/Urdu
+        "urd": set(range(0x0600, 0x06FF)) | set(range(0xFB50, 0xFDFF)),
+    }
+
+    # Devanagari — needs lexical disambiguation between Hindi/Marathi
+    DEVANAGARI_RANGE = set(range(0x0900, 0x097F))
+    LATIN_RANGE = set(range(0x0041, 0x007B))
+
+    def predict_single(self, text: str) -> Tuple[Optional[str], float]:
+        """
+        Returns (predicted_lang, confidence).
+        confidence=0.0 → uncertain (pass to ML model).
+        """
+        char_codepoints = [ord(c) for c in text if c.isalpha()]
+        if not char_codepoints:
+            return None, 0.0
+
+        total = len(char_codepoints)
+
+        # Check unique scripts with unambiguous mapping
+        for lang, codepoints in self.DISTINCTIVE_CHARS.items():
+            matches = sum(cp in codepoints for cp in char_codepoints)
+            ratio = matches / total
+            if ratio > 0.75:
+                return lang, min(0.98, 0.7 + ratio * 0.28)
+
+        # Devanagari → Hindi/Marathi ambiguous, return low confidence
+        deva_matches = sum(cp in self.DEVANAGARI_RANGE for cp in char_codepoints)
+        if deva_matches / total > 0.75:
+            return "hin", 0.55  # slight prior toward Hindi (larger corpus)
+
+        # Latin → likely English
+        latin_matches = sum(cp in self.LATIN_RANGE for cp in char_codepoints)
+        if latin_matches / total > 0.85:
+            return "eng", 0.90
+
+        return None, 0.0
+
+    def predict(self, texts: List[str]) -> List[Tuple[Optional[str], float]]:
+        return [self.predict_single(t) for t in texts]
+
+
+# ─────────────────────────────────────────────
+# Ensemble Model
+# ─────────────────────────────────────────────
+
+class EnsembleLIDModel:
+    """
+    Multi-model ensemble for language identification.
+
+    Architecture:
+      1. Unicode heuristic: fast pre-filter, handles 40-60% of inputs
+      2. CharNgram model: strong char-level baseline
+      3. FastText model: fast neural model with subword features
+      4. Weighted voting: combine model probabilities
+
+    Calibration:
+      - Per-language confidence thresholds
+      - Devanagari disambiguation module (Hindi vs Marathi)
+      - Confusable pair re-ranking
+    """
+
+    SUPPORTED_STRATEGIES = ("weighted_vote", "stacking", "router")
+
+    def __init__(
+        self,
+        ngram_model=None,
+        fasttext_model=None,
+        transformer_model=None,
+        weights: Optional[Dict[str, float]] = None,
+        strategy: str = "weighted_vote",
+    ):
+        self.ngram_model = ngram_model
+        self.fasttext_model = fasttext_model
+        self.transformer_model = transformer_model
+        self.heuristic_model = UnicodeHeuristicModel()
+        self.weights = weights or {"ngram": 0.35, "fasttext": 0.50, "transformer": 0.15}
+        self.strategy = strategy
+        self.all_langs = ALL_LANGUAGE_CODES
+        self.stacking_meta: Optional[Any] = None
+        self.is_fitted = False
+
+    @property
+    def active_models(self) -> Dict[str, Any]:
+        models = {}
+        if self.ngram_model and getattr(self.ngram_model, "is_fitted", False):
+            models["ngram"] = self.ngram_model
+        if self.fasttext_model and getattr(self.fasttext_model, "is_fitted", False):
+            models["fasttext"] = self.fasttext_model
+        if self.transformer_model and getattr(self.transformer_model, "is_fitted", False):
+            models["transformer"] = self.transformer_model
+        return models
+
+    def fit_stacking(self, val_df: pd.DataFrame):
+        """
+        Fit meta-learner (stacking) on validation set predictions.
+        """
+        from sklearn.linear_model import LogisticRegression
+
+        logger.info("Fitting stacking meta-learner on validation set ...")
+        stacked_features = self._get_stacked_features(val_df["text"].tolist())
+        labels = val_df["lang"].tolist()
+
+        meta = LogisticRegression(C=1.0, max_iter=500, solver="lbfgs", multi_class="multinomial")
+        meta.fit(stacked_features, labels)
+        self.stacking_meta = meta
+        logger.success("Stacking meta-learner fitted.")
+
+    def _get_stacked_features(self, texts: List[str]) -> np.ndarray:
+        """Concatenate all model probability vectors."""
+        features = []
+        for name, model in self.active_models.items():
+            if hasattr(model, "predict_proba"):
+                probs = model.predict_proba(texts)
+                if isinstance(probs, tuple):
+                    probs, langs = probs
+                    # Reindex to self.all_langs
+                    full = np.zeros((len(texts), len(self.all_langs)))
+                    for j, lang in enumerate(langs):
+                        if lang in self.all_langs:
+                            k = self.all_langs.index(lang)
+                            full[:, k] = probs[:, j]
+                    probs = full
+                features.append(probs)
+        if not features:
+            raise RuntimeError("No active models for stacking")
+        return np.hstack(features)
+
+    def predict(self, texts: List[str]) -> List[str]:
+        results = self.predict_with_confidence(texts)
+        return [r["predicted_lang"] for r in results]
+
+    def predict_with_confidence(self, texts: List[str]) -> List[Dict]:
+        """Full ensemble inference with detailed outputs."""
+        results = []
+        heuristic_preds = self.heuristic_model.predict(texts)
+
+        # Get model probabilities
+        model_probs = {}
+        for name, model in self.active_models.items():
+            if hasattr(model, "predict_proba"):
+                try:
+                    probs = model.predict_proba(texts)
+                    if isinstance(probs, tuple):
+                        prob_matrix, langs = probs
+                        model_probs[name] = (prob_matrix, langs)
+                    else:
+                        # ngram model returns matrix with .classes_
+                        classes = model.classes_ if hasattr(model, "classes_") else self.all_langs
+                        model_probs[name] = (probs, list(classes))
+                except Exception as e:
+                    logger.warning(f"Model {name} failed: {e}")
+
+        for i, text in enumerate(texts):
+            h_lang, h_conf = heuristic_preds[i]
+
+            # High-confidence heuristic → use directly
+            if h_lang and h_conf >= 0.92 and h_lang not in DEVANAGARI_LANGUAGES:
+                results.append(self._build_result(text, h_lang, h_conf, {}, "heuristic"))
+                continue
+
+            # Weighted voting across ML models
+            combined = np.zeros(len(self.all_langs))
+            total_weight = 0.0
+
+            for name, (prob_matrix, langs) in model_probs.items():
+                w = self.weights.get(name, 1.0)
+                total_weight += w
+                prob_row = prob_matrix[i]
+                for j, lang in enumerate(langs):
+                    if lang in self.all_langs:
+                        k = self.all_langs.index(lang)
+                        combined[k] += w * (prob_row[j] if j < len(prob_row) else 0)
+
+            if total_weight > 0:
+                combined /= total_weight
+
+            # Blend in heuristic prior if partially confident
+            if h_lang and h_conf >= 0.55:
+                k = self.all_langs.index(h_lang) if h_lang in self.all_langs else -1
+                if k >= 0:
+                    alpha = 0.3
+                    combined[k] = (1 - alpha) * combined[k] + alpha * h_conf
+
+            best_idx = int(np.argmax(combined))
+            pred_lang = self.all_langs[best_idx]
+            confidence = float(combined[best_idx])
+
+            # Devanagari disambiguation (Hindi vs Marathi)
+            if pred_lang in DEVANAGARI_LANGUAGES:
+                pred_lang, confidence = self._disambiguate_devanagari(
+                    text, combined, pred_lang, confidence
+                )
+
+            source_contributions = {
+                name: self._get_model_top_pred(name, model_probs, i)
+                for name in model_probs
+            }
+            results.append(
+                self._build_result(text, pred_lang, confidence, source_contributions, "ensemble")
+            )
+
+        return results
+
+    def _disambiguate_devanagari(
+        self, text: str, probs: np.ndarray, pred: str, conf: float
+    ) -> Tuple[str, float]:
+        """
+        Use lexical markers to disambiguate Hindi vs Marathi.
+        Both use Devanagari, so char-level features alone aren't enough.
+        """
+        # Marathi-specific words/patterns
+        marathi_markers = ["आहे", "आहेत", "होते", "केले", "त्यांनी", "आणि", "मराठी"]
+        hindi_markers = ["है", "हैं", "था", "थे", "में", "और", "से", "हिंदी"]
+
+        mar_score = sum(1 for m in marathi_markers if m in text)
+        hin_score = sum(1 for m in hindi_markers if m in text)
+
+        if mar_score > hin_score + 1:
+            k = self.all_langs.index("mar") if "mar" in self.all_langs else -1
+            return "mar", min(0.95, conf + 0.1 * mar_score)
+        elif hin_score > mar_score + 1:
+            k = self.all_langs.index("hin") if "hin" in self.all_langs else -1
+            return "hin", min(0.95, conf + 0.05 * hin_score)
+        return pred, conf
+
+    def _get_model_top_pred(self, name: str, model_probs: Dict, idx: int) -> Dict:
+        if name not in model_probs:
+            return {}
+        prob_matrix, langs = model_probs[name]
+        prob_row = prob_matrix[idx]
+        best = int(np.argmax(prob_row))
+        return {
+            "lang": langs[best] if best < len(langs) else "?",
+            "confidence": float(prob_row[best]),
+        }
+
+    def _build_result(
+        self, text: str, lang: str, conf: float,
+        contributions: Dict, source: str
+    ) -> Dict:
+        info = LANGUAGE_REGISTRY.get(lang)
+        return {
+            "text_preview": text[:80] + "..." if len(text) > 80 else text,
+            "predicted_lang": lang,
+            "predicted_lang_name": info.name if info else lang,
+            "script": info.script if info else "Unknown",
+            "direction": info.direction if info else "LTR",
+            "confidence": round(conf, 4),
+            "is_confident": conf >= 0.75,
+            "source": source,
+            "model_contributions": contributions,
+        }
+
+    def evaluate(self, test_df: pd.DataFrame) -> Dict:
+        texts = test_df["text"].tolist()
+        true_labels = test_df["lang"].tolist()
+        results = self.predict_with_confidence(texts)
+        pred_labels = [r["predicted_lang"] for r in results]
+
+        accuracy = accuracy_score(true_labels, pred_labels)
+        macro_f1 = f1_score(true_labels, pred_labels, average="macro", zero_division=0)
+        report = classification_report(true_labels, pred_labels, output_dict=True, zero_division=0)
+
+        logger.info(f"\n{classification_report(true_labels, pred_labels, zero_division=0)}")
+        logger.success(f"Ensemble → Accuracy: {accuracy:.4f} | Macro-F1: {macro_f1:.4f}")
+
+        return {
+            "accuracy": accuracy,
+            "macro_f1": macro_f1,
+            "per_language": report,
+            "predictions": pred_labels,
+            "true_labels": true_labels,
+            "detailed_results": results,
+        }
+
+    def save(self, path: str):
+        Path(path).parent.mkdir(parents=True, exist_ok=True)
+        data = {
+            "weights": self.weights,
+            "strategy": self.strategy,
+            "stacking_meta": self.stacking_meta,
+            "all_langs": self.all_langs,
+        }
+        with open(path, "wb") as f:
+            pickle.dump(data, f)
+        logger.success(f"Ensemble config saved: {path}")

fasttext_model.py

@@ -0,0 +1,257 @@
+"""
+FastText-based Language Identification
+Uses Facebook's fastText for ultra-fast, high-accuracy language detection.
+Two modes:
+  1. Pretrained lid.176.ftz (off-the-shelf, 176 languages)
+  2. Custom-trained on Pralekha (domain-specific, higher accuracy on Indic)
+"""
+from __future__ import annotations
+
+import os
+import re
+import tempfile
+import urllib.request
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+
+import numpy as np
+import pandas as pd
+from loguru import logger
+from tqdm.auto import tqdm
+
+try:
+    import fasttext
+    FASTTEXT_AVAILABLE = True
+except ImportError:
+    FASTTEXT_AVAILABLE = False
+    logger.warning("fasttext not installed. `pip install fasttext-wheel`")
+
+from src.utils.languages import LANGUAGE_REGISTRY, CODE_TO_LABEL
+
+
+PRETRAINED_MODEL_URL = "https://dl.fbaipublicfiles.com/fasttext/supervised-models/lid.176.ftz"
+LANG_PREFIX = "__label__"
+
+# Pralekha lang codes → fastText ISO codes (for pretrained model)
+PRALEKHA_TO_FT = {
+    "ben": "bn", "eng": "en", "guj": "gu", "hin": "hi",
+    "kan": "kn", "mal": "ml", "mar": "mr", "ori": "or",
+    "pan": "pa", "tam": "ta", "tel": "te", "urd": "ur",
+}
+FT_TO_PRALEKHA = {v: k for k, v in PRALEKHA_TO_FT.items()}
+
+
+class FastTextLIDModel:
+    """
+    FastText-based Language Identification with two modes:
+      - mode='pretrained': Use Facebook's lid.176.ftz (quick, general)
+      - mode='custom'    : Train from scratch on Pralekha (best Indic accuracy)
+    """
+
+    def __init__(
+        self,
+        mode: str = "pretrained",   # "pretrained" | "custom"
+        model_dir: str = "./artifacts/fasttext",
+        dim: int = 100,
+        epoch: int = 25,
+        lr: float = 0.1,
+        word_ngrams: int = 3,
+        min_count: int = 5,
+        thread: int = 8,
+    ):
+        assert mode in ("pretrained", "custom")
+        self.mode = mode
+        self.model_dir = Path(model_dir)
+        self.model_dir.mkdir(parents=True, exist_ok=True)
+        self.dim = dim
+        self.epoch = epoch
+        self.lr = lr
+        self.word_ngrams = word_ngrams
+        self.min_count = min_count
+        self.thread = thread
+        self.model = None
+        self.is_fitted = False
+        self._target_langs = list(PRALEKHA_TO_FT.keys())
+
+    # ── Pretrained Mode ───────────────────────────────────────────────────────
+
+    def load_pretrained(self) -> "FastTextLIDModel":
+        """Download and load Facebook's lid.176.ftz pretrained model."""
+        assert FASTTEXT_AVAILABLE, "fasttext not installed"
+        model_path = self.model_dir / "lid.176.ftz"
+
+        if not model_path.exists():
+            logger.info(f"Downloading pretrained FastText LID model ...")
+            urllib.request.urlretrieve(
+                PRETRAINED_MODEL_URL,
+                str(model_path),
+                reporthook=self._download_progress,
+            )
+            logger.success("Download complete.")
+
+        self.model = fasttext.load_model(str(model_path))
+        self.is_fitted = True
+        self.mode = "pretrained"
+        logger.success("Pretrained FastText model loaded.")
+        return self
+
+    def _download_progress(self, block_num, block_size, total_size):
+        downloaded = block_num * block_size
+        pct = min(100, downloaded * 100 / total_size)
+        if block_num % 50 == 0:
+            logger.debug(f"Download: {pct:.1f}%")
+
+    # ── Custom Training Mode ──────────────────────────────────────────────────
+
+    def fit(
+        self,
+        train_df: pd.DataFrame,
+        val_df: Optional[pd.DataFrame] = None,
+    ) -> "FastTextLIDModel":
+        """Train custom FastText model on Pralekha data."""
+        assert FASTTEXT_AVAILABLE, "fasttext not installed"
+
+        logger.info(f"Preparing FastText training data ({len(train_df):,} samples) ...")
+        train_file = self.model_dir / "train.txt"
+        self._write_fasttext_format(train_df, train_file)
+
+        logger.info(f"Training FastText | dim={self.dim} | epoch={self.epoch} | lr={self.lr}")
+        self.model = fasttext.train_supervised(
+            input=str(train_file),
+            dim=self.dim,
+            epoch=self.epoch,
+            lr=self.lr,
+            wordNgrams=self.word_ngrams,
+            minCount=self.min_count,
+            thread=self.thread,
+            verbose=2,
+            loss="softmax",
+        )
+
+        model_path = self.model_dir / "custom_lid.bin"
+        self.model.save_model(str(model_path))
+        self.is_fitted = True
+        self.mode = "custom"
+
+        if val_df is not None:
+            results = self.evaluate(val_df)
+            logger.success(f"Val Accuracy: {results['accuracy']:.4f} | F1: {results['macro_f1']:.4f}")
+
+        return self
+
+    def _write_fasttext_format(self, df: pd.DataFrame, path: Path):
+        """Write data in fastText supervised format: __label__LANG text"""
+        with open(path, "w", encoding="utf-8") as f:
+            for _, row in tqdm(df.iterrows(), total=len(df), desc="Writing FT data"):
+                text = self._clean_for_fasttext(row["text"])
+                lang = row["lang"]
+                label = f"{LANG_PREFIX}{lang}"
+                f.write(f"{label} {text}\n")
+
+    @staticmethod
+    def _clean_for_fasttext(text: str) -> str:
+        """Minimal cleaning for fastText (preserve Unicode)."""
+        text = re.sub(r"\s+", " ", text).strip()
+        text = re.sub(r"[\r\n\t]", " ", text)
+        return text
+
+    # ── Inference ─────────────────────────────────────────────────────────────
+
+    def predict(self, texts: List[str]) -> List[str]:
+        """Predict language codes."""
+        assert self.is_fitted
+        results = []
+        for text in texts:
+            pred = self._predict_single(text)
+            results.append(pred)
+        return results
+
+    def _predict_single(self, text: str) -> str:
+        text = self._clean_for_fasttext(text)
+        labels, probs = self.model.predict(text, k=1)
+        raw_label = labels[0].replace(LANG_PREFIX, "")
+
+        if self.mode == "pretrained":
+            return FT_TO_PRALEKHA.get(raw_label, raw_label)
+        return raw_label
+
+    def predict_proba(self, texts: List[str]) -> Tuple[np.ndarray, List[str]]:
+        """
+        Returns (prob_matrix [n × n_classes], class_names).
+        """
+        assert self.is_fitted
+        all_labels = self._target_langs
+        n = len(texts)
+        n_classes = len(all_labels)
+        prob_matrix = np.zeros((n, n_classes))
+
+        for i, text in enumerate(texts):
+            text_clean = self._clean_for_fasttext(text)
+            labels, probs = self.model.predict(text_clean, k=-1)
+            for label, prob in zip(labels, probs):
+                raw = label.replace(LANG_PREFIX, "")
+                mapped = FT_TO_PRALEKHA.get(raw, raw) if self.mode == "pretrained" else raw
+                if mapped in all_labels:
+                    j = all_labels.index(mapped)
+                    prob_matrix[i, j] = prob
+
+        # Normalize
+        row_sums = prob_matrix.sum(axis=1, keepdims=True)
+        prob_matrix = np.where(row_sums > 0, prob_matrix / row_sums, prob_matrix)
+
+        return prob_matrix, all_labels
+
+    def predict_with_confidence(self, texts: List[str]) -> List[Dict]:
+        """Rich predictions with top-K confidences."""
+        results = []
+        for text in texts:
+            text_clean = self._clean_for_fasttext(text)
+            labels, probs = self.model.predict(text_clean, k=5)
+            top_k = []
+            for label, prob in zip(labels, probs):
+                raw = label.replace(LANG_PREFIX, "")
+                mapped = FT_TO_PRALEKHA.get(raw, raw) if self.mode == "pretrained" else raw
+                info = LANGUAGE_REGISTRY.get(mapped)
+                top_k.append({
+                    "lang": mapped,
+                    "name": info.name if info else mapped,
+                    "probability": float(prob),
+                })
+
+            results.append({
+                "text": text[:100],
+                "predicted_lang": top_k[0]["lang"],
+                "confidence": top_k[0]["probability"],
+                "is_confident": top_k[0]["probability"] >= 0.7,
+                "top_5": top_k,
+            })
+        return results
+
+    def evaluate(self, test_df: pd.DataFrame) -> Dict:
+        """Full evaluation."""
+        from sklearn.metrics import classification_report, f1_score, accuracy_score
+
+        texts = test_df["text"].tolist()
+        true_labels = test_df["lang"].tolist()
+        pred_labels = self.predict(texts)
+
+        accuracy = accuracy_score(true_labels, pred_labels)
+        macro_f1 = f1_score(true_labels, pred_labels, average="macro", zero_division=0)
+        report = classification_report(true_labels, pred_labels, output_dict=True, zero_division=0)
+
+        logger.info(f"\n{classification_report(true_labels, pred_labels, zero_division=0)}")
+        logger.success(f"Accuracy: {accuracy:.4f} | Macro-F1: {macro_f1:.4f}")
+
+        return {
+            "accuracy": accuracy,
+            "macro_f1": macro_f1,
+            "per_language": report,
+            "predictions": pred_labels,
+            "true_labels": true_labels,
+        }
+
+    def load(self, path: str) -> "FastTextLIDModel":
+        assert FASTTEXT_AVAILABLE
+        self.model = fasttext.load_model(path)
+        self.is_fitted = True
+        return self

languages.py

@@ -0,0 +1,182 @@
+"""
+Language Registry & Constants for AI4Bharat Pralekha Dataset
+Covers 12 Indic languages + English with full Unicode metadata.
+"""
+from __future__ import annotations
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Tuple
+import unicodedata
+
+
+# ─────────────────────────────────────────────
+# Script Unicode Block Definitions
+# ─────────────────────────────────────────────
+SCRIPT_UNICODE_BLOCKS: Dict[str, List[Tuple[int, int]]] = {
+    "Devanagari":  [(0x0900, 0x097F), (0xA8E0, 0xA8FF)],
+    "Bengali":     [(0x0980, 0x09FF)],
+    "Gurmukhi":    [(0x0A00, 0x0A7F)],
+    "Gujarati":    [(0x0A80, 0x0AFF)],
+    "Odia":        [(0x0B00, 0x0B7F)],
+    "Tamil":       [(0x0B80, 0x0BFF)],
+    "Telugu":      [(0x0C00, 0x0C7F)],
+    "Kannada":     [(0x0C80, 0x0CFF)],
+    "Malayalam":   [(0x0D00, 0x0D7F)],
+    "Sinhala":     [(0x0D80, 0x0DFF)],
+    "Arabic":      [(0x0600, 0x06FF), (0x0750, 0x077F), (0xFB50, 0xFDFF), (0xFE70, 0xFEFF)],
+    "Latin":       [(0x0000, 0x007F), (0x0080, 0x00FF), (0x0100, 0x017F)],
+    "Cyrillic":    [(0x0400, 0x04FF)],
+    "CJK":         [(0x4E00, 0x9FFF), (0x3400, 0x4DBF)],
+}
+
+SCRIPT_TO_DIRECTION: Dict[str, str] = {
+    "Devanagari": "LTR",
+    "Bengali":    "LTR",
+    "Gurmukhi":   "LTR",
+    "Gujarati":   "LTR",
+    "Odia":       "LTR",
+    "Tamil":      "LTR",
+    "Telugu":     "LTR",
+    "Kannada":    "LTR",
+    "Malayalam":  "LTR",
+    "Arabic":     "RTL",
+    "Latin":      "LTR",
+}
+
+
+@dataclass
+class LanguageInfo:
+    code: str                    # Pralekha subset code (ben, hin, ...)
+    name: str                    # Full English name
+    script: str                  # Primary script name
+    iso639_1: Optional[str]      # ISO 639-1 two-letter code
+    iso639_3: str                # ISO 639-3 three-letter code
+    label_id: int                # Numeric label for ML models
+    unicode_ranges: List[Tuple[int, int]] = field(default_factory=list)
+    direction: str = "LTR"
+    family: str = "Indo-Aryan"
+    native_name: str = ""
+    romanization_scheme: Optional[str] = None  # IAST, ISO 15919, etc.
+    shares_script_with: List[str] = field(default_factory=list)
+
+    def char_in_script(self, char: str) -> bool:
+        cp = ord(char)
+        return any(lo <= cp <= hi for lo, hi in self.unicode_ranges)
+
+
+# ─────────────────────────────────────────────
+# Language Registry (12 Indic + English)
+# ─────────────────────────────────────────────
+LANGUAGE_REGISTRY: Dict[str, LanguageInfo] = {
+    "ben": LanguageInfo(
+        code="ben", name="Bengali", script="Bengali",
+        iso639_1="bn", iso639_3="ben", label_id=0,
+        unicode_ranges=SCRIPT_UNICODE_BLOCKS["Bengali"],
+        direction="LTR", family="Indo-Aryan",
+        native_name="বাংলা",
+        romanization_scheme="ISO 15919",
+    ),
+    "eng": LanguageInfo(
+        code="eng", name="English", script="Latin",
+        iso639_1="en", iso639_3="eng", label_id=1,
+        unicode_ranges=SCRIPT_UNICODE_BLOCKS["Latin"],
+        direction="LTR", family="Germanic",
+        native_name="English",
+    ),
+    "guj": LanguageInfo(
+        code="guj", name="Gujarati", script="Gujarati",
+        iso639_1="gu", iso639_3="guj", label_id=2,
+        unicode_ranges=SCRIPT_UNICODE_BLOCKS["Gujarati"],
+        direction="LTR", family="Indo-Aryan",
+        native_name="ગુજરાતી",
+        romanization_scheme="ISO 15919",
+    ),
+    "hin": LanguageInfo(
+        code="hin", name="Hindi", script="Devanagari",
+        iso639_1="hi", iso639_3="hin", label_id=3,
+        unicode_ranges=SCRIPT_UNICODE_BLOCKS["Devanagari"],
+        direction="LTR", family="Indo-Aryan",
+        native_name="हिन्दी",
+        romanization_scheme="IAST",
+        shares_script_with=["mar"],
+    ),
+    "kan": LanguageInfo(
+        code="kan", name="Kannada", script="Kannada",
+        iso639_1="kn", iso639_3="kan", label_id=4,
+        unicode_ranges=SCRIPT_UNICODE_BLOCKS["Kannada"],
+        direction="LTR", family="Dravidian",
+        native_name="ಕನ್ನಡ",
+    ),
+    "mal": LanguageInfo(
+        code="mal", name="Malayalam", script="Malayalam",
+        iso639_1="ml", iso639_3="mal", label_id=5,
+        unicode_ranges=SCRIPT_UNICODE_BLOCKS["Malayalam"],
+        direction="LTR", family="Dravidian",
+        native_name="മലയാളം",
+    ),
+    "mar": LanguageInfo(
+        code="mar", name="Marathi", script="Devanagari",
+        iso639_1="mr", iso639_3="mar", label_id=6,
+        unicode_ranges=SCRIPT_UNICODE_BLOCKS["Devanagari"],
+        direction="LTR", family="Indo-Aryan",
+        native_name="मराठी",
+        romanization_scheme="IAST",
+        shares_script_with=["hin"],
+    ),
+    "ori": LanguageInfo(
+        code="ori", name="Odia", script="Odia",
+        iso639_1="or", iso639_3="ori", label_id=7,
+        unicode_ranges=SCRIPT_UNICODE_BLOCKS["Odia"],
+        direction="LTR", family="Indo-Aryan",
+        native_name="ଓଡ଼ିଆ",
+    ),
+    "pan": LanguageInfo(
+        code="pan", name="Punjabi", script="Gurmukhi",
+        iso639_1="pa", iso639_3="pan", label_id=8,
+        unicode_ranges=SCRIPT_UNICODE_BLOCKS["Gurmukhi"],
+        direction="LTR", family="Indo-Aryan",
+        native_name="ਪੰਜਾਬੀ",
+    ),
+    "tam": LanguageInfo(
+        code="tam", name="Tamil", script="Tamil",
+        iso639_1="ta", iso639_3="tam", label_id=9,
+        unicode_ranges=SCRIPT_UNICODE_BLOCKS["Tamil"],
+        direction="LTR", family="Dravidian",
+        native_name="தமிழ்",
+    ),
+    "tel": LanguageInfo(
+        code="tel", name="Telugu", script="Telugu",
+        iso639_1="te", iso639_3="tel", label_id=10,
+        unicode_ranges=SCRIPT_UNICODE_BLOCKS["Telugu"],
+        direction="LTR", family="Dravidian",
+        native_name="తెలుగు",
+    ),
+    "urd": LanguageInfo(
+        code="urd", name="Urdu", script="Arabic",
+        iso639_1="ur", iso639_3="urd", label_id=11,
+        unicode_ranges=SCRIPT_UNICODE_BLOCKS["Arabic"],
+        direction="RTL", family="Indo-Aryan",
+        native_name="اردو",
+        romanization_scheme="ALA-LC",
+    ),
+}
+
+# Derived mappings
+LABEL_TO_CODE: Dict[int, str] = {v.label_id: k for k, v in LANGUAGE_REGISTRY.items()}
+CODE_TO_LABEL: Dict[str, int] = {k: v.label_id for k, v in LANGUAGE_REGISTRY.items()}
+ALL_LANGUAGE_CODES: List[str] = sorted(LANGUAGE_REGISTRY.keys())
+NUM_CLASSES: int = len(LANGUAGE_REGISTRY)
+
+# Script families for analysis
+DRAVIDIAN_LANGUAGES = {"kan", "mal", "tam", "tel"}
+INDO_ARYAN_LANGUAGES = {"ben", "guj", "hin", "mar", "ori", "pan", "urd"}
+DEVANAGARI_LANGUAGES = {"hin", "mar"}
+RTL_LANGUAGES = {"urd"}
+
+# Confusable pairs (share script or similar features)
+CONFUSABLE_PAIRS = [
+    ("hin", "mar"),   # Both Devanagari
+    ("tam", "mal"),   # Related Dravidian scripts
+    ("kan", "tel"),   # Related Dravidian scripts
+    ("hin", "urd"),   # Same spoken language, different scripts
+    ("ben", "ori"),   # Visually similar scripts
+]

ngram_model.py

@@ -0,0 +1,284 @@
+"""
+Character N-gram Language Identification Model
+High-accuracy baseline using TF-IDF character n-grams + Logistic Regression.
+Achieves ~98%+ accuracy on clean Indic text.
+"""
+from __future__ import annotations
+
+import pickle
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+
+import numpy as np
+import pandas as pd
+from loguru import logger
+from sklearn.calibration import CalibratedClassifierCV
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.linear_model import LogisticRegression
+from sklearn.metrics import classification_report, f1_score
+from sklearn.pipeline import Pipeline
+from sklearn.preprocessing import LabelEncoder
+
+from src.utils.languages import LANGUAGE_REGISTRY, LABEL_TO_CODE, CODE_TO_LABEL
+
+
+class CharNgramLIDModel:
+    """
+    Character N-gram Language Identification using TF-IDF + LogReg.
+
+    Key design choices:
+      - analyzer='char_wb': treats word boundaries specially
+      - ngram_range=(2,4): captures morpheme-level patterns
+      - sublinear_tf=True: dampens frequency explosion
+      - Calibration: isotonic regression for reliable probabilities
+      - Per-language confidence thresholds for rejection
+    """
+
+    def __init__(
+        self,
+        ngram_range: Tuple[int, int] = (2, 4),
+        max_features: int = 150_000,
+        C: float = 5.0,
+        max_iter: int = 1000,
+        calibrate: bool = True,
+        n_jobs: int = -1,
+    ):
+        self.ngram_range = ngram_range
+        self.max_features = max_features
+        self.C = C
+        self.max_iter = max_iter
+        self.calibrate = calibrate
+        self.n_jobs = n_jobs
+        self.pipeline: Optional[Pipeline] = None
+        self.label_encoder = LabelEncoder()
+        self.classes_: Optional[np.ndarray] = None
+        self.confidence_thresholds: Dict[str, float] = {}
+        self.is_fitted = False
+
+    def _build_pipeline(self) -> Pipeline:
+        vectorizer = TfidfVectorizer(
+            analyzer="char_wb",
+            ngram_range=self.ngram_range,
+            max_features=self.max_features,
+            sublinear_tf=True,
+            strip_accents=None,     # preserve Indic diacritics!
+            min_df=2,
+            max_df=0.95,
+            dtype=np.float32,
+        )
+        base_clf = LogisticRegression(
+            C=self.C,
+            max_iter=self.max_iter,
+            solver="saga",
+            multi_class="multinomial",
+            n_jobs=self.n_jobs,
+            verbose=0,
+        )
+        if self.calibrate:
+            clf = CalibratedClassifierCV(base_clf, method="isotonic", cv=3)
+        else:
+            clf = base_clf
+
+        return Pipeline([
+            ("tfidf", vectorizer),
+            ("clf", clf),
+        ])
+
+    def fit(
+        self,
+        train_df: pd.DataFrame,
+        val_df: Optional[pd.DataFrame] = None,
+    ) -> "CharNgramLIDModel":
+        logger.info(
+            f"Training CharNgramLID | n_gram={self.ngram_range} | "
+            f"features={self.max_features} | C={self.C}"
+        )
+        texts = train_df["text"].tolist()
+        labels = self.label_encoder.fit_transform(train_df["lang"].tolist())
+        self.classes_ = self.label_encoder.classes_
+
+        self.pipeline = self._build_pipeline()
+        self.pipeline.fit(texts, labels)
+        self.is_fitted = True
+
+        if val_df is not None:
+            self._compute_confidence_thresholds(val_df)
+            val_preds = self.predict(val_df["text"].tolist())
+            f1 = f1_score(val_df["lang"].tolist(), val_preds, average="macro")
+            logger.success(f"Val Macro-F1: {f1:.4f}")
+
+        return self
+
+    def predict(self, texts: List[str]) -> List[str]:
+        assert self.is_fitted, "Model not fitted!"
+        label_ids = self.pipeline.predict(texts)
+        return self.label_encoder.inverse_transform(label_ids).tolist()
+
+    def predict_proba(self, texts: List[str]) -> np.ndarray:
+        """Returns probability matrix [n_samples × n_classes]."""
+        assert self.is_fitted
+        return self.pipeline.predict_proba(texts)
+
+    def predict_with_confidence(
+        self, texts: List[str]
+    ) -> List[Dict]:
+        """
+        Returns per-sample predictions with confidence and
+        top-K alternatives.
+        """
+        probs = self.predict_proba(texts)
+        results = []
+        for i, prob_row in enumerate(probs):
+            top_k_idx = np.argsort(prob_row)[::-1][:5]
+            top_k = [
+                {
+                    "lang": self.classes_[j],
+                    "name": LANGUAGE_REGISTRY.get(self.classes_[j], None),
+                    "probability": float(prob_row[j]),
+                }
+                for j in top_k_idx
+            ]
+            best = top_k[0]
+            threshold = self.confidence_thresholds.get(best["lang"], 0.5)
+            results.append({
+                "text": texts[i][:100] + "..." if len(texts[i]) > 100 else texts[i],
+                "predicted_lang": best["lang"],
+                "confidence": best["probability"],
+                "is_confident": best["probability"] >= threshold,
+                "top_5": top_k,
+            })
+        return results
+
+    def _compute_confidence_thresholds(
+        self, val_df: pd.DataFrame, target_precision: float = 0.95
+    ):
+        """
+        Compute per-language confidence thresholds that achieve
+        target_precision on the validation set.
+        """
+        probs = self.predict_proba(val_df["text"].tolist())
+        true_langs = val_df["lang"].tolist()
+        pred_langs = self.predict(val_df["text"].tolist())
+
+        for lang in self.classes_:
+            lang_idx = np.where(self.classes_ == lang)[0]
+            if len(lang_idx) == 0:
+                continue
+            idx = lang_idx[0]
+            lang_probs = probs[:, idx]
+            is_correct = np.array([p == lang for p in pred_langs])
+            is_predicted = np.array([p == lang for p in pred_langs])
+
+            # Find threshold where precision ≥ target
+            thresholds = np.linspace(0.3, 0.99, 50)
+            best_t = 0.5
+            for t in thresholds:
+                mask = (lang_probs >= t) & is_predicted
+                if mask.sum() == 0:
+                    break
+                prec = is_correct[mask].mean()
+                if prec >= target_precision:
+                    best_t = t
+                    break
+            self.confidence_thresholds[lang] = best_t
+
+        logger.debug(f"Confidence thresholds: {self.confidence_thresholds}")
+
+    def evaluate(self, test_df: pd.DataFrame) -> Dict:
+        """Full evaluation with per-language metrics."""
+        texts = test_df["text"].tolist()
+        true_labels = test_df["lang"].tolist()
+        pred_labels = self.predict(texts)
+        probs = self.predict_proba(texts)
+
+        overall_f1 = f1_score(true_labels, pred_labels, average="macro")
+        report = classification_report(
+            true_labels, pred_labels,
+            target_names=self.classes_,
+            output_dict=True,
+        )
+
+        # Confidence calibration: ECE (Expected Calibration Error)
+        confidences = probs.max(axis=1)
+        correct = np.array(true_labels) == np.array(pred_labels)
+        ece = self._compute_ece(confidences, correct)
+
+        logger.info(f"\n{classification_report(true_labels, pred_labels)}")
+        logger.success(f"Macro F1: {overall_f1:.4f} | ECE: {ece:.4f}")
+
+        return {
+            "macro_f1": overall_f1,
+            "ece": ece,
+            "per_language": report,
+            "predictions": pred_labels,
+            "true_labels": true_labels,
+            "confidences": confidences.tolist(),
+        }
+
+    @staticmethod
+    def _compute_ece(confidences: np.ndarray, correct: np.ndarray, n_bins: int = 10) -> float:
+        bins = np.linspace(0, 1, n_bins + 1)
+        ece = 0.0
+        for lo, hi in zip(bins[:-1], bins[1:]):
+            mask = (confidences >= lo) & (confidences < hi)
+            if mask.sum() == 0:
+                continue
+            avg_conf = confidences[mask].mean()
+            avg_acc = correct[mask].mean()
+            ece += mask.mean() * abs(avg_conf - avg_acc)
+        return float(ece)
+
+    def save(self, path: str):
+        Path(path).parent.mkdir(parents=True, exist_ok=True)
+        with open(path, "wb") as f:
+            pickle.dump({
+                "pipeline": self.pipeline,
+                "label_encoder": self.label_encoder,
+                "classes_": self.classes_,
+                "confidence_thresholds": self.confidence_thresholds,
+                "config": {
+                    "ngram_range": self.ngram_range,
+                    "max_features": self.max_features,
+                    "C": self.C,
+                },
+            }, f)
+        logger.success(f"Model saved: {path}")
+
+    @classmethod
+    def load(cls, path: str) -> "CharNgramLIDModel":
+        with open(path, "rb") as f:
+            data = pickle.load(f)
+        cfg = data.get("config", {})
+        model = cls(
+            ngram_range=cfg.get("ngram_range", (2, 4)),
+            max_features=cfg.get("max_features", 150_000),
+            C=cfg.get("C", 5.0),
+        )
+        model.pipeline = data["pipeline"]
+        model.label_encoder = data["label_encoder"]
+        model.classes_ = data["classes_"]
+        model.confidence_thresholds = data.get("confidence_thresholds", {})
+        model.is_fitted = True
+        logger.success(f"Model loaded: {path}")
+        return model
+
+    def get_top_features_per_language(self, n: int = 20) -> Dict[str, List[str]]:
+        """Extract most discriminative n-grams per language."""
+        assert self.is_fitted
+        tfidf = self.pipeline.named_steps["tfidf"]
+        clf = self.pipeline.named_steps["clf"]
+        feature_names = np.array(tfidf.get_feature_names_out())
+
+        # Get the underlying LR if calibrated
+        if hasattr(clf, "calibrated_classifiers_"):
+            coef = np.mean(
+                [c.estimator.coef_ for c in clf.calibrated_classifiers_], axis=0
+            )
+        else:
+            coef = clf.coef_
+
+        result = {}
+        for i, lang in enumerate(self.classes_):
+            top_idx = np.argsort(coef[i])[::-1][:n]
+            result[lang] = feature_names[top_idx].tolist()
+        return result

preprocessor.py

@@ -0,0 +1,268 @@
+"""
+Advanced Text Preprocessing for Multilingual Indic Text
+Handles Unicode normalization, noise removal, script detection preprocessing.
+"""
+from __future__ import annotations
+
+import re
+import unicodedata
+from dataclasses import dataclass
+from typing import Dict, List, Optional, Tuple
+
+import numpy as np
+from loguru import logger
+
+from src.utils.languages import SCRIPT_UNICODE_BLOCKS, LANGUAGE_REGISTRY
+
+
+# ─────────────────────────────────────────────
+# Unicode Script Detection
+# ─────────────────────────────────────────────
+
+def get_char_script(char: str) -> str:
+    """Get Unicode script name for a single character."""
+    try:
+        cat = unicodedata.category(char)
+        name = unicodedata.name(char, "")
+        # Extract script from Unicode name (e.g. "DEVANAGARI LETTER A" → "Devanagari")
+        parts = name.split()
+        if not parts:
+            return "Unknown"
+        first = parts[0].capitalize()
+        # Multi-word scripts
+        for script in SCRIPT_UNICODE_BLOCKS:
+            if script.upper() in name.upper():
+                return script
+        return first
+    except (ValueError, TypeError):
+        return "Unknown"
+
+
+def get_dominant_script(text: str) -> Tuple[str, Dict[str, float]]:
+    """
+    Detect the dominant script in text.
+    Returns (dominant_script, script_distribution).
+    """
+    script_counts: Dict[str, int] = {}
+    total_alpha = 0
+
+    for char in text:
+        if not char.isalpha():
+            continue
+        total_alpha += 1
+        cp = ord(char)
+
+        matched = False
+        for script, ranges in SCRIPT_UNICODE_BLOCKS.items():
+            for lo, hi in ranges:
+                if lo <= cp <= hi:
+                    script_counts[script] = script_counts.get(script, 0) + 1
+                    matched = True
+                    break
+            if matched:
+                break
+        if not matched:
+            script_counts["Other"] = script_counts.get("Other", 0) + 1
+
+    if total_alpha == 0:
+        return "Unknown", {}
+
+    dist = {s: c / total_alpha for s, c in script_counts.items()}
+    dominant = max(dist, key=dist.get) if dist else "Unknown"
+    return dominant, dist
+
+
+# ─────────────────────────────────────────────
+# Preprocessor Config
+# ─────────────────────────────────────────────
+
+@dataclass
+class PreprocessorConfig:
+    normalize_unicode: bool = True
+    nfc_form: str = "NFC"
+    remove_urls: bool = True
+    remove_emails: bool = True
+    remove_html_tags: bool = True
+    normalize_whitespace: bool = True
+    remove_zero_width: bool = True
+    handle_mixed_numerals: bool = True
+    preserve_script_punctuation: bool = True
+    lowercase_latin: bool = False
+    max_length: Optional[int] = 2000
+    min_length: int = 10
+
+
+# ─────────────────────────────────────────────
+# Main Preprocessor
+# ─────────────────────────────────────────────
+
+class IndicTextPreprocessor:
+    """
+    Comprehensive text preprocessor for Indic multilingual corpora.
+    Handles Unicode normalization, noise removal, and script-aware cleaning.
+    """
+
+    # Regex patterns
+    _URL_RE = re.compile(
+        r"https?://\S+|www\.\S+|ftp://\S+", re.UNICODE
+    )
+    _EMAIL_RE = re.compile(
+        r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"
+    )
+    _HTML_RE = re.compile(r"<[^>]+>")
+    _WHITESPACE_RE = re.compile(r"\s+", re.UNICODE)
+    _ZERO_WIDTH_RE = re.compile(
+        r"[\u200b\u200c\u200d\u200e\u200f\u202a-\u202e\ufeff\u00ad]"
+    )
+    _MULTI_PUNCT_RE = re.compile(r"([!?।॥,;.]){2,}")
+
+    # Indic numeral mappings to ASCII (for normalization)
+    _NUMERAL_MAP = str.maketrans(
+        "".join([
+            "०१२३४५६७८९",   # Devanagari
+            "০১২৩৪৫৬৭৮৯",   # Bengali
+            "੦੧੨੩੪੫੬੭੮੯",   # Gurmukhi
+            "૦૧૨૩૪૫૬૭૮૯",   # Gujarati
+            "୦୧୨୩୪୫୬୭୮୯",   # Odia
+            "௦௧௨௩௪௫௬௭௮௯",   # Tamil
+            "౦౧౨౩౪౫౬౭౮౯",   # Telugu
+            "೦೧೨೩೪೫೬೭೮೯",   # Kannada
+            "൦൧൨൩൪൫൬൭൮൯",   # Malayalam
+        ]),
+        "0123456789" * 9
+    )
+
+    def __init__(self, config: Optional[PreprocessorConfig] = None):
+        self.config = config or PreprocessorConfig()
+
+    def preprocess(self, text: str) -> str:
+        """Full preprocessing pipeline."""
+        if not text or not isinstance(text, str):
+            return ""
+
+        cfg = self.config
+
+        # 1. Unicode normalization
+        if cfg.normalize_unicode:
+            text = unicodedata.normalize(cfg.nfc_form, text)
+
+        # 2. Remove zero-width and invisible characters
+        if cfg.remove_zero_width:
+            text = self._ZERO_WIDTH_RE.sub("", text)
+
+        # 3. Remove HTML tags
+        if cfg.remove_html_tags:
+            text = self._HTML_RE.sub(" ", text)
+
+        # 4. Remove URLs
+        if cfg.remove_urls:
+            text = self._URL_RE.sub(" ", text)
+
+        # 5. Remove emails
+        if cfg.remove_emails:
+            text = self._EMAIL_RE.sub(" ", text)
+
+        # 6. Numeral normalization (optional)
+        if cfg.handle_mixed_numerals:
+            text = text.translate(self._NUMERAL_MAP)
+
+        # 7. Collapse multiple punctuation
+        text = self._MULTI_PUNCT_RE.sub(r"\1", text)
+
+        # 8. Normalize whitespace
+        if cfg.normalize_whitespace:
+            text = self._WHITESPACE_RE.sub(" ", text).strip()
+
+        # 9. Lowercase Latin (optional, e.g. for English)
+        if cfg.lowercase_latin:
+            text = self._selective_lowercase(text)
+
+        # 10. Length truncation
+        if cfg.max_length and len(text) > cfg.max_length:
+            text = text[:cfg.max_length]
+
+        return text
+
+    def _selective_lowercase(self, text: str) -> str:
+        """Lowercase only Latin characters, preserve others."""
+        result = []
+        for ch in text:
+            cp = ord(ch)
+            # Latin Basic + Extended
+            if 0x0041 <= cp <= 0x005A or 0x00C0 <= cp <= 0x00D6:
+                result.append(ch.lower())
+            else:
+                result.append(ch)
+        return "".join(result)
+
+    def batch_preprocess(
+        self, texts: List[str], show_progress: bool = True
+    ) -> List[str]:
+        """Preprocess a batch of texts."""
+        if show_progress:
+            from tqdm.auto import tqdm
+            return [self.preprocess(t) for t in tqdm(texts, desc="Preprocessing")]
+        return [self.preprocess(t) for t in texts]
+
+    def quality_score(self, text: str) -> float:
+        """
+        Returns a text quality score in [0, 1].
+        Considers: alpha ratio, script consistency, length adequacy.
+        """
+        if not text:
+            return 0.0
+
+        alpha = sum(c.isalpha() for c in text)
+        total = len(text)
+        if total == 0:
+            return 0.0
+
+        alpha_ratio = alpha / total
+
+        _, script_dist = get_dominant_script(text)
+        script_concentration = max(script_dist.values()) if script_dist else 0.0
+
+        length_score = min(1.0, len(text) / 100)
+
+        return 0.3 * alpha_ratio + 0.4 * script_concentration + 0.3 * length_score
+
+    def compute_text_features(self, text: str) -> Dict:
+        """Rich feature set for a single text (used in analysis)."""
+        dominant, dist = get_dominant_script(text)
+        n_chars = len(text)
+        n_alpha = sum(c.isalpha() for c in text)
+        n_digits = sum(c.isdigit() for c in text)
+        n_punct = sum(unicodedata.category(c).startswith("P") for c in text)
+        n_spaces = sum(c.isspace() for c in text)
+        unique_chars = len(set(text))
+        type_token_ratio = unique_chars / max(n_chars, 1)
+
+        # Character entropy
+        from collections import Counter as _Counter
+        char_counts = _Counter(text)
+        total_f = sum(char_counts.values())
+        char_entropy = -sum(
+            (c / total_f) * np.log2(c / total_f)
+            for c in char_counts.values() if c > 0
+        ) if total_f > 0 else 0.0
+
+        return {
+            "n_chars": n_chars,
+            "n_alpha": n_alpha,
+            "n_digits": n_digits,
+            "n_punct": n_punct,
+            "n_spaces": n_spaces,
+            "alpha_ratio": n_alpha / max(n_chars, 1),
+            "digit_ratio": n_digits / max(n_chars, 1),
+            "unique_char_ratio": type_token_ratio,
+            "dominant_script": dominant,
+            "script_distribution": dist,
+            "script_count": len(dist),
+            "is_mixed_script": len(dist) > 1 and max(dist.values()) < 0.95,
+            "quality_score": self.quality_score(text),
+            "n_words": len(text.split()),
+            "avg_word_len": (
+                np.mean([len(w) for w in text.split()]) if text.split() else 0
+            ),
+            "char_entropy": float(char_entropy),
+        }

requirements.txt

@@ -0,0 +1,19 @@
+# Core
+numpy>=1.26.0
+pandas>=2.2.0
+scikit-learn>=1.4.0
+scipy>=1.12.0
+
+# NLP / Models
+fasttext-wheel>=0.9.2
+
+# Unicode & text processing
+regex>=2023.12.25
+
+# API & UI
+gradio>=4.24.0
+plotly>=5.20.0
+
+# Utilities
+loguru>=0.7.2
+tqdm>=4.66.0

script_analyzer.py

@@ -0,0 +1,462 @@
+"""
+Advanced Script Analysis Engine
+Deep Unicode-based analysis: script distribution, mixing, directionality,
+transliteration detection, noise analysis, and linguistic features.
+"""
+from __future__ import annotations
+
+import re
+import unicodedata
+from collections import Counter, defaultdict
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Set, Tuple
+
+import numpy as np
+from loguru import logger
+
+from src.utils.languages import (
+    SCRIPT_UNICODE_BLOCKS,
+    SCRIPT_TO_DIRECTION,
+    LANGUAGE_REGISTRY,
+)
+
+
+# ─────────────────────────────────────────────
+# Unicode Block Database
+# ─────────────────────────────────────────────
+
+UNICODE_BLOCK_NAMES: List[Tuple[int, int, str]] = [
+    (0x0000, 0x007F, "Basic Latin"),
+    (0x0080, 0x00FF, "Latin-1 Supplement"),
+    (0x0100, 0x017F, "Latin Extended-A"),
+    (0x0180, 0x024F, "Latin Extended-B"),
+    (0x0250, 0x02AF, "IPA Extensions"),
+    (0x0300, 0x036F, "Combining Diacritical Marks"),
+    (0x0370, 0x03FF, "Greek"),
+    (0x0400, 0x04FF, "Cyrillic"),
+    (0x0500, 0x052F, "Cyrillic Supplement"),
+    (0x0600, 0x06FF, "Arabic"),
+    (0x0700, 0x074F, "Syriac"),
+    (0x0750, 0x077F, "Arabic Supplement"),
+    (0x0900, 0x097F, "Devanagari"),
+    (0x0980, 0x09FF, "Bengali"),
+    (0x0A00, 0x0A7F, "Gurmukhi"),
+    (0x0A80, 0x0AFF, "Gujarati"),
+    (0x0B00, 0x0B7F, "Oriya"),
+    (0x0B80, 0x0BFF, "Tamil"),
+    (0x0C00, 0x0C7F, "Telugu"),
+    (0x0C80, 0x0CFF, "Kannada"),
+    (0x0D00, 0x0D7F, "Malayalam"),
+    (0x0D80, 0x0DFF, "Sinhala"),
+    (0x0E00, 0x0E7F, "Thai"),
+    (0x0F00, 0x0FFF, "Tibetan"),
+    (0x1000, 0x109F, "Myanmar"),
+    (0x10A0, 0x10FF, "Georgian"),
+    (0x1100, 0x11FF, "Hangul Jamo"),
+    (0x1E00, 0x1EFF, "Latin Extended Additional"),
+    (0x2000, 0x206F, "General Punctuation"),
+    (0x20A0, 0x20CF, "Currency Symbols"),
+    (0x2100, 0x214F, "Letterlike Symbols"),
+    (0x2200, 0x22FF, "Mathematical Operators"),
+    (0x3000, 0x303F, "CJK Symbols and Punctuation"),
+    (0x3040, 0x309F, "Hiragana"),
+    (0x30A0, 0x30FF, "Katakana"),
+    (0x4E00, 0x9FFF, "CJK Unified Ideographs"),
+    (0xFB00, 0xFB4F, "Alphabetic Presentation Forms"),
+    (0xFB50, 0xFDFF, "Arabic Presentation Forms-A"),
+    (0xFE70, 0xFEFF, "Arabic Presentation Forms-B"),
+    (0xFF00, 0xFFEF, "Halfwidth and Fullwidth Forms"),
+]
+
+
+def get_unicode_block(cp: int) -> str:
+    for lo, hi, name in UNICODE_BLOCK_NAMES:
+        if lo <= cp <= hi:
+            return name
+    return f"U+{cp:04X} (unknown block)"
+
+
+# ─────────────────────────────────────────────
+# Data Structures
+# ─────────────────────────────────────────────
+
+@dataclass
+class CharacterAnalysis:
+    char: str
+    codepoint: int
+    unicode_name: str
+    unicode_category: str
+    unicode_block: str
+    script: Optional[str]
+    is_alpha: bool
+    is_digit: bool
+    is_punct: bool
+    is_combining: bool
+
+
+@dataclass
+class ScriptSegment:
+    script: str
+    start: int
+    end: int
+    text: str
+    char_count: int
+
+
+@dataclass
+class ScriptAnalysisResult:
+    text: str
+    total_chars: int
+    alpha_chars: int
+    digit_chars: int
+    punct_chars: int
+    space_chars: int
+    other_chars: int
+
+    # Script distribution
+    script_distribution: Dict[str, float]
+    dominant_script: str
+    dominant_script_ratio: float
+
+    # Mixing analysis
+    is_mixed_script: bool
+    mixing_score: float           # 0=pure, 1=maximally mixed
+    script_segments: List[ScriptSegment]
+    script_transitions: int
+
+    # Directionality
+    primary_direction: str
+    has_bidi: bool
+    rtl_ratio: float
+
+    # Noise indicators
+    noise_score: float
+    has_zero_width: bool
+    has_control_chars: bool
+    has_excessive_punct: bool
+    url_count: int
+    emoji_count: int
+    latin_ratio: float            # in otherwise Indic text → potential transliteration
+
+    # Unicode statistics
+    unicode_block_distribution: Dict[str, float]
+    unique_chars: int
+    char_entropy: float
+
+    # Language hints
+    script_to_lang_hints: Dict[str, List[str]]
+
+    def to_dict(self) -> Dict:
+        return {
+            "total_chars": self.total_chars,
+            "alpha_chars": self.alpha_chars,
+            "alpha_ratio": self.alpha_chars / max(self.total_chars, 1),
+            "dominant_script": self.dominant_script,
+            "dominant_script_ratio": self.dominant_script_ratio,
+            "is_mixed_script": self.is_mixed_script,
+            "mixing_score": self.mixing_score,
+            "script_transitions": self.script_transitions,
+            "primary_direction": self.primary_direction,
+            "has_bidi": self.has_bidi,
+            "noise_score": self.noise_score,
+            "char_entropy": self.char_entropy,
+            "script_distribution": self.script_distribution,
+            "script_to_lang_hints": self.script_to_lang_hints,
+        }
+
+
+# ─────────────────────────────────────────────
+# Script Analyser
+# ─────────────────────────────────────────────
+
+class ScriptAnalyzer:
+    """
+    Deep script and Unicode analysis for multilingual text.
+    Designed for Indic languages with mixed-script handling.
+    """
+
+    _URL_RE = re.compile(r"https?://\S+|www\.\S+")
+    _EMOJI_RE = re.compile(
+        "["
+        "\U0001F600-\U0001F64F"
+        "\U0001F300-\U0001F5FF"
+        "\U0001F680-\U0001F6FF"
+        "\U0001F1E0-\U0001F1FF"
+        "\U00002702-\U000027B0"
+        "\U000024C2-\U0001F251"
+        "]+",
+        re.UNICODE
+    )
+    _ZW_RE = re.compile(r"[\u200b-\u200f\u202a-\u202e\ufeff\u00ad]")
+    _CTRL_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]")
+
+    # Script → language hints
+    SCRIPT_LANG_HINTS: Dict[str, List[str]] = {
+        "Devanagari": ["hin", "mar"],
+        "Bengali": ["ben"],
+        "Gurmukhi": ["pan"],
+        "Gujarati": ["guj"],
+        "Oriya": ["ori"],
+        "Tamil": ["tam"],
+        "Telugu": ["tel"],
+        "Kannada": ["kan"],
+        "Malayalam": ["mal"],
+        "Arabic": ["urd"],
+        "Latin": ["eng"],
+    }
+
+    def analyze(self, text: str) -> ScriptAnalysisResult:
+        """Comprehensive script analysis of input text."""
+        if not text:
+            return self._empty_result(text)
+
+        chars = list(text)
+        n = len(chars)
+
+        # Character category counts
+        alpha = sum(c.isalpha() for c in chars)
+        digits = sum(c.isdigit() for c in chars)
+        spaces = sum(c.isspace() for c in chars)
+        puncts = sum(unicodedata.category(c).startswith("P") for c in chars)
+        other = n - alpha - digits - spaces - puncts
+
+        # Script assignment per alpha char
+        script_of: List[Optional[str]] = [self._assign_script(c) for c in chars]
+        alpha_scripts = [s for c, s in zip(chars, script_of) if c.isalpha() and s]
+
+        # Script distribution
+        script_counter = Counter(alpha_scripts)
+        total_alpha = max(len(alpha_scripts), 1)
+        script_dist = {s: c / total_alpha for s, c in script_counter.most_common()}
+
+        dominant = script_counter.most_common(1)[0][0] if script_counter else "Unknown"
+        dom_ratio = script_dist.get(dominant, 0.0)
+
+        # Segment extraction
+        segments = self._extract_segments(chars, script_of)
+        transitions = max(0, len(segments) - 1)
+
+        # Mixing score: entropy-based
+        mixing_score = self._script_entropy(script_dist)
+
+        # Directionality
+        rtl_langs = {"Arabic", "Hebrew"}
+        rtl_chars = sum(1 for s in alpha_scripts if s in rtl_langs)
+        rtl_ratio = rtl_chars / max(total_alpha, 1)
+        primary_direction = "RTL" if rtl_ratio > 0.5 else "LTR"
+        has_bidi = 0.05 < rtl_ratio < 0.95
+
+        # Unicode block distribution
+        block_counter: Counter = Counter()
+        for c in chars:
+            block = get_unicode_block(ord(c))
+            block_counter[block] += 1
+        block_dist = {b: c / n for b, c in block_counter.most_common(10)}
+
+        # Noise analysis
+        url_count = len(self._URL_RE.findall(text))
+        emoji_count = len(self._EMOJI_RE.findall(text))
+        has_zero_width = bool(self._ZW_RE.search(text))
+        has_control = bool(self._CTRL_RE.search(text))
+        has_excessive_punct = (puncts / max(n, 1)) > 0.15
+        latin_ratio = script_dist.get("Basic Latin", 0.0) + script_dist.get("Latin", 0.0)
+
+        noise_components = [
+            0.3 * min(1.0, url_count * 0.5),
+            0.2 * int(has_zero_width),
+            0.2 * int(has_control),
+            0.15 * int(has_excessive_punct),
+            0.15 * min(1.0, emoji_count * 0.2),
+        ]
+        noise_score = sum(noise_components)
+
+        # Char entropy
+        char_freq = Counter(text)
+        total_f = sum(char_freq.values())
+        entropy = -sum(
+            (c / total_f) * np.log2(c / total_f)
+            for c in char_freq.values() if c > 0
+        )
+
+        # Lang hints based on detected scripts
+        lang_hints: Dict[str, List[str]] = {
+            s: self.SCRIPT_LANG_HINTS.get(s, [])
+            for s in script_dist
+        }
+
+        return ScriptAnalysisResult(
+            text=text,
+            total_chars=n,
+            alpha_chars=alpha,
+            digit_chars=digits,
+            punct_chars=puncts,
+            space_chars=spaces,
+            other_chars=other,
+            script_distribution=script_dist,
+            dominant_script=dominant,
+            dominant_script_ratio=dom_ratio,
+            is_mixed_script=len(script_dist) > 1 and dom_ratio < 0.95,
+            mixing_score=mixing_score,
+            script_segments=segments,
+            script_transitions=transitions,
+            primary_direction=primary_direction,
+            has_bidi=has_bidi,
+            rtl_ratio=rtl_ratio,
+            noise_score=noise_score,
+            has_zero_width=has_zero_width,
+            has_control_chars=has_control,
+            has_excessive_punct=has_excessive_punct,
+            url_count=url_count,
+            emoji_count=emoji_count,
+            latin_ratio=latin_ratio,
+            unicode_block_distribution=block_dist,
+            unique_chars=len(set(text)),
+            char_entropy=float(entropy),
+            script_to_lang_hints=lang_hints,
+        )
+
+    def analyze_batch(self, texts: List[str]) -> List[ScriptAnalysisResult]:
+        return [self.analyze(t) for t in texts]
+
+    def _assign_script(self, char: str) -> Optional[str]:
+        if not char.isalpha():
+            return None
+        cp = ord(char)
+        for block_start, block_end, block_name in UNICODE_BLOCK_NAMES:
+            if block_start <= cp <= block_end:
+                return block_name
+        return "Unknown"
+
+    def _extract_segments(
+        self, chars: List[str], scripts: List[Optional[str]]
+    ) -> List[ScriptSegment]:
+        """Extract contiguous script segments from text."""
+        segments = []
+        if not chars:
+            return segments
+
+        current_script = None
+        start = 0
+        seg_chars = []
+
+        for i, (c, s) in enumerate(zip(chars, scripts)):
+            if not c.isalpha():
+                seg_chars.append(c)
+                continue
+            if s != current_script:
+                if current_script is not None and seg_chars:
+                    segments.append(ScriptSegment(
+                        script=current_script,
+                        start=start,
+                        end=i,
+                        text="".join(seg_chars),
+                        char_count=sum(1 for ch in seg_chars if ch.isalpha()),
+                    ))
+                current_script = s
+                start = i
+                seg_chars = [c]
+            else:
+                seg_chars.append(c)
+
+        if current_script and seg_chars:
+            segments.append(ScriptSegment(
+                script=current_script,
+                start=start,
+                end=len(chars),
+                text="".join(seg_chars),
+                char_count=sum(1 for ch in seg_chars if ch.isalpha()),
+            ))
+
+        return [s for s in segments if s.char_count >= 3]
+
+    def _script_entropy(self, dist: Dict[str, float]) -> float:
+        """Shannon entropy over script distribution (normalized to [0,1])."""
+        if len(dist) <= 1:
+            return 0.0
+        probs = list(dist.values())
+        entropy = -sum(p * np.log2(p) for p in probs if p > 0)
+        max_entropy = np.log2(len(probs))
+        return float(entropy / max_entropy) if max_entropy > 0 else 0.0
+
+    def _empty_result(self, text: str) -> ScriptAnalysisResult:
+        return ScriptAnalysisResult(
+            text=text, total_chars=0, alpha_chars=0, digit_chars=0,
+            punct_chars=0, space_chars=0, other_chars=0,
+            script_distribution={}, dominant_script="Unknown",
+            dominant_script_ratio=0.0, is_mixed_script=False,
+            mixing_score=0.0, script_segments=[], script_transitions=0,
+            primary_direction="LTR", has_bidi=False, rtl_ratio=0.0,
+            noise_score=0.0, has_zero_width=False, has_control_chars=False,
+            has_excessive_punct=False, url_count=0, emoji_count=0,
+            latin_ratio=0.0, unicode_block_distribution={},
+            unique_chars=0, char_entropy=0.0, script_to_lang_hints={},
+        )
+
+    def compare_texts(self, text1: str, text2: str) -> Dict:
+        """Compare script profiles of two texts."""
+        r1 = self.analyze(text1)
+        r2 = self.analyze(text2)
+
+        scripts1 = set(r1.script_distribution.keys())
+        scripts2 = set(r2.script_distribution.keys())
+        shared = scripts1 & scripts2
+
+        jaccard = len(shared) / max(len(scripts1 | scripts2), 1)
+
+        return {
+            "script_overlap": list(shared),
+            "jaccard_similarity": jaccard,
+            "dominant_scripts_match": r1.dominant_script == r2.dominant_script,
+            "text1_dominant": r1.dominant_script,
+            "text2_dominant": r2.dominant_script,
+            "text1_mixing_score": r1.mixing_score,
+            "text2_mixing_score": r2.mixing_score,
+        }
+
+    def detect_transliteration(self, text: str) -> Dict:
+        """
+        Detect if text is a transliteration of an Indic language
+        written in Latin script (Romanized Indic).
+        """
+        result = self.analyze(text)
+
+        # If predominantly Latin but contains Indic vocabulary patterns
+        latin_dominated = result.dominant_script in ("Basic Latin", "Latin Extended-A")
+        has_indic_patterns = self._has_indic_romanization_patterns(text)
+
+        # Common Romanization markers
+        has_aspirates = bool(re.search(r"\b\w*(kh|gh|ch|jh|th|dh|ph|bh)\w*\b", text, re.I))
+        has_retroflex = bool(re.search(r"\b\w*[ṭḍṇṣ]\w*\b", text))
+        has_nasal_markers = bool(re.search(r"\b\w*(ng|ny|ñ|ṅ)\w*\b", text, re.I))
+
+        is_likely_transliteration = (
+            latin_dominated and (has_indic_patterns or has_aspirates)
+        )
+
+        return {
+            "is_transliteration": is_likely_transliteration,
+            "confidence": 0.8 if (is_likely_transliteration and has_aspirates) else 0.5,
+            "evidence": {
+                "latin_dominated": latin_dominated,
+                "has_indic_patterns": has_indic_patterns,
+                "has_aspirates": has_aspirates,
+                "has_retroflex": has_retroflex,
+                "has_nasal_markers": has_nasal_markers,
+            }
+        }
+
+    def _has_indic_romanization_patterns(self, text: str) -> bool:
+        """Common romanization patterns across Indic languages."""
+        patterns = [
+            r"\b(hai|hain|nahin|nahi|aur|yeh|woh|kya|kaun|kahan|kyun|mera|tera|apna)\b",  # Hindi
+            r"\b(amma|appa|anna|akka|vanakkam|nandri|romba|paaru|paar)\b",   # Tamil
+            r"\b(ahe|aahe|nahi|tumhi|amhi|tya|tyanche|mhanje)\b",           # Marathi
+        ]
+        for p in patterns:
+            if re.search(p, text, re.I):
+                return True
+        return False
+
+
+# Singleton for reuse
+default_analyzer = ScriptAnalyzer()

transformer_model.py

@@ -0,0 +1,303 @@
+"""
+Transformer-based Language Identification Model
+Uses Google MuRIL (Multilingual Universal Representations for Indian Languages)
+or XLM-RoBERTa for high-accuracy sequence classification.
+
+Requires GPU for practical training. Set enabled: true in config.
+"""
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+
+import numpy as np
+import pandas as pd
+from loguru import logger
+
+try:
+    import torch
+    from torch.utils.data import Dataset, DataLoader
+    from transformers import (
+        AutoTokenizer,
+        AutoModelForSequenceClassification,
+        TrainingArguments,
+        Trainer,
+        EarlyStoppingCallback,
+    )
+    from transformers import DataCollatorWithPadding
+    TRANSFORMERS_AVAILABLE = True
+except ImportError:
+    TRANSFORMERS_AVAILABLE = False
+    logger.warning("transformers/torch not installed. Transformer model unavailable.")
+
+from src.utils.languages import (
+    LANGUAGE_REGISTRY,
+    CODE_TO_LABEL,
+    LABEL_TO_CODE,
+    NUM_CLASSES,
+    ALL_LANGUAGE_CODES,
+)
+
+
+# ─────────────────────────────────────────────
+# Dataset Wrapper
+# ─────────────────────────────────────────────
+
+class IndicLIDDataset:
+    """Wraps a pandas DataFrame for HuggingFace Trainer."""
+
+    def __init__(self, df: pd.DataFrame, tokenizer, max_length: int = 128):
+        self.texts = df["text"].tolist()
+        self.labels = df["label"].tolist()
+        self.tokenizer = tokenizer
+        self.max_length = max_length
+
+    def __len__(self):
+        return len(self.texts)
+
+    def __getitem__(self, idx):
+        encoding = self.tokenizer(
+            self.texts[idx],
+            max_length=self.max_length,
+            truncation=True,
+            padding="max_length",
+            return_tensors="pt",
+        )
+        return {
+            "input_ids": encoding["input_ids"].squeeze(),
+            "attention_mask": encoding["attention_mask"].squeeze(),
+            "labels": torch.tensor(self.labels[idx], dtype=torch.long),
+        }
+
+
+# ─────────────────────────────────────────────
+# Transformer LID Model
+# ─────────────────────────────────────────────
+
+class TransformerLIDModel:
+    """
+    Fine-tuned transformer for language identification.
+    Defaults to MuRIL (designed for Indic languages).
+
+    MuRIL advantages over XLM-R for this task:
+    - Pretrained on 17 Indian languages + transliterated text
+    - Better subword coverage for Indic scripts
+    - Transliteration-aware pretraining
+    """
+
+    SUPPORTED_MODELS = {
+        "muril": "google/muril-base-cased",
+        "xlmr": "xlm-roberta-base",
+        "xlmr-large": "xlm-roberta-large",
+        "indicbert": "ai4bharat/indic-bert",
+    }
+
+    def __init__(
+        self,
+        base_model: str = "muril",
+        max_length: int = 128,
+        batch_size: int = 32,
+        learning_rate: float = 2e-5,
+        num_epochs: int = 3,
+        warmup_ratio: float = 0.1,
+        weight_decay: float = 0.01,
+        output_dir: str = "./artifacts/models/transformer",
+        fp16: bool = True,
+    ):
+        assert TRANSFORMERS_AVAILABLE, "transformers + torch required"
+        self.model_name = self.SUPPORTED_MODELS.get(base_model, base_model)
+        self.max_length = max_length
+        self.batch_size = batch_size
+        self.learning_rate = learning_rate
+        self.num_epochs = num_epochs
+        self.warmup_ratio = warmup_ratio
+        self.weight_decay = weight_decay
+        self.output_dir = Path(output_dir)
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+        self.fp16 = fp16 and torch.cuda.is_available()
+
+        self.tokenizer = None
+        self.model = None
+        self.trainer = None
+        self.is_fitted = False
+
+        # Label mapping
+        self.id2label = {i: l for i, l in LABEL_TO_CODE.items()}
+        self.label2id = {l: i for i, l in self.id2label.items()}
+
+    def _load_base_model(self):
+        logger.info(f"Loading tokenizer and model: {self.model_name}")
+        self.tokenizer = AutoTokenizer.from_pretrained(
+            self.model_name,
+            use_fast=True,
+        )
+        self.model = AutoModelForSequenceClassification.from_pretrained(
+            self.model_name,
+            num_labels=NUM_CLASSES,
+            id2label=self.id2label,
+            label2id=self.label2id,
+            ignore_mismatched_sizes=True,
+        )
+        device = "cuda" if torch.cuda.is_available() else "cpu"
+        logger.info(f"Using device: {device}")
+        self.model.to(device)
+
+    def fit(
+        self,
+        train_df: pd.DataFrame,
+        val_df: Optional[pd.DataFrame] = None,
+    ) -> "TransformerLIDModel":
+        self._load_base_model()
+
+        train_dataset = IndicLIDDataset(train_df, self.tokenizer, self.max_length)
+        val_dataset = IndicLIDDataset(val_df, self.tokenizer, self.max_length) if val_df is not None else None
+
+        training_args = TrainingArguments(
+            output_dir=str(self.output_dir),
+            num_train_epochs=self.num_epochs,
+            per_device_train_batch_size=self.batch_size,
+            per_device_eval_batch_size=self.batch_size * 2,
+            learning_rate=self.learning_rate,
+            warmup_ratio=self.warmup_ratio,
+            weight_decay=self.weight_decay,
+            evaluation_strategy="epoch" if val_dataset else "no",
+            save_strategy="epoch",
+            load_best_model_at_end=val_dataset is not None,
+            metric_for_best_model="eval_loss",
+            fp16=self.fp16,
+            logging_steps=100,
+            logging_dir=str(self.output_dir / "logs"),
+            dataloader_num_workers=4,
+            seed=42,
+            report_to="none",  # Disable wandb/mlflow unless configured
+        )
+
+        self.trainer = Trainer(
+            model=self.model,
+            args=training_args,
+            train_dataset=train_dataset,
+            eval_dataset=val_dataset,
+            tokenizer=self.tokenizer,
+            callbacks=[EarlyStoppingCallback(early_stopping_patience=2)] if val_dataset else [],
+            compute_metrics=self._compute_metrics,
+        )
+
+        logger.info("Starting transformer training ...")
+        self.trainer.train()
+        self.is_fitted = True
+        logger.success("Training complete!")
+        return self
+
+    def _compute_metrics(self, eval_pred):
+        from sklearn.metrics import f1_score, accuracy_score
+        logits, labels = eval_pred
+        predictions = np.argmax(logits, axis=-1)
+        acc = accuracy_score(labels, predictions)
+        f1 = f1_score(labels, predictions, average="macro", zero_division=0)
+        return {"accuracy": acc, "macro_f1": f1}
+
+    def predict(self, texts: List[str]) -> List[str]:
+        assert self.is_fitted
+        probs, _ = self.predict_proba(texts)
+        label_ids = np.argmax(probs, axis=1)
+        return [self.id2label[i] for i in label_ids]
+
+    def predict_proba(self, texts: List[str]) -> Tuple[np.ndarray, List[str]]:
+        assert self.is_fitted
+        import torch
+        from torch.nn.functional import softmax
+
+        device = next(self.model.parameters()).device
+        all_probs = []
+
+        for i in range(0, len(texts), self.batch_size):
+            batch_texts = texts[i:i + self.batch_size]
+            encodings = self.tokenizer(
+                batch_texts,
+                max_length=self.max_length,
+                truncation=True,
+                padding=True,
+                return_tensors="pt",
+            ).to(device)
+
+            with torch.no_grad():
+                outputs = self.model(**encodings)
+                probs = softmax(outputs.logits, dim=-1).cpu().numpy()
+            all_probs.append(probs)
+
+        prob_matrix = np.vstack(all_probs)
+        class_names = [self.id2label[i] for i in range(NUM_CLASSES)]
+        return prob_matrix, class_names
+
+    def predict_with_confidence(self, texts: List[str]) -> List[Dict]:
+        probs, class_names = self.predict_proba(texts)
+        results = []
+        for i, prob_row in enumerate(probs):
+            top_k_idx = np.argsort(prob_row)[::-1][:5]
+            top_k = [
+                {
+                    "lang": class_names[j],
+                    "name": LANGUAGE_REGISTRY.get(class_names[j], None) and LANGUAGE_REGISTRY[class_names[j]].name,
+                    "probability": float(prob_row[j]),
+                }
+                for j in top_k_idx
+            ]
+            best = top_k[0]
+            info = LANGUAGE_REGISTRY.get(best["lang"])
+            results.append({
+                "text_preview": texts[i][:80],
+                "predicted_lang": best["lang"],
+                "predicted_lang_name": info.name if info else best["lang"],
+                "script": info.script if info else "?",
+                "direction": info.direction if info else "LTR",
+                "confidence": best["probability"],
+                "is_confident": best["probability"] >= 0.75,
+                "top_5": top_k,
+            })
+        return results
+
+    def evaluate(self, test_df: pd.DataFrame) -> Dict:
+        from sklearn.metrics import classification_report, f1_score, accuracy_score
+        texts = test_df["text"].tolist()
+        true_labels = test_df["lang"].tolist()
+        pred_labels = self.predict(texts)
+        probs, _ = self.predict_proba(texts)
+
+        accuracy = accuracy_score(true_labels, pred_labels)
+        macro_f1 = f1_score(true_labels, pred_labels, average="macro", zero_division=0)
+        report = classification_report(true_labels, pred_labels, output_dict=True, zero_division=0)
+
+        logger.info(f"\n{classification_report(true_labels, pred_labels, zero_division=0)}")
+        logger.success(f"Accuracy: {accuracy:.4f} | Macro-F1: {macro_f1:.4f}")
+
+        return {
+            "accuracy": accuracy,
+            "macro_f1": macro_f1,
+            "per_language": report,
+            "predictions": pred_labels,
+            "true_labels": true_labels,
+            "confidences": probs.max(axis=1).tolist(),
+        }
+
+    def save(self, path: str):
+        self.model.save_pretrained(path)
+        self.tokenizer.save_pretrained(path)
+        logger.success(f"Transformer model saved: {path}")
+
+    @classmethod
+    def load(cls, path: str) -> "TransformerLIDModel":
+        assert TRANSFORMERS_AVAILABLE
+        obj = cls.__new__(cls)
+        obj.output_dir = Path(path)
+        obj.tokenizer = AutoTokenizer.from_pretrained(path)
+        obj.model = AutoModelForSequenceClassification.from_pretrained(path)
+        obj.is_fitted = True
+        obj.id2label = {i: l for i, l in LABEL_TO_CODE.items()}
+        obj.batch_size = 32
+        obj.max_length = 128
+        obj.fp16 = False
+        device = "cuda" if torch.cuda.is_available() else "cpu"
+        obj.model.to(device)
+        logger.success(f"Transformer model loaded: {path}")
+        return obj