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