python-services/caller_info_extractor.py

# caller_info_extractor.py
# Extracts caller fields from conversation based on optional per-session field spec.
# Uses regex for immediate hot-path extraction when fields are configured.

import re
import logging
from typing import List, Dict, Any, Optional

from language_config import normalize_language

logger = logging.getLogger("caller_info_extractor")

PHONE_PATTERNS = [
    r'\b(?:\+92|0092|92)[\s\-]?3\d{2}[\s\-]?\d{7}\b(?!\d)',
    r'\b(?:\+966|00966|966)[\s\-]?5\d{8}\b(?!\d)',
    r'\b(?:\+971|00971|971)[\s\-]?5[024568]\d{7}\b(?!\d)',
    r'\b(?:\+91|0091|91)[\s\-]?[6789]\d{9}\b(?!\d)',
    r'\b(?:\+1|1)[\s\-]?\(?[2-9]\d{2}\)?[\s\-]?[2-9]\d{2}[\s\-]?\d{4}\b(?!\d)',
    r'\b0\d{2,3}[\s\-]?\d{7,8}\b(?!\d)',
    r'\b\d{10,13}\b(?!\d)',
]

NAME_PATTERNS_EN = [
    r"(?:my name is|i am|i'm|this is|call me)\s+([A-Za-z][a-z]+(?:\s+[A-Za-z][a-z]+)?)",
    r"([A-Za-z][a-z]+(?:\s+[A-Za-z][a-z]+)?)\s+(?:here|speaking|calling)",
]

NAME_PATTERNS_ROMAN_UR = [
    r"(?:mera\s+na(?:am|me?)|apna\s+naam?)\s+([A-Za-z][a-z]{2,}(?:\s+[A-Za-z][a-z]+)?)\s+(?:hai|he|hun|hoon|ho)\b",
    r"(?:mera\s+na(?:am|me?)|apna\s+naam?)\s+([A-Za-z][a-z]{2,}(?:\s+[A-Za-z][a-z]+)?)",
    r"([A-Za-z][a-z]+)\s+(?:ye|yeh|ye\s+mera|mera)\s+na(?:am|me?)",
    r"\bna(?:am|me?)\s+(?:and\s+)?([A-Za-z][a-z]{2,})",
    r"(?:main|mein)\s+([A-Za-z][a-z]{2,})\s+(?:hoon|hun|ho)",
    r"(?:mujhe|humein)\s+([A-Za-z][a-z]{2,})\s+(?:kehte|bolte|bulao|kahte)",
]

NAME_PATTERNS_UR = [
    r"(?:میرا نام|میں ہوں)\s+([\u0600-\u06FF\s]+)",
    r"([\u0600-\u06FF\s]+)\s+(?:بات کر رہا ہوں|بات کر رہی ہوں|بول رہا ہوں)",
]

ID_PATTERNS = [
    r'\b\d{5}[\s\-]?\d{7}[\s\-]?\d\b',  # CNIC Pakistan
    r'\b[A-Z]{0,3}\d{4,12}\b',           # Generic student/employee ID
]

NAME_STOP_WORDS = {
    "he", "hai", "hun", "hoon", "ho", "and", "ye", "yeh", "mera", "name",
    "naam", "number", "num", "the", "my", "is", "am", "me", "id",
}


def extract_phone_from_text(text: str) -> Optional[str]:
    for pattern in PHONE_PATTERNS:
        match = re.search(pattern, text, re.IGNORECASE)
        if match:
            phone = re.sub(r'[\s\-\(\)\+]', '', match.group())
            digit_count = sum(c.isdigit() for c in phone)
            if digit_count < 7 or digit_count > 15:
                logger.warning(f"Rejected implausible phone number ({digit_count} digits): {phone}")
                continue
            logger.info(f"Regex pipeline intercepted telephone credentials: {phone}")
            return phone
    return None


def extract_name_from_text(text: str, language: str = "en") -> Optional[str]:
    lang = normalize_language(language)

    latin_chars = sum(1 for c in text if c.isascii() and c.isalpha())
    total_chars = sum(1 for c in text if c.isalpha())
    is_roman_script = total_chars > 0 and (latin_chars / total_chars) > 0.5

    if lang == "ur" and is_roman_script:
        target_patterns = NAME_PATTERNS_ROMAN_UR + NAME_PATTERNS_EN
    elif lang == "ur":
        target_patterns = NAME_PATTERNS_UR
    else:
        target_patterns = NAME_PATTERNS_EN + NAME_PATTERNS_ROMAN_UR

    for pattern in target_patterns:
        match = re.search(pattern, text, re.IGNORECASE)
        if match:
            name = match.group(1).strip()
            if 2 <= len(name) <= 40 and name.lower() not in NAME_STOP_WORDS:
                logger.info(f"Regex pipeline intercepted identity credentials: {name} [Lang: {lang}]")
                return name
    return None


def extract_id_from_text(text: str) -> Optional[str]:
    for pattern in ID_PATTERNS:
        match = re.search(pattern, text, re.IGNORECASE)
        if match:
            value = match.group().strip()
            if len(value) >= 4:
                logger.info(f"Regex pipeline intercepted ID credentials: {value}")
                return value
    return None


def extract_generic_text_field(text: str) -> Optional[str]:
    match = re.search(r'\b([A-Za-z0-9][A-Za-z0-9\-]{2,20})\b', text)
    if match:
        return match.group(1)
    return None


class CallerInfoCollector:
    """Tracks caller profile fields defined by per-session collection spec."""

    def __init__(self, fields_spec: Optional[List[Dict[str, Any]]] = None):
        self.fields_spec: List[Dict[str, Any]] = fields_spec or []
        self.collected: Dict[str, Optional[str]] = {
            f["key"]: None for f in self.fields_spec if f.get("key")
        }
        self.language: Optional[str] = "en"

    def is_enabled(self) -> bool:
        return bool(self.fields_spec)

    def process_message(self, text: str, language: str = "en") -> None:
        if not self.is_enabled():
            return
        self.language = language
        for field in self.fields_spec:
            key = field.get("key")
            if not key or self.collected.get(key):
                continue
            field_type = (field.get("type") or "text").lower()
            value = None
            if field_type == "name":
                value = extract_name_from_text(text, language=language)
            elif field_type == "phone":
                value = extract_phone_from_text(text)
            elif field_type == "id":
                value = extract_id_from_text(text)
            else:
                value = extract_generic_text_field(text)
            if value:
                self.collected[key] = value

    def update_from_llm_extraction(self, extracted: Dict[str, Any]) -> None:
        if not extracted or not self.is_enabled():
            return
        for field in self.fields_spec:
            key = field.get("key")
            if not key or self.collected.get(key):
                continue
            if extracted.get(key):
                clean = str(extracted[key]).strip()
                if clean and clean.lower() != "unknown":
                    self.collected[key] = clean

    def to_dict(self) -> Dict[str, Any]:
        if not self.is_enabled():
            return {}
        return {k: v for k, v in self.collected.items()}

    def missing_fields(self) -> List[str]:
        if not self.is_enabled():
            return []
        return [f["key"] for f in self.fields_spec if f.get("key") and not self.collected.get(f["key"])]

    def is_complete(self) -> bool:
        return self.is_enabled() and len(self.missing_fields()) == 0

    def get_field_labels(self) -> List[Dict[str, str]]:
        return [
            {
                "key": f["key"],
                "label": f.get("label", f["key"]),
                "label_ur": f.get("label_ur", f.get("label", f["key"])),
                "type": f.get("type", "text"),
            }
            for f in self.fields_spec
            if f.get("key")
        ]