refac
This commit is contained in:
@@ -647,40 +647,24 @@ async def _check_calendar_alerts(app) -> None:
|
||||
except Exception:
|
||||
log.debug(f'Failed to mark event {event.id} as alerted', exc_info=True)
|
||||
|
||||
# Send webhook notification if user has one configured
|
||||
# Send target notification if user has one configured
|
||||
try:
|
||||
webui_name = getattr(app.state, 'WEBUI_NAME', 'Open WebUI')
|
||||
enable_user_webhooks = await Config.get('ui.enable_user_webhooks')
|
||||
|
||||
if enable_user_webhooks:
|
||||
user = await Users.get_user_by_id(event.user_id)
|
||||
if user and user.settings:
|
||||
webhook_url = (
|
||||
user.settings.get('ui', {}).get('notifications', {}).get('webhook_url', None)
|
||||
if isinstance(user.settings, dict)
|
||||
else getattr(getattr(user.settings, 'ui', None), 'get', lambda *a: None)(
|
||||
'notifications', {}
|
||||
).get('webhook_url', None)
|
||||
if hasattr(user.settings, 'ui')
|
||||
else None
|
||||
)
|
||||
if webhook_url:
|
||||
from open_webui.utils.webhook import post_webhook
|
||||
|
||||
time_str = f'in {minutes_until} min' if minutes_until > 0 else 'now'
|
||||
await post_webhook(
|
||||
webui_name,
|
||||
webhook_url,
|
||||
f'{event.title} — starting {time_str}',
|
||||
{
|
||||
'action': 'calendar_alert',
|
||||
'title': event.title,
|
||||
'minutes_until': minutes_until,
|
||||
'event_id': event.id,
|
||||
},
|
||||
)
|
||||
time_str = f'in {minutes_until} min' if minutes_until > 0 else 'now'
|
||||
await publish_event(
|
||||
app,
|
||||
EVENTS.CALENDAR_ALERT,
|
||||
subject_id=event.id,
|
||||
subject_type='calendar.event',
|
||||
source='scheduler',
|
||||
data={
|
||||
**alert_data,
|
||||
'user_id': event.user_id,
|
||||
'message': f'{event.title} — starting {time_str}',
|
||||
},
|
||||
message=event.title,
|
||||
)
|
||||
except Exception:
|
||||
log.debug(f'Failed to send webhook for calendar alert {event.id}', exc_info=True)
|
||||
log.debug(f'Failed to send notification for calendar alert {event.id}', exc_info=True)
|
||||
|
||||
|
||||
async def _record_run(
|
||||
|
||||
@@ -1,18 +1,22 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
import uuid
|
||||
from typing import Any
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from open_webui.events import EVENT_DEFINITIONS_BY_NAME, NOTIFICATION_EVENTS
|
||||
from open_webui.models.config import Config
|
||||
from open_webui.models.users import Users
|
||||
from open_webui.retrieval.web.utils import validate_url
|
||||
from open_webui.utils.webhook import post_webhook
|
||||
|
||||
|
||||
VALID_EVENTS = {'chat.finished', 'chat.failed'}
|
||||
VALID_EVENTS = set(NOTIFICATION_EVENTS)
|
||||
LEGACY_EVENTS = {'chat.finished', 'chat.failed'}
|
||||
VALID_DELIVERY = {'away', 'always'}
|
||||
CHANNEL_MESSAGE_EVENT = 'channel.message'
|
||||
|
||||
DEFAULT_TARGET_ID = 'webhook'
|
||||
log = logging.getLogger(__name__)
|
||||
@@ -26,12 +30,6 @@ def _normalize_target(target: dict[str, Any], existing: dict[str, Any] | None =
|
||||
if target_type != 'webhook':
|
||||
raise ValueError('Unsupported notification target type')
|
||||
|
||||
name = str(target.get('name') or existing.get('name') or 'Webhook').strip() or 'Webhook'
|
||||
slug = ''.join(ch.lower() if ch.isalnum() else '-' for ch in name).strip('-')
|
||||
while '--' in slug:
|
||||
slug = slug.replace('--', '-')
|
||||
target_id = str(target.get('id') or existing.get('id') or slug[:48] or uuid.uuid4().hex[:8]).strip()
|
||||
|
||||
config = dict(existing.get('config') or {})
|
||||
config.update(target.get('config') or {})
|
||||
url = str(config.get('url') or '').strip()
|
||||
@@ -42,26 +40,33 @@ def _normalize_target(target: dict[str, Any], existing: dict[str, Any] | None =
|
||||
validate_url(url)
|
||||
config['url'] = url
|
||||
|
||||
target_id = str(target.get('id') or existing.get('id') or '').strip()
|
||||
if not target_id:
|
||||
hostname = urlparse(url).hostname or 'webhook'
|
||||
target_id = re.sub(r'[^a-zA-Z0-9_-]+', '-', hostname).strip('-').lower() or 'target'
|
||||
|
||||
events = target['events'] if 'events' in target else existing.get('events', [])
|
||||
if events is None:
|
||||
events = []
|
||||
if not isinstance(events, list):
|
||||
raise ValueError('events must be a list')
|
||||
events = [str(event) for event in events]
|
||||
unsupported = [event for event in events if event not in VALID_EVENTS]
|
||||
if unsupported:
|
||||
raise ValueError(f'unsupported notification event: {unsupported[0]}')
|
||||
cleaned_events = []
|
||||
for event in events:
|
||||
event = str(event)
|
||||
if event not in VALID_EVENTS:
|
||||
raise ValueError(f'unsupported notification event: {event}')
|
||||
if event not in cleaned_events:
|
||||
cleaned_events.append(event)
|
||||
|
||||
delivery = str(target.get('delivery') or existing.get('delivery') or 'away')
|
||||
delivery = str(target.get('delivery') or existing.get('delivery') or 'away').strip()
|
||||
if delivery not in VALID_DELIVERY:
|
||||
raise ValueError('Invalid notification delivery mode')
|
||||
|
||||
return {
|
||||
'id': target_id,
|
||||
'type': target_type,
|
||||
'name': name,
|
||||
'enabled': bool(target.get('enabled', existing.get('enabled', True))),
|
||||
'events': events,
|
||||
'events': cleaned_events,
|
||||
'delivery': delivery,
|
||||
'config': config,
|
||||
'created_at': int(existing.get('created_at') or now),
|
||||
@@ -69,12 +74,20 @@ def _normalize_target(target: dict[str, Any], existing: dict[str, Any] | None =
|
||||
}
|
||||
|
||||
|
||||
def _public_target(target: dict[str, Any]) -> dict[str, Any]:
|
||||
def _public_target(target: dict[str, Any], default_target_id: str | None = None) -> dict[str, Any]:
|
||||
config = dict(target.get('config') or {})
|
||||
if config.get('url'):
|
||||
url = str(config['url'])
|
||||
config['url'] = '********' if len(url) <= 12 else f'{url[:8]}...{url[-4:]}'
|
||||
return {**target, 'config': config}
|
||||
url = str(config.pop('url', '') or '')
|
||||
if url:
|
||||
parsed = urlparse(url)
|
||||
if parsed.hostname:
|
||||
path = parsed.path or ''
|
||||
suffix = path[-4:] if len(path) > 4 else path
|
||||
config['url_masked'] = f'{parsed.scheme}://{parsed.hostname}/...{suffix}'
|
||||
else:
|
||||
config['url_masked'] = '****'
|
||||
else:
|
||||
config['url_masked'] = ''
|
||||
return {**target, 'config': config, 'is_default': target.get('id') == default_target_id}
|
||||
|
||||
|
||||
async def _load_notifications(user_id: str) -> dict[str, Any]:
|
||||
@@ -87,17 +100,15 @@ async def _load_notifications(user_id: str) -> dict[str, Any]:
|
||||
notifications = dict(settings.get('notifications') or {})
|
||||
targets = notifications.get('targets')
|
||||
|
||||
legacy_url = str(
|
||||
notifications.get('webhook_url') or settings.get('ui', {}).get('notifications', {}).get('webhook_url') or ''
|
||||
).strip()
|
||||
|
||||
if not isinstance(targets, list) or not targets:
|
||||
legacy_url = str(
|
||||
notifications.get('webhook_url')
|
||||
or settings.get('ui', {}).get('notifications', {}).get('webhook_url')
|
||||
or ''
|
||||
).strip()
|
||||
if legacy_url:
|
||||
target = _normalize_target(
|
||||
{
|
||||
'id': DEFAULT_TARGET_ID,
|
||||
'name': 'Webhook',
|
||||
'type': 'webhook',
|
||||
'enabled': True,
|
||||
'events': sorted(VALID_EVENTS),
|
||||
@@ -105,59 +116,91 @@ async def _load_notifications(user_id: str) -> dict[str, Any]:
|
||||
'config': {'url': legacy_url},
|
||||
}
|
||||
)
|
||||
notifications = {**notifications, 'targets': [target], 'default_target_id': DEFAULT_TARGET_ID}
|
||||
notifications = {
|
||||
**notifications,
|
||||
'targets': [target],
|
||||
'default_target_id': DEFAULT_TARGET_ID,
|
||||
'legacy_notification_events_migrated': True,
|
||||
}
|
||||
await Users.update_user_settings_by_id(user_id, {'notifications': notifications})
|
||||
else:
|
||||
notifications['targets'] = [target for target in targets if isinstance(target, dict)]
|
||||
notifications.setdefault(
|
||||
'default_target_id', notifications['targets'][0].get('id') if notifications['targets'] else None
|
||||
)
|
||||
if legacy_url and not notifications.get('legacy_notification_events_migrated'):
|
||||
changed = False
|
||||
for target in notifications['targets']:
|
||||
if (
|
||||
target.get('id') == DEFAULT_TARGET_ID
|
||||
and str((target.get('config') or {}).get('url') or '').strip() == legacy_url
|
||||
and set(target.get('events') or []) == LEGACY_EVENTS
|
||||
):
|
||||
target['events'] = sorted(VALID_EVENTS)
|
||||
changed = True
|
||||
notifications['legacy_notification_events_migrated'] = True
|
||||
if changed:
|
||||
await Users.update_user_settings_by_id(user_id, {'notifications': notifications})
|
||||
|
||||
return notifications
|
||||
|
||||
|
||||
async def list_targets(user_id: str) -> dict[str, Any]:
|
||||
notifications = await _load_notifications(user_id)
|
||||
default_target_id = notifications.get('default_target_id')
|
||||
return {
|
||||
'targets': [_public_target(target) for target in notifications.get('targets') or []],
|
||||
'default_target_id': notifications.get('default_target_id'),
|
||||
'targets': [_public_target(target, default_target_id) for target in notifications.get('targets') or []],
|
||||
}
|
||||
|
||||
|
||||
async def create_target(user_id: str, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
notifications = await _load_notifications(user_id)
|
||||
targets = notifications.get('targets') or []
|
||||
has_explicit_id = bool(str(payload.get('id') or '').strip())
|
||||
target = _normalize_target(payload)
|
||||
if any(existing.get('id') == target['id'] for existing in targets):
|
||||
target['id'] = f'{target["id"]}-{uuid.uuid4().hex[:6]}'
|
||||
if any(str(existing.get('id', '')).lower() == target['id'].lower() for existing in targets):
|
||||
if has_explicit_id:
|
||||
raise ValueError('notification target id already exists')
|
||||
base = target['id']
|
||||
suffix = 2
|
||||
while any(str(existing.get('id', '')).lower() == target['id'].lower() for existing in targets):
|
||||
target['id'] = f'{base}-{suffix}'
|
||||
suffix += 1
|
||||
targets.append(target)
|
||||
notifications['targets'] = targets
|
||||
notifications.setdefault('default_target_id', target['id'])
|
||||
await Users.update_user_settings_by_id(user_id, {'notifications': notifications})
|
||||
return _public_target(target)
|
||||
return _public_target(target, notifications.get('default_target_id'))
|
||||
|
||||
|
||||
async def update_target(user_id: str, target_id: str, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
notifications = await _load_notifications(user_id)
|
||||
targets = notifications.get('targets') or []
|
||||
for index, existing in enumerate(targets):
|
||||
if existing.get('id') == target_id:
|
||||
if str(existing.get('id', '')).lower() == target_id.lower():
|
||||
updated = _normalize_target({'id': target_id, **payload}, existing=existing)
|
||||
if any(
|
||||
idx != index and str(target.get('id', '')).lower() == updated['id'].lower()
|
||||
for idx, target in enumerate(targets)
|
||||
):
|
||||
raise ValueError('notification target id already exists')
|
||||
targets[index] = updated
|
||||
notifications['targets'] = targets
|
||||
if str(notifications.get('default_target_id') or '').lower() == target_id.lower():
|
||||
notifications['default_target_id'] = updated['id']
|
||||
await Users.update_user_settings_by_id(user_id, {'notifications': notifications})
|
||||
return _public_target(updated)
|
||||
return _public_target(updated, notifications.get('default_target_id'))
|
||||
raise ValueError('Notification target not found')
|
||||
|
||||
|
||||
async def delete_target(user_id: str, target_id: str) -> bool:
|
||||
notifications = await _load_notifications(user_id)
|
||||
targets = notifications.get('targets') or []
|
||||
next_targets = [target for target in targets if target.get('id') != target_id]
|
||||
next_targets = [target for target in targets if str(target.get('id', '')).lower() != target_id.lower()]
|
||||
if len(next_targets) == len(targets):
|
||||
return False
|
||||
notifications['targets'] = next_targets
|
||||
if notifications.get('default_target_id') == target_id:
|
||||
if str(notifications.get('default_target_id') or '').lower() == target_id.lower():
|
||||
notifications['default_target_id'] = next_targets[0].get('id') if next_targets else None
|
||||
await Users.update_user_settings_by_id(user_id, {'notifications': notifications})
|
||||
return True
|
||||
@@ -165,19 +208,33 @@ async def delete_target(user_id: str, target_id: str) -> bool:
|
||||
|
||||
async def set_default_target(user_id: str, target_id: str) -> dict[str, Any]:
|
||||
notifications = await _load_notifications(user_id)
|
||||
if not any(target.get('id') == target_id for target in notifications.get('targets') or []):
|
||||
raise ValueError('Notification target not found')
|
||||
notifications['default_target_id'] = target_id
|
||||
await Users.update_user_settings_by_id(user_id, {'notifications': notifications})
|
||||
return await list_targets(user_id)
|
||||
for target in notifications.get('targets') or []:
|
||||
if str(target.get('id', '')).lower() == target_id.lower():
|
||||
notifications['default_target_id'] = target['id']
|
||||
await Users.update_user_settings_by_id(user_id, {'notifications': notifications})
|
||||
return _public_target(target, target['id'])
|
||||
raise ValueError('Notification target not found')
|
||||
|
||||
|
||||
def get_notification_event_catalog() -> list[dict[str, str]]:
|
||||
return [
|
||||
{
|
||||
'event': event_name,
|
||||
'label': EVENT_DEFINITIONS_BY_NAME[event_name].message or event_name,
|
||||
'description': EVENT_DEFINITIONS_BY_NAME[event_name].description or '',
|
||||
}
|
||||
for event_name in NOTIFICATION_EVENTS
|
||||
]
|
||||
|
||||
|
||||
def _find_target(notifications: dict[str, Any], target: str = '') -> dict[str, Any] | None:
|
||||
targets = notifications.get('targets') or []
|
||||
target = target.strip()
|
||||
target_id = target or str(notifications.get('default_target_id') or '')
|
||||
if not target_id:
|
||||
return None
|
||||
for item in targets:
|
||||
if item.get('id') == target_id or item.get('name') == target:
|
||||
if str(item.get('id', '')).lower() == target_id.lower():
|
||||
return item
|
||||
return None
|
||||
|
||||
@@ -235,7 +292,7 @@ async def dispatch_notification_event(app: Any, event: Any) -> None:
|
||||
for user_id in event_user_ids(event):
|
||||
try:
|
||||
notifications = await _load_notifications(user_id)
|
||||
is_active = await Users.is_user_active(user_id)
|
||||
is_active = False if event.event == CHANNEL_MESSAGE_EVENT else await Users.is_user_active(user_id)
|
||||
|
||||
for target in notifications.get('targets') or []:
|
||||
if not target.get('enabled', True):
|
||||
@@ -246,8 +303,14 @@ async def dispatch_notification_event(app: Any, event: Any) -> None:
|
||||
continue
|
||||
|
||||
data = event.model_dump()
|
||||
title = event.message or data.get('message') or event.event
|
||||
message = str((event.data or {}).get('message') or title)
|
||||
definition = EVENT_DEFINITIONS_BY_NAME.get(event.event)
|
||||
title = event.message or (definition.message if definition else event.event)
|
||||
message = str(
|
||||
(event.data or {}).get('message')
|
||||
or (event.data or {}).get('preview')
|
||||
or (event.data or {}).get('content_preview')
|
||||
or title
|
||||
)
|
||||
await _send_webhook(app_name, target, message, data, title)
|
||||
except Exception:
|
||||
log.exception('Notification delivery failed for user %s and event %s', user_id, event.event)
|
||||
|
||||
Reference in New Issue
Block a user