Blame for python-services/language_detector.py

# language_detector.py
# v4 — URDU-ONLY MODE + arabic_clarify removed
#
# v3 locked to "ur" unconditionally from turn 1.
# v4 removes arabic_needs_clarification / get_arabic_clarification_flag()
# entirely (committed per user mandate — Urdu-only mode, no need for
# Arabic/English/mixed-script distinction). All callers of
# session.get_arabic_clarification_flag() have been updated accordingly
# (prompt_builder.py, voice_pipeline.py, llm_server.py, conversation_manager.py).
#
# The original detection/voting/locking logic is preserved below as
# commented-out code (not deleted) so it can be restored when per-agent
# language selection is reintroduced.

import logging

from language_config import (
    DEFAULT_LANGUAGE,
    DEVANAGARI,
    GURMUKHI,
    ROMAN_URDU_MARKERS,
    detect_roman_urdu,
    has_arabic_script,
    normalize_language,
)
from utils import detect_language_from_content

logger = logging.getLogger("language_detector")

AMBIGUOUS_WORDS = {
    "hello", "hi", "ok", "okay", "yes", "no", "please",
    "sorry", "thanks", "thank", "bye", "good", "hey", "help", "helo",
}

MIN_CHARS_FOR_DETECTION = 8

# Urdu-only mode: the language this session always resolves to.
URDU_ONLY_LANGUAGE = "ur"


def is_ambiguous_message(text: str) -> bool:
    """
    Retained for future multi-language mode — not used while
    URDU_ONLY_LANGUAGE forcing is active.
    """
    text_clean = text.strip().lower()
    if len(text_clean) < MIN_CHARS_FOR_DETECTION:
        words = set(text_clean.split())
        if words.issubset(AMBIGUOUS_WORDS):
            return True
    return False


def _has_urdu_script_evidence(text: str) -> bool:
    """
    Retained for future multi-language mode — not used while
    URDU_ONLY_LANGUAGE forcing is active.
    """
    if DEVANAGARI.search(text) or GURMUKHI.search(text):
        return True
    if has_arabic_script(text):
        return True
    if detect_roman_urdu(text):
        return True
    words = set(text.lower().split())
    if words.intersection(ROMAN_URDU_MARKERS):
        return True
    return False


def detect_language_from_text(text: str) -> tuple[str, float]:
    """
    Retained for future multi-language mode — not used while
    URDU_ONLY_LANGUAGE forcing is active.
    """
    if not text or not text.strip():
        return DEFAULT_LANGUAGE, 0.0

    lang = normalize_language(detect_language_from_content(text))

    if lang == "ur":
        if DEVANAGARI.search(text) or GURMUKHI.search(text):
            return "ur", 0.95
        if has_arabic_script(text):
            return "ur", 0.75
        return "ur", 0.90

    if is_ambiguous_message(text):
        return DEFAULT_LANGUAGE, 0.30

    return "en", 0.70


class SmartLanguageDetector:
    """
    URDU-ONLY MODE v4: locks to "ur" immediately and unconditionally.

    arabic_needs_clarification and get_arabic_clarification_flag() have
    been removed entirely — Urdu-only mode has no need to distinguish
    Arabic script subtypes. All callers updated accordingly.

    process_message() still tracks turn_count for logging. No voting,
    no pivot logic, no arabic script classification.
    """

    def __init__(self):
        self.detected_language = URDU_ONLY_LANGUAGE
        self.confidence = 0.95
        self.turn_count = 0
        self.language_votes: dict[str, float] = {URDU_ONLY_LANGUAGE: 1.0}
        self.locked = True

    def process_message(self, text: str) -> str:
        """
        Process user message and return resolved language code.
        Urdu-only mode: always returns "ur". Tracks turn_count for logging.
        """
        self.turn_count += 1

        logger.info(
            f"Turn {self.turn_count}: language=ur (forced, Urdu-only mode)"
        )

        return self.detected_language

        # ── ORIGINAL EN/UR VOTING LOGIC (disabled — Urdu-only mode) ────────
        #
        # lang, confidence = detect_language_from_text(text)
        # lang = normalize_language(lang)
        #
        # if self.locked:
        #     if confidence > 0.92 and lang != self.detected_language:
        #         logger.info(
        #             f"Strong language pivot detected: {self.detected_language} → {lang}"
        #         )
        #         self.detected_language = lang
        #         self.confidence = confidence
        #         self.language_votes = {lang: confidence + 1.0}
        #     return self.detected_language
        #
        # self.language_votes[lang] = self.language_votes.get(lang, 0.0) + confidence
        # best_lang = max(self.language_votes, key=self.language_votes.get)
        # best_score = self.language_votes[best_lang]
        #
        # should_lock = (
        #     confidence > 0.88
        #     or (self.turn_count >= 2 and best_score > 1.35)
        #     or (self.turn_count >= 2 and _has_urdu_script_evidence(text))
        # )
        #
        # self.detected_language = normalize_language(best_lang)
        # self.confidence = confidence
        #
        # if should_lock:
        #     self.locked = True
        #     logger.info(f"Session conversation language locked to: {self.detected_language}")
        #
        # return self.detected_language

        # ── ORIGINAL ARABIC CLARIFICATION TRACKING (disabled — Urdu-only) ──
        # arabic_needs_clarification was set here via:
        #     if get_arabic_clarification_flag(text):
        #         self.arabic_needs_clarification = True
        #     elif classify_arabic_script(text) == "clear_urdu":
        #         self.arabic_needs_clarification = False
        # Re-enable alongside multi-language mode if needed.

    def get_language(self) -> str:
        return normalize_language(self.detected_language)

    def is_locked(self) -> bool:
        return self.locked

    def reset(self):
        self.__init__()


def detect_language(text: str) -> str:
    """Stateless wrapper for compatibility. Urdu-only mode: always returns 'ur'."""
    return URDU_ONLY_LANGUAGE

    # ── ORIGINAL (disabled — Urdu-only mode) ───────────────────────────────
    # return normalize_language(detect_language_from_content(text))