"""Branded HTML email templates that mirror the subscription Mini App look. The web app uses a dark theme with a configurable accent colour admin-configured accent colour and logo. The same accent + logo are reused here so emails feel like part of the product. All copy goes through the shared `JsonI18n` instance so translations live in ``locales/.json`` next to the rest of the bot strings. """ from __future__ import annotations import html import re from dataclasses import dataclass from typing import Optional, Sequence, Tuple from urllib.parse import urlsplit from bot.middlewares.i18n import JsonI18n, get_i18n_instance, normalize_locale_language_code from config.settings import Settings _BG = "#05070a" _CARD_BG = "#0e1116" _BORDER = "#1a1f27" _TEXT = "#e6e9ef" _TEXT_MUTED = "#9aa3b2" _TEXT_DIM = "#5d6573" _DEFAULT_ACCENT = "#00fe7a" _HEX_RE = re.compile(r"^#(?:[0-9a-fA-F]{3}|[0-9a-fA-F]{6})$") @dataclass(frozen=True) class EmailContent: subject: str text: str html: str def _safe_color(value: Optional[str]) -> str: if not value: return _DEFAULT_ACCENT candidate = value.strip() if _HEX_RE.match(candidate): return candidate return _DEFAULT_ACCENT def _public_logo_url(settings: Settings) -> Optional[str]: """Email recipients can't reach the in-app /webapp-logo proxy, so only a stored public https URL can be used directly. Anything else is dropped.""" if getattr(settings, "WEBAPP_LOGO_USE_EMOJI", False): return None raw = (settings.WEBAPP_LOGO_URL or "").strip() if not raw: return None parsed = urlsplit(raw) if parsed.scheme != "https" or not parsed.hostname: return None return raw def _brand_title(settings: Settings) -> str: title = (settings.WEBAPP_TITLE or "").strip() return title or "Subscription" def _normalize_lang(language_code: Optional[str], settings: Settings) -> str: return normalize_locale_language_code( language_code or settings.DEFAULT_LANGUAGE or "ru", prefer_known_base=False, ) def _resolve_i18n(i18n: Optional[JsonI18n]) -> JsonI18n: return i18n or get_i18n_instance() def _t_html(i18n: JsonI18n, lang: str, key: str, **kwargs) -> str: """Translate for HTML context: format args are HTML-escaped, the translated template itself is treated as already-safe HTML (locale files are author-controlled and may include simple inline tags like ).""" safe_kwargs = {k: html.escape(str(v)) for k, v in kwargs.items()} return i18n.gettext(lang, key, **safe_kwargs) def _t_text(i18n: JsonI18n, lang: str, key: str, **kwargs) -> str: return i18n.gettext(lang, key, **kwargs) def _layout( *, settings: Settings, preheader: str, heading: str, intro_html: str, body_html: str, footer_html: str, ) -> str: accent = _safe_color(settings.WEBAPP_PRIMARY_COLOR) brand_title = html.escape(_brand_title(settings)) logo_url = _public_logo_url(settings) logo_block = "" if logo_url: logo_block = ( f'' ) return f""" {html.escape(heading)}
{html.escape(preheader)}
{logo_block}
{brand_title}

{html.escape(heading)}

{intro_html}
{body_html}
{footer_html}
""" # noqa: E501 def _info_rows_html(rows: Sequence[Tuple[str, str]]) -> str: if not rows: return "" last = len(rows) - 1 cells = [] for index, (label, value) in enumerate(rows): border = "" if index == last else f"border-bottom:1px solid {_BORDER};" cells.append( f"" f'{html.escape(label)}' # noqa: E501 f"{html.escape(value)}" # noqa: E501 f"" ) return ( f'' # noqa: E501 + "".join(cells) + "
" ) def _cta_button_html(*, label: str, url: str, accent: str) -> str: safe_label = html.escape(label) safe_url = html.escape(url, quote=True) # Accent green is light, so contrast text is dark; works for the default and similar light accents. # noqa: E501 return ( f'' f'
' # noqa: E501 f'{safe_label}' f"
" ) def _format_amount(amount: float, currency: str) -> str: rounded = round(float(amount), 2) if rounded.is_integer(): body = f"{int(rounded)}" else: body = f"{rounded:.2f}" suffix = (currency or "").strip() return f"{body} {suffix}".strip() def _format_traffic(traffic_gb: Optional[float]) -> str: if traffic_gb is None: return "—" value = float(traffic_gb) return str(int(value)) if value.is_integer() else f"{value:g}" def _format_minutes(seconds: int) -> int: return max(1, int(seconds) // 60) def render_login_code( settings: Settings, *, code: str, language_code: Optional[str], magic_link: Optional[str] = None, purpose: str = "login", i18n: Optional[JsonI18n] = None, ) -> EmailContent: i18n = _resolve_i18n(i18n) lang = _normalize_lang(language_code, settings) minutes = _format_minutes(settings.EMAIL_CODE_TTL_SECONDS) accent = _safe_color(settings.WEBAPP_PRIMARY_COLOR) brand = _brand_title(settings) template_prefix = "email_set_password_code" if purpose == "set_password" else "email_login_code" safe_magic_link = (magic_link or "").strip() if template_prefix == "email_login_code" else "" subject = _t_text(i18n, lang, f"{template_prefix}_subject", code=code) preheader = _t_text(i18n, lang, f"{template_prefix}_preheader", minutes=minutes) heading = _t_text(i18n, lang, f"{template_prefix}_heading") intro = _t_text(i18n, lang, f"{template_prefix}_intro") expiry_html = _t_html(i18n, lang, f"{template_prefix}_expiry_html", minutes=minutes) security = _t_text(i18n, lang, f"{template_prefix}_security") footer = _t_html(i18n, lang, "email_footer_auto", brand=brand) text_lines = [_t_text(i18n, lang, f"{template_prefix}_text", code=code, minutes=minutes)] if safe_magic_link: text_lines.append(_t_text(i18n, lang, "email_login_code_text_magic", url=safe_magic_link)) code_block = ( f'' # noqa: E501 f'
' # noqa: E501 f"
" # noqa: E501 f"{html.escape(code)}" f"
" ) magic_block = "" if safe_magic_link: cta_label = _t_text(i18n, lang, "email_login_code_magic_cta") divider_label = _t_text(i18n, lang, "email_login_code_magic_or") magic_intro = _t_text(i18n, lang, "email_login_code_magic_intro") magic_hint = _t_text(i18n, lang, "email_login_code_magic_hint") divider_html = ( f'' # noqa: E501 f"" f'' # noqa: E501 f'' # noqa: E501 f'' # noqa: E501 f"
 {html.escape(divider_label)} 
" ) magic_block = ( divider_html + f'

{html.escape(magic_intro)}

' # noqa: E501 + _cta_button_html(label=cta_label, url=safe_magic_link, accent=accent) + f'

{html.escape(magic_hint)}

' # noqa: E501 ) body_html = ( code_block + f'

{expiry_html}

' # noqa: E501 + f'

{html.escape(security)}

' # noqa: E501 + magic_block ) rendered = _layout( settings=settings, preheader=preheader, heading=heading, intro_html=html.escape(intro), body_html=body_html, footer_html=footer, ) return EmailContent(subject=subject, text="\n".join(text_lines), html=rendered) def render_account_merged( settings: Settings, *, language_code: Optional[str], primary_user_id: Optional[int], removed_user_id: Optional[int], final_end_date_text: str, i18n: Optional[JsonI18n] = None, ) -> EmailContent: i18n = _resolve_i18n(i18n) lang = _normalize_lang(language_code, settings) brand = _brand_title(settings) primary = "—" if primary_user_id is None else f"#{primary_user_id}" removed = "—" if removed_user_id is None else f"#{removed_user_id}" end_date = final_end_date_text or "—" subject = _t_text(i18n, lang, "email_account_merged_subject") preheader = _t_text(i18n, lang, "email_account_merged_preheader") heading = _t_text(i18n, lang, "email_account_merged_heading") intro = _t_text(i18n, lang, "email_account_merged_intro") note = _t_text(i18n, lang, "email_account_merged_note") footer = _t_html(i18n, lang, "email_footer_auto", brand=brand) text = _t_text( i18n, lang, "email_account_merged_text", primary=primary, removed=removed, end_date=end_date, ) rows = [ (_t_text(i18n, lang, "email_account_merged_row_kept"), primary), (_t_text(i18n, lang, "email_account_merged_row_removed"), removed), (_t_text(i18n, lang, "email_account_merged_row_end_date"), end_date), ] body_html = ( _info_rows_html(rows) + f'

{html.escape(note)}

' # noqa: E501 ) rendered = _layout( settings=settings, preheader=preheader, heading=heading, intro_html=html.escape(intro), body_html=body_html, footer_html=footer, ) return EmailContent(subject=subject, text=text, html=rendered) def render_payment_success( settings: Settings, *, language_code: Optional[str], sale_mode: str, months: int, traffic_gb: Optional[float], amount: float, currency: str, end_date_text: str, dashboard_url: Optional[str], provider_label: Optional[str] = None, i18n: Optional[JsonI18n] = None, ) -> EmailContent: i18n = _resolve_i18n(i18n) lang = _normalize_lang(language_code, settings) accent = _safe_color(settings.WEBAPP_PRIMARY_COLOR) brand = _brand_title(settings) is_traffic = (sale_mode or "").split("@", 1)[0].split("|", 1)[0] in { "traffic", "traffic_package", "topup", "premium_topup", } amount_text = _format_amount(amount, currency) safe_dashboard_url = (dashboard_url or "").strip() end_date = end_date_text or "—" traffic_label = _format_traffic(traffic_gb) subject = _t_text(i18n, lang, "email_payment_success_subject") preheader = _t_text(i18n, lang, "email_payment_success_preheader") heading = _t_text(i18n, lang, "email_payment_success_heading") footer_note = _t_text(i18n, lang, "email_payment_success_footer_note") footer = _t_html(i18n, lang, "email_footer_auto", brand=brand) cta_label = _t_text(i18n, lang, "email_payment_success_cta") if is_traffic: intro = _t_text(i18n, lang, "email_payment_success_intro_traffic", traffic_gb=traffic_label) period_label = _t_text(i18n, lang, "email_payment_success_row_traffic") period_value = _t_text( i18n, lang, "email_payment_success_traffic_value", traffic_gb=traffic_label ) text = _t_text( i18n, lang, "email_payment_success_text_traffic", amount=amount_text, traffic_gb=traffic_label, end_date=end_date, ) else: months_int = int(months or 0) intro = _t_text(i18n, lang, "email_payment_success_intro_subscription", months=months_int) period_label = _t_text(i18n, lang, "email_payment_success_row_period") period_value = _t_text( i18n, lang, "email_payment_success_period_value", months=months_int, ) text = _t_text( i18n, lang, "email_payment_success_text_subscription", amount=amount_text, months=months_int, end_date=end_date, ) rows: list[Tuple[str, str]] = [ (period_label, period_value), (_t_text(i18n, lang, "email_payment_success_row_amount"), amount_text), (_t_text(i18n, lang, "email_payment_success_row_end_date"), end_date), ] if provider_label: rows.append((_t_text(i18n, lang, "email_payment_success_row_method"), provider_label)) text_lines = [text] if safe_dashboard_url: text_lines.append( _t_text(i18n, lang, "email_payment_success_text_dashboard", url=safe_dashboard_url) ) body_parts = [_info_rows_html(rows)] if safe_dashboard_url: body_parts.append(_cta_button_html(label=cta_label, url=safe_dashboard_url, accent=accent)) body_parts.append( f'

{html.escape(footer_note)}

' # noqa: E501 ) rendered = _layout( settings=settings, preheader=preheader, heading=heading, intro_html=html.escape(intro), body_html="".join(body_parts), footer_html=footer, ) return EmailContent(subject=subject, text="\n".join(text_lines), html=rendered) def render_subscription_expiring( settings: Settings, *, language_code: Optional[str], days_left: int, end_date_text: str, dashboard_url: Optional[str], i18n: Optional[JsonI18n] = None, ) -> EmailContent: i18n = _resolve_i18n(i18n) lang = _normalize_lang(language_code, settings) accent = _safe_color(settings.WEBAPP_PRIMARY_COLOR) brand = _brand_title(settings) safe_dashboard_url = (dashboard_url or "").strip() days = max(0, int(days_left)) end_date = end_date_text or "—" if days == 0: suffix = "today" elif days == 1: suffix = "tomorrow" else: suffix = "days" subject = _t_text(i18n, lang, f"email_subscription_expiring_subject_{suffix}", days=days) heading = _t_text(i18n, lang, f"email_subscription_expiring_heading_{suffix}", days=days) preheader = _t_text(i18n, lang, f"email_subscription_expiring_preheader_{suffix}", days=days) intro = _t_text(i18n, lang, f"email_subscription_expiring_intro_{suffix}", days=days) note = _t_text(i18n, lang, "email_subscription_expiring_note") footer = _t_html(i18n, lang, "email_footer_auto", brand=brand) cta_label = _t_text(i18n, lang, "email_subscription_expiring_cta") rows = [ (_t_text(i18n, lang, "email_subscription_expiring_row_days_left"), str(days)), (_t_text(i18n, lang, "email_subscription_expiring_row_end_date"), end_date), ] text_lines = [ _t_text(i18n, lang, "email_subscription_expiring_text", heading=heading, end_date=end_date), ] if safe_dashboard_url: text_lines.append( _t_text(i18n, lang, "email_subscription_expiring_text_renew", url=safe_dashboard_url) ) body_parts = [_info_rows_html(rows)] if safe_dashboard_url: body_parts.append(_cta_button_html(label=cta_label, url=safe_dashboard_url, accent=accent)) body_parts.append( f'

{html.escape(note)}

' # noqa: E501 ) rendered = _layout( settings=settings, preheader=preheader, heading=heading, intro_html=html.escape(intro), body_html="".join(body_parts), footer_html=footer, ) return EmailContent(subject=subject, text="\n".join(text_lines), html=rendered) def _subscription_lifecycle_title( i18n: JsonI18n, lang: str, notification_key: str, *, days_left: Optional[int], hours_before: Optional[int], ) -> str: if notification_key == "before_2d_autorenew": return _t_text(i18n, lang, "email_subscription_lifecycle_subject_autorenew") if notification_key == "expired": return _t_text(i18n, lang, "email_subscription_lifecycle_subject_expired") if notification_key == "expired_24h_after": return _t_text(i18n, lang, "email_subscription_lifecycle_subject_expired_after") if hours_before is not None: return _t_text( i18n, lang, "email_subscription_lifecycle_subject_before_hours", hours=hours_before, ) return _t_text( i18n, lang, "email_subscription_lifecycle_subject_before_days", days=max(0, int(days_left or 0)), ) def render_subscription_lifecycle_notification( settings: Settings, *, language_code: Optional[str], notification_key: str, message_text: str, end_date_text: str, dashboard_url: Optional[str], mirrored_from_telegram: bool = False, days_left: Optional[int] = None, hours_before: Optional[int] = None, i18n: Optional[JsonI18n] = None, ) -> EmailContent: i18n = _resolve_i18n(i18n) lang = _normalize_lang(language_code, settings) accent = _safe_color(settings.WEBAPP_PRIMARY_COLOR) brand = _brand_title(settings) safe_dashboard_url = (dashboard_url or "").strip() end_date = end_date_text or "—" subject = _subscription_lifecycle_title( i18n, lang, notification_key, days_left=days_left, hours_before=hours_before, ) intro_key = ( "email_subscription_lifecycle_intro_mirrored" if mirrored_from_telegram else "email_subscription_lifecycle_intro_direct" ) intro = _t_text(i18n, lang, intro_key) footer = _t_html(i18n, lang, "email_footer_auto", brand=brand) cta_label = _t_text(i18n, lang, "email_subscription_lifecycle_cta") rows = [ (_t_text(i18n, lang, "email_subscription_lifecycle_row_end_date"), end_date), ] message_html = ( f'
{html.escape(message_text or "")}
' ) body_parts = [_info_rows_html(rows), message_html] if safe_dashboard_url: body_parts.append(_cta_button_html(label=cta_label, url=safe_dashboard_url, accent=accent)) rendered = _layout( settings=settings, preheader=subject, heading=subject, intro_html=html.escape(intro), body_html="".join(body_parts), footer_html=footer, ) text_lines = [subject, "", message_text] if safe_dashboard_url: text_lines.extend( [ "", _t_text( i18n, lang, "email_subscription_lifecycle_text_renew", url=safe_dashboard_url, ), ] ) return EmailContent(subject=subject, text="\n".join(text_lines), html=rendered) def _support_email( settings: Settings, i18n: Optional[JsonI18n], language: Optional[str], *, subject: str, heading: str, intro: str, rows: Sequence[Tuple[str, str]], body_preview: str, ticket_url: Optional[str], cta_label: str, ) -> EmailContent: lang = _normalize_lang(language, settings) brand = _brand_title(settings) accent = _safe_color(settings.WEBAPP_PRIMARY_COLOR) safe_url = (ticket_url or "").strip() footer = _t_html(_resolve_i18n(i18n), lang, "email_footer_auto", brand=brand) preview_block = ( f'
{html.escape(body_preview or "")}
' ) body_parts = [_info_rows_html(rows), preview_block] if safe_url: body_parts.append(_cta_button_html(label=cta_label, url=safe_url, accent=accent)) rendered = _layout( settings=settings, preheader=intro, heading=heading, intro_html=html.escape(intro), body_html="".join(body_parts), footer_html=footer, ) text_lines = [intro, "", *[f"{label}: {value}" for label, value in rows], "", body_preview] if safe_url: text_lines.extend(["", safe_url]) return EmailContent(subject=subject, text="\n".join(text_lines), html=rendered) def render_support_new_ticket_admin( settings: Settings, i18n: Optional[JsonI18n], language: Optional[str], *, ticket_id: int, user_display: str, subject: str, body_preview: str, snapshot_rows: Sequence[Tuple[str, str]], ticket_url: Optional[str], ) -> EmailContent: rows = [ ("Ticket", f"#{ticket_id}"), ("User", user_display), ("Subject", subject), *snapshot_rows, ] return _support_email( settings, i18n, language, subject=f"New support ticket #{ticket_id}", heading=f"New support ticket #{ticket_id}", intro="A user opened a new support ticket.", rows=rows, body_preview=body_preview, ticket_url=ticket_url, cta_label="Open ticket", ) def render_support_user_reply_admin( settings: Settings, i18n: Optional[JsonI18n], language: Optional[str], *, ticket_id: int, user_display: str, subject: str, body_preview: str, snapshot_rows: Sequence[Tuple[str, str]], ticket_url: Optional[str], ) -> EmailContent: rows = [ ("Ticket", f"#{ticket_id}"), ("User", user_display), ("Subject", subject), *snapshot_rows, ] return _support_email( settings, i18n, language, subject=f"New user reply in ticket #{ticket_id}", heading=f"User replied in ticket #{ticket_id}", intro="A user sent a new support message.", rows=rows, body_preview=body_preview, ticket_url=ticket_url, cta_label="Open ticket", ) def render_support_admin_reply_user( settings: Settings, i18n: Optional[JsonI18n], language: Optional[str], *, ticket_id: int, subject: str, body_preview: str, ticket_url: Optional[str], ) -> EmailContent: return _support_email( settings, i18n, language, subject=f"New reply for ticket #{ticket_id}", heading=f"New reply for ticket #{ticket_id}", intro="Support has replied to your ticket.", rows=[("Ticket", f"#{ticket_id}"), ("Subject", subject)], body_preview=body_preview, ticket_url=ticket_url, cta_label="Open in Mini App", ) def render_support_ticket_closed_user( settings: Settings, i18n: Optional[JsonI18n], language: Optional[str], *, ticket_id: int, subject: str, body_preview: str = "", ticket_url: Optional[str], ) -> EmailContent: return _support_email( settings, i18n, language, subject=f"Ticket #{ticket_id} was closed", heading=f"Ticket #{ticket_id} was closed", intro="Your support ticket has been closed.", rows=[("Ticket", f"#{ticket_id}"), ("Subject", subject)], body_preview=body_preview or "The ticket is closed.", ticket_url=ticket_url, cta_label="Open in Mini App", )