Coverage for src/ai_lls_lib/core/processor.py: 99%
270 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-24 12:44 +0000
1"""
2Bulk CSV processing for phone verification
3"""
5import csv
6import os
7from collections.abc import Iterable, Iterator, Sequence
8from concurrent.futures import ThreadPoolExecutor
9from dataclasses import dataclass, field
10from io import StringIO
12from aws_lambda_powertools import Logger
14from ..providers.exceptions import ProviderError, ProviderFailureError
15from .models import PhoneVerification
16from .verifier import PhoneVerifier
18logger = Logger()
20# Key under which csv.DictReader stores fields beyond the header count (ragged rows).
21# Never a real header name, so it cannot collide with customer columns.
22_SURPLUS_KEY = "__surplus_fields__"
25def _dict_reader(source: StringIO, fieldnames: Sequence[str] | None = None) -> csv.DictReader:
26 """DictReader that tolerates ragged rows.
28 Extra fields are kept under _SURPLUS_KEY instead of None (which crashes DictWriter),
29 and missing fields read as "" instead of None (which crashes .strip()).
30 """
31 return csv.DictReader(source, fieldnames=fieldnames, restkey=_SURPLUS_KEY, restval="")
34def _fold_surplus(row: dict, headers: Sequence[str]) -> bool:
35 """Move surplus fields from a ragged row into its last original column.
37 Customer data is never discarded: surplus values are re-joined with commas into the
38 last header's cell so they survive into the results file. Surplus made only of empty
39 strings (a trailing separator) is dropped as noise. Returns True if data was folded.
40 Call this AFTER reading the phone value so the lookup key is never altered.
41 """
42 surplus = row.pop(_SURPLUS_KEY, None)
43 if not surplus or not headers:
44 return False
45 if not any(value != "" for value in surplus):
46 return False
47 last = headers[-1]
48 row[last] = ",".join([row.get(last, ""), *surplus])
49 return True
52# Output marker for rows the provider never answered for. Distinct from "unknown",
53# which is a genuine classification returned by the provider.
54VERIFICATION_FAILED = "verification_failed"
57@dataclass
58class ProcessingReport:
59 """Outcome of a bulk run, split by failure kind.
61 invalid_rows are customer data problems (a per-row skip). provider_failures and
62 unexpected_failures are rows the service never got an answer for; they are the
63 caller's signal that a job must not be reported as fully verified.
64 """
66 results: list[PhoneVerification] = field(default_factory=list)
67 total_rows: int = 0
68 empty_rows: int = 0
69 invalid_rows: int = 0
70 provider_failures: int = 0
71 unexpected_failures: int = 0
72 failed_phones: set[str] = field(default_factory=set)
74 @property
75 def verified(self) -> int:
76 return len(self.results)
78 @property
79 def failures(self) -> int:
80 return self.provider_failures + self.unexpected_failures
82 @property
83 def attempted(self) -> int:
84 return self.verified + self.failures + self.invalid_rows
86 @property
87 def failure_rate(self) -> float:
88 return self.failures / self.attempted if self.attempted else 0.0
91class BulkProcessor:
92 """Process CSV files for bulk phone verification"""
94 def __init__(
95 self,
96 verifier: PhoneVerifier,
97 max_failure_rate: float = 0.2,
98 max_consecutive_failures: int = 20,
99 max_workers: int | None = None,
100 ):
101 """
102 Args:
103 verifier: PhoneVerifier used for each row. Its provider and cache must be
104 thread-safe when max_workers > 1 (the shipped ones are).
105 max_failure_rate: abort the job (ProviderFailureError) when the share of rows the
106 provider never answered for exceeds this, measured over attempted rows
107 max_consecutive_failures: abort immediately after this many provider failures
108 in a row, so a total outage does not burn a call per row
109 max_workers: bounded thread pool size for provider calls; defaults from
110 LLS_BULK_MAX_WORKERS (10). 1 runs rows serially on the calling thread.
111 Per-row blocking HTTP at ~150 ms gives ~7 rows/s serial; 10 workers
112 raise that roughly tenfold, which is what makes 20k-row files fit a
113 Lambda budget. Input order is preserved in the results.
114 """
115 self.verifier = verifier
116 self.max_failure_rate = max_failure_rate
117 self.max_consecutive_failures = max_consecutive_failures
118 if max_workers is None:
119 raw = os.environ.get("LLS_BULK_MAX_WORKERS", "")
120 max_workers = int(raw) if raw else 10
121 self.max_workers = max(1, max_workers)
123 def _record_row_failure(
124 self, exc: Exception, row_num: int, phone: str, report: ProcessingReport, consecutive: int
125 ) -> int:
126 """Classify a per-row exception into the report and return the new consecutive count.
128 Raises ProviderFailureError as soon as consecutive provider failures hit the limit.
129 """
130 if isinstance(exc, ProviderError):
131 report.provider_failures += 1
132 report.failed_phones.add(phone)
133 logger.error(f"Provider failure at row {row_num}: {exc}")
134 consecutive += 1
135 elif isinstance(exc, ValueError):
136 report.invalid_rows += 1
137 logger.warning(f"Invalid phone at row {row_num}: {exc}")
138 return 0
139 else:
140 report.unexpected_failures += 1
141 report.failed_phones.add(phone)
142 logger.error(f"Verification failed at row {row_num}: {exc}")
143 consecutive += 1
145 if consecutive >= self.max_consecutive_failures:
146 raise ProviderFailureError(
147 f"Aborting: provider failed {consecutive} rows in a row (last: {exc})",
148 failures=report.failures,
149 attempted=report.attempted,
150 )
151 return consecutive
153 def _check_failure_rate(self, report: ProcessingReport) -> None:
154 """Raise ProviderFailureError if too many rows were never verified."""
155 if report.failures and report.failure_rate > self.max_failure_rate:
156 raise ProviderFailureError(
157 f"Provider failed {report.failures} of {report.attempted} rows "
158 f"({report.failure_rate:.0%}); job cannot be reported as verified",
159 failures=report.failures,
160 attempted=report.attempted,
161 )
163 def process_csv(self, csv_text: str, phone_column: str = "phone") -> list[PhoneVerification]:
164 """
165 Process CSV text content.
166 Returns list of verification results.
168 Raises ProviderFailureError when the provider failed for too many rows; see
169 process_csv_report for the per-kind counts.
170 """
171 return self.process_csv_report(csv_text, phone_column).results
173 def process_csv_report(self, csv_text: str, phone_column: str = "phone") -> ProcessingReport:
174 """
175 Process CSV text content and return a ProcessingReport with results and counts.
177 Raises ProviderFailureError when the provider failed for more than max_failure_rate
178 of attempted rows, or for max_consecutive_failures rows in a row.
179 """
180 report = ProcessingReport()
181 results = report.results
182 consecutive = 0
184 try:
185 # Strip UTF-8 BOM if present (Excel on Windows adds this)
186 csv_text = csv_text.lstrip("\ufeff")
188 # Use StringIO to parse CSV text
189 csv_file = StringIO(csv_text)
190 reader = _dict_reader(csv_file)
192 # Find phone column (case-insensitive)
193 headers = reader.fieldnames or []
194 phone_col = self._find_phone_column(headers, phone_column)
196 if not phone_col:
197 raise ValueError(f"Phone column '{phone_column}' not found in CSV")
199 logger.info(
200 f"Starting CSV processing using phone column '{phone_col}' "
201 f"with {self.max_workers} worker(s)"
202 )
204 # Collect the work list first so the pool can be fed in input order
205 work: list[tuple[int, str]] = []
206 for row_num, row in enumerate(reader, start=2): # Start at 2 (header is 1)
207 report.total_rows += 1
208 phone = row.get(phone_col, "").strip()
209 if not phone:
210 report.empty_rows += 1
211 logger.warning(f"Empty phone at row {row_num}")
212 continue
213 work.append((row_num, phone))
215 # Each outcome is either a PhoneVerification or the exception for that row.
216 # Ordered iteration keeps results in input order regardless of completion order.
217 for row_num, phone, outcome in self._verify_all(work):
218 if isinstance(outcome, PhoneVerification):
219 results.append(outcome)
220 consecutive = 0
221 if len(results) % 100 == 0:
222 logger.info(f"Processed {len(results)} phones (at row {row_num})")
223 else:
224 consecutive = self._record_row_failure(
225 outcome, row_num, phone, report, consecutive
226 )
228 self._check_failure_rate(report)
229 logger.info(
230 f"Completed processing: {report.verified} verified, "
231 f"{report.invalid_rows} invalid, {report.failures} not verified"
232 )
234 except Exception as e:
235 logger.error(f"CSV processing failed: {str(e)}")
236 raise
238 return report
240 def _verify_one(self, phone: str) -> PhoneVerification | Exception:
241 """Verify a single phone, returning the exception instead of raising."""
242 try:
243 return self.verifier.verify(phone)
244 except Exception as e: # classified by the caller
245 return e
247 def _verify_all(
248 self, work: list[tuple[int, str]]
249 ) -> Iterator[tuple[int, str, PhoneVerification | Exception]]:
250 """Yield (row_num, phone, outcome) in input order, using the bounded worker pool.
252 Serial when max_workers is 1. Otherwise rows are submitted in order and outcomes
253 yielded in order; if the consumer stops early (a ProviderFailureError from the
254 failure classifier), the remaining queued rows are cancelled instead of run.
255 """
256 if self.max_workers == 1 or len(work) <= 1:
257 for row_num, phone in work:
258 yield row_num, phone, self._verify_one(phone)
259 return
261 workers = min(self.max_workers, len(work))
262 executor = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="bulk-verify")
263 try:
264 futures = [executor.submit(self._verify_one, phone) for _, phone in work]
265 for (row_num, phone), future in zip(work, futures, strict=True):
266 yield row_num, phone, future.result()
267 finally:
268 executor.shutdown(wait=True, cancel_futures=True)
270 def _find_phone_column(self, headers: list[str] | Sequence[str], preferred: str) -> str | None:
271 """Find phone column in headers (case-insensitive)"""
272 # First try exact match
273 for header in headers:
274 if header.lower() == preferred.lower():
275 return header
277 # Common phone column names
278 phone_patterns = [
279 "phone",
280 "phone_number",
281 "phonenumber",
282 "mobile",
283 "cell",
284 "telephone",
285 "tel",
286 "number",
287 "contact",
288 ]
290 for header in headers:
291 header_lower = header.lower()
292 for pattern in phone_patterns:
293 if pattern in header_lower:
294 logger.info(f"Using column '{header}' as phone column")
295 return header
297 return None
299 def generate_results_csv(
300 self,
301 original_csv_text: str,
302 results: list[PhoneVerification],
303 failed_phones: set[str] | None = None,
304 ) -> str:
305 """
306 Generate CSV with original data plus verification results.
307 Adds columns: line_type, dnc, known_litigator
308 Rows listed in failed_phones (raw phone values the provider never answered for)
309 are marked verification_failed rather than unknown.
310 Returns CSV text string.
311 """
312 failed_phones = failed_phones or set()
313 # Create lookup dict
314 results_map = {r.phone_number: r for r in results}
316 # Parse original CSV (strip UTF-8 BOM if present)
317 original_csv_text = original_csv_text.lstrip("\ufeff")
318 input_file = StringIO(original_csv_text)
319 reader = _dict_reader(input_file)
320 headers = list(reader.fieldnames or [])
322 # Add new columns
323 output_headers = headers + ["line_type", "dnc", "known_litigator"]
325 # Create output CSV in memory
326 output = StringIO()
327 writer = csv.DictWriter(output, fieldnames=output_headers)
328 writer.writeheader()
330 phone_col = self._find_phone_column(headers, "phone")
331 folded_rows = 0
333 for row in reader:
334 # Read the lookup key before folding so surplus never alters it
335 phone = row.get(phone_col, "").strip()
336 if _fold_surplus(row, headers):
337 folded_rows += 1
339 # Try to normalize for lookup
340 try:
341 normalized = self.verifier.normalize_phone(phone)
342 if phone in failed_phones:
343 row["line_type"] = VERIFICATION_FAILED
344 row["dnc"] = ""
345 row["known_litigator"] = ""
346 elif normalized in results_map:
347 result = results_map[normalized]
348 row["line_type"] = result.line_type.value
349 row["dnc"] = "true" if result.dnc else "false"
350 row["known_litigator"] = "true" if result.known_litigator else "false"
351 else:
352 row["line_type"] = "unknown"
353 row["dnc"] = ""
354 row["known_litigator"] = ""
355 except Exception:
356 row["line_type"] = "invalid"
357 row["dnc"] = ""
358 row["known_litigator"] = ""
360 writer.writerow(row)
362 if folded_rows:
363 logger.warning(f"Folded surplus fields into last column on {folded_rows} ragged rows")
365 # Return CSV text
366 return output.getvalue()
368 def process_csv_stream(
369 self, lines: Iterable[str], phone_column: str = "phone", batch_size: int = 100
370 ) -> Iterator[list[PhoneVerification]]:
371 """
372 Process CSV lines as a stream, yielding batches of results.
373 Memory-efficient for large files.
375 Args:
376 lines: Iterator of CSV lines (including header)
377 phone_column: Column name containing phone numbers
378 batch_size: Number of results to accumulate before yielding
380 Yields:
381 Batches of PhoneVerification results
382 """
383 lines_list = list(lines) # Need to iterate twice - once for headers, once for data
385 if not lines_list:
386 logger.error("Empty CSV stream")
387 return
389 # Parse header (strip UTF-8 BOM if present)
390 header_line = lines_list[0].lstrip("\ufeff")
391 reader = _dict_reader(StringIO(header_line))
392 headers = reader.fieldnames or []
393 phone_col = self._find_phone_column(headers, phone_column)
395 if not phone_col:
396 raise ValueError(f"Phone column '{phone_column}' not found in CSV")
398 batch = []
399 row_num = 2 # Start at 2 (header is 1)
400 total_processed = 0
401 report = ProcessingReport()
402 consecutive = 0
404 # Process data lines
405 for line in lines_list[1:]:
406 if not line.strip():
407 continue
409 phone = ""
410 try:
411 # Parse single line
412 row = next(_dict_reader(StringIO(line), fieldnames=headers))
413 phone = row.get(phone_col, "").strip()
415 if not phone:
416 logger.warning(f"Empty phone at row {row_num}")
417 row_num += 1
418 continue
420 # Verify phone
421 result = self.verifier.verify(phone)
422 batch.append(result)
423 total_processed += 1
424 consecutive = 0
426 # Yield batch when full
427 if len(batch) >= batch_size:
428 logger.info(
429 f"Processed batch of {len(batch)} phones (total: {total_processed}, at row {row_num})"
430 )
431 yield batch
432 batch = []
434 except Exception as e:
435 consecutive = self._record_row_failure(e, row_num, phone, report, consecutive)
436 finally:
437 row_num += 1
439 # Yield remaining results
440 if batch:
441 logger.info(f"Processed final batch of {len(batch)} phones (total: {total_processed})")
442 yield batch
444 # Results were yielded, not kept; the report carries counts only
445 report_attempted = total_processed + report.failures + report.invalid_rows
446 if report.failures and report.failures / report_attempted > self.max_failure_rate:
447 raise ProviderFailureError(
448 f"Provider failed {report.failures} of {report_attempted} rows; "
449 "job cannot be reported as verified",
450 failures=report.failures,
451 attempted=report_attempted,
452 )
454 logger.info(f"Stream processing completed. Total processed: {total_processed}")
456 def generate_results_csv_stream(
457 self,
458 original_lines: Iterable[str],
459 results_stream: Iterator[list[PhoneVerification]],
460 phone_column: str = "phone",
461 failed_phones: set[str] | None = None,
462 ) -> Iterator[str]:
463 """
464 Generate CSV results as a stream, line by line.
465 Memory-efficient for large files.
467 Args:
468 original_lines: Iterator of original CSV lines
469 results_stream: Iterator of batched PhoneVerification results
470 phone_column: Column name containing phone numbers
471 failed_phones: raw phone values the provider never answered for; marked
472 verification_failed rather than unknown
474 Yields:
475 CSV lines with verification results added
476 """
477 failed_phones = failed_phones or set()
478 lines_iter = iter(original_lines)
480 # Read and yield modified header
481 try:
482 header_line = next(lines_iter).lstrip("\ufeff")
483 reader = _dict_reader(StringIO(header_line))
484 headers = list(reader.fieldnames or [])
486 # Add new columns
487 output_headers = headers + ["line_type", "dnc", "known_litigator"]
488 yield ",".join(output_headers) + "\n"
490 phone_col = self._find_phone_column(headers, phone_column)
492 except StopIteration:
493 return
495 # Build results lookup from stream
496 results_map = {}
497 for batch in results_stream:
498 for result in batch:
499 results_map[result.phone_number] = result
501 # Reset lines iterator
502 lines_iter = iter(original_lines)
503 next(lines_iter) # Skip header
505 # Process and yield data lines
506 for line in lines_iter:
507 if not line.strip():
508 continue
510 row = next(_dict_reader(StringIO(line), fieldnames=headers))
511 # Read the lookup key before folding so surplus never alters it
512 phone = row.get(phone_col, "").strip()
513 _fold_surplus(row, headers)
515 # Add verification results
516 try:
517 normalized = self.verifier.normalize_phone(phone)
518 if phone in failed_phones:
519 row["line_type"] = VERIFICATION_FAILED
520 row["dnc"] = ""
521 row["known_litigator"] = ""
522 elif normalized in results_map:
523 result = results_map[normalized]
524 row["line_type"] = result.line_type.value
525 row["dnc"] = "true" if result.dnc else "false"
526 row["known_litigator"] = "true" if result.known_litigator else "false"
527 else:
528 row["line_type"] = "unknown"
529 row["dnc"] = ""
530 row["known_litigator"] = ""
531 except Exception:
532 row["line_type"] = "invalid"
533 row["dnc"] = ""
534 row["known_litigator"] = ""
536 # Write row
537 output = StringIO()
538 writer = csv.DictWriter(output, fieldnames=output_headers)
539 writer.writerow(row)
540 yield output.getvalue()