"""Author RAG Chatbot SaaS — Email Service (Gmail SMTP). All outgoing emails go through this service. RULE: Email sends are always queued as Celery tasks (non-blocking). RULE: Never call this directly from API handlers — use tasks/email_task.py. Templates are inline HTML for zero external dependency. """ import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText import structlog from app.config import get_settings logger = structlog.get_logger(__name__) cfg = get_settings() class EmailService: """Sends transactional emails via Gmail SMTP (STARTTLS on 587, SSL on 465).""" def _smtp_meta(self): """Return resolved from/support addresses for outgoing mail.""" from app.services.smtp_config_service import resolve_smtp_config_sync creds = resolve_smtp_config_sync() if creds: return creds.from_name, creds.from_address, creds.support_email return cfg.EMAIL_FROM_NAME, str(cfg.EMAIL_FROM_ADDRESS or cfg.SMTP_USERNAME or ""), str( cfg.SUPPORT_EMAIL or cfg.SMTP_USERNAME or "", ) def _send(self, to: str, subject: str, html_body: str, reply_to: str | None = None) -> None: """Send a single email. Args: to: Recipient email address. subject: Email subject line. html_body: Full HTML body content. reply_to: Optional Reply-To header for reader replies. """ msg = MIMEMultipart("alternative") msg["Subject"] = subject from_name, from_addr, _ = self._smtp_meta() msg["From"] = f"{from_name} <{from_addr}>" msg["To"] = to if reply_to: msg["Reply-To"] = reply_to msg.attach(MIMEText(html_body, "html")) from app.services.smtp_config_service import resolve_smtp_config_sync from app.services.smtp_transport import send_raw_message creds = resolve_smtp_config_sync() if not creds: raise smtplib.SMTPException("SMTP is not configured") try: send_raw_message(creds, from_addr, to, msg.as_string()) logger.info("Email sent", to=to, subject=subject) except smtplib.SMTPException as e: logger.error("Failed to send email", to=to, subject=subject, error=str(e)) raise def send_welcome(self, to: str, name: str) -> None: """Send welcome email after successful registration. Args: to: Author's email address. name: Author's display name. """ subject = "Welcome to AuthorBot! 🎉" body = _render_template("welcome", {"name": name}) self._send(to, subject, body) def send_account_locked(self, to: str) -> None: """Notify user their account has been temporarily locked. Args: to: Author's email address. """ subject = "AuthorBot: Account temporarily locked" body = _render_template("account_locked", {"minutes": cfg.LOCKOUT_DURATION_MINUTES}) self._send(to, subject, body) def send_subscription_granted(self, to: str, name: str, plan: str, expires_at: str) -> None: """Notify author that access has been granted. Args: to: Author's email address. name: Author's display name. plan: Subscription plan name. expires_at: Formatted expiry date string. """ subject = "Your AuthorBot chatbot is now active!" body = _render_template("subscription_granted", { "name": name, "plan": plan, "expires_at": expires_at }) self._send(to, subject, body) def send_subscription_revoked(self, to: str, name: str, reason: str) -> None: """Notify author that access has been revoked. Args: to: Author's email address. name: Author's display name. reason: Revocation reason. """ subject = "AuthorBot: Your chatbot access has been revoked" body = _render_template("subscription_revoked", {"name": name, "reason": reason}) self._send(to, subject, body) def send_subscription_expiry_warning( self, to: str, name: str, days_remaining: int, expires_at: str ) -> None: """Send expiry warning email (7 days and 1 day before). Args: to: Author's email address. name: Author's display name. days_remaining: Number of days until expiry. expires_at: Formatted expiry date string. """ subject = f"AuthorBot: Your subscription expires in {days_remaining} day(s)" body = _render_template("expiry_warning", { "name": name, "days_remaining": days_remaining, "expires_at": expires_at }) self._send(to, subject, body) def send_token_budget_warning( self, to: str, name: str, pct_used: int, tokens_used: int, tokens_total: int, forecast_date: str ) -> None: """Send token budget warning email. Args: to: Author's email address. name: Author's display name. pct_used: Percentage of budget consumed. tokens_used: Raw tokens consumed this month. tokens_total: Total monthly budget. forecast_date: Estimated date of exhaustion. """ subject = f"AuthorBot: You've used {pct_used}% of your monthly tokens" body = _render_template("token_warning", { "name": name, "pct_used": pct_used, "tokens_used": f"{tokens_used:,}", "tokens_total": f"{tokens_total:,}", "forecast_date": forecast_date, }) self._send(to, subject, body) def send_weekly_digest( self, to: str, name: str, chat_count: int, top_book: str, token_pct: int, top_country: str ) -> None: """Send weekly summary digest email. Args: to: Author's email address. name: Author's display name. chat_count: Total chats this week. top_book: Title of the most-engaged book. token_pct: Token usage percentage this month. top_country: Country with most visitors. """ subject = f"AuthorBot Weekly: {chat_count} chats this week 📚" body = _render_template("weekly_digest", { "name": name, "chat_count": chat_count, "top_book": top_book, "token_pct": token_pct, "top_country": top_country, }) self._send(to, subject, body) def send_publish_help_request( self, *, author_name: str, author_email: str, book_title: str, platform_names: list[str], message: str, submitted_at: str, book_id: str | None = None, isbn: str | None = None, book_author: str | None = None, source_url: str | None = None, ) -> None: """Send publish-help request to the support inbox. Args: author_name: Display name of the requesting author. author_email: Author's account email. book_title: Title of the book needing distribution help. platform_names: Human-readable platform names. message: Optional note from the author. submitted_at: ISO timestamp of the request. book_id: Optional book UUID. isbn: Optional ISBN. book_author: Optional book author name. source_url: Optional source listing URL. """ dest = self._smtp_meta()[2] or cfg.SUPPORT_EMAIL if not dest: raise ValueError("SUPPORT_EMAIL is not configured") subject = f"Publish help: {book_title} ({len(platform_names)} platform(s))" body = _render_template("publish_help", { "author_name": author_name, "author_email": author_email, "book_title": book_title, "book_id": book_id or "—", "isbn": isbn or "—", "book_author": book_author or "—", "source_url": source_url or "—", "platforms": ", ".join(platform_names), "message": message or "(no message)", "submitted_at": submitted_at, }) self._send(str(dest), subject, body) def send_contact_request( self, author_name: str, author_email: str, category: str, message: str, submitted_at: str, subject: str | None = None, ) -> None: """Send a general contact message to the support inbox.""" dest = self._smtp_meta()[2] or cfg.SUPPORT_EMAIL if not dest: raise ValueError("SUPPORT_EMAIL is not configured") mail_subject = subject or f"Contact: {category} — {author_name}" body = _render_template("contact_request", { "author_name": author_name, "author_email": author_email, "category": category, "subject": mail_subject, "message": message, "submitted_at": submitted_at, }) self._send(str(dest), mail_subject, body) def send_contact_reply( self, *, author_name: str, author_email: str, subject: str, original_message: str, reply_body: str, replied_at: str, ) -> None: """Notify an author that SuperAdmin replied to their support message.""" mail_subject = f"Re: {subject}" body = _render_template("contact_reply", { "author_name": author_name, "subject": subject, "original_message": original_message, "reply_body": reply_body, "replied_at": replied_at, }) self._send(author_email, mail_subject, body) def send_author_reader_email( self, *, to_email: str, reader_name: str | None, author_name: str, author_email: str, subject: str, message: str, sent_at: str, ) -> None: """Send a personal message from an author to a reader who opted in.""" import html greeting = reader_name or "there" body = _render_template("author_reader_email", { "greeting": html.escape(greeting), "author_name": html.escape(author_name), "message": html.escape(message), "sent_at": sent_at, }) self._send(to_email, subject, body, reply_to=author_email) def send_account_credentials( self, *, to: str, name: str, login_url: str, temp_password: str | None = None, user_chose_password: bool = False, ) -> None: """Send login credentials after Stripe payment or SuperAdmin support action.""" import html subject = "Your AuthorBot account is ready" body = _render_template("account_credentials", { "name": html.escape(name), "login_url": html.escape(login_url), "email": html.escape(to), "temp_password": html.escape(temp_password) if temp_password else None, "user_chose_password": user_chose_password, "forgot_url": html.escape(f"{login_url.rstrip('/login')}/reset-password"), }) self._send(to, subject, body) def send_payment_failed_notice(self, to: str, name: str, billing_url: str) -> None: """Notify author that a subscription payment failed.""" import html subject = "Action required: payment failed" body = _render_template("payment_failed", { "name": html.escape(name), "billing_url": html.escape(billing_url), }) self._send(to, subject, body) def send_subscription_cancelled_notice(self, to: str, name: str) -> None: """Notify author that their subscription was cancelled.""" import html subject = "Your AuthorBot subscription has ended" body = _render_template("subscription_cancelled_notice", { "name": html.escape(name), }) self._send(to, subject, body) def send_publish_help_confirmation( self, to: str, name: str, book_title: str, platform_names: list[str], ) -> None: """Confirm receipt of a publish-help request to the author. Args: to: Author's email address. name: Author's display name. book_title: Title of the book. platform_names: Platforms requested for help. """ subject = "We received your publishing help request" body = _render_template("publish_help_confirmation", { "name": name, "book_title": book_title, "platforms": ", ".join(platform_names), }) self._send(to, subject, body) # ─── Template Renderer ──────────────────────────────────────────────────────── def _render_template(template_name: str, context: dict) -> str: """Render an HTML email template with context variables. Args: template_name: Name of the template (maps to a function below). context: Dictionary of variables to inject. Returns: Rendered HTML string. """ templates = { "welcome": _tpl_welcome, "account_locked": _tpl_account_locked, "subscription_granted": _tpl_subscription_granted, "subscription_revoked": _tpl_subscription_revoked, "expiry_warning": _tpl_expiry_warning, "token_warning": _tpl_token_warning, "weekly_digest": _tpl_weekly_digest, "publish_help": _tpl_publish_help, "publish_help_confirmation": _tpl_publish_help_confirmation, "contact_request": _tpl_contact_request, "contact_reply": _tpl_contact_reply, "author_reader_email": _tpl_author_reader_email, "account_credentials": _tpl_account_credentials, "payment_failed": _tpl_payment_failed, "subscription_cancelled_notice": _tpl_subscription_cancelled_notice, } renderer = templates.get(template_name) if not renderer: raise ValueError(f"Unknown email template: {template_name}") return renderer(context) def _base_html(title: str, content: str) -> str: """Wrap content in a branded HTML email shell.""" return f"""
Your AuthorBot account is ready. Start by uploading your first book and configuring your chatbot.
Your chatbot will help readers discover and fall in love with your books — automatically.
""") def _tpl_account_locked(ctx: dict) -> str: return _base_html("Account Locked", f"""Too many failed login attempts. Your account is locked for {ctx['minutes']} minutes.
If this wasn't you, please contact support immediately.
""") def _tpl_subscription_granted(ctx: dict) -> str: return _base_html("Subscription Active", f"""Hi {ctx['name']}, your {ctx['plan']} subscription is now active.
Access expires: {ctx['expires_at']}
""") def _tpl_subscription_revoked(ctx: dict) -> str: return _base_html("Access Revoked", f"""Hi {ctx['name']}, your chatbot access has been revoked.
Reason: {ctx['reason']}
Please contact us to restore access.
""") def _tpl_expiry_warning(ctx: dict) -> str: return _base_html("Subscription Expiring Soon", f"""Hi {ctx['name']}, your subscription expires on {ctx['expires_at']}.
Contact us to renew and keep your chatbot running.
""") def _tpl_token_warning(ctx: dict) -> str: return _base_html("Token Budget Warning", f"""Hi {ctx['name']},
Usage: {ctx['tokens_used']} / {ctx['tokens_total']} tokens
At this pace, budget runs out around {ctx['forecast_date']}.
""") def _tpl_weekly_digest(ctx: dict) -> str: return _base_html("Weekly Digest", f"""Hi {ctx['name']}, here's what happened this week:
Author: {ctx['author_name']} <{ctx['author_email']}>
Book: {ctx['book_title']}
Book author: {ctx['book_author']}
ISBN: {ctx['isbn']}
Book ID: {ctx['book_id']}
Source URL: {ctx['source_url']}
Platforms: {ctx['platforms']}
Submitted: {ctx['submitted_at']}
{ctx['message']}
Hi {ctx['name']},
Thanks for reaching out about {ctx['book_title']}. Our team will help you publish on: {ctx['platforms']}.
We aim to reply within 24 hours.
""") def _tpl_contact_request(ctx: dict) -> str: return _base_html("Contact Request", f"""
Author: {ctx['author_name']} <{ctx['author_email']}>
Category: {ctx['category']}
Subject: {ctx.get('subject', '—')}
Submitted: {ctx['submitted_at']}
{ctx['message']}
Our team replied to your message about {ctx['subject']}.
{ctx['reply_body']}
{ctx['original_message']}
Log in to your AuthorBot dashboard to view this reply in the Messages tab.
""") def _tpl_author_reader_email(ctx: dict) -> str: return _base_html("Message from your author", f"""{ctx['author_name']} sent you a message via their book chatbot:
{ctx['message']}
Reply to this email to reach {ctx['author_name']} directly.
""") def _tpl_account_credentials(ctx: dict) -> str: pwd_block = "" if ctx.get("temp_password"): pwd_block = f"""
Email: {ctx['email']}
Temporary password: {ctx['temp_password']}
Please change this password after your first login.
""" elif ctx.get("user_chose_password"): pwd_block = f"""
Email: {ctx['email']}
Use the password you set during checkout.
Forgot it? Reset your password
""" else: pwd_block = f"""Email: {ctx['email']}
""" return _base_html("Your Account", f"""Your AuthorBot account is active. Sign in to upload books and configure your chatbot.
{pwd_block} """) def _tpl_payment_failed(ctx: dict) -> str: return _base_html("Payment Failed", f"""Hi {ctx['name']}, we could not process your latest subscription payment.
Update your payment method within 7 days to keep your chatbot running.
""") def _tpl_subscription_cancelled_notice(ctx: dict) -> str: return _base_html("Subscription Ended", f"""Hi {ctx['name']}, your AuthorBot subscription has been cancelled and your chatbot access has ended.
Contact support if you'd like to reactivate.
""")