ai_lls_lib

47class PhoneVerification(BaseModel):
48    """Result of phone number verification"""
49
50    phone_number: str = Field(..., description="E.164 formatted phone number")
51    line_type: LineType = Field(..., description="Type of phone line")
52    dnc: bool = Field(..., description="Whether number is on DNC list")
53    known_litigator: bool = Field(False, description="Whether number belongs to a known litigator")
54    cached: bool = Field(..., description="Whether result came from cache")
55    verified_at: datetime = Field(..., description="When verification occurred")
56    source: VerificationSource = Field(..., description="Source of verification data")
57
58    class Config:
59        json_encoders = {datetime: lambda v: v.isoformat()}

58    class Config:
59        json_encoders = {datetime: lambda v: v.isoformat()}

62class BulkJob(BaseModel):
63    """Bulk processing job metadata"""
64
65    job_id: str = Field(..., description="Unique job identifier")
66    status: JobStatus = Field(..., description="Current job status")

69class BulkJobStatus(BulkJob):
70    """Extended bulk job status with progress info"""
71
72    total_rows: int | None = Field(None, description="Total rows to process")
73    processed_rows: int | None = Field(None, description="Rows processed so far")
74    result_url: str | None = Field(None, description="S3 URL of results")
75    created_at: datetime = Field(..., description="Job creation time")
76    completed_at: datetime | None = Field(None, description="Job completion time")
77    error: str | None = Field(None, description="Error message if failed")

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

39class VerificationResult(NamedTuple):
40    """Raw verification result returned by a provider"""
41
42    line_type: LineType
43    dnc: bool
44    known_litigator: bool = False

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

30class JobStatus(StrEnum):
31    """Bulk job status enumeration"""
32
33    PENDING = "pending"
34    PROCESSING = "processing"
35    COMPLETED = "completed"
36    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        provider_result = self.provider.verify_phone(normalized)
61
62        result = PhoneVerification(
63            phone_number=normalized,
64            line_type=provider_result.line_type,
65            dnc=provider_result.dnc,
66            known_litigator=provider_result.known_litigator,
67            cached=False,
68            verified_at=datetime.now(UTC),
69            source=VerificationSource.API,
70        )
71
72        # Store in cache if available
73        if self.cache:
74            try:
75                self.cache.set(normalized, result)
76            except Exception as e:
77                logger.warning(f"Failed to cache result: {e}")
78                # Continue without caching - don't fail the verification
79
80        return result
81
82    def _check_line_type(self, phone: str) -> LineType:
83        """
84        Check line type (for backwards compatibility with CLI).
85        Delegates to provider.
86        """
87        return self.provider.verify_phone(phone).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        return self.provider.verify_phone(phone).dnc

 92class BulkProcessor:
 93    """Process CSV files for bulk phone verification"""
 94
 95    def __init__(
 96        self,
 97        verifier: PhoneVerifier,
 98        max_failure_rate: float = 0.2,
 99        max_consecutive_failures: int = 20,
100        max_workers: int | None = None,
101    ):
102        """
103        Args:
104            verifier: PhoneVerifier used for each row. Its provider and cache must be
105                thread-safe when max_workers > 1 (the shipped ones are).
106            max_failure_rate: abort the job (ProviderFailureError) when the share of rows the
107                provider never answered for exceeds this, measured over attempted rows
108            max_consecutive_failures: abort immediately after this many provider failures
109                in a row, so a total outage does not burn a call per row
110            max_workers: bounded thread pool size for provider calls; defaults from
111                LLS_BULK_MAX_WORKERS (10). 1 runs rows serially on the calling thread.
112                Per-row blocking HTTP at ~150 ms gives ~7 rows/s serial; 10 workers
113                raise that roughly tenfold, which is what makes 20k-row files fit a
114                Lambda budget. Input order is preserved in the results.
115        """
116        self.verifier = verifier
117        self.max_failure_rate = max_failure_rate
118        self.max_consecutive_failures = max_consecutive_failures
119        if max_workers is None:
120            raw = os.environ.get("LLS_BULK_MAX_WORKERS", "")
121            max_workers = int(raw) if raw else 10
122        self.max_workers = max(1, max_workers)
123
124    def _record_row_failure(
125        self, exc: Exception, row_num: int, phone: str, report: ProcessingReport, consecutive: int
126    ) -> int:
127        """Classify a per-row exception into the report and return the new consecutive count.
128
129        Raises ProviderFailureError as soon as consecutive provider failures hit the limit.
130        """
131        if isinstance(exc, ProviderError):
132            report.provider_failures += 1
133            report.failed_phones.add(phone)
134            logger.error(f"Provider failure at row {row_num}: {exc}")
135            consecutive += 1
136        elif isinstance(exc, ValueError):
137            report.invalid_rows += 1
138            logger.warning(f"Invalid phone at row {row_num}: {exc}")
139            return 0
140        else:
141            report.unexpected_failures += 1
142            report.failed_phones.add(phone)
143            logger.error(f"Verification failed at row {row_num}: {exc}")
144            consecutive += 1
145
146        if consecutive >= self.max_consecutive_failures:
147            raise ProviderFailureError(
148                f"Aborting: provider failed {consecutive} rows in a row (last: {exc})",
149                failures=report.failures,
150                attempted=report.attempted,
151            )
152        return consecutive
153
154    def _check_failure_rate(self, report: ProcessingReport) -> None:
155        """Raise ProviderFailureError if too many rows were never verified."""
156        if report.failures and report.failure_rate > self.max_failure_rate:
157            raise ProviderFailureError(
158                f"Provider failed {report.failures} of {report.attempted} rows "
159                f"({report.failure_rate:.0%}); job cannot be reported as verified",
160                failures=report.failures,
161                attempted=report.attempted,
162            )
163
164    def process_csv(self, csv_text: str, phone_column: str = "phone") -> list[PhoneVerification]:
165        """
166        Process CSV text content.
167        Returns list of verification results.
168
169        Raises ProviderFailureError when the provider failed for too many rows; see
170        process_csv_report for the per-kind counts.
171        """
172        return self.process_csv_report(csv_text, phone_column).results
173
174    def process_csv_report(self, csv_text: str, phone_column: str = "phone") -> ProcessingReport:
175        """
176        Process CSV text content and return a ProcessingReport with results and counts.
177
178        Raises ProviderFailureError when the provider failed for more than max_failure_rate
179        of attempted rows, or for max_consecutive_failures rows in a row.
180        """
181        report = ProcessingReport()
182        results = report.results
183        consecutive = 0
184
185        try:
186            # Strip UTF-8 BOM if present (Excel on Windows adds this)
187            csv_text = csv_text.lstrip("\ufeff")
188
189            # Use StringIO to parse CSV text
190            csv_file = StringIO(csv_text)
191            reader = _dict_reader(csv_file)
192
193            # Find phone column (case-insensitive)
194            headers = reader.fieldnames or []
195            phone_col = self._find_phone_column(headers, phone_column)
196
197            if not phone_col:
198                raise ValueError(f"Phone column '{phone_column}' not found in CSV")
199
200            logger.info(
201                f"Starting CSV processing using phone column '{phone_col}' "
202                f"with {self.max_workers} worker(s)"
203            )
204
205            # Collect the work list first so the pool can be fed in input order
206            work: list[tuple[int, str]] = []
207            for row_num, row in enumerate(reader, start=2):  # Start at 2 (header is 1)
208                report.total_rows += 1
209                phone = row.get(phone_col, "").strip()
210                if not phone:
211                    report.empty_rows += 1
212                    logger.warning(f"Empty phone at row {row_num}")
213                    continue
214                work.append((row_num, phone))
215
216            # Each outcome is either a PhoneVerification or the exception for that row.
217            # Ordered iteration keeps results in input order regardless of completion order.
218            for row_num, phone, outcome in self._verify_all(work):
219                if isinstance(outcome, PhoneVerification):
220                    results.append(outcome)
221                    consecutive = 0
222                    if len(results) % 100 == 0:
223                        logger.info(f"Processed {len(results)} phones (at row {row_num})")
224                else:
225                    consecutive = self._record_row_failure(
226                        outcome, row_num, phone, report, consecutive
227                    )
228
229            self._check_failure_rate(report)
230            logger.info(
231                f"Completed processing: {report.verified} verified, "
232                f"{report.invalid_rows} invalid, {report.failures} not verified"
233            )
234
235        except Exception as e:
236            logger.error(f"CSV processing failed: {str(e)}")
237            raise
238
239        return report
240
241    def _verify_one(self, phone: str) -> PhoneVerification | Exception:
242        """Verify a single phone, returning the exception instead of raising."""
243        try:
244            return self.verifier.verify(phone)
245        except Exception as e:  # classified by the caller
246            return e
247
248    def _verify_all(
249        self, work: list[tuple[int, str]]
250    ) -> Iterator[tuple[int, str, PhoneVerification | Exception]]:
251        """Yield (row_num, phone, outcome) in input order, using the bounded worker pool.
252
253        Serial when max_workers is 1. Otherwise rows are submitted in order and outcomes
254        yielded in order; if the consumer stops early (a ProviderFailureError from the
255        failure classifier), the remaining queued rows are cancelled instead of run.
256        """
257        if self.max_workers == 1 or len(work) <= 1:
258            for row_num, phone in work:
259                yield row_num, phone, self._verify_one(phone)
260            return
261
262        workers = min(self.max_workers, len(work))
263        executor = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="bulk-verify")
264        try:
265            futures = [executor.submit(self._verify_one, phone) for _, phone in work]
266            for (row_num, phone), future in zip(work, futures, strict=True):
267                yield row_num, phone, future.result()
268        finally:
269            executor.shutdown(wait=True, cancel_futures=True)
270
271    def _find_phone_column(self, headers: list[str] | Sequence[str], preferred: str) -> str | None:
272        """Find phone column in headers (case-insensitive)"""
273        # First try exact match
274        for header in headers:
275            if header.lower() == preferred.lower():
276                return header
277
278        # Common phone column names
279        phone_patterns = [
280            "phone",
281            "phone_number",
282            "phonenumber",
283            "mobile",
284            "cell",
285            "telephone",
286            "tel",
287            "number",
288            "contact",
289        ]
290
291        for header in headers:
292            header_lower = header.lower()
293            for pattern in phone_patterns:
294                if pattern in header_lower:
295                    logger.info(f"Using column '{header}' as phone column")
296                    return header
297
298        return None
299
300    def generate_results_csv(
301        self,
302        original_csv_text: str,
303        results: list[PhoneVerification],
304        failed_phones: set[str] | None = None,
305    ) -> str:
306        """
307        Generate CSV with original data plus verification results.
308        Adds columns: line_type, dnc, known_litigator
309        Rows listed in failed_phones (raw phone values the provider never answered for)
310        are marked verification_failed rather than unknown.
311        Returns CSV text string.
312        """
313        failed_phones = failed_phones or set()
314        # Create lookup dict
315        results_map = {r.phone_number: r for r in results}
316
317        # Parse original CSV (strip UTF-8 BOM if present)
318        original_csv_text = original_csv_text.lstrip("\ufeff")
319        input_file = StringIO(original_csv_text)
320        reader = _dict_reader(input_file)
321        headers = list(reader.fieldnames or [])
322
323        # Add new columns
324        output_headers = headers + ["line_type", "dnc", "known_litigator"]
325
326        # Create output CSV in memory
327        output = StringIO()
328        writer = csv.DictWriter(output, fieldnames=output_headers)
329        writer.writeheader()
330
331        phone_col = self._find_phone_column(headers, "phone")
332        folded_rows = 0
333
334        for row in reader:
335            # Read the lookup key before folding so surplus never alters it
336            phone = row.get(phone_col, "").strip()
337            if _fold_surplus(row, headers):
338                folded_rows += 1
339
340            # Try to normalize for lookup
341            try:
342                normalized = self.verifier.normalize_phone(phone)
343                if phone in failed_phones:
344                    row["line_type"] = VERIFICATION_FAILED
345                    row["dnc"] = ""
346                    row["known_litigator"] = ""
347                elif normalized in results_map:
348                    result = results_map[normalized]
349                    row["line_type"] = result.line_type.value
350                    row["dnc"] = "true" if result.dnc else "false"
351                    row["known_litigator"] = "true" if result.known_litigator else "false"
352                else:
353                    row["line_type"] = "unknown"
354                    row["dnc"] = ""
355                    row["known_litigator"] = ""
356            except Exception:
357                row["line_type"] = "invalid"
358                row["dnc"] = ""
359                row["known_litigator"] = ""
360
361            writer.writerow(row)
362
363        if folded_rows:
364            logger.warning(f"Folded surplus fields into last column on {folded_rows} ragged rows")
365
366        # Return CSV text
367        return output.getvalue()
368
369    def process_csv_stream(
370        self, lines: Iterable[str], phone_column: str = "phone", batch_size: int = 100
371    ) -> Iterator[list[PhoneVerification]]:
372        """
373        Process CSV lines as a stream, yielding batches of results.
374        Memory-efficient for large files.
375
376        Args:
377            lines: Iterator of CSV lines (including header)
378            phone_column: Column name containing phone numbers
379            batch_size: Number of results to accumulate before yielding
380
381        Yields:
382            Batches of PhoneVerification results
383        """
384        lines_list = list(lines)  # Need to iterate twice - once for headers, once for data
385
386        if not lines_list:
387            logger.error("Empty CSV stream")
388            return
389
390        # Parse header (strip UTF-8 BOM if present)
391        header_line = lines_list[0].lstrip("\ufeff")
392        reader = _dict_reader(StringIO(header_line))
393        headers = reader.fieldnames or []
394        phone_col = self._find_phone_column(headers, phone_column)
395
396        if not phone_col:
397            raise ValueError(f"Phone column '{phone_column}' not found in CSV")
398
399        batch = []
400        row_num = 2  # Start at 2 (header is 1)
401        total_processed = 0
402        report = ProcessingReport()
403        consecutive = 0
404
405        # Process data lines
406        for line in lines_list[1:]:
407            if not line.strip():
408                continue
409
410            phone = ""
411            try:
412                # Parse single line
413                row = next(_dict_reader(StringIO(line), fieldnames=headers))
414                phone = row.get(phone_col, "").strip()
415
416                if not phone:
417                    logger.warning(f"Empty phone at row {row_num}")
418                    row_num += 1
419                    continue
420
421                # Verify phone
422                result = self.verifier.verify(phone)
423                batch.append(result)
424                total_processed += 1
425                consecutive = 0
426
427                # Yield batch when full
428                if len(batch) >= batch_size:
429                    logger.info(
430                        f"Processed batch of {len(batch)} phones (total: {total_processed}, at row {row_num})"
431                    )
432                    yield batch
433                    batch = []
434
435            except Exception as e:
436                consecutive = self._record_row_failure(e, row_num, phone, report, consecutive)
437            finally:
438                row_num += 1
439
440        # Yield remaining results
441        if batch:
442            logger.info(f"Processed final batch of {len(batch)} phones (total: {total_processed})")
443            yield batch
444
445        # Results were yielded, not kept; the report carries counts only
446        report_attempted = total_processed + report.failures + report.invalid_rows
447        if report.failures and report.failures / report_attempted > self.max_failure_rate:
448            raise ProviderFailureError(
449                f"Provider failed {report.failures} of {report_attempted} rows; "
450                "job cannot be reported as verified",
451                failures=report.failures,
452                attempted=report_attempted,
453            )
454
455        logger.info(f"Stream processing completed. Total processed: {total_processed}")
456
457    def generate_results_csv_stream(
458        self,
459        original_lines: Iterable[str],
460        results_stream: Iterator[list[PhoneVerification]],
461        phone_column: str = "phone",
462        failed_phones: set[str] | None = None,
463    ) -> Iterator[str]:
464        """
465        Generate CSV results as a stream, line by line.
466        Memory-efficient for large files.
467
468        Args:
469            original_lines: Iterator of original CSV lines
470            results_stream: Iterator of batched PhoneVerification results
471            phone_column: Column name containing phone numbers
472            failed_phones: raw phone values the provider never answered for; marked
473                verification_failed rather than unknown
474
475        Yields:
476            CSV lines with verification results added
477        """
478        failed_phones = failed_phones or set()
479        lines_iter = iter(original_lines)
480
481        # Read and yield modified header
482        try:
483            header_line = next(lines_iter).lstrip("\ufeff")
484            reader = _dict_reader(StringIO(header_line))
485            headers = list(reader.fieldnames or [])
486
487            # Add new columns
488            output_headers = headers + ["line_type", "dnc", "known_litigator"]
489            yield ",".join(output_headers) + "\n"
490
491            phone_col = self._find_phone_column(headers, phone_column)
492
493        except StopIteration:
494            return
495
496        # Build results lookup from stream
497        results_map = {}
498        for batch in results_stream:
499            for result in batch:
500                results_map[result.phone_number] = result
501
502        # Reset lines iterator
503        lines_iter = iter(original_lines)
504        next(lines_iter)  # Skip header
505
506        # Process and yield data lines
507        for line in lines_iter:
508            if not line.strip():
509                continue
510
511            row = next(_dict_reader(StringIO(line), fieldnames=headers))
512            # Read the lookup key before folding so surplus never alters it
513            phone = row.get(phone_col, "").strip()
514            _fold_surplus(row, headers)
515
516            # Add verification results
517            try:
518                normalized = self.verifier.normalize_phone(phone)
519                if phone in failed_phones:
520                    row["line_type"] = VERIFICATION_FAILED
521                    row["dnc"] = ""
522                    row["known_litigator"] = ""
523                elif normalized in results_map:
524                    result = results_map[normalized]
525                    row["line_type"] = result.line_type.value
526                    row["dnc"] = "true" if result.dnc else "false"
527                    row["known_litigator"] = "true" if result.known_litigator else "false"
528                else:
529                    row["line_type"] = "unknown"
530                    row["dnc"] = ""
531                    row["known_litigator"] = ""
532            except Exception:
533                row["line_type"] = "invalid"
534                row["dnc"] = ""
535                row["known_litigator"] = ""
536
537            # Write row
538            output = StringIO()
539            writer = csv.DictWriter(output, fieldnames=output_headers)
540            writer.writerow(row)
541            yield output.getvalue()