Coverage for custom_components/supernotify/transports/sms.py: 100%

1from __future__ import annotations 

2 

3import logging 

4import unicodedata 

5from typing import TYPE_CHECKING, Any 

6 

7from homeassistant.components.notify.const import ATTR_DATA, ATTR_TARGET 

8 

9from custom_components.supernotify.const import ( 

10 ATTR_PHONE, 

11 TRANSPORT_SMS, 

12) 

13from custom_components.supernotify.model import ( 

14 DebugTrace, 

15 MessageOnlyPolicy, 

16 SuppressionReason, 

17 TransportConfig, 

18 TransportFeature, 

19) 

20from custom_components.supernotify.options import ( 

21 OPTION_MESSAGE_USAGE, 

22 OPTION_SIMPLIFY_TEXT, 

23 OPTION_STRIP_URLS, 

24) 

25from custom_components.supernotify.target import TargetEntityCategory 

26from custom_components.supernotify.transport import ( 

27 Transport, 

28) 

29 

30if TYPE_CHECKING: 

31 from custom_components.supernotify.envelope import Envelope 

32 

33RE_VALID_PHONE = r"^(\+\d{1,3})?\s?\(?\d{1,4}\)?[\s.-]?\d{3}[\s.-]?\d{4}$" 

34 

35_LOGGER = logging.getLogger(__name__) 

36 

37 

38class SMSTransport(Transport): 

39 name = TRANSPORT_SMS 

40 action_configured = True 

41 MAX_MESSAGE_LENGTH = 158 

42 

43 def __init__(self, *args: Any, **kwargs: Any) -> None: 

44 super().__init__(*args, **kwargs) 

45 

46 @property 

47 def supported_features(self) -> TransportFeature: 

48 return TransportFeature.MESSAGE 

49 

50 @property 

51 def default_config(self) -> TransportConfig: 

52 config = super().default_config 

53 config.delivery_defaults.options = { 

54 OPTION_SIMPLIFY_TEXT: True, 

55 OPTION_STRIP_URLS: False, 

56 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.COMBINE_TITLE, 

57 } 

58 for module in ( 

59 "homeassistant.components.twilio_sms.notify", 

60 "custom_components.mikrotik_sms.notify", 

61 ): 

62 action: str | None = self.hass_api.find_service("notify", module) 

63 if action: 

64 config.delivery_defaults.action = action 

65 _LOGGER.info("SUPERNOTIFY SMS action defaults to %s", action) 

66 break 

67 return config 

68 

69 @property 

70 def unique_target_categories(self) -> list[str | TargetEntityCategory]: 

71 return [ATTR_PHONE] 

72 

73 def validate_action(self, action: str | None) -> bool: 

74 """Override in subclass if transport has fixed action or doesn't require one""" 

75 return action is not None 

76 

77 def simplify(self, text: str | None, strip_urls: bool = False) -> str | None: 

78 """Remove emoji and other pictographic symbols, but not currency signs, brackets or maths 

79 symbols as for a voice assistant, since SMS can carry these and the message needs them""" 

80 if not text: 

81 return None 

82 if strip_urls: 

83 text = self._strip_urls(text) 

84 text = unicodedata.normalize("NFC", text) 

85 return "".join(c for c in text if unicodedata.category(c) not in ("So", "Sk", "Mn")) 

86 

87 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: 

88 _LOGGER.debug("SUPERNOTIFY notify_sms: %s", envelope.delivery_name) 

89 

90 data: dict[str, Any] = envelope.data or {} 

91 # resolved_targets(), not the typed `.phone` getter: envelope.target is already 

92 # scoped to this delivery by Delivery.select_targets(), so this also picks up a 

93 # `sms:`/`{sms: ...}`-qualified number that isn't shaped like a validated one 

94 # (e.g. a short code) 

95 mobile_numbers = envelope.target.resolved_targets() if envelope.target else [] 

96 

97 if not envelope.message: 

98 _LOGGER.warning("SUPERNOTIFY notify_sms: No message to send") 

99 envelope.skipped = 1 

100 envelope.skip_reason = SuppressionReason.INVALID_ACTION_DATA 

101 return False 

102 

103 message: str = envelope.message or "" 

104 if len(message) > self.MAX_MESSAGE_LENGTH: 

105 _LOGGER.debug( 

106 "SUPERNOTIFY notify_sms: Message too long (%d characters), truncating to %d characters", 

107 len(message), 

108 self.MAX_MESSAGE_LENGTH, 

109 ) 

110 

111 action_data = {"message": message[: self.MAX_MESSAGE_LENGTH], ATTR_TARGET: mobile_numbers} 

112 if data and data.get("data"): 

113 action_data[ATTR_DATA] = data.get("data", {}) 

114 

115 return await self.call_action(envelope, action_data=action_data)