Coverage for custom_components/supernotify/methods/mobile_push.py: 92%
107 statements
« prev ^ index » next coverage.py v7.6.8, created at 2024-12-28 14:21 +0000
1import logging
2import re
3from typing import Any
5import httpx
6from bs4 import BeautifulSoup
7from homeassistant.components.notify.const import ATTR_DATA
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 METHOD_MOBILE_PUSH,
21 CommandType,
22 QualifiedTargetType,
23 RecipientType,
24)
25from custom_components.supernotify.delivery_method import DeliveryMethod
26from custom_components.supernotify.envelope import Envelope
28RE_VALID_MOBILE_APP = r"mobile_app_[A-Za-z0-9_]+"
30_LOGGER = logging.getLogger(__name__)
33class MobilePushDeliveryMethod(DeliveryMethod):
34 method = METHOD_MOBILE_PUSH
36 def __init__(self, *args, **kwargs) -> None:
37 super().__init__(*args, **kwargs)
38 self.action_titles: dict[str, str] = {}
40 def select_target(self, target: str) -> bool:
41 return re.fullmatch(RE_VALID_MOBILE_APP, target) is not None
43 def validate_action(self, action: str | None) -> bool:
44 return action is None
46 def recipient_target(self, recipient: dict) -> list[str]:
47 if CONF_PERSON in recipient:
48 services: list[str] = [md.get(CONF_NOTIFY_ACTION) for md in recipient.get(CONF_MOBILE_DEVICES, [])]
49 return list(filter(None, services))
50 return []
52 async def action_title(self, url: str) -> str | None:
53 if url in self.action_titles:
54 return self.action_titles[url]
55 try:
56 async with httpx.AsyncClient(timeout=httpx.Timeout(timeout=5.0)) as client:
57 resp: httpx.Response = await client.get(url, follow_redirects=True, timeout=5)
58 html = BeautifulSoup(resp.text, features="html.parser")
59 if html.title and html.title.string:
60 self.action_titles[url] = html.title.string
61 return html.title.string
62 except Exception as e:
63 _LOGGER.debug("SUPERNOTIFY failed to retrieve url title at %s: %s", url, e)
64 return None
66 async def deliver(self, envelope: Envelope) -> bool:
67 if not envelope.targets:
68 _LOGGER.warning("SUPERNOTIFY No targets provided for mobile_push")
69 return False
70 data: dict[str, Any] = envelope.data or {}
71 # TODO: category not passed anywhere
72 category = data.get(ATTR_ACTION_CATEGORY, "general")
73 action_groups = envelope.action_groups
75 _LOGGER.info("SUPERNOTIFY notify_mobile: %s -> %s", envelope.title, envelope.targets)
77 media = envelope.media or {}
78 camera_entity_id = media.get(ATTR_MEDIA_CAMERA_ENTITY_ID)
79 clip_url: str | None = self.abs_url(media.get(ATTR_MEDIA_CLIP_URL))
80 snapshot_url: str | None = self.abs_url(media.get(ATTR_MEDIA_SNAPSHOT_URL))
81 # options = data.get(CONF_OPTIONS, {})
83 match envelope.priority:
84 case custom_components.supernotify.PRIORITY_CRITICAL:
85 push_priority = "critical"
86 case custom_components.supernotify.PRIORITY_HIGH:
87 push_priority = "time-sensitive"
88 case custom_components.supernotify.PRIORITY_MEDIUM:
89 push_priority = "active"
90 case custom_components.supernotify.PRIORITY_LOW:
91 push_priority = "passive"
92 case _:
93 push_priority = "active"
94 _LOGGER.warning("SUPERNOTIFY Unexpected priority %s", envelope.priority)
96 data.setdefault("actions", [])
97 data.setdefault("push", {})
98 data["push"]["interruption-level"] = push_priority
99 if push_priority == "critical":
100 data["push"].setdefault("sound", {})
101 data["push"]["sound"].setdefault("name", "default")
102 data["push"]["sound"]["critical"] = 1
103 data["push"]["sound"].setdefault("volume", 1.0)
104 else:
105 # critical notifications can't be grouped on iOS
106 category = category or camera_entity_id or "appd"
107 data.setdefault("group", category)
109 if camera_entity_id:
110 data["entity_id"] = camera_entity_id
111 # data['actions'].append({'action':'URI','title':'View Live','uri':'/cameras/%s' % device}
112 if clip_url:
113 data["video"] = clip_url
114 if snapshot_url:
115 data["image"] = snapshot_url
117 data.setdefault("actions", [])
118 for action in envelope.actions:
119 app_url: str | None = self.abs_url(action.get(ATTR_ACTION_URL))
120 if app_url:
121 app_url_title = action.get(ATTR_ACTION_URL_TITLE) or self.action_title(app_url) or "Click for Action"
122 action[ATTR_ACTION_URL_TITLE] = app_url_title
123 data["actions"].append(action)
124 if camera_entity_id:
125 data["actions"].append({
126 "action": f"SUPERNOTIFY_SNOOZE_EVERYONE_CAMERA_{camera_entity_id}",
127 "title": f"Snooze camera notifications for {camera_entity_id}",
128 "behavior": "textInput",
129 "textInputButtonTitle": "Minutes to snooze",
130 "textInputPlaceholder": "60",
131 })
132 for group, actions in self.context.mobile_actions.items():
133 if action_groups is None or group in action_groups:
134 data["actions"].extend(actions)
135 if not data["actions"]:
136 del data["actions"]
137 action_data = envelope.core_action_data()
138 action_data[ATTR_DATA] = data
139 hits = 0
140 for mobile_target in envelope.targets:
141 full_target = mobile_target if mobile_target.startswith("notify.") else f"notify.{mobile_target}"
142 if await self.call_action(envelope, full_target, action_data=action_data):
143 hits += 1
144 else:
145 simple_target = (
146 mobile_target if not mobile_target.startswith("notify.") else mobile_target.replace("notify.", "")
147 )
148 _LOGGER.warning("SUPERNOTIFY Failed to send to %s, snoozing for a day", simple_target)
149 # somewhat hacky way to tie the mobile device back to a recipient to please the snoozing api
150 for recipient in self.context.people.values():
151 for md in recipient.get(CONF_MOBILE_DEVICES, []):
152 if md.get(CONF_NOTIFY_ACTION) in (simple_target, mobile_target):
153 self.context.snoozer.register_snooze(
154 CommandType.SNOOZE,
155 target_type=QualifiedTargetType.ACTION,
156 target=simple_target,
157 recipient_type=RecipientType.USER,
158 recipient=recipient[CONF_PERSON],
159 snooze_for=24 * 60 * 60,
160 reason="Action Failure",
161 )
162 return hits > 0