"""Configurable email notifications for support tickets (SMTP, off by default).""" from __future__ import annotations import json import logging import os import smtplib from email.message import EmailMessage from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) SETTING_KEY = "ticket_email" PASSWORD_MASK = "********" DEFAULT_CONFIG: Dict[str, Any] = { "enabled": False, "provider": "smtp", "smtp_host": "", "smtp_port": 587, "smtp_user": "", "smtp_password": "", "smtp_use_tls": True, "from_email": "", "from_name": "PCA Support", "support_inbox_email": "", "reply_to_email": "", "events": { "ticket_created": True, "ticket_assigned": True, "ticket_resolved": False, "customer_reply": False, "staff_reply": False, "staff_login": False, }, } def _table_exists(cursor, table_name: str) -> bool: cursor.execute( """ SELECT 1 FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = %s LIMIT 1 """, (table_name,), ) return cursor.fetchone() is not None def ensure_notification_settings_schema(cursor) -> None: cursor.execute( """ CREATE TABLE IF NOT EXISTS pcaa_notification_settings ( setting_key VARCHAR(64) NOT NULL PRIMARY KEY, config_json JSON NOT NULL, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, updated_by VARCHAR(100) NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ ) def _merge_defaults(raw: Optional[Dict[str, Any]]) -> Dict[str, Any]: base = json.loads(json.dumps(DEFAULT_CONFIG)) if not raw: return base for key, value in raw.items(): if key == "events" and isinstance(value, dict): base["events"].update(value) else: base[key] = value return base def _env_fallback() -> Dict[str, Any]: host = str(os.getenv("TICKET_SMTP_HOST", "") or "").strip() if not host: return {} return { "enabled": str(os.getenv("TICKET_EMAIL_ENABLED", "")).lower() in ("1", "true", "yes"), "smtp_host": host, "smtp_port": int(os.getenv("TICKET_SMTP_PORT", "587") or 587), "smtp_user": str(os.getenv("TICKET_SMTP_USER", "") or "").strip(), "smtp_password": str(os.getenv("TICKET_SMTP_PASSWORD", "") or "").strip(), "smtp_use_tls": str(os.getenv("TICKET_SMTP_USE_TLS", "true")).lower() not in ("0", "false", "no"), "from_email": str(os.getenv("TICKET_FROM_EMAIL", "") or "").strip(), "from_name": str(os.getenv("TICKET_FROM_NAME", "PCA Support") or "PCA Support").strip(), "support_inbox_email": str(os.getenv("TICKET_SUPPORT_INBOX", "") or "").strip(), "reply_to_email": str(os.getenv("TICKET_REPLY_TO", "") or "").strip(), } def get_ticket_email_config(cursor, *, include_secrets: bool = False) -> Dict[str, Any]: ensure_notification_settings_schema(cursor) stored: Dict[str, Any] = {} if _table_exists(cursor, "pcaa_notification_settings"): cursor.execute( "SELECT config_json FROM pcaa_notification_settings WHERE setting_key = %s LIMIT 1", (SETTING_KEY,), ) row = cursor.fetchone() if row and row.get("config_json"): raw = row["config_json"] stored = raw if isinstance(raw, dict) else json.loads(raw) config = _merge_defaults(stored) env = _env_fallback() for key, value in env.items(): if value not in (None, "", False) and not config.get(key): config[key] = value if include_secrets: return config masked = dict(config) if masked.get("smtp_password"): masked["smtp_password"] = PASSWORD_MASK masked["has_smtp_password"] = True else: masked["has_smtp_password"] = False return masked def save_ticket_email_config(cursor, payload: Dict[str, Any], updated_by: Optional[str] = None) -> Dict[str, Any]: ensure_notification_settings_schema(cursor) current = get_ticket_email_config(cursor, include_secrets=True) incoming = _merge_defaults(payload or {}) password = str(incoming.get("smtp_password") or "").strip() if not password or password == PASSWORD_MASK: incoming["smtp_password"] = current.get("smtp_password") or "" config = _merge_defaults(incoming) if config.get("enabled"): events = dict(config.get("events") or {}) recommended = ("ticket_created", "ticket_assigned", "customer_reply", "staff_login") if not any(events.get(key) for key in recommended): for key in recommended: events[key] = True config["events"] = events cursor.execute( """ INSERT INTO pcaa_notification_settings (setting_key, config_json, updated_by) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE config_json = VALUES(config_json), updated_by = VALUES(updated_by) """, (SETTING_KEY, json.dumps(config), updated_by), ) return get_ticket_email_config(cursor, include_secrets=False) def _smtp_send(config: Dict[str, Any], to_addrs: List[str], subject: str, body: str) -> None: recipients = [str(a).strip() for a in to_addrs if str(a or "").strip()] if not recipients: raise ValueError("No recipients") host = str(config.get("smtp_host") or "").strip() if not host: raise ValueError("SMTP host is not configured") port = int(config.get("smtp_port") or 587) user = str(config.get("smtp_user") or "").strip() password = str(config.get("smtp_password") or "").strip() use_tls = bool(config.get("smtp_use_tls", True)) from_email = str(config.get("from_email") or user or "").strip() if not from_email: raise ValueError("From email is not configured") from_name = str(config.get("from_name") or "PCA Support").strip() reply_to = str(config.get("reply_to_email") or "").strip() msg = EmailMessage() msg["Subject"] = subject msg["From"] = f"{from_name} <{from_email}>" if from_name else from_email msg["To"] = ", ".join(recipients) if reply_to: msg["Reply-To"] = reply_to msg.set_content(body) with smtplib.SMTP(host, port, timeout=30) as smtp: if use_tls: smtp.starttls() if user and password: smtp.login(user, password) smtp.send_message(msg) def send_test_email(cursor, to_email: str) -> Dict[str, Any]: config = get_ticket_email_config(cursor, include_secrets=True) if not config.get("enabled"): raise ValueError("Ticket email notifications are disabled. Enable them first.") _smtp_send( config, [to_email], "PCA Support — test email", "This is a test email from the PCA support ticket notification system.", ) return {"success": True, "sent_to": to_email} def _event_enabled(config: Dict[str, Any], event: str) -> bool: if not config.get("enabled"): return False events = config.get("events") or {} return bool(events.get(event)) def _ticket_label(ticket: Dict[str, Any]) -> str: return str(ticket.get("ticket_number") or f"#{ticket.get('id')}") def try_send_ticket_email( cursor, event: str, ticket: Dict[str, Any], *, assignee_email: Optional[str] = None, message_preview: Optional[str] = None, ) -> None: try: config = get_ticket_email_config(cursor, include_secrets=True) if not _event_enabled(config, event): return label = _ticket_label(ticket) bid = ticket.get("business_id") subject = ticket.get("subject") or ticket.get("issue_text") or "Support ticket" body_lines = [ f"Ticket: {label}", f"Business: {bid} — {ticket.get('business_name') or ''}", f"Subject: {subject}", f"Status: {ticket.get('status')}", ] if ticket.get("assigned_to"): body_lines.append(f"Assigned to: {ticket.get('assigned_to')}") if ticket.get("due_at"): body_lines.append(f"Due: {ticket.get('due_at')}") if message_preview: body_lines.extend(["", "Message:", str(message_preview).strip()]) body = "\n".join(body_lines) mail_subject = f"[PCA] {event.replace('_', ' ').title()} — {label}" if event == "ticket_created": mail_subject = f"[PCA] New support ticket — {label}" body = ( f"A customer raised a new support ticket.\n\n" f"{body}\n\n" "Open PCA Master → Support to review and assign it." ) elif event == "ticket_assigned": mail_subject = f"[PCA] New assignment — {label}" body = ( f"You have been assigned a support ticket.\n\n" f"{body}\n\n" "Open PCA Master → Support → My assigned tickets to view it." ) recipients: List[str] = [] if event == "ticket_created": inbox = str(config.get("support_inbox_email") or "").strip() if inbox: recipients.append(inbox) else: logger.info( "ticket email skipped event=ticket_created ticket=%s (support_inbox_email not set)", label, ) return elif event == "ticket_assigned": email = str(assignee_email or "").strip() if email: recipients.append(email) else: logger.info( "ticket email skipped event=ticket_assigned ticket=%s (assignee has no email)", label, ) return elif event == "ticket_resolved": customer = str(ticket.get("user_email") or "").strip() if customer: recipients.append(customer) elif event == "customer_reply": inbox = str(config.get("support_inbox_email") or "").strip() if inbox: recipients.append(inbox) if assignee_email: recipients.append(assignee_email) elif event == "staff_reply": customer = str(ticket.get("user_email") or "").strip() if customer: recipients.append(customer) recipients = list(dict.fromkeys([r for r in recipients if r])) if not recipients: logger.info("ticket email skipped event=%s ticket=%s (no recipients)", event, label) return _smtp_send(config, recipients, mail_subject, body) logger.info("ticket email sent event=%s ticket=%s to=%s", event, label, recipients) except Exception as exc: logger.warning("ticket email failed event=%s ticket=%s: %s", event, ticket.get("id"), exc) def _assigned_active_tickets(cursor, user_id: int) -> List[Dict[str, Any]]: cursor.execute( """ SELECT ticket_number, business_id, subject, issue_text, status, due_at FROM reported_issues WHERE assigned_to_user_id = %s AND status IN ('open', 'in_progress') ORDER BY due_at ASC, created_at DESC LIMIT 25 """, (int(user_id),), ) return list(cursor.fetchall() or []) def _is_staff_login_email_target(user: Dict[str, Any]) -> bool: if not user: return False if user.get("is_master") in (True, 1, "1"): return True if user.get("has_master_panel") not in (True, 1, "1"): return False team = str(user.get("staff_team") or "").strip().lower() return team in ("developer", "support") def try_send_staff_login_email(cursor, user: Dict[str, Any]) -> None: """Email developer/support staff their assigned ticket digest on login.""" try: if not _is_staff_login_email_target(user): return email = str(user.get("email") or "").strip() user_id = user.get("id") if not email or not user_id: return config = get_ticket_email_config(cursor, include_secrets=True) if not _event_enabled(config, "staff_login"): return username = ( str(user.get("full_name") or "").strip() or str(user.get("username") or "").strip() or "there" ) tickets = _assigned_active_tickets(cursor, int(user_id)) lines = [ f"Hello {username},", "", "You logged in to the PCA master panel.", "", ] if tickets: lines.append(f"You have {len(tickets)} assigned ticket(s) open:") lines.append("") for row in tickets: label = row.get("ticket_number") or "Ticket" subj = row.get("subject") or row.get("issue_text") or "" subj = str(subj).strip().replace("\n", " ")[:100] due = row.get("due_at") due_txt = f" · due {due}" if due else "" lines.append(f"- {label} · BID {row.get('business_id')} · {subj}{due_txt}") else: lines.append("You have no open assigned tickets right now.") lines.extend(["", "— PCA Support"]) body = "\n".join(lines) subject = f"[PCA] Login digest — {len(tickets)} assigned ticket(s)" _smtp_send(config, [email], subject, body) logger.info("staff login email sent to %s (%s tickets)", email, len(tickets)) except Exception as exc: logger.warning("staff login email failed for user %s: %s", user.get("id"), exc)