Minecon724/anonsubmit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add email notifications

Browse source
This commit is contained in:
Minecon724 2025-04-07 07:34:15 +02:00
commit 194311ff92
Signed by: Minecon724
GPG key ID: A02E6E67AB961189
9 changed files with 473 additions and 89 deletions
Download patch file Download diff file Expand all files Collapse all files

118
src/anonchat/admin_routes.py
View file

@ -2,7 +2,9 @@ from flask import request, render_template, redirect, url_for, flash, session
from functools import wraps
from . import app, db, limiter, csrf
from .models import Inquiry, Message, Settings, Admin
from .webhooks import WebhookError, inquiry_created
from .notifiers import webhooks, emails
from .notifiers.webhooks import WebhookError
from .notifiers.emails import EmailNotificationError
def is_admin():
return 'admin_authenticated' in session and session['admin_authenticated']
@ -84,33 +86,6 @@ def admin_dashboard():
return render_template('admin/dashboard.html', inquiries_with_data=inquiries_with_data)
@app.route('/admin/settings/webhook', methods=['POST'])
@admin_required
def admin_settings_webhook():
settings = Settings.query.first()
if not settings:
settings = Settings()
db.session.add(settings)
# Update webhook settings
settings.webhook_enabled = request.form.get('webhook_enabled') == 'on'
settings.webhook_url = request.form.get('webhook_url', '')
settings.webhook_secret = request.form.get('webhook_secret', '')
db.session.commit()
# Test webhook if enabled and URL is provided
if settings.webhook_enabled and settings.webhook_url:
try:
inquiry_created('1234abcd1234abcd', 'This is a test message', is_async=False)
flash('Webhook settings saved and test sent successfully')
except WebhookError as e:
flash(f'Webhook test failed: {str(e)}')
else:
flash('Webhook settings saved')
return redirect(url_for('admin_settings'))
@app.route('/admin/settings', methods=['GET'])
@admin_required
def admin_settings():
@ -149,4 +124,91 @@ def admin_settings_password():
flash('Password updated successfully', 'success')
return redirect(url_for('admin_settings'))
@app.route('/admin/settings/notification', methods=['POST'])
@admin_required
def admin_settings_notification():
settings = Settings.query.first()
if not settings:
settings = Settings()
db.session.add(settings)
settings.notification_show_message = request.form.get('notification_show_message') == 'on'
db.session.commit()
flash('Notification settings saved', 'success')
return redirect(url_for('admin_settings'))
@app.route('/admin/settings/webhook', methods=['POST'])
@admin_required
def admin_settings_webhook():
settings = Settings.query.first()
if not settings:
settings = Settings()
db.session.add(settings)
# Update webhook settings
settings.webhook_enabled = request.form.get('webhook_enabled') == 'on'
settings.webhook_url = request.form.get('webhook_url', '')
settings.webhook_secret = request.form.get('webhook_secret', '')
db.session.commit()
# Test webhook if enabled and URL is provided
if settings.webhook_enabled and settings.webhook_url:
try:
webhooks.send_webhook(
event_type='inquiry_created',
data={
'inquiry_id': '1234abcd1234abcd',
'message': 'This is a test message'
},
is_async=False
)
flash('Webhook settings saved and test sent successfully')
except WebhookError as e:
flash(f'Webhook test failed: {str(e)}', 'error')
else:
flash('Webhook settings saved', 'success')
return redirect(url_for('admin_settings'))
@app.route('/admin/settings/email', methods=['POST'])
@admin_required
def admin_settings_email():
settings = Settings.query.first()
if not settings:
settings = Settings()
db.session.add(settings)
settings.email_notifications_enabled = request.form.get('email_notifications_enabled') == 'on'
settings.smtp_server = request.form.get('smtp_server', '')
settings.smtp_use_ssl = request.form.get('smtp_use_ssl') == 'on'
settings.smtp_username = request.form.get('smtp_username', '')
settings.smtp_password = request.form.get('smtp_password', '')
settings.recipient_email = request.form.get('recipient_email', '')
db.session.commit()
if settings.email_notifications_enabled:
result = emails.verify_email_settings(settings)
if result['errors']:
flash(f'Email settings verification failed: {", ".join(result["errors"])}', 'error')
else:
try:
emails.send_email(
title='Inquiry #1234abcd1234abcd Created',
inquiry_id='1234abcd1234abcd',
data={
'message': 'This is a test message'
},
is_async=False
)
flash('Email settings saved and test sent successfully')
except EmailNotificationError as e:
flash(f'Email test failed: {str(e)}', 'error')
else:
flash('Email notifications svaed', 'success')
return redirect(url_for('admin_settings'))

48
src/anonchat/migrations/add_email_settings.py Normal file
View file

@ -0,0 +1,48 @@
from sqlalchemy import Column, Boolean, String, inspect
from sqlalchemy.sql import text
from flask import current_app
def run_migration(db):
"""Add email notification settings columns to the Settings table if they don't exist."""
current_app.logger.info("Running migration: add_email_settings")
# Check if the table exists first
inspector = inspect(db.engine)
if 'settings' not in inspector.get_table_names():
current_app.logger.info("Settings table doesn't exist yet, skipping migration")
return False
# Check existing columns
columns = [col['name'] for col in inspector.get_columns('settings')]
# Track if we need to commit changes
changes_made = False
# Dictionary of columns to add with their types
columns_to_add = {
'email_notifications_enabled': 'BOOLEAN DEFAULT FALSE',
'smtp_server': 'VARCHAR(255)',
'smtp_use_ssl': 'BOOLEAN DEFAULT FALSE',
'smtp_username': 'VARCHAR(255)',
'smtp_password': 'VARCHAR(255)',
'recipient_email': 'VARCHAR(255)',
'notification_show_message': 'BOOLEAN DEFAULT FALSE'
}
# Add each column if it doesn't exist
for column_name, column_type in columns_to_add.items():
if column_name not in columns:
current_app.logger.info(f"Adding {column_name} column to settings table")
with db.engine.connect() as conn:
conn.execute(text(f"ALTER TABLE settings ADD COLUMN {column_name} {column_type}"))
conn.commit()
changes_made = True
else:
current_app.logger.info(f"{column_name} column already exists in settings table")
if changes_made:
current_app.logger.info("Migration completed successfully")
else:
current_app.logger.info("No changes needed, all columns already exist")
return changes_made

14
src/anonchat/models/settings.py
View file

@ -2,6 +2,18 @@ from .. import db
class Settings(db.Model):
id = db.Column(db.Integer, primary_key=True)
# Webhook settings
webhook_enabled = db.Column(db.Boolean, default=False)
webhook_url = db.Column(db.String(255), nullable=True)
webhook_secret = db.Column(db.String(255), nullable=True)
webhook_secret = db.Column(db.String(255), nullable=True)
# Email notification settings
email_notifications_enabled = db.Column(db.Boolean, default=False)
smtp_server = db.Column(db.String(255), nullable=True)
smtp_use_ssl = db.Column(db.Boolean, default=False)
smtp_username = db.Column(db.String(255), nullable=True)
smtp_password = db.Column(db.String(255), nullable=True) # Consider storing this securely
recipient_email = db.Column(db.String(255), nullable=True)
notification_show_message = db.Column(db.Boolean, default=False)

80
src/anonchat/notifiers/__init__.py Normal file
View file

@ -0,0 +1,80 @@
from flask import current_app
from .emails import send_email
from .webhooks import send_webhook
MESSAGE_PREVIEW_LENGTH = 30
EVENT_TYPES = {
'inquiry_created': 'Inquiry #{} Created',
'inquiry_reopened': 'Inquiry #{} Reopened',
'inquiry_closed': 'Inquiry #{} Closed',
'inquiry_message': 'Inquiry #{} Message'
}
### Abstract methods ###
def inquiry_created(inquiry_id: str, message: str = None, is_async: bool = True) -> None:
_send_event(
event_type='inquiry_created',
inquiry_id=inquiry_id,
message=message,
is_async=is_async
)
def inquiry_reopened(inquiry_id: str, is_async: bool = True) -> None:
_send_event(
event_type='inquiry_reopened',
inquiry_id=inquiry_id,
is_async=is_async
)
def inquiry_closed(inquiry_id: str, is_async: bool = True) -> None:
_send_event(
event_type='inquiry_closed',
inquiry_id=inquiry_id,
is_async=is_async
)
def inquiry_message(inquiry_id: str, message: str = None, is_async: bool = True) -> None:
_send_event(
event_type='inquiry_message',
inquiry_id=inquiry_id,
message=message,
is_async=is_async
)
### Private methods ###
def _send_event(event_type: str, inquiry_id: str, message: str = None, is_async: bool = True) -> None:
_send_email_ignore_errors(
title=EVENT_TYPES[event_type].format(inquiry_id),
inquiry_id=inquiry_id,
data={
'message': _get_message_preview(message)
},
is_async=is_async
)
_send_webhook_ignore_errors(
event_type=event_type,
data={
'inquiry_id': inquiry_id,
'message': _get_message_preview(message)
},
is_async=is_async
)
def _send_email_ignore_errors(title, inquiry_id, data, is_async=True):
try:
send_email(title, inquiry_id, data, is_async)
except:
pass
def _send_webhook_ignore_errors(event_type, data, is_async=True):
try:
send_webhook(event_type, data, is_async)
except:
pass
def _get_message_preview(message: str) -> str:
return (message[:MESSAGE_PREVIEW_LENGTH] + '...') if len(message) > MESSAGE_PREVIEW_LENGTH else message

156
src/anonchat/notifiers/emails.py Normal file
View file

@ -0,0 +1,156 @@
import smtplib
from email.message import EmailMessage
import threading
from datetime import datetime
from flask import current_app
from ..models import Settings
class EmailNotificationError(Exception):
"""Exception raised for errors in the email notification process.
Attributes:
message -- explanation of the error
original_exception -- the original exception that caused this error (optional)
"""
def __init__(self, message, original_exception=None):
self.message = message
self.original_exception = original_exception
super().__init__(self.message)
def verify_email_settings(settings: Settings = None):
"""
Verifies that email notification settings are properly configured.
Returns:
dict: A dictionary containing:
- 'valid' (bool): Whether settings are valid for sending emails
- 'errors' (list): List of error messages if any settings are invalid
- 'settings' (dict): Current settings state (without sensitive values)
"""
if settings is None:
settings = Settings.query.first()
result = {
'valid': False,
'errors': [],
'settings': {}
}
# Check if settings exist and notifications are enabled
if not settings:
result['errors'].append("Settings not found in database")
return result
result['settings']['email_notifications_enabled'] = settings.email_notifications_enabled
if not settings.email_notifications_enabled:
result['errors'].append("Email notifications are disabled")
return result
# Required fields
required_fields = {
'smtp_server': settings.smtp_server,
'smtp_username': settings.smtp_username,
'recipient_email': settings.recipient_email
}
# Check required fields
for field, value in required_fields.items():
result['settings'][field] = value
if not value:
result['errors'].append(f"Missing required setting: {field}")
# Test SMTP connection if desired (commented out to avoid actual connection attempts)
# This could be implemented as a separate function that calls this one first
# Mark as valid if no errors
result['valid'] = len(result['errors']) == 0
return result
def send_email(title: str, inquiry_id: str, data: dict = {}, is_async=False):
"""Sends an email notification, either synchronously or asynchronously."""
app = current_app._get_current_object()
if is_async:
_send_email_async(app, title, inquiry_id, data)
else:
_send_email_worker(app, title, inquiry_id, data)
def _send_email_async(app, title: str, inquiry_id: str, data: dict = {}):
"""Queue email to be sent in a background thread"""
thread = threading.Thread(target=_send_email_worker, args=(app, title, inquiry_id, data))
thread.daemon = True # Thread will exit when main thread exits
thread.start()
def _send_email_worker(app, title: str, inquiry_id: str, data: dict = {}):
"""Worker function that sends the email notification."""
try:
with app.app_context():
settings = Settings.query.first()
if not settings or not settings.email_notifications_enabled:
return False
verification_result = verify_email_settings(settings)
if not verification_result['valid']:
raise EmailNotificationError(f"Invalid email settings: {', '.join(verification_result['errors'])}")
# Construct email message
msg = EmailMessage()
msg['Subject'] = "[AnonChat] " + title
msg['From'] = settings.smtp_username
msg['To'] = settings.recipient_email
content = f"""
Title: {title}
Timestamp: {datetime.utcnow().isoformat()} UTC
Inquiry ID: {inquiry_id}
{data.get('message', '') if settings.notification_show_message else ''}"""
msg.set_content(content)
_send_smtp_message(settings, msg)
app.logger.info(f"Email notification sent for title: {title}")
except Exception as e:
app.logger.error(f"Email error: {e}")
if isinstance(e, EmailNotificationError):
raise e
raise EmailNotificationError(f"Exception: {e}", original_exception=e)
def _get_smtp_server(smtp_server: str):
smtp_server_parts = smtp_server.split(':')
smtp_host = smtp_server_parts[0]
if len(smtp_server_parts) > 1:
smtp_port = int(smtp_server_parts[1])
else:
smtp_port = 0
return (smtp_host, smtp_port)
def _send_smtp_message(settings, msg):
smtp_server = None
smtp_host, smtp_port = _get_smtp_server(settings.smtp_server)
try:
if settings.smtp_use_ssl:
smtp_server = smtplib.SMTP_SSL(smtp_host, smtp_port, timeout=10)
else:
smtp_server = smtplib.SMTP(smtp_host, smtp_port, timeout=10)
if settings.smtp_username and settings.smtp_password:
smtp_server.login(settings.smtp_username, settings.smtp_password)
smtp_server.send_message(msg)
except smtplib.SMTPException as e:
raise EmailNotificationError(f"SMTP Error: {e}", original_exception=e)
finally:
if smtp_server:
smtp_server.quit()

32
src/anonchat/webhooks.py → src/anonchat/notifiers/webhooks.py
View file

@ -4,10 +4,9 @@ import hmac
import hashlib
import json
import threading
import logging
from datetime import datetime
from flask import current_app
from .models import Settings
from ..models import Settings
class WebhookError(Exception):
"""Exception raised for errors in the webhook process.
@ -25,7 +24,7 @@ class WebhookError(Exception):
super().__init__(self.message)
def _send_webhook(event_type, data, is_async=False):
def send_webhook(event_type, data, is_async=False):
app = current_app._get_current_object()
if is_async:
@ -85,29 +84,4 @@ def _send_webhook_worker(event_type, data, app):
if isinstance(e, WebhookError):
raise e
raise WebhookError(f"Exception: {e}", original_exception=e)
def inquiry_created(inquiry_id, message, is_async=True):
_send_webhook('inquiry_created', {
'inquiry_id': inquiry_id,
'message': message
}, is_async)
def inquiry_reopened(inquiry_id, is_async=True):
_send_webhook('inquiry_reopened', {
'inquiry_id': inquiry_id
}, is_async)
def inquiry_closed(inquiry_id, is_async=True):
_send_webhook('inquiry_closed', {
'inquiry_id': inquiry_id
}, is_async)
def inquiry_message(inquiry_id, message, is_async=True):
_send_webhook('inquiry_message', {
'inquiry_id': inquiry_id,
'message': message
}, is_async)
raise WebhookError(f"Exception: {e}", original_exception=e)

37
src/anonchat/routes.py
View file

@ -1,6 +1,6 @@
from flask import request, jsonify, render_template, redirect, url_for, flash, session, make_response
from . import app, db, limiter, csrf, webhooks
from .webhooks import WebhookError
from . import app, db, limiter, csrf, notifiers
from .models import Inquiry, Message, Settings, Admin
import os
import json
@ -79,11 +79,8 @@ def create_inquiry():
db.session.add(message)
db.session.commit()
# Send webhook for new inquiry
try:
webhooks.inquiry_created(new_inquiry.id, initial_message)
except WebhookError as e:
app.logger.error(f"Error sending webhook for new inquiry: {str(e)}")
# Send notifications for new inquiry
notifiers.inquiry_created(new_inquiry.id, initial_message)
# Add to recent inquiries cookie
return redirect(url_for('inquiry', inquiry_id=new_inquiry.id))
@ -101,11 +98,8 @@ def inquiry(inquiry_id):
inquiry.reopen()
db.session.commit()
# Send webhook for reopened inquiry
try:
webhooks.inquiry_reopened(inquiry_id)
except WebhookError as e:
app.logger.error(f"Error sending webhook for reopened inquiry: {str(e)}")
# Send notifications for reopened inquiry
notifiers.inquiry_reopened(inquiry_id)
flash('Inquiry reopened successfully.')
@ -121,10 +115,7 @@ def inquiry(inquiry_id):
db.session.commit()
if not is_admin_user: # Admins don't need to be notified of their own messages
try:
webhooks.inquiry_message(inquiry_id, message_content)
except WebhookError as e:
app.logger.error(f"Error sending webhook for message: {str(e)}")
notifiers.inquiry_message(inquiry_id, message_content)
response = redirect(url_for('inquiry', inquiry_id=inquiry_id))
return add_inquiry_to_cookie(response, inquiry_id) if not is_admin_user else response
@ -237,11 +228,8 @@ def close_inquiry(inquiry_id):
inquiry.close()
db.session.commit()
# Send webhook for closed inquiry
try:
webhooks.inquiry_closed(inquiry_id)
except WebhookError as e:
app.logger.error(f"Error sending webhook for closed inquiry: {str(e)}")
# Send notifications for closed inquiry
notifiers.inquiry_closed(inquiry_id)
# Redirect back to the inquiry or admin dashboard based on referrer
referrer = request.referrer
@ -260,11 +248,8 @@ def reopen_inquiry(inquiry_id):
inquiry.reopen()
db.session.commit()
# Send webhook for reopened inquiry
try:
webhooks.inquiry_reopened(inquiry_id)
except WebhookError as e:
app.logger.error(f"Error sending webhook for reopened inquiry: {str(e)}")
# Send notifications for reopened inquiry
notifiers.inquiry_reopened(inquiry_id)
flash('Inquiry reopened successfully.')

4
src/anonchat/static/css/styles.css
View file

@ -27,8 +27,6 @@ a:visited {
border-radius: 5px;
box-shadow: 0 2px 5px rgba(0,0,0,0.3);
}
main {
}
footer {
font-size: small;
color: #999;
@ -40,7 +38,7 @@ h1 {
form {
margin-bottom: 1rem;
}
textarea, input[type="text"], input[type="password"] {
textarea, input[type="text"], input[type="password"], input[type="email"] {
width: 100%;
padding: 0.5rem;
margin-bottom: 1rem;

73
src/anonchat/templates/admin/settings.html
View file

@ -29,9 +29,29 @@
</div>
<br>
<div class="notification-settings">
<h3>Notification Settings</h3>
<div class="webhook-info">
<h3>Webhook Settings</h3>
<div style="margin-top: 1rem;">
<form method="POST" action="{{ url_for('admin_settings_notification') }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}"/>
<label>
<input type="checkbox" name="notification_show_message" {% if settings.notification_show_message %}checked{% endif %}>
Include message content in email notifications
</label>
<br>
<button type="submit">Save Settings</button>
</form>
</div>
</div>
<br>
<div class="webhook-settings">
<h3>Webhook Notification Settings</h3>
<div style="margin-bottom: 2rem;">
<p>Configure webhook settings to receive notifications for new inquiries and responses.</p>
@ -84,6 +104,55 @@
<br>
<div class="email-settings">
<h3>Email Notification Settings</h3>
<div style="margin-bottom: 2rem;">
<p>Configure email settings to receive notifications for new inquiries and responses.</p>
</div>
<form method="POST" action="{{ url_for('admin_settings_email') }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}"/>
<div>
<label>
<input type="checkbox" name="email_notifications_enabled" {% if settings.email_notifications_enabled %}checked{% endif %}>
Enable Email Notifications
</label>
</div>
<div style="margin-top: 1rem;">
<label for="smtp_server">SMTP Server:</label>
<input type="text" id="smtp_server" name="smtp_server" value="{{ settings.smtp_server or '' }}" placeholder="smtp.example.com">
</div>
<div style="margin-top: 1rem;">
<label>
<input type="checkbox" name="smtp_use_ssl" {% if settings.smtp_use_ssl %}checked{% endif %}>
Use SSL
</label>
</div>
<div style="margin-top: 1rem;">
<label for="smtp_username">SMTP Username / Sender Email:</label>
<input type="text" id="smtp_username" name="smtp_username" value="{{ settings.smtp_username or '' }}" placeholder="username@example.com">
</div>
<div style="margin-top: 1rem;">
<label for="smtp_password">SMTP Password:</label>
<input type="password" id="smtp_password" name="smtp_password" value="{{ settings.smtp_password or '' }}" placeholder="Password">
</div>
<div style="margin-top: 1rem;">
<label for="recipient_email">Recipient Email:</label>
<input type="email" id="recipient_email" name="recipient_email" value="{{ settings.recipient_email or '' }}" placeholder="admin@example.com">
</div>
<button type="submit" style="margin-top: 1rem;">Save Email Settings</button>
</form>
</div>
<br>
<div class="settings-info">
<h3>System Configuration</h3>
<p>The following settings are configured through environment variables:</p>