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 logger.error(f"Phone normalization failed: {str(e)}") 

46 raise ValueError(f"Invalid phone format: {phone}") from e 

47 

48 def verify(self, phone: str) -> PhoneVerification: 

49 """Verify phone number for line type and DNC status""" 

50 normalized = self.normalize_phone(phone) 

51 

52 # Check cache first if available 

53 if self.cache: 

54 cached = self.cache.get(normalized) 

55 if cached: 

56 return cached 

57 

58 # Use provider to verify 

59 provider_result = self.provider.verify_phone(normalized) 

60 

61 result = PhoneVerification( 

62 phone_number=normalized, 

63 line_type=provider_result.line_type, 

64 dnc=provider_result.dnc, 

65 known_litigator=provider_result.known_litigator, 

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 return self.provider.verify_phone(phone).line_type 

87 

88 def _check_dnc(self, phone: str) -> bool: 

89 """ 

90 Check DNC status (for backwards compatibility with CLI). 

91 Delegates to provider. 

92 """ 

93 return self.provider.verify_phone(phone).dnc