Coverage for custom_components/supernotify/transports/ntfy.py: 92%

1"""ntfy transport for SuperNotify. 

2 

3Sends push notifications via ntfy (https://ntfy.sh) or self-hosted instance. 

4ntfy is an official HA integration since 2025.5. 

5Uses the ntfy.publish action, targeting the `notify` entity the integration creates for each 

6topic, or the topic's device with `ntfy_device_id`. 

7 

8Only the fields ntfy.publish accepts are passed on: its schema rejects any other key, so the 

9rest of the notification's data, meant for other deliveries, is dropped. 

10 

11Supported data: keys (all optional): 

12 ntfy_device_id str device_id of an ntfy topic, used when no ntfy `notify` 

13 entity is targeted 

14 ntfy_priority int 5=urgent, 4=high, 3=default, 2=low, 1=min 

15 ntfy_tags list[str] tag/emoji shortcodes (e.g. ["warning", "house"]) 

16 ntfy_click str URL opened on notification tap 

17 ntfy_attach_image bool grab image via shared pipeline and attach to ntfy. 

18 Used only when no snapshot_url is already in media. 

19 Requires media_web_path configured and image saved within it. 

20 ntfy_filename str attachment filename (default: snapshot.jpg) 

21 ntfy_icon str JPEG/PNG icon URL 

22 ntfy_markdown bool enable Markdown rendering (default: false) 

23 ntfy_delay str delivery delay: "10m", "1h", "2h30m", or "HH:MM" 

24 ntfy_sequence_id str message ID for subsequent updates/cancellations 

25 ntfy_email str email forwarding (e.g. "user@example.com") 

26 ntfy_actions list[dict] action buttons, max 3 (see examples below) 

27 

28ntfy_actions -- supported types: 

29 view: {action: view, label: "Open", url: "https://...", clear: false} 

30 http: {action: http, label: "POST", url: "https://...", method: post, headers: {}, body: ""} 

31 broadcast: {action: broadcast, label: "Intent", intent: "io.heckel.ntfy.USER_ACTION", extras: {}} 

32 copy: {action: copy, label: "Copy", value: "text to copy"} 

33""" 

34 

35from __future__ import annotations 

36 

37import logging 

38import re 

39from typing import TYPE_CHECKING, Any, ClassVar 

40 

41from homeassistant.const import ATTR_DEVICE_ID, ATTR_ENTITY_ID 

42from homeassistant.helpers.typing import ConfigType 

43 

44from custom_components.supernotify.common import boolify 

45from custom_components.supernotify.const import ( 

46 TRANSPORT_NTFY, 

47) 

48from custom_components.supernotify.model import DebugTrace, SuppressionReason, TargetRequired, TransportConfig, TransportFeature 

49from custom_components.supernotify.options import MEDIA_OPTIONS, DeliveryOption 

50from custom_components.supernotify.target import TargetEntityCategory 

51from custom_components.supernotify.transport import Transport 

52 

53if TYPE_CHECKING: 

54 from custom_components.supernotify.envelope import Envelope 

55 from custom_components.supernotify.hass_api import HomeAssistantAPI 

56 

57_LOGGER = logging.getLogger(__name__) 

58 

59HA_NTFY_DOMAIN = "ntfy" 

60 

61_PRIORITY_MAP = { 

62 "critical": 5, # urgent/max 

63 "high": 4, # high 

64 "medium": 3, # default 

65 "low": 2, # low 

66 "minimum": 1, # min 

67} 

68 

69_DELAY_RE = re.compile(r"(?:(\d+)h)?(?:(\d+)m)?(?:(\d+)s)?") 

70 

71_REQUIRED_ACTION_KEYS = {"action", "label"} 

72 

73 

74def _parse_delay(delay: str) -> str: 

75 """Convert user-friendly delay to HA offset format HH:MM or HH:MM:SS. 

76 

77 Accepts: "10m", "1h", "1h30m", "00:10", "01:30:00" 

78 Returns: "HH:MM" or "HH:MM:SS" 

79 """ 

80 if re.match(r"^\d{1,2}:\d{2}(:\d{2})?$", delay): 

81 return delay # already in HH:MM or HH:MM:SS format 

82 m = _DELAY_RE.fullmatch(delay.strip()) 

83 if m and (m.group(1) or m.group(2) or m.group(3)): 

84 hours = int(m.group(1) or 0) 

85 minutes = int(m.group(2) or 0) 

86 seconds = int(m.group(3) or 0) 

87 if seconds: 

88 return f"{hours:02d}:{minutes:02d}:{seconds:02d}" 

89 return f"{hours:02d}:{minutes:02d}" 

90 _LOGGER.warning("SUPERNOTIFY ntfy: unrecognized delay format '%s', passing as-is", delay) 

91 return delay 

92 

93 

94def _validate_actions(actions: list) -> list: 

95 """Validate ntfy action buttons, dropping malformed entries with a warning.""" 

96 valid = [] 

97 for i, a in enumerate(actions): 

98 if not isinstance(a, dict) or not _REQUIRED_ACTION_KEYS.issubset(a): 

99 missing = _REQUIRED_ACTION_KEYS - set(a) if isinstance(a, dict) else _REQUIRED_ACTION_KEYS 

100 _LOGGER.warning( 

101 "SUPERNOTIFY ntfy: action[%d] missing required keys %s, skipped", 

102 i, 

103 missing, 

104 ) 

105 continue 

106 valid.append(a) 

107 return valid 

108 

109 

110class NtfyTransport(Transport): 

111 """Notify via ntfy push notification service.""" 

112 

113 name = TRANSPORT_NTFY 

114 integration_domain = HA_NTFY_DOMAIN 

115 declared_options: ClassVar[list[DeliveryOption]] = [*MEDIA_OPTIONS] 

116 

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

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

119 

120 @property 

121 def supported_features(self) -> TransportFeature: 

122 return ( 

123 TransportFeature.MESSAGE 

124 | TransportFeature.TITLE 

125 | TransportFeature.IMAGES 

126 | TransportFeature.ACTIONS 

127 | TransportFeature.SNAPSHOT_IMAGE 

128 ) 

129 

130 @property 

131 def default_config(self) -> TransportConfig: 

132 config = super().default_config 

133 config.delivery_defaults.action = "ntfy.publish" 

134 # a delivery may instead fix its topic with ntfy_device_id in data, with no target - like 

135 # mqtt's topic, Notification._target_required() stops the standard delivery firing unprompted 

136 config.delivery_defaults.target_required = TargetRequired.OPTIONAL 

137 return config 

138 

139 @property 

140 def unique_target_categories(self) -> list[str | TargetEntityCategory]: 

141 # the notify entity for each topic is the integration's own, claimed ahead of the 

142 # generic notify_entity transport, like telegram's 

143 return [TargetEntityCategory(domain="notify", platform=HA_NTFY_DOMAIN)] 

144 

145 def is_viable(self, hass_api: HomeAssistantAPI) -> bool: 

146 return hass_api.find_config_entry_data(HA_NTFY_DOMAIN) is not None 

147 

148 def build_standard_deliveries(self, hass_api: HomeAssistantAPI) -> dict[str, ConfigType]: 

149 return {self.name: {}} 

150 

151 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: 

152 _LOGGER.debug("SUPERNOTIFY ntfy %s", envelope.message) 

153 

154 raw_data: dict[str, Any] = dict(envelope.data) if envelope.data else {} 

155 

156 device_id = raw_data.pop("ntfy_device_id", None) 

157 priority_ovr = raw_data.pop("ntfy_priority", None) 

158 tags = raw_data.pop("ntfy_tags", []) 

159 click_url = raw_data.pop("ntfy_click", None) or envelope.notification_uri(absolute=True) 

160 attach_image = boolify(raw_data.pop("ntfy_attach_image", False), default=False) 

161 filename = raw_data.pop("ntfy_filename", "snapshot.jpg") 

162 icon = raw_data.pop("ntfy_icon", None) 

163 markdown = boolify(raw_data.pop("ntfy_markdown", False), default=False) 

164 delay = raw_data.pop("ntfy_delay", None) 

165 sequence_id = raw_data.pop("ntfy_sequence_id", None) 

166 email = raw_data.pop("ntfy_email", None) 

167 actions = raw_data.pop("ntfy_actions", []) 

168 

169 entity_ids: list[str] = envelope.target.domain_entity_ids("notify") if envelope.target else [] 

170 if not device_id and not entity_ids: 

171 _LOGGER.warning("SUPERNOTIFY ntfy: no ntfy notify entity or ntfy_device_id to send to") 

172 envelope.skipped = 1 

173 envelope.skip_reason = SuppressionReason.NO_TARGET 

174 return False 

175 

176 # Validate ntfy_priority range (1-5) 

177 if priority_ovr is not None: 

178 try: 

179 priority_ovr = int(priority_ovr) 

180 if not 1 <= priority_ovr <= 5: 

181 _LOGGER.warning("SUPERNOTIFY ntfy: ntfy_priority %s out of range 1-5, using mapping", priority_ovr) 

182 priority_ovr = None 

183 except (TypeError, ValueError) as e: # py3.13 compat 

184 _LOGGER.warning("SUPERNOTIFY ntfy: invalid ntfy_priority %r, using mapping: %s", priority_ovr, e) 

185 priority_ovr = None 

186 

187 ntfy_priority = priority_ovr or _PRIORITY_MAP.get(envelope.priority or "medium", 3) 

188 

189 action_data = envelope.core_action_data() 

190 action_data["priority"] = ntfy_priority 

191 

192 if tags: 

193 action_data["tags"] = tags 

194 if click_url: 

195 action_data["click"] = click_url 

196 if icon: 

197 action_data["icon"] = icon 

198 if markdown: 

199 action_data["markdown"] = True 

200 if delay: 

201 action_data["delay"] = _parse_delay(delay) 

202 if sequence_id: 

203 action_data["sequence_id"] = sequence_id 

204 if email: 

205 action_data["email"] = email 

206 if actions: 

207 if not isinstance(actions, list): 

208 _LOGGER.warning("SUPERNOTIFY ntfy: ntfy_actions must be a list, ignored") 

209 else: 

210 action_data["actions"] = _validate_actions(actions)[:3] 

211 

212 # Image attachment: snapshot_url passthrough > grab_image for camera 

213 image_url: str | None = await self.image_url(envelope, grab=attach_image) 

214 if image_url: 

215 action_data["attach"] = image_url 

216 action_data["filename"] = filename 

217 

218 if raw_data: 

219 _LOGGER.debug("SUPERNOTIFY ntfy: dropping data keys not accepted by ntfy.publish: %s", sorted(raw_data)) 

220 

221 # targeted entities win over a fixed device, like mqtt's topic 

222 target_data: dict[str, Any] = {ATTR_ENTITY_ID: entity_ids} if entity_ids else {ATTR_DEVICE_ID: device_id} 

223 

224 return await self.call_action(envelope, action_data=action_data, target_data=target_data)