Coverage for custom_components/supernotify/transports/mobile_push.py: 93%

1import logging 

2from datetime import timedelta 

3from typing import Any 

4 

5import httpx 

6from bs4 import BeautifulSoup 

7from homeassistant.components.notify.const import ATTR_DATA 

8 

9import custom_components.supernotify 

10from custom_components.supernotify import ( 

11 ATTR_ACTION_CATEGORY, 

12 ATTR_ACTION_URL, 

13 ATTR_ACTION_URL_TITLE, 

14 ATTR_MEDIA_CAMERA_ENTITY_ID, 

15 ATTR_MEDIA_CLIP_URL, 

16 ATTR_MEDIA_SNAPSHOT_URL, 

17 ATTR_MOBILE_APP_ID, 

18 CONF_MOBILE_APP_ID, 

19 CONF_MOBILE_DEVICES, 

20 CONF_PERSON, 

21 OPTION_MESSAGE_USAGE, 

22 OPTION_SIMPLIFY_TEXT, 

23 OPTION_STRIP_URLS, 

24 OPTION_TARGET_CATEGORIES, 

25 TRANSPORT_MOBILE_PUSH, 

26) 

27from custom_components.supernotify.envelope import Envelope 

28from custom_components.supernotify.model import ( 

29 CommandType, 

30 MessageOnlyPolicy, 

31 QualifiedTargetType, 

32 RecipientType, 

33 TargetRequired, 

34 TransportConfig, 

35) 

36from custom_components.supernotify.transport import ( 

37 Transport, 

38) 

39 

40_LOGGER = logging.getLogger(__name__) 

41 

42 

43class MobilePushTransport(Transport): 

44 name = TRANSPORT_MOBILE_PUSH 

45 

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

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

48 self.action_titles: dict[str, str] = {} 

49 

50 @property 

51 def default_config(self) -> TransportConfig: 

52 config = TransportConfig() 

53 config.delivery_defaults.target_required = TargetRequired.ALWAYS 

54 config.delivery_defaults.options = { 

55 OPTION_SIMPLIFY_TEXT: False, 

56 OPTION_STRIP_URLS: False, 

57 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD, 

58 OPTION_TARGET_CATEGORIES: [ATTR_MOBILE_APP_ID], 

59 } 

60 return config 

61 

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

63 return action is None 

64 

65 async def action_title(self, url: str) -> str | None: 

66 if url in self.action_titles: 

67 return self.action_titles[url] 

68 try: 

69 async with httpx.AsyncClient(timeout=httpx.Timeout(timeout=5.0)) as client: 

70 resp: httpx.Response = await client.get(url, follow_redirects=True, timeout=5) 

71 html = BeautifulSoup(resp.text, features="html.parser") 

72 if html.title and html.title.string: 

73 self.action_titles[url] = html.title.string 

74 return html.title.string 

75 except Exception as e: 

76 _LOGGER.debug("SUPERNOTIFY failed to retrieve url title at %s: %s", url, e) 

77 return None 

78 

79 async def deliver(self, envelope: Envelope) -> bool: 

80 if not envelope.target.mobile_app_ids: 

81 _LOGGER.warning("SUPERNOTIFY No targets provided for mobile_push") 

82 return False 

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

84 # TODO: category not passed anywhere 

85 category = data.get(ATTR_ACTION_CATEGORY, "general") 

86 action_groups = envelope.action_groups 

87 

88 _LOGGER.debug("SUPERNOTIFY notify_mobile: %s -> %s", envelope.title, envelope.target.mobile_app_ids) 

89 

90 media = envelope.media or {} 

91 camera_entity_id = media.get(ATTR_MEDIA_CAMERA_ENTITY_ID) 

92 clip_url: str | None = self.hass_api.abs_url(media.get(ATTR_MEDIA_CLIP_URL)) 

93 snapshot_url: str | None = self.hass_api.abs_url(media.get(ATTR_MEDIA_SNAPSHOT_URL)) 

94 # options = data.get(CONF_OPTIONS, {}) 

95 

96 match envelope.priority: 

97 case custom_components.supernotify.PRIORITY_CRITICAL: 

98 push_priority = "critical" 

99 case custom_components.supernotify.PRIORITY_HIGH: 

100 push_priority = "time-sensitive" 

101 case custom_components.supernotify.PRIORITY_MEDIUM: 

102 push_priority = "active" 

103 case custom_components.supernotify.PRIORITY_LOW: 

104 push_priority = "passive" 

105 case _: 

106 push_priority = "active" 

107 _LOGGER.warning("SUPERNOTIFY Unexpected priority %s", envelope.priority) 

108 

109 data.setdefault("actions", []) 

110 data.setdefault("push", {}) 

111 data["push"]["interruption-level"] = push_priority 

112 if push_priority == "critical": 

113 data["push"].setdefault("sound", {}) 

114 data["push"]["sound"].setdefault("name", "default") 

115 data["push"]["sound"]["critical"] = 1 

116 data["push"]["sound"].setdefault("volume", 1.0) 

117 else: 

118 # critical notifications can't be grouped on iOS 

119 category = category or camera_entity_id or "appd" 

120 data.setdefault("group", category) 

121 

122 if camera_entity_id: 

123 data["entity_id"] = camera_entity_id 

124 # data['actions'].append({'action':'URI','title':'View Live','uri':'/cameras/%s' % device} 

125 if clip_url: 

126 data["video"] = clip_url 

127 if snapshot_url: 

128 data["image"] = snapshot_url 

129 

130 data.setdefault("actions", []) 

131 for action in envelope.actions: 

132 app_url: str | None = self.hass_api.abs_url(action.get(ATTR_ACTION_URL)) 

133 if app_url: 

134 app_url_title = action.get(ATTR_ACTION_URL_TITLE) or self.action_title(app_url) or "Click for Action" 

135 action[ATTR_ACTION_URL_TITLE] = app_url_title 

136 data["actions"].append(action) 

137 if camera_entity_id: 

138 data["actions"].append({ 

139 "action": f"SUPERNOTIFY_SNOOZE_EVERYONE_CAMERA_{camera_entity_id}", 

140 "title": f"Snooze camera notifications for {camera_entity_id}", 

141 "behavior": "textInput", 

142 "textInputButtonTitle": "Minutes to snooze", 

143 "textInputPlaceholder": "60", 

144 }) 

145 for group, actions in self.context.mobile_actions.items(): 

146 if action_groups is None or group in action_groups: 

147 data["actions"].extend(actions) 

148 if not data["actions"]: 

149 del data["actions"] 

150 action_data = envelope.core_action_data() 

151 action_data[ATTR_DATA] = data 

152 hits = 0 

153 for mobile_target in envelope.target.mobile_app_ids: 

154 full_target = mobile_target if mobile_target.startswith("notify.") else f"notify.{mobile_target}" 

155 if await self.call_action(envelope, qualified_action=full_target, action_data=action_data, implied_target=True): 

156 hits += 1 

157 else: 

158 simple_target = ( 

159 mobile_target if not mobile_target.startswith("notify.") else mobile_target.replace("notify.", "") 

160 ) 

161 _LOGGER.warning("SUPERNOTIFY Failed to send to %s, snoozing for a day", simple_target) 

162 if self.people_registry: 

163 # somewhat hacky way to tie the mobile device back to a recipient to please the snoozing api 

164 for recipient in self.people_registry.people.values(): 

165 for md in recipient.get(CONF_MOBILE_DEVICES, []): 

166 if md.get(CONF_MOBILE_APP_ID) in (simple_target, mobile_target): 

167 self.context.snoozer.register_snooze( 

168 CommandType.SNOOZE, 

169 target_type=QualifiedTargetType.MOBILE, 

170 target=simple_target, 

171 recipient_type=RecipientType.USER, 

172 recipient=recipient[CONF_PERSON], 

173 snooze_for=timedelta(days=1), 

174 reason="Action Failure", 

175 ) 

176 return hits > 0