"""SMTP-рассылка важных уведомлений пользователям."""
from __future__ import annotations
import asyncio
import html as html_lib
import logging
import re
import smtplib
from email.message import EmailMessage
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from config import Settings
from services.db import Database
logger = logging.getLogger(__name__)
_TAG_RE = re.compile(r"<[^>]+>")
_WS_RE = re.compile(r"[ \t]+\n")
# Фирменные цвета VPN Service (email-safe, без «фиолетового AI»)
_ACCENT = "#0d9f6e"
_ACCENT_DARK = "#0a7a54"
_BG = "#eef3f0"
_CARD = "#ffffff"
_TEXT = "#12201a"
_MUTED = "#5c7167"
_LINE = "#d5e0da"
_WARN_BG = "#fff8e8"
_WARN_BORDER = "#e8c96a"
_WARN_TEXT = "#6b5420"
def html_to_plain(html: str) -> str:
text = (html or "").replace("
", "\n").replace("
", "\n").replace("
", "\n")
text = text.replace("", "\n").replace("", "\n").replace("", "\n")
text = _TAG_RE.sub("", text)
text = (
text.replace(" ", " ")
.replace("<", "<")
.replace(">", ">")
.replace("&", "&")
.replace(""", '"')
)
text = _WS_RE.sub("\n", text)
return text.strip()
def _e(value: str) -> str:
return html_lib.escape(value or "", quote=True)
def smtp_configured(settings: Settings) -> bool:
return bool((settings.smtp_host or "").strip() and (settings.smtp_from or "").strip())
def render_email_layout(
*,
title: str,
body_html: str,
preheader: str = "",
footer_note: str = "",
webapp_url: str = "",
cta_url: str = "",
cta_label: str = "",
) -> str:
"""Брендированная HTML-обёртка для писем (таблицы + inline CSS)."""
title_e = _e(title)
pre_e = _e(preheader or title)
footer_e = _e(
footer_note
or "Это письмо отправлено автоматически. Отвечать на него не нужно."
)
brand_url = (webapp_url or "").rstrip("/")
brand_link = (
f''
f"Открыть кабинет"
if brand_url
else ""
)
cta_block = ""
if cta_url and cta_label:
cta_block = f"""
|
|
"""
return f"""
{title_e}
{pre_e}
|
🦊 VPN Service
|
|
Безопасный доступ · личный кабинет
|
|
{title_e}
{body_html}
|
{cta_block}
{footer_e}
{"
" + brand_link if brand_link else ""}
|
|
© VPN Service
|
"""
def render_info_box(html_inner: str, *, kind: str = "info") -> str:
if kind == "warn":
bg, border, color = _WARN_BG, _WARN_BORDER, _WARN_TEXT
else:
bg, border, color = "#f0faf5", _LINE, _MUTED
return f"""
"""
def render_password_reset_email(*, reset_url: str, webapp_url: str = "") -> str:
body = f"""
Мы получили запрос на сброс пароля для вашего аккаунта VPN Service.
Нажмите кнопку ниже, чтобы задать новый пароль. Ссылка действует 1 час.
{render_info_box(
"Безопасность. Если вы не запрашивали сброс — просто проигнорируйте это письмо. "
"Пароль не изменится, пока вы сами не подтвердите действие по ссылке.",
kind="warn",
)}
Кнопка не работает? Скопируйте ссылку в браузер:
{_e(reset_url)}
"""
return render_email_layout(
title="Сброс пароля",
preheader="Ссылка для сброса пароля действует 1 час",
body_html=body,
cta_url=reset_url,
cta_label="Сбросить пароль",
webapp_url=webapp_url,
footer_note="Письмо отправлено, потому что для аккаунта указан этот email.",
)
def render_smtp_test_email(*, webapp_url: str = "") -> str:
body = f"""
Поздравляем — SMTP настроен правильно.
Это тестовое письмо из админ-панели VPN Service. Теперь можно отправлять
сброс пароля и важные уведомления пользователям.
{render_info_box(
"Что дальше. Убедитесь, что у пользователей сохранён email в кабинете, "
"и проверьте папку «Спам», если письмо не видно во входящих.",
kind="info",
)}
"""
return render_email_layout(
title="SMTP работает",
preheader="Тестовое письмо — почта настроена верно",
body_html=body,
webapp_url=webapp_url,
cta_url=(webapp_url or "").rstrip("/") + "/admin" if webapp_url else "",
cta_label="Открыть админку" if webapp_url else "",
footer_note="Тест из раздела Админка → SMTP.",
)
def render_notification_email(
*,
title: str,
body_html: str,
webapp_url: str = "",
preheader: str = "",
) -> str:
"""Обернуть произвольное HTML-уведомление в фирменный шаблон."""
return render_email_layout(
title=title,
preheader=preheader or title,
body_html=f'{body_html}
',
webapp_url=webapp_url,
footer_note="Уведомление аккаунта VPN Service. Отключить email можно в кабинете → Аккаунт.",
)
def send_email_sync(
settings: Settings,
*,
to: str,
subject: str,
html: str,
plain: str | None = None,
) -> None:
host = (settings.smtp_host or "").strip()
from_addr = (settings.smtp_from or "").strip()
to_addr = (to or "").strip()
if not host or not from_addr or not to_addr:
raise ValueError("SMTP не настроен или нет получателя")
msg = EmailMessage()
msg["Subject"] = subject
msg["From"] = from_addr
msg["To"] = to_addr
body_plain = plain if plain is not None else html_to_plain(html)
msg.set_content(body_plain or subject)
if html.strip():
msg.add_alternative(html, subtype="html")
port = int(settings.smtp_port or 587)
user = (settings.smtp_user or "").strip()
password = settings.smtp_password or ""
use_ssl = bool(settings.smtp_ssl)
use_tls = bool(settings.smtp_tls) and not use_ssl
if use_ssl:
with smtplib.SMTP_SSL(host, port, timeout=30) as smtp:
if user:
smtp.login(user, password)
smtp.send_message(msg)
return
with smtplib.SMTP(host, port, timeout=30) as smtp:
smtp.ehlo()
if use_tls:
smtp.starttls()
smtp.ehlo()
if user:
smtp.login(user, password)
smtp.send_message(msg)
async def send_email(
settings: Settings,
*,
to: str,
subject: str,
html: str,
plain: str | None = None,
) -> None:
await asyncio.to_thread(
send_email_sync,
settings,
to=to,
subject=subject,
html=html,
plain=plain,
)
async def notify_important(
*,
db: Database,
settings: Settings,
telegram_id: int,
html: str,
subject: str,
bot=None,
) -> None:
"""Telegram (если есть) + email (если SMTP и email пользователя заданы)."""
user = db.get_user(telegram_id)
if user and user.banned:
return
want_tg = True if not user else bool(user.notify_telegram)
want_email = True if not user else bool(user.notify_email)
if want_tg and bot is not None and telegram_id > 0:
try:
await bot.send_message(telegram_id, html, parse_mode="HTML")
except Exception: # noqa: BLE001
logger.warning("telegram notify failed tg=%s", telegram_id)
if not want_email or not smtp_configured(settings):
return
email = (user.email if user else None) or ""
if not email.strip():
return
try:
# Telegram получает компактный HTML; в письме — фирменный шаблон
title = (subject or "VPN Service").replace("— VPN Service", "").strip() or "Уведомление"
mail_html = render_notification_email(
title=title,
body_html=html,
webapp_url=settings.webapp_url or "",
preheader=html_to_plain(html)[:120],
)
await send_email(settings, to=email.strip(), subject=subject, html=mail_html)
except Exception: # noqa: BLE001
logger.warning("email notify failed tg=%s to=%s", telegram_id, email, exc_info=True)