gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement notification system with Apprise integration and credential checks

Browse Source
This commit is contained in:
Christian Krakau-Louis
2025-04-11 01:55:10 +02:00
parent 0dad3cbb95
commit ce91dc8c79
16 changed files with 820 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/utils/config_validator.py
+12 -2
View File
@@ -5,14 +5,23 @@ This file serves as a backward-compatible interface to the config_validator pack
"""
# Import and re-export all functions from the new package
from app.utils.config_validator.validators import validate_email_config, validate_storage_configs, check_all_configs
from app.utils.config_validator.validators import (
validate_email_config,
validate_storage_configs,
validate_notification_config,
check_all_configs
)
from app.utils.config_validator.masking import mask_sensitive_value
from app.utils.config_validator.providers import get_provider_status
from app.utils.config_validator.settings_display import get_settings_for_display, dump_all_settings
from app.utils.config_validator.settings_display import (
get_settings_for_display,
dump_all_settings
)
__all__ = [
'validate_email_config',
'validate_storage_configs',
'validate_notification_config',
'mask_sensitive_value',
'get_provider_status',
'get_settings_for_display',
@@ -20,3 +29,4 @@ __all__ = [
'check_all_configs'
]
app/utils/config_validator/__init__.py
+2
View File
@@ -5,6 +5,7 @@ Configuration validation package for the application.
from app.utils.config_validator.validators import (
validate_email_config,
validate_storage_configs,
validate_notification_config,
check_all_configs
)
from app.utils.config_validator.masking import mask_sensitive_value
@@ -17,6 +18,7 @@ from app.utils.config_validator.settings_display import (
__all__ = [
'validate_email_config',
'validate_storage_configs',
'validate_notification_config',
'mask_sensitive_value',
'get_provider_status',
'get_settings_for_display',
app/utils/config_validator/providers.py
+19
View File
@@ -11,6 +11,24 @@ def get_provider_status():
"""
providers = {}
# Add Notification configuration - Make sure this provider is near the top of the list
providers["Notifications"] = {
"name": "Notifications",
"icon": "fa-solid fa-bell",
"configured": bool(getattr(settings, 'notification_urls', None)),
"enabled": True,
"description": "Send system notifications via various services",
"details": {
"services": str(len(getattr(settings, 'notification_urls', []))) + " service(s) configured" if getattr(settings, 'notification_urls', None) else "Not configured",
"task_failure": getattr(settings, 'notify_on_task_failure', True),
"credential_failure": getattr(settings, 'notify_on_credential_failure', True),
"startup": getattr(settings, 'notify_on_startup', True),
"shutdown": getattr(settings, 'notify_on_shutdown', False)
},
"testable": True,
"test_endpoint": "/api/diagnostic/test-notification"
}
# Add AI services first
providers["OpenAI"] = {
"name": "OpenAI",
@@ -253,4 +271,5 @@ def get_provider_status():
}
}
return providers
app/utils/config_validator/settings_display.py
+21
View File
@@ -23,6 +23,20 @@ def dump_all_settings():
value = f"{value[:visible_start]}{'*' * (len(value) - visible_start - visible_end)}{value[-visible_end:]}"
else:
value = f"{value[:2]}{'*' * (len(value) - 4)}{value[-2:]}" if isinstance(value, str) and len(value) > 4 else "****"
# Special handling for notification URLs
if key == 'notification_urls' and value:
try:
from app.utils.notification import _mask_sensitive_url
if isinstance(value, list):
masked_urls = [_mask_sensitive_url(url) for url in value]
logger.info(f"{key}: {masked_urls}")
else:
logger.info(f"{key}: {_mask_sensitive_url(value)}")
continue # Skip the default logging
except (ImportError, AttributeError):
pass # Fall back to default logging if _mask_sensitive_url is not available
logger.info(f"{key}: {value}")
logger.info("--- END OF SETTINGS DUMP ---")
@@ -169,6 +183,13 @@ def get_settings_for_display(show_values=False):
"Monitoring": [
"uptime_kuma_url",
"uptime_kuma_ping_interval"
],
"Notifications": [
"notification_urls",
"notify_on_task_failure",
"notify_on_credential_failure",
"notify_on_startup",
"notify_on_shutdown"
]
}
app/utils/config_validator/validators.py
+39 -1
View File
@@ -136,6 +136,36 @@ def validate_storage_configs():
return issues
def validate_notification_config():
"""Check notification configuration"""
issues = []
# Check if any notification URLs are configured
if not getattr(settings, 'notification_urls', None):
issues.append("No notification URLs configured")
else:
try:
# Try initializing Apprise to validate URLs
import apprise
a = apprise.Apprise()
for url in settings.notification_urls:
try:
if not a.add(url):
issues.append(f"Invalid notification URL format: {url}")
except Exception as e:
issues.append(f"Error with notification URL: {str(e)}")
except ImportError:
issues.append("Apprise module not installed")
if not issues:
logger.info("Notification configuration valid")
else:
logger.warning(f"Notification configuration issues: {', '.join(issues)}")
return issues
def check_all_configs():
"""Run all configuration validations and log results"""
from app.utils.config_validator.settings_display import dump_all_settings
@@ -161,8 +191,16 @@ def check_all_configs():
else:
logger.info(f"{provider.capitalize()} configuration OK")
# Check notification configuration
notification_issues = validate_notification_config()
if notification_issues:
logger.warning(f"Notification configuration issues: {', '.join(notification_issues)}")
else:
logger.info("Notification configuration OK")
# Return all identified issues
return {
'email': email_issues,
'storage': storage_issues
'storage': storage_issues,
'notification': notification_issues
}
app/utils/notification.py
+185
View File
@@ -0,0 +1,185 @@
import apprise
import logging
from typing import List, Optional, Dict, Any, Union
from app.config import settings
logger = logging.getLogger(__name__)
# Global Apprise instance
_apprise = None
def init_apprise() -> apprise.Apprise:
"""Initialize the Apprise instance with configured notification services"""
global _apprise
if _apprise is None:
_apprise = apprise.Apprise()
# Add all configured notification services
if settings.notification_urls:
for url in settings.notification_urls:
try:
_apprise.add(url)
logger.info(f"Added notification service: {_mask_sensitive_url(url)}")
except Exception as e:
logger.error(f"Failed to add notification service: {str(e)}")
else:
logger.warning("No notification services configured")
return _apprise
def _mask_sensitive_url(url: str) -> str:
"""Mask sensitive parts of notification URLs for logging"""
# Simple masking for common URL formats with credentials
import re
# Match patterns like user:pass@host or token in URL parameters
masked = re.sub(r'://([^:]+):([^@]+)@', r'://\1:****@', url)
masked = re.sub(r'(discord://)[^/]+/[^/]+', r'\1webhook_id/****', masked)
masked = re.sub(r'(tgram://)[^/]+/[^/]+', r'\1bot_token/****', masked)
masked = re.sub(r'([?&](token|key|api_key|password|secret)=)([^&]+)', r'\1****', masked)
return masked
def send_notification(
title: str,
message: str,
notification_type: str = "info",
tags: Optional[List[str]] = None,
attachments: Optional[List[str]] = None,
data: Optional[Dict[str, Any]] = None
) -> bool:
"""
Send a notification through all configured channels
Args:
title: The notification title
message: The notification body message
notification_type: Type of notification (info, success, warning, failure)
tags: Optional list of tags for filtering notifications
attachments: Optional list of file paths to attach
data: Optional additional data for the notification
Returns:
bool: True if notification was sent successfully to at least one service
"""
if not settings.notification_urls:
logger.debug(f"Notification not sent (no services configured): {title}")
return False
try:
apprise_obj = init_apprise()
# Set notification type
notify_type = apprise.NotifyType.INFO
if notification_type == "success":
notify_type = apprise.NotifyType.SUCCESS
elif notification_type in ("warning", "warn"):
notify_type = apprise.NotifyType.WARNING
elif notification_type in ("failure", "error", "failed"):
notify_type = apprise.NotifyType.FAILURE
# Send the notification to each service individually for better error reporting
if not apprise_obj.servers: # Access servers as an attribute, not a method
logger.warning("No notification servers available despite having URLs configured")
return False
total_services = len(apprise_obj.servers)
successful_services = 0
for server in apprise_obj.servers: # Iterate through the list directly
try:
service_name = str(server).split("://")[0] if "://" in str(server) else str(server)
service_result = server.notify(
title=title,
body=message,
notify_type=notify_type,
attach=attachments
)
if service_result:
successful_services += 1
logger.debug(f"Notification sent via {service_name}")
else:
logger.warning(f"Failed to send notification via {service_name}")
except Exception as e:
logger.error(f"Error sending notification via {str(server)}: {str(e)}")
overall_result = successful_services > 0
if overall_result:
logger.debug(f"Notification sent: '{title}' (successful: {successful_services}/{total_services})")
else:
logger.warning(f"Failed to send notification to ALL services: '{title}' (0/{total_services})")
return overall_result
except Exception as e:
logger.exception(f"Error sending notification: {e}")
return False
def notify_celery_failure(task_name: str, task_id: str, exc: Exception, args: list, kwargs: dict) -> bool:
"""Send a notification about a failed Celery task"""
if not settings.notify_on_task_failure:
return False
title = f"Task Failed: {task_name}"
message = f"""
Task {task_name} ({task_id}) failed with error:
{type(exc).__name__}: {str(exc)}
Arguments: {args}
Keyword arguments: {kwargs}
"""
return send_notification(
title=title,
message=message,
notification_type="failure",
tags=["celery", "failure", task_name]
)
def notify_credential_failure(service_name: str, error: str) -> bool:
"""Send a notification about a credential failure"""
if not settings.notify_on_credential_failure:
return False
title = f"Credential Failure: {service_name}"
message = f"""
The credentials for {service_name} have failed:
{error}
Please check and update the credentials in the system settings.
"""
return send_notification(
title=title,
message=message,
notification_type="warning",
tags=["credentials", "warning", service_name]
)
def notify_startup() -> bool:
"""Send a notification that the application has started"""
if not settings.notify_on_startup:
return False
title = f"DocuElevate Started"
message = f"DocuElevate has been started successfully on {settings.external_hostname}"
return send_notification(
title=title,
message=message,
notification_type="success",
tags=["system", "startup"]
)
def notify_shutdown() -> bool:
"""Send a notification that the application is shutting down"""
if not settings.notify_on_shutdown:
return False
title = f"DocuElevate Shutting Down"
message = f"DocuElevate on {settings.external_hostname} is shutting down"
return send_notification(
title=title,
message=message,
notification_type="info",
tags=["system", "shutdown"]
)