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 ntfy.publish action with device_id target (topics configured in HA integration). 

6 

7Supported data: keys (all optional except ntfy_device_id): 

8 ntfy_device_id str device_id of the ntfy topic configured in HA (required) 

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

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

11 ntfy_click str URL opened on notification tap 

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

13 Used only when no snapshot_url is already in media. 

14 Requires media_web_path configured and image saved within it. 

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

16 ntfy_icon str JPEG/PNG icon URL 

17 ntfy_markdown bool enable Markdown rendering (default: false) 

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

19 ntfy_sequence_id str message ID for subsequent updates/cancellations 

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

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

22 

23ntfy_actions -- supported types: 

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

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

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

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

28""" 

29 

30from __future__ import annotations 

31 

32import logging 

33import re 

34from typing import TYPE_CHECKING, Any, ClassVar 

35 

36from homeassistant.const import ATTR_DEVICE_ID 

37from homeassistant.helpers.typing import ConfigType 

38 

39from custom_components.supernotify.common import boolify 

40from custom_components.supernotify.const import ( 

41 ATTR_MEDIA_SNAPSHOT_URL, 

42 TRANSPORT_NTFY, 

43) 

44from custom_components.supernotify.model import DebugTrace, TargetRequired, TransportConfig, TransportFeature 

45from custom_components.supernotify.options import MEDIA_OPTIONS, DeliveryOption 

46from custom_components.supernotify.transport import Transport 

47 

48if TYPE_CHECKING: 

49 from custom_components.supernotify.envelope import Envelope 

50 from custom_components.supernotify.hass_api import HomeAssistantAPI 

51 

52_LOGGER = logging.getLogger(__name__) 

53 

54HA_NTFY_DOMAIN = "ntfy" 

55 

56_PRIORITY_MAP = { 

57 "critical": 5, # urgent/max 

58 "high": 4, # high 

59 "medium": 3, # default 

60 "low": 2, # low 

61 "minimum": 1, # min 

62} 

63 

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

65 

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

67 

68 

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

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

71 

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

73 Returns: "HH:MM" or "HH:MM:SS" 

74 """ 

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

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

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

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

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

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

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

82 if seconds: 

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

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

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

86 return delay 

87 

88 

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

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

91 valid = [] 

92 for i, a in enumerate(actions): 

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

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

95 _LOGGER.warning( 

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

97 i, 

98 missing, 

99 ) 

100 continue 

101 valid.append(a) 

102 return valid 

103 

104 

105class NtfyTransport(Transport): 

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

107 

108 name = TRANSPORT_NTFY 

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

110 

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

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

113 

114 @property 

115 def supported_features(self) -> TransportFeature: 

116 return ( 

117 TransportFeature.MESSAGE 

118 | TransportFeature.TITLE 

119 | TransportFeature.IMAGES 

120 | TransportFeature.ACTIONS 

121 | TransportFeature.SNAPSHOT_IMAGE 

122 ) 

123 

124 @property 

125 def default_config(self) -> TransportConfig: 

126 config = TransportConfig() 

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

128 config.delivery_defaults.inclusion = self.inclusion_mode 

129 config.delivery_defaults.target_required = TargetRequired.NEVER 

130 return config 

131 

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

133 return hass_api.find_config_entry_data(HA_NTFY_DOMAIN) is not None 

134 

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

136 return {self.name: {}} 

137 

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

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

140 

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

142 

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

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

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

146 click_url = raw_data.pop("ntfy_click", None) 

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

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

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

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

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

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

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

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

155 

156 if not device_id: 

157 _LOGGER.warning("SUPERNOTIFY ntfy: ntfy_device_id not configured in delivery data") 

158 return False 

159 

160 # Validate ntfy_priority range (1-5) 

161 if priority_ovr is not None: 

162 try: 

163 priority_ovr = int(priority_ovr) 

164 if not 1 <= priority_ovr <= 5: 

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

166 priority_ovr = None 

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

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

169 priority_ovr = None 

170 

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

172 

173 action_data = envelope.core_action_data() 

174 action_data["priority"] = ntfy_priority 

175 

176 if tags: 

177 action_data["tags"] = tags 

178 if click_url: 

179 action_data["click"] = click_url 

180 if icon: 

181 action_data["icon"] = icon 

182 if markdown: 

183 action_data["markdown"] = True 

184 if delay: 

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

186 if sequence_id: 

187 action_data["sequence_id"] = sequence_id 

188 if email: 

189 action_data["email"] = email 

190 if actions: 

191 if not isinstance(actions, list): 

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

193 else: 

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

195 

196 # Image attachment: snapshot_url passthrough > grab_image for camera 

197 if envelope.media: 

198 snapshot_url = envelope.media.get(ATTR_MEDIA_SNAPSHOT_URL) 

199 if snapshot_url: 

200 action_data["attach"] = self.hass_api.abs_url(snapshot_url) 

201 action_data["filename"] = filename 

202 elif attach_image: 

203 image_path = await envelope.grab_image() 

204 if image_path: 

205 image_url = await self.context.media_storage.object_url(image_path) 

206 if image_url: 

207 action_data["attach"] = image_url 

208 action_data["filename"] = filename 

209 

210 # Residual generic keys (non ntfy_*) passed to payload 

211 action_data.update(raw_data) 

212 

213 target_data = {ATTR_DEVICE_ID: device_id} 

214 

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