Merge pull request #60 from christianlouis/58-add-notification-framework

feat: Implement notification system with Apprise integration and cred…
This commit is contained in:
Christian Krakau-Louis
2025-04-11 01:56:48 +02:00
committed by GitHub
16 changed files with 820 additions and 6 deletions
+20
View File
@@ -124,6 +124,26 @@ SFTP_PASSWORD=your_secure_sftp_password
# SFTP_PRIVATE_KEY_PASSPHRASE=optional_passphrase # SFTP_PRIVATE_KEY_PASSPHRASE=optional_passphrase
SFTP_FOLDER=/Documents/Uploads SFTP_FOLDER=/Documents/Uploads
# **Notification Settings**
# Configure notification services using Apprise URL format
# See https://github.com/caronc/apprise#supported-notifications
# Examples:
# - Discord: discord://webhook_id/webhook_token
# - Telegram: tgram://bot_token/chat_id
# - Email: mailto://user:pass@example.com
# - Pushover: pover://user_key/app_token
# - Slack: slack://tokenA/tokenB/tokenC
# - Matrix: matrix://username:password@domain/#room
# You can specify multiple notification URLs by separating them with commas
NOTIFICATION_URLS=discord://webhook_id/webhook_token,mailto://user:pass@gmail.com,tgram://bot_token/chat_id
# Control when notifications are sent
NOTIFY_ON_TASK_FAILURE=True
NOTIFY_ON_CREDENTIAL_FAILURE=True
NOTIFY_ON_STARTUP=True
NOTIFY_ON_SHUTDOWN=False
# Uptime Kuma # Uptime Kuma
UPTIME_KUMA_URL=https://status.example.com/api/push/abcdef123456?status=up UPTIME_KUMA_URL=https://status.example.com/api/push/abcdef123456?status=up
UPTIME_KUMA_PING_INTERVAL=5 UPTIME_KUMA_PING_INTERVAL=5
+49
View File
@@ -49,3 +49,52 @@ async def diagnostic_settings(request: Request, current_user: dict = Depends(get
"settings": safe_settings, "settings": safe_settings,
"message": "Full settings have been dumped to application logs" "message": "Full settings have been dumped to application logs"
} }
@router.post("/diagnostic/test-notification")
@require_login
async def test_notification(request: Request):
# Add request_time to request.state
import datetime
request.state.request_time = datetime.datetime.utcnow().isoformat()
"""
Send a test notification through all configured notification channels
"""
from app.utils.notification import send_notification
try:
notification_urls = getattr(settings, 'notification_urls', [])
if not notification_urls:
return {
"status": "warning",
"message": "No notification services configured. Add notification URLs to your configuration."
}
# Send a test notification
hostname = settings.external_hostname or "Document Processor"
result = send_notification(
title=f"Test Notification from {hostname}",
message=f"This is a test notification sent at {request.state.request_time}. If you're receiving this, notifications are working!",
notification_type="success",
tags=["test", "notification", "diagnostic"]
)
if result:
logger.info("Test notification sent successfully")
return {
"status": "success",
"message": f"Test notification sent successfully to {len(notification_urls)} service(s)",
"services_count": len(notification_urls)
}
else:
logger.warning("Test notification send attempt returned False")
return {
"status": "error",
"message": "Failed to send test notification. Check application logs for details."
}
except Exception as e:
logger.exception(f"Error sending test notification: {e}")
return {
"status": "error",
"message": f"Error sending notification: {str(e)}"
}
+22
View File
@@ -18,3 +18,25 @@ celery.conf.task_default_queue = 'document_processor'
celery.conf.task_routes = { celery.conf.task_routes = {
"app.tasks.*": {"queue": "document_processor"}, "app.tasks.*": {"queue": "document_processor"},
} }
# Task failure notification handler
from celery.signals import task_failure
@task_failure.connect
def task_failure_handler(sender=None, task_id=None, exception=None, args=None,
kwargs=None, traceback=None, einfo=None, **kw):
"""Handler for Celery task failures to send notifications"""
if getattr(settings, 'notify_on_task_failure', True):
try:
# Import here to avoid circular imports
from app.utils.notification import notify_celery_failure
notify_celery_failure(
task_name=sender.name if sender else "Unknown",
task_id=task_id or "N/A",
exc=exception,
args=args or [],
kwargs=kwargs or {}
)
except Exception as e:
import logging
logging.exception(f"Failed to send task failure notification: {e}")
+16
View File
@@ -32,6 +32,7 @@ from app.tasks.upload_to_email import upload_to_email
from app.tasks.imap_tasks import pull_all_inboxes from app.tasks.imap_tasks import pull_all_inboxes
from app.tasks.send_to_all import send_to_all_destinations from app.tasks.send_to_all import send_to_all_destinations
from app.tasks.uptime_kuma_tasks import ping_uptime_kuma from app.tasks.uptime_kuma_tasks import ping_uptime_kuma
from app.tasks.check_credentials import check_credentials
celery.conf.task_routes = { celery.conf.task_routes = {
"app.tasks.*": {"queue": "default"}, "app.tasks.*": {"queue": "default"},
@@ -44,6 +45,9 @@ def test_task():
# If you want Celery Beat to run the poll task every minute, add: # If you want Celery Beat to run the poll task every minute, add:
from celery.schedules import crontab from celery.schedules import crontab
# Run the check_credentials task at startup
check_credentials.apply_async(countdown=10) # Run 10 seconds after worker starts
celery.conf.beat_schedule = { celery.conf.beat_schedule = {
"poll-inboxes-every-minute": { "poll-inboxes-every-minute": {
"task": "app.tasks.imap_tasks.pull_all_inboxes", "task": "app.tasks.imap_tasks.pull_all_inboxes",
@@ -56,6 +60,18 @@ celery.conf.beat_schedule = {
"schedule": crontab(minute=f"*/{settings.uptime_kuma_ping_interval}"), "schedule": crontab(minute=f"*/{settings.uptime_kuma_ping_interval}"),
"options": {"expires": 55}, # Ensure tasks don't pile up "options": {"expires": 55}, # Ensure tasks don't pile up
} if settings.uptime_kuma_url else None, } if settings.uptime_kuma_url else None,
# Check credentials every 5 minutes
"check-credentials-regularly": {
"task": "app.tasks.check_credentials.check_credentials",
"schedule": crontab(minute="*/5"), # Every 5 minutes
"options": {"expires": 240}, # 4 minutes expiry
},
# Also keep daily check for logs and statistics purposes
"check-credentials-daily": {
"task": "app.tasks.check_credentials.check_credentials",
"schedule": crontab(hour="0", minute="0"), # Midnight
"options": {"expires": 3600}, # 1 hour expiry
}
} }
# Remove None entries from beat_schedule # Remove None entries from beat_schedule
+35 -1
View File
@@ -1,7 +1,8 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
from pydantic_settings import BaseSettings from pydantic_settings import BaseSettings
from typing import Optional, List, Dict, Any from typing import Optional, List, Dict, Any, Union
from pydantic import Field, validator
import os import os
from datetime import datetime from datetime import datetime
@@ -128,6 +129,39 @@ class Settings(BaseSettings):
# Feature flags # Feature flags
allow_file_delete: bool = True # Default to allowing file deletion from database allow_file_delete: bool = True # Default to allowing file deletion from database
# Notification settings
notification_urls: Union[List[str], str] = Field(
default_factory=list,
description="List of Apprise notification URLs (e.g., discord://, telegram://, etc.)"
)
notify_on_task_failure: bool = Field(
default=True,
description="Send notifications when Celery tasks fail"
)
notify_on_credential_failure: bool = Field(
default=True,
description="Send notifications when credential checks fail"
)
notify_on_startup: bool = Field(
default=True,
description="Send notifications when application starts"
)
notify_on_shutdown: bool = Field(
default=False,
description="Send notifications when application shuts down"
)
@validator('notification_urls', pre=True)
def parse_notification_urls(cls, v):
"""Parse notification URLs from string or list"""
if isinstance(v, str):
if ',' in v:
return [url.strip() for url in v.split(',') if url.strip()]
elif v.strip():
return [v.strip()]
return []
return v
# Get build date from environment or file # Get build date from environment or file
@property @property
def build_date(self) -> str: def build_date(self) -> str:
+15
View File
@@ -14,6 +14,7 @@ from pathlib import Path
from app.database import init_db from app.database import init_db
from app.config import settings from app.config import settings
from app.utils.config_validator import check_all_configs from app.utils.config_validator import check_all_configs
from app.utils.notification import init_apprise, send_notification, notify_startup, notify_shutdown
# Import the routers - now using views directly instead of frontend # Import the routers - now using views directly instead of frontend
from app.views import router as frontend_router from app.views import router as frontend_router
@@ -68,6 +69,20 @@ async def startup_event():
logging.info("Application started with valid configuration") logging.info("Application started with valid configuration")
logging.info("Router organization: Using refactored API routers from app/api/ directory") logging.info("Router organization: Using refactored API routers from app/api/ directory")
# Initialize notification system
init_apprise()
# Send startup notification
notify_startup()
@app.on_event("shutdown")
async def shutdown_event():
"""Run shutdown tasks for the application"""
logging.info("Application shutting down")
# Send shutdown notification
notify_shutdown()
# Custom 404 - we can still return the Jinja2 template, or the old static file: # Custom 404 - we can still return the Jinja2 template, or the old static file:
@app.exception_handler(404) @app.exception_handler(404)
+267
View File
@@ -0,0 +1,267 @@
import logging
from app.celery_app import celery
from app.config import settings
from app.utils.notification import notify_credential_failure
import time
import os
import json
import asyncio
import inspect
# Import the test functions from API routes
from app.api.openai import test_openai_connection
from app.api.azure import test_azure_connection
from app.api.dropbox import test_dropbox_token
from app.api.google_drive import test_google_drive_token
from app.api.onedrive import test_onedrive_token
# Import config validation utilities
from app.utils.config_validator import validate_storage_configs, get_provider_status
# Create an enhanced mock Request object for API functions that expect it
class MockRequest:
"""Mock request object with session and other attributes needed for API functions"""
def __init__(self):
self.session = {"user": {"id": "credential_checker", "name": "System Credential Checker"}}
self.app = None
self.headers = {}
self.query_params = {}
self.path_params = {}
async def json(self):
return {}
async def form(self):
return {}
logger = logging.getLogger(__name__)
# Path to store failure counts
FAILURE_STATE_FILE = os.path.join(settings.workdir, 'credential_failures.json')
def get_failure_state():
"""Read the failure state from file"""
try:
if os.path.exists(FAILURE_STATE_FILE):
with open(FAILURE_STATE_FILE, 'r') as f:
return json.load(f)
except Exception as e:
logger.error(f"Error reading failure state file: {e}")
# Default empty state
return {}
def save_failure_state(state):
"""Save failure state to file"""
try:
with open(FAILURE_STATE_FILE, 'w') as f:
json.dump(state, f)
except Exception as e:
logger.error(f"Error saving failure state file: {e}")
# Helper function to get the inner function without the decorator
def unwrap_decorated_function(func):
"""Get the original function from a decorated function"""
if hasattr(func, "__wrapped__"):
return unwrap_decorated_function(func.__wrapped__)
return func
# Create synchronous versions of the test functions that bypass authentication
def sync_test_openai_connection():
"""Synchronous wrapper for the OpenAI test function that bypasses auth"""
# Get the original function without the @require_login decorator
inner_func = unwrap_decorated_function(test_openai_connection)
request = MockRequest()
if inspect.iscoroutinefunction(inner_func):
return asyncio.run(inner_func(request))
return inner_func(request)
def sync_test_azure_connection():
"""Synchronous wrapper for the Azure test function that bypasses auth"""
inner_func = unwrap_decorated_function(test_azure_connection)
request = MockRequest()
if inspect.iscoroutinefunction(inner_func):
return asyncio.run(inner_func(request))
return inner_func(request)
def sync_test_dropbox_token():
"""Synchronous wrapper for the Dropbox test function that bypasses auth"""
inner_func = unwrap_decorated_function(test_dropbox_token)
request = MockRequest()
if inspect.iscoroutinefunction(inner_func):
return asyncio.run(inner_func(request))
return inner_func(request)
def sync_test_google_drive_token():
"""Synchronous wrapper for the Google Drive test function that bypasses auth"""
inner_func = unwrap_decorated_function(test_google_drive_token)
request = MockRequest()
if inspect.iscoroutinefunction(inner_func):
return asyncio.run(inner_func(request))
return inner_func(request)
def sync_test_onedrive_token():
"""Synchronous wrapper for the OneDrive test function that bypasses auth"""
inner_func = unwrap_decorated_function(test_onedrive_token)
request = MockRequest()
if inspect.iscoroutinefunction(inner_func):
return asyncio.run(inner_func(request))
return inner_func(request)
@celery.task
def check_credentials():
"""Check all configured credentials and notify if any are invalid"""
logger.info("Starting credential check task")
# Load current failure state
failure_state = get_failure_state()
# Track failures
failures = []
# Get provider configurations from config_validator
provider_status = get_provider_status()
storage_configs = validate_storage_configs()
# Define services with their test functions and configuration status
services = [
{
"name": "OpenAI",
"check_func": sync_test_openai_connection,
"configured": provider_status.get("OpenAI", {}).get("configured", False),
"config_issues": [] # OpenAI isn't in storage_configs
},
{
"name": "Azure Document Intelligence",
"check_func": sync_test_azure_connection,
"configured": provider_status.get("Azure AI", {}).get("configured", False),
"config_issues": [] # Azure isn't in storage_configs
},
{
"name": "Dropbox",
"check_func": sync_test_dropbox_token,
"configured": provider_status.get("Dropbox", {}).get("configured", False),
"config_issues": storage_configs.get("dropbox", [])
},
{
"name": "Google Drive",
"check_func": sync_test_google_drive_token,
"configured": provider_status.get("Google Drive", {}).get("configured", False),
"config_issues": storage_configs.get("google_drive", [])
},
{
"name": "OneDrive",
"check_func": sync_test_onedrive_token,
"configured": provider_status.get("OneDrive", {}).get("configured", False),
"config_issues": storage_configs.get("onedrive", [])
}
]
# Check each service
results = {}
current_time = int(time.time())
for service in services:
service_name = service["name"]
logger.info(f"Checking credentials for {service_name}")
# Skip services that aren't configured
if not service["configured"]:
config_issues = service["config_issues"]
issue_msg = f"Not properly configured" + (f": {', '.join(config_issues)}" if config_issues else "")
logger.info(f"Skipping {service_name}: {issue_msg}")
results[service_name] = {
"status": "unconfigured",
"message": issue_msg
}
continue
try:
# Call the synchronized test function and get the result
result = service["check_func"]()
# All test functions return a dict with "status" field
is_valid = result.get("status") == "success"
error_message = result.get("message", "Unknown error")
# Store the result
results[service_name] = {
"status": "valid" if is_valid else "invalid",
"message": error_message
}
if not is_valid:
failures.append(service_name)
# Get current failure count for this service
service_state = failure_state.get(service_name, {"count": 0, "last_notified": 0})
service_state["count"] = service_state.get("count", 0) + 1
# Only notify if we haven't reached the notification threshold (3 failures)
# or if this is the first failure after a recovery
if service_state["count"] <= 3 or service_state.get("recovered", False):
notify_credential_failure(service_name, error_message)
service_state["last_notified"] = current_time
service_state["recovered"] = False
logger.warning(f"{service_name} credentials check failed ({service_state['count']} times): {error_message}")
else:
# We're in cooldown mode
logger.warning(f"{service_name} credentials check failed ({service_state['count']} times): {error_message} - notification suppressed")
# Update failure state
failure_state[service_name] = service_state
else:
logger.info(f"{service_name} credentials are valid")
# Check if this was previously failing and now recovered
if service_name in failure_state and failure_state[service_name].get("count", 0) > 0:
logger.info(f"{service_name} has recovered after {failure_state[service_name]['count']} failures")
# Mark it as recovered and reset count
failure_state[service_name] = {"count": 0, "recovered": True, "last_notified": 0}
elif service_name in failure_state:
# Just make sure recovered flag is cleared if it was there
failure_state[service_name]["recovered"] = True
except Exception as e:
logger.error(f"Error checking {service_name} credentials: {e}", exc_info=True)
failures.append(service_name)
error_message = f"Exception during credential check: {str(e)}"
# Get current failure count for this service
service_state = failure_state.get(service_name, {"count": 0, "last_notified": 0})
service_state["count"] = service_state.get("count", 0) + 1
# Only notify if we haven't reached the notification threshold or if we just recovered
if service_state["count"] <= 3 or service_state.get("recovered", False):
notify_credential_failure(service_name, error_message)
service_state["last_notified"] = current_time
service_state["recovered"] = False
# Update failure state
failure_state[service_name] = service_state
# Store the error result
results[service_name] = {
"status": "error",
"message": error_message
}
# Save updated failure state
save_failure_state(failure_state)
# Count only services that were actually checked (configured services)
configured_services = [s for s in services if s["configured"]]
num_configured = len(configured_services)
# Summarize results
logger.info(f"Credential check completed. Configured services: {num_configured}, Valid: {num_configured - len(failures)}, Invalid: {len(failures)}")
return {
"checked": num_configured,
"unconfigured": len(services) - num_configured,
"failures": len(failures),
"results": results,
"failure_state": failure_state
}
+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 # 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.masking import mask_sensitive_value
from app.utils.config_validator.providers import get_provider_status 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__ = [ __all__ = [
'validate_email_config', 'validate_email_config',
'validate_storage_configs', 'validate_storage_configs',
'validate_notification_config',
'mask_sensitive_value', 'mask_sensitive_value',
'get_provider_status', 'get_provider_status',
'get_settings_for_display', 'get_settings_for_display',
@@ -20,3 +29,4 @@ __all__ = [
'check_all_configs' 'check_all_configs'
] ]
+2
View File
@@ -5,6 +5,7 @@ Configuration validation package for the application.
from app.utils.config_validator.validators import ( from app.utils.config_validator.validators import (
validate_email_config, validate_email_config,
validate_storage_configs, validate_storage_configs,
validate_notification_config,
check_all_configs check_all_configs
) )
from app.utils.config_validator.masking import mask_sensitive_value from app.utils.config_validator.masking import mask_sensitive_value
@@ -17,6 +18,7 @@ from app.utils.config_validator.settings_display import (
__all__ = [ __all__ = [
'validate_email_config', 'validate_email_config',
'validate_storage_configs', 'validate_storage_configs',
'validate_notification_config',
'mask_sensitive_value', 'mask_sensitive_value',
'get_provider_status', 'get_provider_status',
'get_settings_for_display', 'get_settings_for_display',
+19
View File
@@ -11,6 +11,24 @@ def get_provider_status():
""" """
providers = {} 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 # Add AI services first
providers["OpenAI"] = { providers["OpenAI"] = {
"name": "OpenAI", "name": "OpenAI",
@@ -253,4 +271,5 @@ def get_provider_status():
} }
} }
return providers return providers
@@ -23,6 +23,20 @@ def dump_all_settings():
value = f"{value[:visible_start]}{'*' * (len(value) - visible_start - visible_end)}{value[-visible_end:]}" value = f"{value[:visible_start]}{'*' * (len(value) - visible_start - visible_end)}{value[-visible_end:]}"
else: else:
value = f"{value[:2]}{'*' * (len(value) - 4)}{value[-2:]}" if isinstance(value, str) and len(value) > 4 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(f"{key}: {value}")
logger.info("--- END OF SETTINGS DUMP ---") logger.info("--- END OF SETTINGS DUMP ---")
@@ -169,6 +183,13 @@ def get_settings_for_display(show_values=False):
"Monitoring": [ "Monitoring": [
"uptime_kuma_url", "uptime_kuma_url",
"uptime_kuma_ping_interval" "uptime_kuma_ping_interval"
],
"Notifications": [
"notification_urls",
"notify_on_task_failure",
"notify_on_credential_failure",
"notify_on_startup",
"notify_on_shutdown"
] ]
} }
+39 -1
View File
@@ -136,6 +136,36 @@ def validate_storage_configs():
return issues 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(): def check_all_configs():
"""Run all configuration validations and log results""" """Run all configuration validations and log results"""
from app.utils.config_validator.settings_display import dump_all_settings from app.utils.config_validator.settings_display import dump_all_settings
@@ -161,8 +191,16 @@ def check_all_configs():
else: else:
logger.info(f"{provider.capitalize()} configuration OK") 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 all identified issues
return { return {
'email': email_issues, 'email': email_issues,
'storage': storage_issues 'storage': storage_issues,
'notification': notification_issues
} }
+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"]
)
+7 -1
View File
@@ -86,6 +86,9 @@ async def status_dashboard(request: Request):
except Exception: except Exception:
container_info = {'is_docker': False, 'id': 'Unknown', 'git_sha': 'Unknown'} container_info = {'is_docker': False, 'id': 'Unknown', 'git_sha': 'Unknown'}
# Get notification URLs for the notification box
notification_urls = getattr(settings, 'notification_urls', [])
return templates.TemplateResponse( return templates.TemplateResponse(
"status_dashboard.html", "status_dashboard.html",
{ {
@@ -95,7 +98,10 @@ async def status_dashboard(request: Request):
"build_date": build_date, "build_date": build_date,
"debug_enabled": getattr(settings, 'debug', False), "debug_enabled": getattr(settings, 'debug', False),
"last_check": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "last_check": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"container_info": container_info "container_info": container_info,
"settings": {
"notification_urls": notification_urls
}
} }
) )
+107
View File
@@ -104,6 +104,28 @@
</button> </button>
{% endif %} {% endif %}
<!-- Notification Test Button -->
{% if name == "Notifications" and provider.configured %}
<button
id="testNotificationBtn"
class="test-generic-btn inline-flex items-center px-2.5 py-1.5 border border-gray-300 text-xs font-medium rounded text-gray-700 bg-white hover:bg-gray-50 focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-indigo-500"
data-endpoint="{{ provider.test_endpoint }}"
data-method="POST">
Test Notifications
</button>
{% endif %}
<!-- Generic Test Button for any provider with test_endpoint -->
{% if provider.testable and provider.configured and provider.test_endpoint and name != "Notifications" %}
<button
class="test-generic-btn inline-flex items-center px-2.5 py-1.5 border border-gray-300 text-xs font-medium rounded text-gray-700 bg-white hover:bg-gray-50 focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-indigo-500"
data-endpoint="{{ provider.test_endpoint }}"
data-method="{{ provider.test_method|default('GET') }}">
Test {{ name }}
</button>
{% endif %}
<!-- Provider-specific buttons -->
{% if name == "Dropbox" %} {% if name == "Dropbox" %}
{% if provider.configured %} {% if provider.configured %}
<button <button
@@ -397,6 +419,91 @@ document.addEventListener('DOMContentLoaded', function() {
}); });
}); });
// Test notifications
const testNotificationBtn = document.getElementById('testNotificationBtn');
if (testNotificationBtn) {
testNotificationBtn.addEventListener('click', function() {
const originalText = this.innerHTML;
this.innerHTML = '<i class="fa-solid fa-spinner fa-spin mr-2"></i> Sending...';
this.disabled = true;
fetch('/api/diagnostic/test-notification', {
method: 'POST',
})
.then(response => response.json())
.then(data => {
if (data.status === 'success') {
showModal('success', 'Test Notification Sent', data.message);
} else if (data.status === 'warning') {
showModal('error', 'Notification Configuration Missing', data.message);
} else {
showModal('error', 'Test Notification Failed', data.message);
}
})
.catch(error => {
showModal('error', 'Connection Error', 'Error testing notifications: ' + error.message);
})
.finally(() => {
// Always restore the button text and enable the button, regardless of success or failure
this.innerHTML = originalText;
this.disabled = false;
});
});
}
// Generic test button functionality
const testGenericBtns = document.querySelectorAll('.test-generic-btn:not(#testNotificationBtn)');
testGenericBtns.forEach(button => {
button.addEventListener('click', function() {
const originalText = this.innerHTML;
const endpoint = this.getAttribute('data-endpoint');
const method = this.getAttribute('data-method') || 'GET';
this.innerHTML = '<i class="fa-solid fa-spinner fa-spin mr-2"></i> Testing...';
this.disabled = true;
fetch(endpoint, {
method: method,
})
.then(response => response.json())
.then(data => {
if (data.status === 'success') {
// If there's token info, we need to handle it specially
if (data.token_info && data.token_info.expires_in_human) {
let message = data.message || 'Connection successful';
message += `<br><br><div class="bg-blue-50 p-3 rounded mt-2">
<span class="font-medium">Token valid for:</span> ${data.token_info.expires_in_human}
</div>`;
modalTitle.textContent = 'Test Successful';
modalMessage.innerHTML = message;
modalIcon.innerHTML = '<i class="fa-solid fa-check text-green-600 fa-2x"></i>';
modalIcon.className = 'mx-auto flex items-center justify-center h-12 w-12 rounded-full bg-green-100 mb-4';
resultModal.classList.remove('hidden');
} else {
// Regular success
showModal('success', 'Test Successful', data.message + (data.account ? ' as ' + data.account : ''));
}
} else if (data.status === 'warning') {
showModal('error', 'Warning', data.message);
} else if (data.needs_reauth) {
showModal('error', 'Authentication Required', 'Your token has expired or is invalid. Please reconfigure this connection.');
} else {
showModal('error', 'Test Failed', data.message);
}
})
.catch(error => {
showModal('error', 'Connection Error', `Error: ${error.message}`);
})
.finally(() => {
// Always restore the button state
this.innerHTML = originalText;
this.disabled = false;
});
});
});
// Test provider connections // Test provider connections
const testButtons = document.querySelectorAll('.test-provider-btn'); const testButtons = document.querySelectorAll('.test-provider-btn');
testButtons.forEach(button => { testButtons.forEach(button => {
+4 -1
View File
@@ -27,4 +27,7 @@ msal>=1.20.0
boto3>=1.28.0 boto3>=1.28.0
# SFTP # SFTP
paramiko>=3.4.0 # SSH/SFTP implementation for Python paramiko>=3.4.0 # SSH/SFTP implementation for Python
# Notification service
apprise>=1.4.0