""" Email sending utilities with support for AWS SES and Yandex SMTP """ import smtplib import logging from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from typing import Optional from config import settings logger = logging.getLogger(__name__) # Try to import boto3 for AWS SES try: import boto3 from botocore.exceptions import ClientError, BotoCoreError BOTO3_AVAILABLE = True except ImportError: BOTO3_AVAILABLE = False logger.warning("[Email] boto3 не установлен. AWS SES недоступен. Установите: pip install boto3") def generate_verification_email(code: str) -> str: """Generate HTML email with verification code""" return f"""

Код подтверждения

Ваш код для регистрации в Nakama:

{code}

Код действителен в течение 15 минут.

Если вы не запрашивали этот код, просто проигнорируйте это письмо.

""" async def send_email_aws_ses(to: str, subject: str, html: str, text: Optional[str] = None): """Send email via AWS SES or Yandex Cloud Postbox (SESv2)""" if not BOTO3_AVAILABLE: raise ValueError("boto3 не установлен. Установите: pip install boto3") if not settings.AWS_SES_ACCESS_KEY_ID or not settings.AWS_SES_SECRET_ACCESS_KEY: raise ValueError("AWS_SES_ACCESS_KEY_ID и AWS_SES_SECRET_ACCESS_KEY должны быть установлены в .env") try: # Check if using Yandex Cloud Postbox (has custom endpoint) is_yandex_cloud = bool(settings.AWS_SES_ENDPOINT_URL and 'yandex' in settings.AWS_SES_ENDPOINT_URL.lower()) # Create client config client_config = { 'aws_access_key_id': settings.AWS_SES_ACCESS_KEY_ID, 'aws_secret_access_key': settings.AWS_SES_SECRET_ACCESS_KEY, 'region_name': settings.AWS_SES_REGION } # Add custom endpoint if specified (for Yandex Cloud Postbox) if settings.AWS_SES_ENDPOINT_URL: client_config['endpoint_url'] = settings.AWS_SES_ENDPOINT_URL # Yandex Cloud Postbox uses SESv2 API # Also check if region is ru-central1 (Yandex Cloud region) if is_yandex_cloud or settings.AWS_SES_REGION == 'ru-central1': endpoint_url = settings.AWS_SES_ENDPOINT_URL or 'https://postbox.cloud.yandex.net' logger.info(f"[Email] Использование Yandex Cloud Postbox (SESv2 API): {endpoint_url}") # Override endpoint if not set if not settings.AWS_SES_ENDPOINT_URL: client_config['endpoint_url'] = endpoint_url sesv2_client = boto3.client('sesv2', **client_config) # Prepare email for SESv2 destination = {'ToAddresses': [to]} content = { 'Simple': { 'Subject': { 'Data': subject, 'Charset': 'UTF-8' }, 'Body': { 'Html': { 'Data': html, 'Charset': 'UTF-8' }, 'Text': { 'Data': text or html.replace('<[^>]*>', ''), 'Charset': 'UTF-8' } } } } # Send email via SESv2 response = sesv2_client.send_email( FromEmailAddress=settings.EMAIL_FROM, Destination=destination, Content=content ) logger.info(f"✅ Email отправлен через Yandex Cloud Postbox на {to}, MessageId: {response.get('MessageId')}") return {"success": True, "to": to, "messageId": response.get('MessageId')} else: # Standard AWS SES (SESv1 API) logger.info(f"[Email] Использование AWS SES (SESv1 API): {settings.AWS_SES_REGION}") ses_client = boto3.client('ses', **client_config) # Prepare email for SESv1 destination = {'ToAddresses': [to]} message = { 'Subject': {'Data': subject, 'Charset': 'UTF-8'}, 'Body': { 'Html': {'Data': html, 'Charset': 'UTF-8'}, 'Text': {'Data': text or html.replace('<[^>]*>', ''), 'Charset': 'UTF-8'} } } # Send email response = ses_client.send_email( Source=settings.EMAIL_FROM, Destination=destination, Message=message ) logger.info(f"✅ Email отправлен через AWS SES на {to}, MessageId: {response.get('MessageId')}") return {"success": True, "to": to, "messageId": response.get('MessageId')} except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') error_msg = e.response.get('Error', {}).get('Message', str(e)) logger.error(f"❌ AWS SES ошибка ({error_code}): {error_msg}") # More informative error messages if error_code == 'MessageRejected': raise ValueError(f"Письмо отклонено: {error_msg}. Проверьте, что адрес {settings.EMAIL_FROM} верифицирован.") elif error_code == 'InvalidParameterValue': raise ValueError(f"Неверный параметр: {error_msg}") elif error_code == 'AccessDenied': raise ValueError(f"Доступ запрещен: {error_msg}. Проверьте права доступа ключей.") raise ValueError(f"AWS SES ошибка ({error_code}): {error_msg}") except Exception as e: logger.error(f"❌ Ошибка отправки email через AWS SES: {e}") import traceback traceback.print_exc() raise async def send_email_smtp(to: str, subject: str, html: str, text: Optional[str] = None): """Send email via SMTP (Yandex, Gmail, etc.)""" try: # Create message msg = MIMEMultipart('alternative') msg['Subject'] = subject msg['From'] = settings.EMAIL_FROM msg['To'] = to # Add text and HTML parts if text: msg.attach(MIMEText(text, 'plain', 'utf-8')) msg.attach(MIMEText(html, 'html', 'utf-8')) # Connect and send # Поддерживаем 'yandex' и 'smtp' как алиасы email_provider = settings.EMAIL_PROVIDER.lower() if email_provider in ['yandex', 'smtp']: logger.info(f"[Email] Отправка через SMTP ({email_provider}): {settings.YANDEX_SMTP_HOST}:{settings.YANDEX_SMTP_PORT}") if not settings.YANDEX_SMTP_USER or not settings.YANDEX_SMTP_PASSWORD: raise ValueError("YANDEX_SMTP_USER и YANDEX_SMTP_PASSWORD должны быть установлены в .env") # Используем SMTP_SSL для порта 465 if settings.YANDEX_SMTP_PORT == 465: server = smtplib.SMTP_SSL(settings.YANDEX_SMTP_HOST, settings.YANDEX_SMTP_PORT) else: server = smtplib.SMTP(settings.YANDEX_SMTP_HOST, settings.YANDEX_SMTP_PORT) if settings.YANDEX_SMTP_SECURE: server.starttls() server.login(settings.YANDEX_SMTP_USER, settings.YANDEX_SMTP_PASSWORD) server.send_message(msg) server.quit() logger.info(f"✅ Email отправлен на {to}") return {"success": True, "to": to} else: raise ValueError(f"Email provider '{settings.EMAIL_PROVIDER}' не поддерживается. Используйте 'aws', 'yandex' или 'smtp'") except Exception as e: logger.error(f"❌ Ошибка отправки email: {e}") # More informative error messages if "Authentication" in str(e) or "credentials" in str(e).lower(): raise ValueError( "Неверные учетные данные SMTP. Для Yandex используйте пароль приложения " "(https://id.yandex.ru/security), а не основной пароль аккаунта." ) elif "Connection" in str(e): raise ValueError( f"Не удалось подключиться к SMTP серверу {settings.YANDEX_SMTP_HOST}:{settings.YANDEX_SMTP_PORT}" ) raise async def send_email(to: str, subject: str, html: str, text: Optional[str] = None): """Send email using configured provider (AWS SES or SMTP)""" email_provider = settings.EMAIL_PROVIDER.lower() logger.info(f"[Email] 🔍 Начало отправки email") logger.info(f"[Email] Provider из настроек: '{settings.EMAIL_PROVIDER}' (lowercase: '{email_provider}')") logger.info(f"[Email] To: {to}") logger.info(f"[Email] From: {settings.EMAIL_FROM}") logger.info(f"[Email] Subject: {subject}") if email_provider == 'aws': logger.info(f"[Email] Выбран AWS SES") logger.info(f"[Email] AWS_SES_ACCESS_KEY_ID: {'установлен' if settings.AWS_SES_ACCESS_KEY_ID else 'НЕ УСТАНОВЛЕН'}") logger.info(f"[Email] AWS_SES_SECRET_ACCESS_KEY: {'установлен' if settings.AWS_SES_SECRET_ACCESS_KEY else 'НЕ УСТАНОВЛЕН'}") logger.info(f"[Email] AWS_SES_REGION: {settings.AWS_SES_REGION}") logger.info(f"[Email] AWS_SES_ENDPOINT_URL: {settings.AWS_SES_ENDPOINT_URL or 'не установлен'}") return await send_email_aws_ses(to, subject, html, text) elif email_provider in ['yandex', 'smtp']: logger.info(f"[Email] Выбран SMTP ({email_provider})") logger.info(f"[Email] YANDEX_SMTP_USER: {'установлен' if settings.YANDEX_SMTP_USER else 'НЕ УСТАНОВЛЕН'}") logger.info(f"[Email] YANDEX_SMTP_PASSWORD: {'установлен' if settings.YANDEX_SMTP_PASSWORD else 'НЕ УСТАНОВЛЕН'}") logger.info(f"[Email] YANDEX_SMTP_HOST: {settings.YANDEX_SMTP_HOST}") logger.info(f"[Email] YANDEX_SMTP_PORT: {settings.YANDEX_SMTP_PORT}") return await send_email_smtp(to, subject, html, text) else: logger.error(f"[Email] ❌ Неподдерживаемый провайдер: '{settings.EMAIL_PROVIDER}'") raise ValueError(f"Email provider '{settings.EMAIL_PROVIDER}' не поддерживается. Используйте 'aws', 'yandex' или 'smtp'") async def send_verification_code(email: str, code: str): """Send verification code to email""" logger.info(f"[Email] 📧 send_verification_code вызван для {email}") logger.info(f"[Email] 📧 Проверка настроек перед отправкой:") logger.info(f"[Email] 📧 EMAIL_PROVIDER (raw): '{settings.EMAIL_PROVIDER}'") logger.info(f"[Email] 📧 EMAIL_PROVIDER (lower): '{settings.EMAIL_PROVIDER.lower()}'") logger.info(f"[Email] 📧 EMAIL_FROM: '{settings.EMAIL_FROM}'") logger.info(f"[Email] 📧 AWS_SES_ACCESS_KEY_ID: {'установлен' if settings.AWS_SES_ACCESS_KEY_ID else 'НЕ УСТАНОВЛЕН'}") logger.info(f"[Email] 📧 AWS_SES_SECRET_ACCESS_KEY: {'установлен' if settings.AWS_SES_SECRET_ACCESS_KEY else 'НЕ УСТАНОВЛЕН'}") logger.info(f"[Email] 📧 AWS_SES_REGION: '{settings.AWS_SES_REGION}'") logger.info(f"[Email] 📧 AWS_SES_ENDPOINT_URL: '{settings.AWS_SES_ENDPOINT_URL}'") logger.info(f"[Email] 📧 YANDEX_SMTP_USER: {'установлен' if settings.YANDEX_SMTP_USER else 'НЕ УСТАНОВЛЕН'}") logger.info(f"[Email] 📧 YANDEX_SMTP_PASSWORD: {'установлен' if settings.YANDEX_SMTP_PASSWORD else 'НЕ УСТАНОВЛЕН'}") subject = "Код подтверждения регистрации - Nakama" html = generate_verification_email(code) text = f"Ваш код подтверждения: {code}. Код действителен 15 минут." return await send_email(email, subject, html, text) async def send_admin_confirmation_code(code: str, action: str, user_info: dict): """Send admin confirmation code to owner email""" action_text = "добавления админа" if action == "add" else "удаления админа" subject = f"Код подтверждения {action_text} - Nakama Moderation" html = f"""

Подтверждение {action_text}

Пользователь: @{user_info.get('username', 'не указан')} ({user_info.get('firstName', '')})

{f"

Номер админа: {user_info['adminNumber']}

" if 'adminNumber' in user_info else ''}

Код подтверждения:

{code}

Код действителен в течение 5 минут.

Если вы не запрашивали это подтверждение, проигнорируйте это письмо.

""" text = f"""Код подтверждения {action_text}: {code} Пользователь: @{user_info.get('username', 'не указан')} Код действителен 5 минут.""" return await send_email(settings.OWNER_EMAIL, subject, html, text)