423 lines
18 KiB
Python
423 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
Administrator Notification Service
|
||
|
||
This module provides functionality to notify system administrators about
|
||
critical issues related to certificate generation, particularly LibreOffice
|
||
unavailability and repeated generation failures.
|
||
"""
|
||
|
||
import logging
|
||
from datetime import datetime, timedelta
|
||
from typing import Optional, Dict, Any
|
||
|
||
_logger = logging.getLogger(__name__)
|
||
|
||
|
||
class AdminNotifier:
|
||
"""
|
||
Service for sending notifications to system administrators.
|
||
|
||
This class handles the creation and sending of notifications to administrators
|
||
when critical issues occur, such as LibreOffice unavailability or repeated
|
||
certificate generation failures.
|
||
"""
|
||
|
||
# Class-level cache for tracking notification history
|
||
_notification_history = {}
|
||
_failure_counts = {}
|
||
|
||
# Notification throttling settings
|
||
THROTTLE_MINUTES = 60 # Don't send same notification more than once per hour
|
||
FAILURE_THRESHOLD = 3 # Notify after 3 consecutive failures
|
||
|
||
@classmethod
|
||
def _should_send_notification(cls, notification_key: str) -> bool:
|
||
"""
|
||
Check if a notification should be sent based on throttling rules.
|
||
|
||
Args:
|
||
notification_key: Unique key identifying the notification type
|
||
|
||
Returns:
|
||
bool: True if notification should be sent, False if throttled
|
||
"""
|
||
now = datetime.now()
|
||
|
||
# Check if we've sent this notification recently
|
||
if notification_key in cls._notification_history:
|
||
last_sent = cls._notification_history[notification_key]
|
||
time_since_last = (now - last_sent).total_seconds() / 60
|
||
|
||
if time_since_last < cls.THROTTLE_MINUTES:
|
||
_logger.debug(
|
||
'Notification throttled: %s (last sent %.1f minutes ago)',
|
||
notification_key, time_since_last
|
||
)
|
||
return False
|
||
|
||
return True
|
||
|
||
@classmethod
|
||
def _record_notification_sent(cls, notification_key: str):
|
||
"""
|
||
Record that a notification was sent.
|
||
|
||
Args:
|
||
notification_key: Unique key identifying the notification type
|
||
"""
|
||
cls._notification_history[notification_key] = datetime.now()
|
||
|
||
@classmethod
|
||
def _increment_failure_count(cls, failure_key: str) -> int:
|
||
"""
|
||
Increment the failure count for a specific operation.
|
||
|
||
Args:
|
||
failure_key: Unique key identifying the failing operation
|
||
|
||
Returns:
|
||
int: Current failure count
|
||
"""
|
||
if failure_key not in cls._failure_counts:
|
||
cls._failure_counts[failure_key] = 0
|
||
|
||
cls._failure_counts[failure_key] += 1
|
||
return cls._failure_counts[failure_key]
|
||
|
||
@classmethod
|
||
def _reset_failure_count(cls, failure_key: str):
|
||
"""
|
||
Reset the failure count for a specific operation.
|
||
|
||
Args:
|
||
failure_key: Unique key identifying the operation
|
||
"""
|
||
if failure_key in cls._failure_counts:
|
||
del cls._failure_counts[failure_key]
|
||
|
||
@classmethod
|
||
def notify_libreoffice_unavailable(
|
||
cls,
|
||
env,
|
||
error_message: str,
|
||
context_data: Optional[Dict[str, Any]] = None
|
||
):
|
||
"""
|
||
Notify administrators that LibreOffice is unavailable.
|
||
|
||
This is a critical notification that should be sent when LibreOffice
|
||
cannot be found or fails to execute, preventing PDF certificate generation.
|
||
|
||
Args:
|
||
env: Odoo environment object
|
||
error_message: Description of the LibreOffice issue
|
||
context_data: Additional context information
|
||
"""
|
||
notification_key = 'libreoffice_unavailable'
|
||
|
||
# Check if we should send this notification (throttling)
|
||
if not cls._should_send_notification(notification_key):
|
||
return
|
||
|
||
try:
|
||
# Get admin users (users with Settings access)
|
||
admin_group = env.ref('base.group_system', raise_if_not_found=False)
|
||
|
||
if not admin_group:
|
||
_logger.warning('Could not find admin group to notify about LibreOffice error')
|
||
return
|
||
|
||
admin_users = admin_group.users
|
||
|
||
if not admin_users:
|
||
_logger.warning('No admin users found to notify about LibreOffice error')
|
||
return
|
||
|
||
# Build context information
|
||
context_info = ""
|
||
if context_data:
|
||
context_parts = []
|
||
for key, value in context_data.items():
|
||
context_parts.append(f"<li><strong>{key}:</strong> {value}</li>")
|
||
if context_parts:
|
||
context_info = f"<ul>{''.join(context_parts)}</ul>"
|
||
|
||
# Create notification message
|
||
subject = '🚨 Survey Certificate: LibreOffice Unavailable'
|
||
body = f"""
|
||
<div style="font-family: Arial, sans-serif; padding: 20px; background-color: #fff3cd; border: 1px solid #ffc107; border-radius: 5px;">
|
||
<h2 style="color: #856404; margin-top: 0;">⚠️ LibreOffice Unavailable</h2>
|
||
|
||
<p style="color: #856404;">
|
||
<strong>Certificate generation is currently unavailable due to LibreOffice issues.</strong>
|
||
</p>
|
||
|
||
<div style="background-color: white; padding: 15px; border-radius: 3px; margin: 15px 0;">
|
||
<h3 style="margin-top: 0; color: #333;">Error Details:</h3>
|
||
<p style="color: #666; font-family: monospace; background-color: #f8f9fa; padding: 10px; border-radius: 3px;">
|
||
{error_message}
|
||
</p>
|
||
</div>
|
||
|
||
{f'<div style="background-color: white; padding: 15px; border-radius: 3px; margin: 15px 0;"><h3 style="margin-top: 0; color: #333;">Context:</h3>{context_info}</div>' if context_info else ''}
|
||
|
||
<div style="background-color: white; padding: 15px; border-radius: 3px; margin: 15px 0;">
|
||
<h3 style="margin-top: 0; color: #333;">📋 Action Required:</h3>
|
||
<p style="color: #666;">Please install LibreOffice on the server to enable PDF certificate generation:</p>
|
||
<ul style="color: #666;">
|
||
<li><strong>Ubuntu/Debian:</strong> <code style="background-color: #f8f9fa; padding: 2px 6px; border-radius: 3px;">sudo apt-get install libreoffice</code></li>
|
||
<li><strong>CentOS/RHEL:</strong> <code style="background-color: #f8f9fa; padding: 2px 6px; border-radius: 3px;">sudo yum install libreoffice</code></li>
|
||
<li><strong>macOS:</strong> <code style="background-color: #f8f9fa; padding: 2px 6px; border-radius: 3px;">brew install --cask libreoffice</code></li>
|
||
</ul>
|
||
<p style="color: #666;"><strong>After installation, restart the Odoo service.</strong></p>
|
||
</div>
|
||
|
||
<div style="background-color: #e7f3ff; padding: 15px; border-radius: 3px; margin: 15px 0; border-left: 4px solid #0066cc;">
|
||
<p style="color: #004085; margin: 0;">
|
||
<strong>ℹ️ Note:</strong> Survey completion will continue to work normally, but certificates will not be generated until LibreOffice is available.
|
||
</p>
|
||
</div>
|
||
|
||
<p style="color: #856404; font-size: 12px; margin-bottom: 0;">
|
||
<em>Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</em>
|
||
</p>
|
||
</div>
|
||
"""
|
||
|
||
# Send notification to each admin
|
||
notification_count = 0
|
||
for admin in admin_users:
|
||
try:
|
||
env['mail.message'].create({
|
||
'subject': subject,
|
||
'body': body,
|
||
'message_type': 'notification',
|
||
'subtype_id': env.ref('mail.mt_note').id,
|
||
'partner_ids': [(4, admin.partner_id.id)],
|
||
'model': 'survey.survey',
|
||
'res_id': 0, # Not tied to a specific survey
|
||
})
|
||
notification_count += 1
|
||
except Exception as e:
|
||
_logger.warning(
|
||
'Failed to send LibreOffice notification to admin %s: %s',
|
||
admin.name, str(e)
|
||
)
|
||
|
||
if notification_count > 0:
|
||
_logger.info(
|
||
'Sent LibreOffice unavailable notification to %d administrators',
|
||
notification_count
|
||
)
|
||
cls._record_notification_sent(notification_key)
|
||
|
||
except Exception as e:
|
||
_logger.error(
|
||
'Failed to notify administrators about LibreOffice unavailability: %s',
|
||
str(e), exc_info=True
|
||
)
|
||
|
||
@classmethod
|
||
def notify_repeated_generation_failures(
|
||
cls,
|
||
env,
|
||
survey_id: int,
|
||
survey_title: str,
|
||
failure_count: int,
|
||
recent_errors: Optional[list] = None
|
||
):
|
||
"""
|
||
Notify administrators about repeated certificate generation failures.
|
||
|
||
This notification is sent when certificate generation fails multiple times
|
||
in a row, indicating a persistent issue that needs attention.
|
||
|
||
Args:
|
||
env: Odoo environment object
|
||
survey_id: ID of the survey experiencing failures
|
||
survey_title: Title of the survey
|
||
failure_count: Number of consecutive failures
|
||
recent_errors: List of recent error messages
|
||
"""
|
||
notification_key = f'repeated_failures_survey_{survey_id}'
|
||
|
||
# Check if we should send this notification (throttling)
|
||
if not cls._should_send_notification(notification_key):
|
||
return
|
||
|
||
try:
|
||
# Get admin users (users with Settings access)
|
||
admin_group = env.ref('base.group_system', raise_if_not_found=False)
|
||
|
||
if not admin_group:
|
||
_logger.warning('Could not find admin group to notify about generation failures')
|
||
return
|
||
|
||
admin_users = admin_group.users
|
||
|
||
if not admin_users:
|
||
_logger.warning('No admin users found to notify about generation failures')
|
||
return
|
||
|
||
# Build error list
|
||
error_list = ""
|
||
if recent_errors:
|
||
error_items = []
|
||
for idx, error in enumerate(recent_errors[-5:], 1): # Show last 5 errors
|
||
error_items.append(f"<li><strong>Error {idx}:</strong> {error}</li>")
|
||
error_list = f"<ul>{''.join(error_items)}</ul>"
|
||
|
||
# Create notification message
|
||
subject = f'⚠️ Survey Certificate: Repeated Generation Failures'
|
||
body = f"""
|
||
<div style="font-family: Arial, sans-serif; padding: 20px; background-color: #f8d7da; border: 1px solid #f5c6cb; border-radius: 5px;">
|
||
<h2 style="color: #721c24; margin-top: 0;">⚠️ Repeated Certificate Generation Failures</h2>
|
||
|
||
<p style="color: #721c24;">
|
||
<strong>Certificate generation has failed {failure_count} consecutive times for a survey.</strong>
|
||
</p>
|
||
|
||
<div style="background-color: white; padding: 15px; border-radius: 3px; margin: 15px 0;">
|
||
<h3 style="margin-top: 0; color: #333;">Survey Information:</h3>
|
||
<ul style="color: #666;">
|
||
<li><strong>Survey ID:</strong> {survey_id}</li>
|
||
<li><strong>Survey Title:</strong> {survey_title}</li>
|
||
<li><strong>Failure Count:</strong> {failure_count}</li>
|
||
</ul>
|
||
</div>
|
||
|
||
{f'<div style="background-color: white; padding: 15px; border-radius: 3px; margin: 15px 0;"><h3 style="margin-top: 0; color: #333;">Recent Errors:</h3>{error_list}</div>' if error_list else ''}
|
||
|
||
<div style="background-color: white; padding: 15px; border-radius: 3px; margin: 15px 0;">
|
||
<h3 style="margin-top: 0; color: #333;">🔍 Possible Causes:</h3>
|
||
<ul style="color: #666;">
|
||
<li>LibreOffice is not installed or not accessible</li>
|
||
<li>Template file is corrupted or invalid</li>
|
||
<li>Placeholder mappings are misconfigured</li>
|
||
<li>Server resource constraints (disk space, memory)</li>
|
||
<li>Permission issues with temporary file directories</li>
|
||
</ul>
|
||
</div>
|
||
|
||
<div style="background-color: white; padding: 15px; border-radius: 3px; margin: 15px 0;">
|
||
<h3 style="margin-top: 0; color: #333;">📋 Recommended Actions:</h3>
|
||
<ol style="color: #666;">
|
||
<li>Check Odoo logs for detailed error messages</li>
|
||
<li>Verify LibreOffice is installed and accessible</li>
|
||
<li>Review the survey's certificate template configuration</li>
|
||
<li>Test certificate generation manually from the survey form</li>
|
||
<li>Check server resources (disk space, memory)</li>
|
||
</ol>
|
||
</div>
|
||
|
||
<div style="background-color: #e7f3ff; padding: 15px; border-radius: 3px; margin: 15px 0; border-left: 4px solid #0066cc;">
|
||
<p style="color: #004085; margin: 0;">
|
||
<strong>ℹ️ Note:</strong> Survey completion continues to work normally, but certificates are not being generated for participants.
|
||
</p>
|
||
</div>
|
||
|
||
<p style="color: #721c24; font-size: 12px; margin-bottom: 0;">
|
||
<em>Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</em>
|
||
</p>
|
||
</div>
|
||
"""
|
||
|
||
# Send notification to each admin
|
||
notification_count = 0
|
||
for admin in admin_users:
|
||
try:
|
||
env['mail.message'].create({
|
||
'subject': subject,
|
||
'body': body,
|
||
'message_type': 'notification',
|
||
'subtype_id': env.ref('mail.mt_note').id,
|
||
'partner_ids': [(4, admin.partner_id.id)],
|
||
'model': 'survey.survey',
|
||
'res_id': survey_id,
|
||
})
|
||
notification_count += 1
|
||
except Exception as e:
|
||
_logger.warning(
|
||
'Failed to send failure notification to admin %s: %s',
|
||
admin.name, str(e)
|
||
)
|
||
|
||
if notification_count > 0:
|
||
_logger.info(
|
||
'Sent repeated failure notification to %d administrators for survey %s',
|
||
notification_count, survey_id
|
||
)
|
||
cls._record_notification_sent(notification_key)
|
||
|
||
except Exception as e:
|
||
_logger.error(
|
||
'Failed to notify administrators about repeated failures: %s',
|
||
str(e), exc_info=True
|
||
)
|
||
|
||
@classmethod
|
||
def track_generation_failure(
|
||
cls,
|
||
env,
|
||
survey_id: int,
|
||
survey_title: str,
|
||
error_message: str
|
||
):
|
||
"""
|
||
Track a certificate generation failure and notify if threshold is reached.
|
||
|
||
This method increments the failure count for a survey and sends a notification
|
||
to administrators if the failure threshold is reached.
|
||
|
||
Args:
|
||
env: Odoo environment object
|
||
survey_id: ID of the survey
|
||
survey_title: Title of the survey
|
||
error_message: Description of the error
|
||
"""
|
||
failure_key = f'survey_{survey_id}_failures'
|
||
|
||
# Increment failure count
|
||
failure_count = cls._increment_failure_count(failure_key)
|
||
|
||
_logger.warning(
|
||
'Certificate generation failure #%d for survey %s: %s',
|
||
failure_count, survey_id, error_message
|
||
)
|
||
|
||
# Check if we've reached the threshold
|
||
if failure_count >= cls.FAILURE_THRESHOLD:
|
||
# Store recent errors
|
||
error_history_key = f'{failure_key}_history'
|
||
if error_history_key not in cls._notification_history:
|
||
cls._notification_history[error_history_key] = []
|
||
|
||
cls._notification_history[error_history_key].append(error_message)
|
||
|
||
# Send notification
|
||
cls.notify_repeated_generation_failures(
|
||
env,
|
||
survey_id,
|
||
survey_title,
|
||
failure_count,
|
||
cls._notification_history.get(error_history_key, [])
|
||
)
|
||
|
||
@classmethod
|
||
def track_generation_success(cls, survey_id: int):
|
||
"""
|
||
Track a successful certificate generation and reset failure count.
|
||
|
||
Args:
|
||
survey_id: ID of the survey
|
||
"""
|
||
failure_key = f'survey_{survey_id}_failures'
|
||
error_history_key = f'{failure_key}_history'
|
||
|
||
# Reset failure count and error history
|
||
cls._reset_failure_count(failure_key)
|
||
if error_history_key in cls._notification_history:
|
||
del cls._notification_history[error_history_key]
|