ai_lls_lib

38class PhoneVerification(BaseModel):
39    """Result of phone number verification"""
40
41    phone_number: str = Field(..., description="E.164 formatted phone number")
42    line_type: LineType = Field(..., description="Type of phone line")
43    dnc: bool = Field(..., description="Whether number is on DNC list")
44    cached: bool = Field(..., description="Whether result came from cache")
45    verified_at: datetime = Field(..., description="When verification occurred")
46    source: VerificationSource = Field(..., description="Source of verification data")
47
48    class Config:
49        json_encoders = {datetime: lambda v: v.isoformat()}

48    class Config:
49        json_encoders = {datetime: lambda v: v.isoformat()}

52class BulkJob(BaseModel):
53    """Bulk processing job metadata"""
54
55    job_id: str = Field(..., description="Unique job identifier")
56    status: JobStatus = Field(..., description="Current job status")

59class BulkJobStatus(BulkJob):
60    """Extended bulk job status with progress info"""
61
62    total_rows: int | None = Field(None, description="Total rows to process")
63    processed_rows: int | None = Field(None, description="Rows processed so far")
64    result_url: str | None = Field(None, description="S3 URL of results")
65    created_at: datetime = Field(..., description="Job creation time")
66    completed_at: datetime | None = Field(None, description="Job completion time")
67    error: str | None = Field(None, description="Error message if failed")

12class LineType(StrEnum):
13    """Phone line type enumeration"""
14
15    MOBILE = "mobile"
16    LANDLINE = "landline"
17    VOIP = "voip"
18    UNKNOWN = "unknown"

21class VerificationSource(StrEnum):
22    """Source of verification data"""
23
24    API = "api"
25    CACHE = "cache"
26    BULK_IMPORT = "bulk_import"

29class JobStatus(StrEnum):
30    """Bulk job status enumeration"""
31
32    PENDING = "pending"
33    PROCESSING = "processing"
34    COMPLETED = "completed"
35    FAILED = "failed"

18class PhoneVerifier:
19    """Verifies phone numbers for line type and DNC status"""
20
21    def __init__(
22        self, cache: DynamoDBCache | None = None, provider: VerificationProvider | None = None
23    ):
24        """
25        Initialize phone verifier.
26
27        Args:
28            cache: Optional DynamoDB cache for storing results
29            provider: Verification provider (defaults to ExternalAPIProvider)
30        """
31        self.cache = cache
32        self.provider = provider or ExternalAPIProvider()
33        logger.debug("PhoneVerifier initialized")
34
35    def normalize_phone(self, phone: str) -> str:
36        """Normalize phone to E.164 format"""
37        try:
38            # Parse with US as default country
39            parsed = phonenumbers.parse(phone, "US")
40            if not phonenumbers.is_valid_number(parsed):
41                raise ValueError(f"Invalid phone number: {phone}")
42
43            # Format as E.164
44            return phonenumbers.format_number(parsed, phonenumbers.PhoneNumberFormat.E164)
45        except Exception as e:
46            logger.error(f"Phone normalization failed: {str(e)}")
47            raise ValueError(f"Invalid phone format: {phone}") from e
48
49    def verify(self, phone: str) -> PhoneVerification:
50        """Verify phone number for line type and DNC status"""
51        normalized = self.normalize_phone(phone)
52
53        # Check cache first if available
54        if self.cache:
55            cached = self.cache.get(normalized)
56            if cached:
57                return cached
58
59        # Use provider to verify
60        line_type, dnc_status = self.provider.verify_phone(normalized)
61
62        result = PhoneVerification(
63            phone_number=normalized,
64            line_type=line_type,
65            dnc=dnc_status,
66            cached=False,
67            verified_at=datetime.now(UTC),
68            source=VerificationSource.API,
69        )
70
71        # Store in cache if available
72        if self.cache:
73            try:
74                self.cache.set(normalized, result)
75            except Exception as e:
76                logger.warning(f"Failed to cache result: {e}")
77                # Continue without caching - don't fail the verification
78
79        return result
80
81    def _check_line_type(self, phone: str) -> LineType:
82        """
83        Check line type (for backwards compatibility with CLI).
84        Delegates to provider.
85        """
86        line_type, _ = self.provider.verify_phone(phone)
87        return line_type
88
89    def _check_dnc(self, phone: str) -> bool:
90        """
91        Check DNC status (for backwards compatibility with CLI).
92        Delegates to provider.
93        """
94        _, dnc_status = self.provider.verify_phone(phone)
95        return dnc_status

 18class BulkProcessor:
 19    """Process CSV files for bulk phone verification"""
 20
 21    def __init__(self, verifier: PhoneVerifier):
 22        self.verifier = verifier
 23
 24    def process_csv(self, csv_text: str, phone_column: str = "phone") -> list[PhoneVerification]:
 25        """
 26        Process CSV text content.
 27        Returns list of verification results.
 28        """
 29        results = []
 30
 31        try:
 32            # Strip UTF-8 BOM if present (Excel on Windows adds this)
 33            csv_text = csv_text.lstrip("\ufeff")
 34
 35            # Use StringIO to parse CSV text
 36            csv_file = StringIO(csv_text)
 37            reader = csv.DictReader(csv_file)
 38
 39            # Find phone column (case-insensitive)
 40            headers = reader.fieldnames or []
 41            phone_col = self._find_phone_column(headers, phone_column)
 42
 43            if not phone_col:
 44                raise ValueError(f"Phone column '{phone_column}' not found in CSV")
 45
 46            logger.info(f"Starting CSV processing using phone column '{phone_col}'")
 47
 48            for row_num, row in enumerate(reader, start=2):  # Start at 2 (header is 1)
 49                try:
 50                    phone = row.get(phone_col, "").strip()
 51                    if not phone:
 52                        logger.warning(f"Empty phone at row {row_num}")
 53                        continue
 54
 55                    # Verify phone
 56                    result = self.verifier.verify(phone)
 57                    results.append(result)
 58
 59                    # Log progress every 100 rows
 60                    if len(results) % 100 == 0:
 61                        logger.info(f"Processed {len(results)} phones (at row {row_num})")
 62
 63                except ValueError as e:
 64                    logger.warning(f"Invalid phone at row {row_num}: {str(e)}")
 65                    continue
 66                except Exception as e:
 67                    logger.error(f"Error processing row {row_num}: {str(e)}")
 68                    continue
 69
 70            logger.info(f"Completed processing {len(results)} valid phones")
 71
 72        except Exception as e:
 73            logger.error(f"CSV processing failed: {str(e)}")
 74            raise
 75
 76        return results
 77
 78    def _find_phone_column(self, headers: list[str] | Sequence[str], preferred: str) -> str | None:
 79        """Find phone column in headers (case-insensitive)"""
 80        # First try exact match
 81        for header in headers:
 82            if header.lower() == preferred.lower():
 83                return header
 84
 85        # Common phone column names
 86        phone_patterns = [
 87            "phone",
 88            "phone_number",
 89            "phonenumber",
 90            "mobile",
 91            "cell",
 92            "telephone",
 93            "tel",
 94            "number",
 95            "contact",
 96        ]
 97
 98        for header in headers:
 99            header_lower = header.lower()
100            for pattern in phone_patterns:
101                if pattern in header_lower:
102                    logger.info(f"Using column '{header}' as phone column")
103                    return header
104
105        return None
106
107    def generate_results_csv(self, original_csv_text: str, results: list[PhoneVerification]) -> str:
108        """
109        Generate CSV with original data plus verification results.
110        Adds columns: line_type, dnc
111        Returns CSV text string.
112        """
113        # Create lookup dict
114        results_map = {r.phone_number: r for r in results}
115
116        # Parse original CSV (strip UTF-8 BOM if present)
117        original_csv_text = original_csv_text.lstrip("\ufeff")
118        input_file = StringIO(original_csv_text)
119        reader = csv.DictReader(input_file)
120        headers = list(reader.fieldnames or [])
121
122        # Add new columns
123        output_headers = headers + ["line_type", "dnc"]
124
125        # Create output CSV in memory
126        output = StringIO()
127        writer = csv.DictWriter(output, fieldnames=output_headers)
128        writer.writeheader()
129
130        phone_col = self._find_phone_column(headers, "phone")
131
132        for row in reader:
133            phone = row.get(phone_col, "").strip()
134
135            # Try to normalize for lookup
136            try:
137                normalized = self.verifier.normalize_phone(phone)
138                if normalized in results_map:
139                    result = results_map[normalized]
140                    row["line_type"] = result.line_type.value
141                    row["dnc"] = "true" if result.dnc else "false"
142                else:
143                    row["line_type"] = "unknown"
144                    row["dnc"] = ""
145            except Exception:
146                row["line_type"] = "invalid"
147                row["dnc"] = ""
148
149            writer.writerow(row)
150
151        # Return CSV text
152        return output.getvalue()
153
154    def process_csv_stream(
155        self, lines: Iterable[str], phone_column: str = "phone", batch_size: int = 100
156    ) -> Iterator[list[PhoneVerification]]:
157        """
158        Process CSV lines as a stream, yielding batches of results.
159        Memory-efficient for large files.
160
161        Args:
162            lines: Iterator of CSV lines (including header)
163            phone_column: Column name containing phone numbers
164            batch_size: Number of results to accumulate before yielding
165
166        Yields:
167            Batches of PhoneVerification results
168        """
169        lines_list = list(lines)  # Need to iterate twice - once for headers, once for data
170
171        if not lines_list:
172            logger.error("Empty CSV stream")
173            return
174
175        # Parse header (strip UTF-8 BOM if present)
176        header_line = lines_list[0].lstrip("\ufeff")
177        reader = csv.DictReader(StringIO(header_line))
178        headers = reader.fieldnames or []
179        phone_col = self._find_phone_column(headers, phone_column)
180
181        if not phone_col:
182            raise ValueError(f"Phone column '{phone_column}' not found in CSV")
183
184        batch = []
185        row_num = 2  # Start at 2 (header is 1)
186        total_processed = 0
187
188        # Process data lines
189        for line in lines_list[1:]:
190            if not line.strip():
191                continue
192
193            try:
194                # Parse single line
195                row = next(csv.DictReader(StringIO(line), fieldnames=headers))
196                phone = row.get(phone_col, "").strip()
197
198                if not phone:
199                    logger.warning(f"Empty phone at row {row_num}")
200                    row_num += 1
201                    continue
202
203                # Verify phone
204                result = self.verifier.verify(phone)
205                batch.append(result)
206                total_processed += 1
207
208                # Yield batch when full
209                if len(batch) >= batch_size:
210                    logger.info(
211                        f"Processed batch of {len(batch)} phones (total: {total_processed}, at row {row_num})"
212                    )
213                    yield batch
214                    batch = []
215
216            except ValueError as e:
217                logger.warning(f"Invalid phone at row {row_num}: {str(e)}")
218            except Exception as e:
219                logger.error(f"Error processing row {row_num}: {str(e)}")
220            finally:
221                row_num += 1
222
223        # Yield remaining results
224        if batch:
225            logger.info(f"Processed final batch of {len(batch)} phones (total: {total_processed})")
226            yield batch
227
228        logger.info(f"Stream processing completed. Total processed: {total_processed}")
229
230    def generate_results_csv_stream(
231        self,
232        original_lines: Iterable[str],
233        results_stream: Iterator[list[PhoneVerification]],
234        phone_column: str = "phone",
235    ) -> Iterator[str]:
236        """
237        Generate CSV results as a stream, line by line.
238        Memory-efficient for large files.
239
240        Args:
241            original_lines: Iterator of original CSV lines
242            results_stream: Iterator of batched PhoneVerification results
243            phone_column: Column name containing phone numbers
244
245        Yields:
246            CSV lines with verification results added
247        """
248        lines_iter = iter(original_lines)
249
250        # Read and yield modified header
251        try:
252            header_line = next(lines_iter).lstrip("\ufeff")
253            reader = csv.DictReader(StringIO(header_line))
254            headers = list(reader.fieldnames or [])
255
256            # Add new columns
257            output_headers = headers + ["line_type", "dnc"]
258            yield ",".join(output_headers) + "\n"
259
260            phone_col = self._find_phone_column(headers, phone_column)
261
262        except StopIteration:
263            return
264
265        # Build results lookup from stream
266        results_map = {}
267        for batch in results_stream:
268            for result in batch:
269                results_map[result.phone_number] = result
270
271        # Reset lines iterator
272        lines_iter = iter(original_lines)
273        next(lines_iter)  # Skip header
274
275        # Process and yield data lines
276        for line in lines_iter:
277            if not line.strip():
278                continue
279
280            row = next(csv.DictReader(StringIO(line), fieldnames=headers))
281            phone = row.get(phone_col, "").strip()
282
283            # Add verification results
284            try:
285                normalized = self.verifier.normalize_phone(phone)
286                if normalized in results_map:
287                    result = results_map[normalized]
288                    row["line_type"] = result.line_type.value
289                    row["dnc"] = "true" if result.dnc else "false"
290                else:
291                    row["line_type"] = "unknown"
292                    row["dnc"] = ""
293            except Exception:
294                row["line_type"] = "invalid"
295                row["dnc"] = ""
296
297            # Write row
298            output = StringIO()
299            writer = csv.DictWriter(output, fieldnames=output_headers)
300            writer.writerow(row)
301            yield output.getvalue()

 17class DynamoDBCache:
 18    """Cache for phone verification results using DynamoDB with TTL"""
 19
 20    def __init__(self, table_name: str, ttl_days: int = 90):
 21        self.table_name = table_name
 22        self.ttl_days = ttl_days
 23        self.dynamodb = boto3.resource("dynamodb")
 24        self.table = self.dynamodb.Table(table_name)
 25
 26    def get(self, phone_number: str) -> PhoneVerification | None:
 27        """Get cached verification result"""
 28        try:
 29            response = self.table.get_item(Key={"phone_number": phone_number})
 30
 31            if "Item" not in response:
 32                logger.info(f"Cache miss for {phone_number[:6]}***")
 33                return None
 34
 35            item: dict[str, Any] = response["Item"]
 36            logger.info(f"Cache hit for {phone_number[:6]}***")
 37
 38            return PhoneVerification(
 39                phone_number=str(item["phone_number"]),
 40                line_type=LineType(str(item["line_type"])),
 41                dnc=bool(item["dnc"]),
 42                cached=True,
 43                verified_at=datetime.fromisoformat(str(item["verified_at"])),
 44                source=VerificationSource.CACHE,
 45            )
 46
 47        except Exception as e:
 48            logger.error(f"Cache get error: {str(e)}")
 49            return None
 50
 51    def set(self, phone_number: str, verification: PhoneVerification) -> None:
 52        """Store verification result in cache"""
 53        try:
 54            ttl = int((datetime.now(UTC) + timedelta(days=self.ttl_days)).timestamp())
 55
 56            self.table.put_item(
 57                Item={
 58                    "phone_number": phone_number,
 59                    "line_type": verification.line_type.value,
 60                    "dnc": verification.dnc,
 61                    "verified_at": verification.verified_at.isoformat(),
 62                    "source": verification.source.value,
 63                    "ttl": ttl,
 64                }
 65            )
 66
 67            logger.info(f"Cached result for {phone_number[:6]}***")
 68
 69        except Exception as e:
 70            logger.error(f"Cache set error: {str(e)}")
 71            # Don't fail the request if cache write fails
 72
 73    def batch_get(self, phone_numbers: list[str]) -> dict[str, PhoneVerification | None]:
 74        """Get multiple cached results"""
 75        results: dict[str, PhoneVerification | None] = {}
 76
 77        # DynamoDB batch get (max 100 items per request)
 78        for i in range(0, len(phone_numbers), 100):
 79            batch = phone_numbers[i : i + 100]
 80
 81            try:
 82                response = self.dynamodb.batch_get_item(
 83                    RequestItems={
 84                        self.table_name: {"Keys": [{"phone_number": phone} for phone in batch]}
 85                    }
 86                )
 87
 88                for item in response.get("Responses", {}).get(self.table_name, []):
 89                    phone = str(item["phone_number"])
 90                    results[phone] = PhoneVerification(
 91                        phone_number=phone,
 92                        line_type=LineType(str(item["line_type"])),
 93                        dnc=bool(item["dnc"]),
 94                        cached=True,
 95                        verified_at=datetime.fromisoformat(str(item["verified_at"])),
 96                        source=VerificationSource.CACHE,
 97                    )
 98
 99            except Exception as e:
100                logger.error(f"Batch cache get error: {str(e)}")
101
102        # Fill in None for misses
103        for phone in phone_numbers:
104            if phone not in results:
105                results[phone] = None
106
107        return results