"""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""" {title}

✦ AuthorBot

{content}
AuthorBot — Intelligent Book Advisor for Authors
You're receiving this because you have an AuthorBot account.
""" def _tpl_welcome(ctx: dict) -> str: return _base_html("Welcome to AuthorBot", f"""

Welcome, {ctx['name']}! 🎉

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"""

Account Temporarily Locked

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"""

Your chatbot is now active! ✓

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"""

Chatbot Access Revoked

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"""

Subscription Expiring in {ctx['days_remaining']} Day(s)

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"""

You've used {ctx['pct_used']}% of your tokens

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"""

Your week in review 📚

Hi {ctx['name']}, here's what happened this week:

""") def _tpl_publish_help(ctx: dict) -> str: return _base_html("Publish Help Request", f"""

New publishing help request

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']}

Message

{ctx['message']}

""") def _tpl_publish_help_confirmation(ctx: dict) -> str: return _base_html("Request Received", f"""

We received your request ✓

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"""

New contact message

Author: {ctx['author_name']} <{ctx['author_email']}>
Category: {ctx['category']}
Subject: {ctx.get('subject', '—')}
Submitted: {ctx['submitted_at']}

Message

{ctx['message']}

""") def _tpl_contact_reply(ctx: dict) -> str: return _base_html("Support Reply", f"""

Hi {ctx['author_name']},

Our team replied to your message about {ctx['subject']}.

Our reply

{ctx['reply_body']}

Your original message

{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"""

Hi {ctx['greeting']},

{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"""

Welcome, {ctx['name']}!

Your AuthorBot account is active. Sign in to upload books and configure your chatbot.

{pwd_block}

Sign in to your dashboard

""") def _tpl_payment_failed(ctx: dict) -> str: return _base_html("Payment Failed", f"""

Payment failed

Hi {ctx['name']}, we could not process your latest subscription payment.

Update your payment method within 7 days to keep your chatbot running.

Manage billing

""") def _tpl_subscription_cancelled_notice(ctx: dict) -> str: return _base_html("Subscription Ended", f"""

Subscription ended

Hi {ctx['name']}, your AuthorBot subscription has been cancelled and your chatbot access has ended.

Contact support if you'd like to reactivate.

""")