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 line_type, dnc_status = self.provider.verify_phone(normalized) 

60 

61 result = PhoneVerification( 

62 phone_number=normalized, 

63 line_type=line_type, 

64 dnc=dnc_status, 

65 cached=False, 

66 verified_at=datetime.now(UTC), 

67 source=VerificationSource.API, 

68 ) 

69 

70 # Store in cache if available 

71 if self.cache: 

72 try: 

73 self.cache.set(normalized, result) 

74 except Exception as e: 

75 logger.warning(f"Failed to cache result: {e}") 

76 # Continue without caching - don't fail the verification 

77 

78 return result 

79 

80 def _check_line_type(self, phone: str) -> LineType: 

81 """ 

82 Check line type (for backwards compatibility with CLI). 

83 Delegates to provider. 

84 """ 

85 line_type, _ = self.provider.verify_phone(phone) 

86 return 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 _, dnc_status = self.provider.verify_phone(phone) 

94 return dnc_status