Coverage for custom_components/supernotify/transports/mobile_push.py: 100%
123 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-04-01 15:06 +0000
1import logging
2import time
3from datetime import timedelta
4from typing import TYPE_CHECKING, Any
6from aiohttp import ClientResponse, ClientSession, ClientTimeout
7from bs4 import BeautifulSoup
8from homeassistant.components.notify.const import ATTR_DATA
10import custom_components.supernotify.const as const
11from custom_components.supernotify.const import (
12 ATTR_ACTION_CATEGORY,
13 ATTR_ACTION_URL,
14 ATTR_ACTION_URL_TITLE,
15 ATTR_DEFAULT,
16 ATTR_MEDIA_CAMERA_ENTITY_ID,
17 ATTR_MEDIA_CLIP_URL,
18 ATTR_MEDIA_SNAPSHOT_URL,
19 ATTR_MOBILE_APP_ID,
20 OPTION_DEVICE_DISCOVERY,
21 OPTION_DEVICE_DOMAIN,
22 OPTION_MESSAGE_USAGE,
23 OPTION_SIMPLIFY_TEXT,
24 OPTION_STRIP_URLS,
25 OPTION_TARGET_CATEGORIES,
26 TRANSPORT_MOBILE_PUSH,
27)
28from custom_components.supernotify.model import (
29 CommandType,
30 DebugTrace,
31 DeliveryConfig,
32 MessageOnlyPolicy,
33 QualifiedTargetType,
34 RecipientType,
35 Target,
36 TargetRequired,
37 TransportConfig,
38 TransportFeature,
39)
40from custom_components.supernotify.transport import (
41 Transport,
42)
44if TYPE_CHECKING:
45 from custom_components.supernotify.envelope import Envelope
46 from custom_components.supernotify.hass_api import HomeAssistantAPI
48_LOGGER = logging.getLogger(__name__)
51class MobilePushTransport(Transport):
52 name = TRANSPORT_MOBILE_PUSH
54 def __init__(self, *args: Any, **kwargs: Any) -> None:
55 super().__init__(*args, **kwargs)
56 self.action_titles: dict[str, str] = {}
57 self.action_title_failures: dict[str, float] = {}
59 @property
60 def supported_features(self) -> TransportFeature:
61 return (
62 TransportFeature.MESSAGE
63 | TransportFeature.TITLE
64 | TransportFeature.ACTIONS
65 | TransportFeature.IMAGES
66 | TransportFeature.VIDEO
67 )
69 def extra_attributes(self) -> dict[str, Any]:
70 return {"action_titles": self.action_titles, "action_title_failures": self.action_title_failures}
72 @property
73 def default_config(self) -> TransportConfig:
74 config = TransportConfig()
75 config.delivery_defaults.target_required = TargetRequired.ALWAYS
76 config.delivery_defaults.options = {
77 OPTION_SIMPLIFY_TEXT: False,
78 OPTION_STRIP_URLS: False,
79 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD,
80 OPTION_TARGET_CATEGORIES: [ATTR_MOBILE_APP_ID],
81 OPTION_DEVICE_DISCOVERY: False,
82 OPTION_DEVICE_DOMAIN: ["mobile_app"],
83 }
84 return config
86 def auto_configure(self, hass_api: HomeAssistantAPI) -> DeliveryConfig | None: # noqa: ARG002
87 return self.delivery_defaults
89 def validate_action(self, action: str | None) -> bool:
90 return action is None
92 async def action_title(self, url: str, retry_timeout: int = 900) -> str | None:
93 """Attempt to create a title for mobile action from the TITLE of the web page at the URL"""
94 if url in self.action_titles:
95 return self.action_titles[url]
96 if url in self.action_title_failures:
97 # don't retry too often
98 if time.time() - self.action_title_failures[url] < retry_timeout:
99 _LOGGER.debug("SUPERNOTIFY skipping retry after previous failure to retrieve url title for ", url)
100 return None
101 try:
102 websession: ClientSession = self.context.hass_api.http_session()
103 resp: ClientResponse = await websession.get(url, timeout=ClientTimeout(total=5.0))
104 body = await resp.content.read()
105 # wrap heavy bs4 parsing in a job to avoid blocking the event loop
106 html = await self.context.hass_api.create_job(BeautifulSoup, body, "html.parser")
107 if html.title and html.title.string:
108 self.action_titles[url] = html.title.string
109 return html.title.string
110 except Exception as e:
111 _LOGGER.warning("SUPERNOTIFY failed to retrieve url title at %s: %s", url, e)
112 self.action_title_failures[url] = time.time()
113 return None
115 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: # noqa: ARG002
116 if not envelope.target.mobile_app_ids:
117 _LOGGER.warning("SUPERNOTIFY No targets provided for mobile_push")
118 return False
119 data: dict[str, Any] = envelope.data or {}
120 category = data.get(ATTR_ACTION_CATEGORY, "general")
121 action_groups = envelope.action_groups
123 _LOGGER.debug("SUPERNOTIFY notify_mobile: %s -> %s", envelope.title, envelope.target.mobile_app_ids)
125 media = envelope.media or {}
126 camera_entity_id = media.get(ATTR_MEDIA_CAMERA_ENTITY_ID)
127 clip_url: str | None = self.hass_api.abs_url(media.get(ATTR_MEDIA_CLIP_URL))
128 snapshot_url: str | None = self.hass_api.abs_url(media.get(ATTR_MEDIA_SNAPSHOT_URL))
129 # options = data.get(CONF_OPTIONS, {})
131 match envelope.priority:
132 case const.PRIORITY_CRITICAL:
133 push_priority = "critical"
134 case const.PRIORITY_HIGH:
135 push_priority = "time-sensitive"
136 case const.PRIORITY_MEDIUM:
137 push_priority = "active"
138 case const.PRIORITY_LOW:
139 push_priority = "passive"
140 case const.PRIORITY_MINIMUM:
141 push_priority = "passive"
142 case _:
143 push_priority = "active"
144 _LOGGER.warning("SUPERNOTIFY Unexpected priority %s", envelope.priority)
146 data.setdefault("actions", [])
147 data.setdefault("push", {})
148 data["push"]["interruption-level"] = push_priority
149 if push_priority == "critical":
150 data["push"].setdefault("sound", {})
151 data["push"]["sound"].setdefault("name", ATTR_DEFAULT)
152 data["push"]["sound"]["critical"] = 1
153 data["push"]["sound"].setdefault("volume", 1.0)
154 else:
155 # critical notifications can't be grouped on iOS
156 category = category or camera_entity_id or "appd"
157 data.setdefault("group", category)
159 if camera_entity_id:
160 data["entity_id"] = camera_entity_id
161 # data['actions'].append({'action':'URI','title':'View Live','uri':'/cameras/%s' % device}
162 if clip_url:
163 data["video"] = clip_url
164 if snapshot_url:
165 data["image"] = snapshot_url
167 data.setdefault("actions", [])
168 for action in envelope.actions:
169 app_url: str | None = self.hass_api.abs_url(action.get(ATTR_ACTION_URL))
170 if app_url:
171 app_url_title = action.get(ATTR_ACTION_URL_TITLE) or self.action_title(app_url) or "Click for Action"
172 action[ATTR_ACTION_URL_TITLE] = app_url_title
173 data["actions"].append(action)
174 if camera_entity_id:
175 data["actions"].append({
176 "action": f"SUPERNOTIFY_SNOOZE_EVERYONE_CAMERA_{camera_entity_id}",
177 "title": f"Snooze camera notifications for {camera_entity_id}",
178 "behavior": "textInput",
179 "textInputButtonTitle": "Minutes to snooze",
180 "textInputPlaceholder": "60",
181 })
182 for group, actions in self.context.mobile_actions.items():
183 if action_groups is None or group in action_groups:
184 data["actions"].extend(actions)
185 if not data["actions"]:
186 del data["actions"]
187 action_data = envelope.core_action_data()
188 action_data[ATTR_DATA] = data
189 hits = 0
190 for mobile_target in envelope.target.mobile_app_ids:
191 full_target = mobile_target if Target.is_notify_entity(mobile_target) else f"notify.{mobile_target}"
192 if await self.call_action(envelope, qualified_action=full_target, action_data=action_data, implied_target=True):
193 hits += 1
194 else:
195 simple_target = (
196 mobile_target if not Target.is_notify_entity(mobile_target) else mobile_target.replace("notify.", "")
197 )
198 _LOGGER.warning("SUPERNOTIFY Failed to send to %s, snoozing for a day", simple_target)
199 if self.people_registry:
200 # somewhat hacky way to tie the mobile device back to a recipient to please the snoozing api
201 for recipient in self.people_registry.enabled_recipients():
202 for md in recipient.mobile_devices:
203 if md in (simple_target, mobile_target):
204 self.context.snoozer.register_snooze(
205 CommandType.SNOOZE,
206 target_type=QualifiedTargetType.MOBILE,
207 target=simple_target,
208 recipient_type=RecipientType.USER,
209 recipient=recipient.entity_id,
210 snooze_for=timedelta(days=1),
211 reason="Action Failure",
212 )
213 return hits > 0