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 

35 

36from homeassistant.const import ATTR_DEVICE_ID 

37 

38from custom_components.supernotify.common import boolify 

39from custom_components.supernotify.const import ( 

40 ATTR_MEDIA_SNAPSHOT_URL, 

41 TRANSPORT_NTFY, 

42) 

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

44from custom_components.supernotify.transport import Transport 

45 

46if TYPE_CHECKING: 

47 from custom_components.supernotify.envelope import Envelope 

48 

49_LOGGER = logging.getLogger(__name__) 

50 

51_PRIORITY_MAP = { 

52 "critical": 5, # urgent/max 

53 "high": 4, # high 

54 "medium": 3, # default 

55 "low": 2, # low 

56 "minimum": 1, # min 

57} 

58 

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

60 

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

62 

63 

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

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

66 

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

68 Returns: "HH:MM" or "HH:MM:SS" 

69 """ 

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

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

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

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

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

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

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

77 if seconds: 

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

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

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

81 return delay 

82 

83 

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

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

86 valid = [] 

87 for i, a in enumerate(actions): 

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

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

90 _LOGGER.warning( 

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

92 i, 

93 missing, 

94 ) 

95 continue 

96 valid.append(a) 

97 return valid 

98 

99 

100class NtfyTransport(Transport): 

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

102 

103 name = TRANSPORT_NTFY 

104 

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

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

107 

108 @property 

109 def supported_features(self) -> TransportFeature: 

110 return ( 

111 TransportFeature.MESSAGE 

112 | TransportFeature.TITLE 

113 | TransportFeature.IMAGES 

114 | TransportFeature.ACTIONS 

115 | TransportFeature.SNAPSHOT_IMAGE 

116 ) 

117 

118 @property 

119 def default_config(self) -> TransportConfig: 

120 config = TransportConfig() 

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

122 config.delivery_defaults.target_required = TargetRequired.NEVER 

123 return config 

124 

125 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: # noqa: ARG002 

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

127 

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

129 

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

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

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

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

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

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

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

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

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

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

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

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

142 

143 if not device_id: 

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

145 return False 

146 

147 # Validate ntfy_priority range (1-5) 

148 if priority_ovr is not None: 

149 try: 

150 priority_ovr = int(priority_ovr) 

151 if not 1 <= priority_ovr <= 5: 

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

153 priority_ovr = None 

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

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

156 priority_ovr = None 

157 

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

159 

160 action_data = envelope.core_action_data() 

161 action_data["priority"] = ntfy_priority 

162 

163 if tags: 

164 action_data["tags"] = tags 

165 if click_url: 

166 action_data["click"] = click_url 

167 if icon: 

168 action_data["icon"] = icon 

169 if markdown: 

170 action_data["markdown"] = True 

171 if delay: 

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

173 if sequence_id: 

174 action_data["sequence_id"] = sequence_id 

175 if email: 

176 action_data["email"] = email 

177 if actions: 

178 if not isinstance(actions, list): 

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

180 else: 

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

182 

183 # Image attachment: snapshot_url passthrough > grab_image for camera 

184 if envelope.media: 

185 snapshot_url = envelope.media.get(ATTR_MEDIA_SNAPSHOT_URL) 

186 if snapshot_url: 

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

188 action_data["filename"] = filename 

189 elif attach_image: 

190 image_path = await envelope.grab_image() 

191 if image_path: 

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

193 if image_url: 

194 action_data["attach"] = image_url 

195 action_data["filename"] = filename 

196 

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

198 action_data.update(raw_data) 

199 

200 target_data = {ATTR_DEVICE_ID: device_id} 

201 

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