Coverage for custom_components/supernotify/methods/mobile_push.py: 92%

1import logging 

2import re 

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 CONF_MOBILE_DEVICES, 

18 CONF_NOTIFY_ACTION, 

19 CONF_PERSON, 

20 CONF_TARGETS_REQUIRED, 

21 METHOD_MOBILE_PUSH, 

22 CommandType, 

23 QualifiedTargetType, 

24 RecipientType, 

25) 

26from custom_components.supernotify.delivery_method import DeliveryMethod 

27from custom_components.supernotify.envelope import Envelope 

28 

29RE_VALID_MOBILE_APP = r"mobile_app_[A-Za-z0-9_]+" 

30 

31_LOGGER = logging.getLogger(__name__) 

32 

33 

34class MobilePushDeliveryMethod(DeliveryMethod): 

35 method = METHOD_MOBILE_PUSH 

36 

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

38 kwargs.setdefault(CONF_TARGETS_REQUIRED, False) # notify entities used 

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

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

41 

42 def select_target(self, target: str) -> bool: 

43 return re.fullmatch(RE_VALID_MOBILE_APP, target) is not None 

44 

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

46 return action is None 

47 

48 def recipient_target(self, recipient: dict[str, Any]) -> list[str]: 

49 if CONF_PERSON in recipient: 

50 services: list[str] = [md.get(CONF_NOTIFY_ACTION) for md in recipient.get(CONF_MOBILE_DEVICES, [])] 

51 return list(filter(None, services)) 

52 return [] 

53 

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

55 if url in self.action_titles: 

56 return self.action_titles[url] 

57 try: 

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

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

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

61 if html.title and html.title.string: 

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

63 return html.title.string 

64 except Exception as e: 

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

66 return None 

67 

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

69 if not envelope.targets: 

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

71 return False 

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

73 # TODO: category not passed anywhere 

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

75 action_groups = envelope.action_groups 

76 

77 _LOGGER.debug("SUPERNOTIFY notify_mobile: %s -> %s", envelope.title, envelope.targets) 

78 

79 media = envelope.media or {} 

80 camera_entity_id = media.get(ATTR_MEDIA_CAMERA_ENTITY_ID) 

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

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

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

84 

85 match envelope.priority: 

86 case custom_components.supernotify.PRIORITY_CRITICAL: 

87 push_priority = "critical" 

88 case custom_components.supernotify.PRIORITY_HIGH: 

89 push_priority = "time-sensitive" 

90 case custom_components.supernotify.PRIORITY_MEDIUM: 

91 push_priority = "active" 

92 case custom_components.supernotify.PRIORITY_LOW: 

93 push_priority = "passive" 

94 case _: 

95 push_priority = "active" 

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

97 

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

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

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

101 if push_priority == "critical": 

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

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

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

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

106 else: 

107 # critical notifications can't be grouped on iOS 

108 category = category or camera_entity_id or "appd" 

109 data.setdefault("group", category) 

110 

111 if camera_entity_id: 

112 data["entity_id"] = camera_entity_id 

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

114 if clip_url: 

115 data["video"] = clip_url 

116 if snapshot_url: 

117 data["image"] = snapshot_url 

118 

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

120 for action in envelope.actions: 

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

122 if app_url: 

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

124 action[ATTR_ACTION_URL_TITLE] = app_url_title 

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

126 if camera_entity_id: 

127 data["actions"].append({ 

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

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

130 "behavior": "textInput", 

131 "textInputButtonTitle": "Minutes to snooze", 

132 "textInputPlaceholder": "60", 

133 }) 

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

135 if action_groups is None or group in action_groups: 

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

137 if not data["actions"]: 

138 del data["actions"] 

139 action_data = envelope.core_action_data() 

140 action_data[ATTR_DATA] = data 

141 hits = 0 

142 for mobile_target in envelope.targets: 

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

144 if await self.call_action(envelope, qualified_action=full_target, action_data=action_data): 

145 hits += 1 

146 else: 

147 simple_target = ( 

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

149 ) 

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

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

152 for recipient in self.context.people.values(): 

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

154 if md.get(CONF_NOTIFY_ACTION) in (simple_target, mobile_target): 

155 self.context.snoozer.register_snooze( 

156 CommandType.SNOOZE, 

157 target_type=QualifiedTargetType.ACTION, 

158 target=simple_target, 

159 recipient_type=RecipientType.USER, 

160 recipient=recipient[CONF_PERSON], 

161 snooze_for=24 * 60 * 60, 

162 reason="Action Failure", 

163 ) 

164 return hits > 0