Coverage for src/ai_lls_lib/core/verifier.py: 100%

1""" 

2Phone verification logic - checks line type and DNC status 

3""" 

4 

5from datetime import UTC, datetime 

6 

7import phonenumbers 

8from aws_lambda_powertools import Logger 

9 

10from ..providers import ExternalAPIProvider, VerificationProvider 

11from .cache import DynamoDBCache 

12from .models import LineType, PhoneVerification, VerificationSource 

13 

14logger = Logger() 

15 

16 

17class PhoneVerifier: 

18 """Verifies phone numbers for line type and DNC status""" 

19 

20 def __init__( 

21 self, cache: DynamoDBCache | None = None, provider: VerificationProvider | None = None 

22 ): 

23 """ 

24 Initialize phone verifier. 

25 

26 Args: 

27 cache: Optional DynamoDB cache for storing results 

28 provider: Verification provider (defaults to ExternalAPIProvider) 

29 """ 

30 self.cache = cache 

31 self.provider = provider or ExternalAPIProvider() 

32 logger.debug("PhoneVerifier initialized") 

33 

34 def normalize_phone(self, phone: str) -> str: 

35 """Normalize phone to E.164 format""" 

36 try: 

37 # Parse with US as default country 

38 parsed = phonenumbers.parse(phone, "US") 

39 if not phonenumbers.is_valid_number(parsed): 

40 raise ValueError(f"Invalid phone number: {phone}") 

41 

42 # Format as E.164 

43 return phonenumbers.format_number(parsed, phonenumbers.PhoneNumberFormat.E164) 

44 except Exception as e: 

45 # Callers classify invalid rows themselves (BulkProcessor logs 

46 # them at WARNING), so there is no ERROR record here. 

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