Coverage for open_webui/utils/notifications.py: 14%

1from __future__ import annotations

3import logging

4import re

5import time

6from typing import Any

7from urllib.parse import urlparse

9from open_webui.events import EVENT_DEFINITIONS_BY_NAME, NOTIFICATION_EVENTS

10from open_webui.models.config import Config

11from open_webui.models.users import Users

12from open_webui.retrieval.web.utils import validate_url

13from open_webui.utils.webhook import post_webhook

16VALID_EVENTS = set(NOTIFICATION_EVENTS)

17LEGACY_EVENTS = {'chat.finished', 'chat.failed'}

18VALID_DELIVERY = {'away', 'always'}

19CHAT_FINISHED_EVENT = 'chat.finished'

20CHAT_FAILED_EVENT = 'chat.failed'

21CHANNEL_MESSAGE_EVENT = 'channel.message'

22CALENDAR_ALERT_EVENT = 'calendar.alert'

24DEFAULT_TARGET_ID = 'webhook'

25DESCRIPTION_DEFAULT = object()

26log = logging.getLogger(__name__)

29def _normalize_target(target: dict[str, Any], existing: dict[str, Any] | None = None) -> dict[str, Any]:

30 existing = existing or {}

31 now = int(time.time())

33 target_type = str(target.get('type') or existing.get('type') or 'webhook').strip()

34 if target_type != 'webhook':

35 raise ValueError('Unsupported notification target type')

37 config = dict(existing.get('config') or {})

38 config.update(target.get('config') or {})

39 url = str(config.get('url') or '').strip()

40 if not url:

41 raise ValueError('Webhook URL is required')

42 if '...' in url:

43 url = str((existing.get('config') or {}).get('url') or '').strip()

44 validate_url(url)

45 config['url'] = url

47 target_id = str(target.get('id') or existing.get('id') or '').strip()

48 if not target_id:

49 target_id = urlparse(url).hostname or 'webhook'

50 target_id = re.sub(r'[^a-zA-Z0-9_-]+', '-', target_id).strip('-').lower() or 'target'

52 events = target['events'] if 'events' in target else existing.get('events', [])

53 if events is None:

54 events = []

55 if not isinstance(events, list):

56 raise ValueError('events must be a list')

57 cleaned_events = []

58 for event in events:

59 event = str(event)

60 if event not in VALID_EVENTS:

61 raise ValueError(f'unsupported notification event: {event}')

62 if event not in cleaned_events:

63 cleaned_events.append(event)

65 delivery = str(target.get('delivery') or existing.get('delivery') or 'away').strip()

66 if delivery not in VALID_DELIVERY:

67 raise ValueError('Invalid notification delivery mode')

69 return {

70 'id': target_id,

71 'type': target_type,

72 'enabled': bool(target.get('enabled', existing.get('enabled', True))),

73 'events': cleaned_events,

74 'delivery': delivery,

75 'config': config,

76 'created_at': int(existing.get('created_at') or now),

77 'updated_at': now,

78 }

81def _public_target(target: dict[str, Any], default_target_id: str | None = None) -> dict[str, Any]:

82 config = dict(target.get('config') or {})

83 url = str(config.pop('url', '') or '')

84 if url:

85 parsed = urlparse(url)

86 if parsed.hostname:

87 path = parsed.path or ''

88 suffix = path[-4:] if len(path) > 4 else path

89 config['url_masked'] = f'{parsed.scheme}://{parsed.hostname}/...{suffix}'

90 else:

91 config['url_masked'] = '****'

92 else:

93 config['url_masked'] = ''

94 return {**target, 'config': config, 'is_default': target.get('id') == default_target_id}

97async def _load_notifications(user_id: str) -> dict[str, Any]:

98 user = await Users.get_user_by_id(user_id)

99 if not user:

100 raise ValueError('User not found')

101

102 settings = getattr(user, 'settings', None)

103 settings = settings.model_dump(exclude_none=True) if hasattr(settings, 'model_dump') else dict(settings or {})

104 notifications = dict(settings.get('notifications') or {})

105 targets = notifications.get('targets')

106

107 legacy_url = str(

108 notifications.get('webhook_url') or settings.get('ui', {}).get('notifications', {}).get('webhook_url') or ''

109 ).strip()

110

111 if not isinstance(targets, list) or not targets:

112 if legacy_url:

113 target = _normalize_target(

114 {

115 'id': DEFAULT_TARGET_ID,

116 'type': 'webhook',

117 'enabled': True,

118 'events': sorted(VALID_EVENTS),

119 'delivery': 'away',

120 'config': {'url': legacy_url},

121 }

122 )

123 notifications = {

124 **notifications,

125 'targets': [target],

126 'default_target_id': DEFAULT_TARGET_ID,

127 'legacy_notification_events_migrated': True,

128 }

129 await Users.update_user_settings_by_id(user_id, {'notifications': notifications})

130 else:

131 notifications['targets'] = [target for target in targets if isinstance(target, dict)]

132 notifications.setdefault(

133 'default_target_id', notifications['targets'][0].get('id') if notifications['targets'] else None

134 )

135 if legacy_url and not notifications.get('legacy_notification_events_migrated'):

136 changed = False

137 for target in notifications['targets']:

138 if ( 138 ↛ anywhereline 138 didn't jump anywhere: it always raised an exception.

139 target.get('id') == DEFAULT_TARGET_ID

140 and str((target.get('config') or {}).get('url') or '').strip() == legacy_url

141 and set(target.get('events') or []) == LEGACY_EVENTS

142 ):

143 target['events'] = sorted(VALID_EVENTS)

144 changed = True

145 notifications['legacy_notification_events_migrated'] = True

146 if changed:

147 await Users.update_user_settings_by_id(user_id, {'notifications': notifications})

148

149 return notifications

150

151

152async def list_targets(user_id: str) -> dict[str, Any]:

153 notifications = await _load_notifications(user_id)

154 default_target_id = notifications.get('default_target_id')

155 return {

156 'targets': [_public_target(target, default_target_id) for target in notifications.get('targets') or []],

157 }

158

159

160async def create_target(user_id: str, payload: dict[str, Any]) -> dict[str, Any]:

161 notifications = await _load_notifications(user_id)

162 targets = notifications.get('targets') or []

163 has_explicit_id = bool(str(payload.get('id') or '').strip())

164 target = _normalize_target(payload)

165 if any(str(existing.get('id', '')).lower() == target['id'].lower() for existing in targets):

166 if has_explicit_id:

167 raise ValueError('notification target id already exists')

168 base = target['id']

169 suffix = 2

170 while any(str(existing.get('id', '')).lower() == target['id'].lower() for existing in targets):

171 target['id'] = f'{base}-{suffix}'

172 suffix += 1

173 targets.append(target)

174 notifications['targets'] = targets

175 notifications.setdefault('default_target_id', target['id'])

176 await Users.update_user_settings_by_id(user_id, {'notifications': notifications})

177 return _public_target(target, notifications.get('default_target_id'))

178

179

180async def update_target(user_id: str, target_id: str, payload: dict[str, Any]) -> dict[str, Any]:

181 notifications = await _load_notifications(user_id)

182 targets = notifications.get('targets') or []

183 for index, existing in enumerate(targets):

184 if str(existing.get('id', '')).lower() == target_id.lower():

185 updated = _normalize_target({'id': target_id, **payload}, existing=existing)

186 if any(

187 idx != index and str(target.get('id', '')).lower() == updated['id'].lower()

188 for idx, target in enumerate(targets)

189 ):

190 raise ValueError('notification target id already exists')

191 targets[index] = updated

192 notifications['targets'] = targets

193 if str(notifications.get('default_target_id') or '').lower() == target_id.lower():

194 notifications['default_target_id'] = updated['id']

195 await Users.update_user_settings_by_id(user_id, {'notifications': notifications})

196 return _public_target(updated, notifications.get('default_target_id'))

197 raise ValueError('Notification target not found')

198

199

200async def delete_target(user_id: str, target_id: str) -> bool:

201 notifications = await _load_notifications(user_id)

202 targets = notifications.get('targets') or []

203 next_targets = [target for target in targets if str(target.get('id', '')).lower() != target_id.lower()]

204 if len(next_targets) == len(targets):

205 return False

206 notifications['targets'] = next_targets

207 if str(notifications.get('default_target_id') or '').lower() == target_id.lower():

208 notifications['default_target_id'] = next_targets[0].get('id') if next_targets else None

209 await Users.update_user_settings_by_id(user_id, {'notifications': notifications})

210 return True

211

212

213async def set_default_target(user_id: str, target_id: str) -> dict[str, Any]:

214 notifications = await _load_notifications(user_id)

215 for target in notifications.get('targets') or []:

216 if str(target.get('id', '')).lower() == target_id.lower():

217 notifications['default_target_id'] = target['id']

218 await Users.update_user_settings_by_id(user_id, {'notifications': notifications})

219 return _public_target(target, target['id'])

220 raise ValueError('Notification target not found')

221

222

223def get_notification_event_catalog() -> list[dict[str, str]]:

224 return [

225 {

226 'event': event_name,

227 'label': EVENT_DEFINITIONS_BY_NAME[event_name].message or event_name,

228 'description': EVENT_DEFINITIONS_BY_NAME[event_name].description or '',

229 }

230 for event_name in NOTIFICATION_EVENTS

231 ]

232

233

234def _find_target(notifications: dict[str, Any], target: str = '') -> dict[str, Any] | None:

235 targets = notifications.get('targets') or []

236 target = target.strip()

237 target_id = target or str(notifications.get('default_target_id') or '')

238 if not target_id:

239 return None

240 for item in targets:

241 if str(item.get('id', '')).lower() == target_id.lower():

242 return item

243 return None

244

245

246async def _send_webhook(

247 app_name: str,

248 target: dict[str, Any],

249 message: str,

250 data: dict[str, Any],

251 title: str = '',

252 description: str | None | object = DESCRIPTION_DEFAULT,

253):

254 url = str((target.get('config') or {}).get('url') or '').strip()

255 if not url:

256 raise ValueError('Webhook URL is required')

257 if description is DESCRIPTION_DEFAULT:

258 description = message if title else None

259 ok = await post_webhook(app_name, url, title or message, data, description=description)

260 if not ok:

261 raise ValueError('Webhook delivery failed')

262

263

264def _notification_webhook_content(event: Any) -> tuple[str, str, dict[str, Any], str | None]:

265 data = event.data or {}

266

267 if event.event == CHAT_FINISHED_EVENT:

268 title = str(data.get('title') or event.message or 'Chat finished')

269 content = str(data.get('message') or '')

270 url = str(data.get('url') or '')

271 chat_id = str(data.get('chat_id') or '')

272 if chat_id and url.endswith(f'/c/{chat_id}'):

273 url = f'{url[: -len(f"/c/{chat_id}")].rstrip("/")}/{chat_id}'

274 body = '\n'.join(part for part in (content, url) if part)

275 return (

276 f'**{title}**',

277 body,

278 {

279 'action': 'chat',

280 'message': content,

281 'title': title,

282 'url': url,

283 },

284 body,

285 )

286

287 if event.event == CHAT_FAILED_EVENT:

288 title = str(event.message or 'Chat failed')

289 content = str(data.get('message') or '')

290 url = str(data.get('url') or '')

291 chat_id = str(data.get('chat_id') or '')

292 if chat_id and url.endswith(f'/c/{chat_id}'):

293 url = f'{url[: -len(f"/c/{chat_id}")].rstrip("/")}/{chat_id}'

294 body = '\n'.join(part for part in (content, url) if part)

295 return (

296 f'**{title}**',

297 body,

298 {

299 'action': 'chat_failed',

300 'message': content,

301 'title': title,

302 'url': url,

303 },

304 body,

305 )

306

307 if event.event == CHANNEL_MESSAGE_EVENT:

308 channel_name = str(data.get('title') or event.message or 'Channel')

309 content = str(data.get('content') or data.get('message') or '')

310 url = str(data.get('url') or '')

311 body = '\n'.join(part for part in (content, url) if part)

312 return (

313 f'**#{channel_name}**',

314 body,

315 {

316 'action': 'channel',

317 'message': content,

318 'title': channel_name,

319 'url': url,

320 },

321 body,

322 )

323

324 if event.event == CALENDAR_ALERT_EVENT:

325 title = str(data.get('title') or event.message or 'Calendar alert')

326 starts_in = str(data.get('starts_in') or '')

327 message = f'**{title}**\nstarting {starts_in}'.strip()

328 return (

329 '',

330 message,

331 {

332 'action': 'calendar_alert',

333 'title': title,

334 'minutes_until': data.get('minutes_until'),

335 'event_id': data.get('event_id') or (event.subject or {}).get('id'),

336 },

337 None,

338 )

339

340 definition = EVENT_DEFINITIONS_BY_NAME.get(event.event)

341 title = event.message or (definition.message if definition else event.event)

342 message = str(data.get('message') or data.get('preview') or data.get('content_preview') or title)

343 return str(title), message, event.model_dump(), message if title else None

344

345

346# LICENSE covers this Open WebUI notification identifier.

347# Do not alter, remove, obscure, or replace it except as LICENSE permits:

348# https://docs.openwebui.com/license.

349async def test_target(user_id: str, target_id: str, app_name: str = 'Open WebUI') -> dict[str, Any]:

350 notifications = await _load_notifications(user_id)

351 target = _find_target(notifications, target_id)

352 if not target:

353 raise ValueError('Notification target not found')

354 await _send_webhook(

355 app_name,

356 target,

357 # LICENSE covers this Open WebUI notification copy.

358 # Do not alter, remove, obscure, or replace it except as LICENSE permits:

359 # https://docs.openwebui.com/license.

360 'This is a test notification from Open WebUI.',

361 {'action': 'test', 'user_id': user_id},

362 'Test notification',

363 )

364 return {'ok': True}

365

366

367# LICENSE covers this Open WebUI notification identifier.

368# Do not alter, remove, obscure, or replace it except as LICENSE permits:

369# https://docs.openwebui.com/license.

370async def notify_target(

371 user_id: str,

372 message: str,

373 target: str = '',

374 title: str = '',

375 app_name: str = 'Open WebUI',

376) -> dict[str, Any]:

377 notifications = await _load_notifications(user_id)

378 item = _find_target(notifications, target)

379 if not item:

380 raise ValueError('Notification target not found')

381 if not item.get('enabled', True):

382 raise ValueError('Notification target is disabled')

383 await _send_webhook(

384 app_name,

385 item,

386 message,

387 {'action': 'notify', 'user_id': user_id, 'message': message, 'title': title},

388 title or 'Notification',

389 )

390 return {'ok': True, 'target_id': item.get('id')}

391

392

393async def dispatch_notification_event(app: Any, event: Any) -> None:

394 if event.event not in VALID_EVENTS or not await Config.get('ui.enable_user_webhooks'): 394 ↛ 397line 394 didn't jump to line 397 because the condition on line 394 was always true

395 return

396

397 from open_webui.events import event_user_ids

398

399 # LICENSE covers this Open WebUI notification identifier.

400 # Do not alter, remove, obscure, or replace it except as LICENSE permits:

401 # https://docs.openwebui.com/license.

402 app_name = getattr(getattr(app, 'state', None), 'WEBUI_NAME', 'Open WebUI')

403 for user_id in event_user_ids(event):

404 try:

405 notifications = await _load_notifications(user_id)

406 is_active = False if event.event == CHANNEL_MESSAGE_EVENT else await Users.is_user_active(user_id)

407

408 for target in notifications.get('targets') or []:

409 if not target.get('enabled', True):

410 continue

411 if event.event not in target.get('events', []):

412 continue

413 if target.get('delivery', 'away') == 'away' and is_active:

414 continue

415

416 title, message, data, description = _notification_webhook_content(event)

417 await _send_webhook(app_name, target, message, data, title, description=description)

418 except Exception:

419 log.exception('Notification delivery failed for user %s and event %s', user_id, event.event)