"""
app/notify.py
=============
Email notification helper for compression job completion.
Delivery uses SMTP settings stored in SQLite (via app.db).
If no SMTP settings have been configured, the send call returns an
informative error rather than silently failing.
Public API
----------
get_smtp_config() -> dict with all SMTP fields (safe for the UI)
send_completion_email(to, results, cancelled) -> (ok: bool, error: str)
SMTP settings keys (stored in the 'settings' table)
----------------------------------------------------
smtp_host — hostname or IP of the SMTP server
smtp_port — port number (str)
smtp_security — 'tls' (STARTTLS) | 'ssl' (SMTPS) | 'none'
smtp_user — login username (optional)
smtp_password — login password (optional, stored as-is)
smtp_from — From: address used in sent mail
"""
import smtplib
import socket
import ssl
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.utils import formatdate, make_msgid
from .db import get_setting
# ---------------------------------------------------------------------------
# SMTP config helper
# ---------------------------------------------------------------------------
def get_smtp_config() -> dict:
"""
Read SMTP settings from the database and return them as a dict.
The password field is replaced with a placeholder so this dict is
safe to serialise and send to the browser.
Returns
-------
{
host, port, security, user, from_addr,
password_set: bool (True if a password is stored)
}
"""
return {
'host': get_setting('smtp_host') or '',
'port': get_setting('smtp_port') or '587',
'security': get_setting('smtp_security') or 'tls',
'user': get_setting('smtp_user') or '',
'from_addr': get_setting('smtp_from') or '',
'password_set': bool(get_setting('smtp_password')),
}
def _load_smtp_config() -> dict:
"""Load full config including the raw password (server-side only)."""
return {
'host': get_setting('smtp_host') or '',
'port': int(get_setting('smtp_port') or 587),
'security': get_setting('smtp_security') or 'tls',
'user': get_setting('smtp_user') or '',
'password': get_setting('smtp_password') or '',
'from_addr': get_setting('smtp_from') or '',
}
# ---------------------------------------------------------------------------
# Send helper
# ---------------------------------------------------------------------------
def send_completion_email(
to_address: str,
results: list[dict],
cancelled: bool,
) -> tuple[bool, str]:
"""
Send a job-completion notification to *to_address* using the SMTP
settings stored in SQLite.
Returns (success, error_message).
"""
if not to_address or '@' not in to_address:
return False, 'Invalid recipient email address'
cfg = _load_smtp_config()
if not cfg['host']:
return False, (
'No SMTP server configured. '
'Please add your SMTP settings in the ⚙ Settings panel.'
)
if not cfg['from_addr']:
return False, (
'No From address configured. '
'Please add your SMTP settings in the ⚙ Settings panel.'
)
# ── Build message ─────────────────────────────────────────────────────
done_files = [r for r in results if r.get('status') == 'done']
error_files = [r for r in results if r.get('status') == 'error']
total = len(results)
hostname = socket.getfqdn()
if cancelled:
subject = (f'VideoPress: compression cancelled '
f'({len(done_files)}/{total} completed) on {hostname}')
elif error_files:
subject = (f'VideoPress: compression complete with '
f'{len(error_files)} error(s) on {hostname}')
else:
subject = (f'VideoPress: compression complete — '
f'{total} file(s) processed on {hostname}')
msg = MIMEMultipart('alternative')
msg['Subject'] = subject
msg['From'] = cfg['from_addr']
msg['To'] = to_address
msg['Date'] = formatdate(localtime=True)
msg['Message-ID'] = make_msgid(domain=hostname)
msg.attach(MIMEText(
_build_plain(hostname, cancelled, done_files, error_files, total),
'plain', 'utf-8',
))
msg.attach(MIMEText(
_build_html(hostname, subject, cancelled, done_files, error_files, total),
'html', 'utf-8',
))
# ── Connect and send ──────────────────────────────────────────────────
try:
security = cfg['security'].lower()
host = cfg['host']
port = cfg['port']
if security == 'ssl':
# SMTPS — wrap in SSL from the start (port 465 typically)
context = ssl.create_default_context()
server = smtplib.SMTP_SSL(host, port, context=context, timeout=15)
else:
# Plain or STARTTLS (port 587 typically)
server = smtplib.SMTP(host, port, timeout=15)
server.ehlo()
if security == 'tls':
context = ssl.create_default_context()
server.starttls(context=context)
server.ehlo()
with server:
if cfg['user'] and cfg['password']:
server.login(cfg['user'], cfg['password'])
server.sendmail(cfg['from_addr'], [to_address], msg.as_bytes())
return True, ''
except smtplib.SMTPAuthenticationError:
return False, (
'Authentication failed — check your username and password. '
'For Gmail/Google Workspace, use an App Password rather than '
'your account password.'
)
except smtplib.SMTPConnectError as exc:
return False, (
f'Could not connect to {host}:{port}. '
f'Check the host, port, and security setting. ({exc})'
)
except smtplib.SMTPRecipientsRefused as exc:
refused = ', '.join(exc.recipients.keys())
return False, f'Recipient address rejected by server: {refused}'
except smtplib.SMTPSenderRefused as exc:
return False, (
f'From address "{cfg["from_addr"]}" was rejected by the server. '
f'Ensure it matches your authenticated account. ({exc.smtp_error.decode(errors="replace")})'
)
except smtplib.SMTPException as exc:
return False, f'SMTP error: {exc}'
except ssl.SSLError as exc:
return False, (
f'SSL/TLS error connecting to {host}:{port} — '
f'try changing the Security setting. ({exc})'
)
except TimeoutError:
return False, (
f'Connection to {host}:{port} timed out. '
f'Check the host and port, and that the server is reachable.'
)
except OSError as exc:
return False, (
f'Network error connecting to {host}:{port} — {exc}. '
f'Check the hostname and that the server is reachable.'
)
except Exception as exc:
return False, f'Unexpected error: {exc}'
# ---------------------------------------------------------------------------
# Email body builders
# ---------------------------------------------------------------------------
def _build_plain(hostname, cancelled, done_files, error_files, total) -> str:
lines = [
'VideoPress Compression Report',
f'Host : {hostname}',
f'Status : {"Cancelled" if cancelled else "Complete"}',
f'Files : {len(done_files)} succeeded, {len(error_files)} failed, {total} total',
'',
]
if done_files:
lines.append('Completed files:')
for r in done_files:
lines.append(
f" ✓ {r.get('filename','?')} "
f"({r.get('output_size_gb','?')} GB, "
f"-{r.get('reduction_pct','?')}%)"
)
lines.append('')
if error_files:
lines.append('Failed files:')
for r in error_files:
lines.append(
f" ✗ {r.get('filename','?')} "
f"— {r.get('message','unknown error')}"
)
lines.append('')
lines += ['—', 'Sent by VideoPress FFmpeg Compressor']
return '\n'.join(lines)
def _build_html(hostname, subject, cancelled, done_files, error_files, total) -> str:
status_colour = (
'#166534' if not cancelled and not error_files
else '#92400e' if cancelled
else '#991b1b'
)
status_label = (
'Cancelled' if cancelled
else 'Complete ✓' if not error_files
else 'Complete with errors'
)
def file_rows(files, icon, bg):
rows = ''
for r in files:
detail = (
f"{r.get('output_size_gb','?')} GB · "
f"-{r.get('reduction_pct','?')}%"
if r.get('status') == 'done'
else r.get('message', 'unknown error')
)
rows += (
f'