# 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") ]