| |
| |
| |
|
|
| 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', |
| r'\b[A-Z]{0,3}\d{4,12}\b', |
| ] |
|
|
| 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") |
| ] |
|
|
