""" Norda Biznes - Email Service ============================= Sends emails via SMTP (OVH Zimbra). Author: Maciej Pienczyn, InPi sp. z o.o. Created: 2025-12-25 """ import os import logging from typing import Optional, List from dotenv import load_dotenv # Load .env file for environment variables load_dotenv() logger = logging.getLogger(__name__) class EmailService: """Service for sending emails via SMTP""" def __init__(self, mail_from: str): """ Initialize Email Service Args: mail_from: Default sender email address (e.g., noreply@nordabiznes.pl) """ self.mail_from = mail_from def send_mail( self, to: List[str], subject: str, body_text: str, body_html: Optional[str] = None, from_address: Optional[str] = None, bcc: Optional[List[str]] = None, extra_headers: Optional[dict] = None, ) -> bool: return self._send_via_smtp(subject, body_html or body_text, to, bcc=bcc, extra_headers=extra_headers) def _send_via_smtp(self, subject, body, to, sender_name=None, bcc=None, extra_headers=None): """Send email via SMTP (OVH Zimbra).""" import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.utils import formataddr smtp_host = os.getenv('SMTP_HOST', 'ssl0.ovh.net') smtp_port = int(os.getenv('SMTP_PORT', '587')) smtp_user = os.getenv('SMTP_USER', '') smtp_pass = os.getenv('SMTP_PASSWORD', '') mail_from = os.getenv('MAIL_FROM', smtp_user) mail_from_name = sender_name or os.getenv('MAIL_FROM_NAME', 'NordaBiznes Portal') if not smtp_user or not smtp_pass: logger.error("SMTP failed: SMTP_USER or SMTP_PASSWORD not configured") return False try: msg = MIMEMultipart('alternative') msg['Subject'] = subject msg['From'] = formataddr((mail_from_name, mail_from)) msg['To'] = ', '.join(to) if isinstance(to, list) else to if bcc: msg['Bcc'] = ', '.join(bcc) if isinstance(bcc, list) else bcc if extra_headers: for k, v in extra_headers.items(): msg[k] = v msg.attach(MIMEText(body, 'html', 'utf-8')) all_recipients = list(to) if isinstance(to, list) else [to] if bcc: all_recipients.extend(bcc if isinstance(bcc, list) else [bcc]) with smtplib.SMTP(smtp_host, smtp_port, timeout=30) as server: server.starttls() server.login(smtp_user, smtp_pass) server.sendmail(mail_from, all_recipients, msg.as_string()) logger.info(f"Email sent successfully via SMTP to {to}") return True except Exception as e: logger.error(f"SMTP send failed: {e}", exc_info=True) return False # Global service instance _email_service: Optional[EmailService] = None def init_email_service(): """ Initialize the global Email Service instance from environment variables Required env vars: SMTP_USER: SMTP username SMTP_PASSWORD: SMTP password MAIL_FROM: Default sender email address """ global _email_service mail_from = os.getenv('MAIL_FROM', 'noreply@nordabiznes.pl') smtp_user = os.getenv('SMTP_USER') if smtp_user: _email_service = EmailService(mail_from) logger.info(f"Email Service initialized (SMTP), sender: {mail_from}") return True else: logger.warning("Email Service not configured - missing SMTP_USER") return False PUBLIC_BASE_URL = os.getenv('PUBLIC_BASE_URL', 'https://nordabiznes.pl').rstrip('/') def _build_unsubscribe_footer(notification_type: str, user_id: int): """Return (footer_html, footer_text, unsubscribe_url) or (None, None, None).""" try: from utils.unsubscribe_tokens import generate_unsubscribe_token, NOTIFICATION_TYPE_TO_COLUMN if notification_type not in NOTIFICATION_TYPE_TO_COLUMN: return None, None, None token = generate_unsubscribe_token(user_id, notification_type) url = f"{PUBLIC_BASE_URL}/unsubscribe?t={token}" footer_html = ( '
' '

' f'Nie chcesz otrzymywać takich e-maili? ' f'' f'Wyłącz ten typ powiadomień jednym kliknięciem.' f' Preferencje dla wszystkich typów: ' f'' f'panel konta.' '

' ) footer_text = ( "\n\n---\n" f"Aby wyłączyć ten typ powiadomień e-mail, otwórz: {url}\n" f"Pełne preferencje: {PUBLIC_BASE_URL}/konto/prywatnosc\n" ) return footer_html, footer_text, url except Exception as e: logger.debug("unsubscribe footer build failed: %s", e) return None, None, None def send_email( to: List[str], subject: str, body_text: str, body_html: Optional[str] = None, from_address: Optional[str] = None, email_type: str = 'notification', user_id: Optional[int] = None, recipient_name: Optional[str] = None, bcc: Optional[List[str]] = None, notification_type: Optional[str] = None, ) -> bool: """ Send email using the global Email Service instance Args: to: List of recipient email addresses (can be single email as list) subject: Email subject body_text: Plain text email body body_html: HTML email body (optional) from_address: Sender email (optional) email_type: Type of email for logging (welcome, password_reset, notification) user_id: User ID for logging (optional) recipient_name: Recipient name for logging (optional) bcc: List of BCC emails (optional, defaults to MAIL_BCC env var) Returns: True if sent successfully, False otherwise """ if not _email_service: logger.error("Email Service not initialized. Call init_email_service() first.") return False # Convert single email to list if needed if isinstance(to, str): to = [to] # Default BCC from env var (empty by default — no admin copies) if bcc is None: default_bcc = os.getenv('MAIL_BCC', '') if default_bcc: bcc = [addr.strip() for addr in default_bcc.split(',') if addr.strip()] # Don't BCC someone who is already a direct recipient bcc = [addr for addr in bcc if addr not in to] # Unsubscribe footer + RFC 8058 List-Unsubscribe headers (dla powiadomień user-facing) extra_headers = None if notification_type and user_id: html_footer, text_footer, unsub_url = _build_unsubscribe_footer(notification_type, user_id) if unsub_url: if body_html: # Dodaj przed zamykającym lub po prostu doklejaj if '' in body_html.lower(): idx = body_html.lower().rfind('') body_html = body_html[:idx] + html_footer + body_html[idx:] else: body_html = body_html + html_footer if body_text: body_text = body_text + text_footer extra_headers = { 'List-Unsubscribe': f'<{unsub_url}>', 'List-Unsubscribe-Post': 'List-Unsubscribe=One-Click', } result = _email_service.send_mail(to, subject, body_text, body_html, from_address, bcc=bcc, extra_headers=extra_headers) # Log email to database _log_email( email_type=email_type, recipient_emails=to, recipient_name=recipient_name, subject=subject, user_id=user_id, sender_email=from_address or _email_service.mail_from if _email_service else None, success=result ) return result def _log_email( email_type: str, recipient_emails: List[str], subject: str, success: bool, recipient_name: Optional[str] = None, user_id: Optional[int] = None, sender_email: Optional[str] = None, error_message: Optional[str] = None ) -> None: """ Log email to database for monitoring. Args: email_type: Type of email (welcome, password_reset, notification) recipient_emails: List of recipient email addresses subject: Email subject success: Whether email was sent successfully recipient_name: Recipient name (optional) user_id: User ID (optional) sender_email: Sender email address (optional) error_message: Error message if failed (optional) """ try: from database import SessionLocal, EmailLog from datetime import datetime db = SessionLocal() try: for email in recipient_emails: log_entry = EmailLog( email_type=email_type, recipient_email=email, recipient_name=recipient_name, subject=subject, user_id=user_id, sender_email=sender_email, status='sent' if success else 'failed', sent_at=datetime.utcnow() if success else None, error_message=error_message if not success else None ) db.add(log_entry) db.commit() logger.info(f"Email logged: {email_type} -> {recipient_emails} (status: {'sent' if success else 'failed'})") except Exception as e: db.rollback() logger.error(f"Failed to log email to database: {e}") finally: db.close() except ImportError: # Database not available (e.g., during testing) logger.warning("Could not log email - database module not available") def is_configured() -> bool: """ Check if Email Service is configured and ready Returns: True if service is initialized, False otherwise """ return _email_service is not None # Auto-initialize on module load init_email_service() # ============================================================ # EMAIL TEMPLATES (v3 design) # ============================================================ def _email_v3_wrap(title: str, subtitle: str, content_html: str) -> str: """Wrap email content in v3 branded shell (header + footer).""" return f'''
NB

{title}

{subtitle}

{content_html}

Norda Biznes Partner

Stowarzyszenie Norda Biznes

ul. 12 Marca 238/5, 84-200 Wejherowo

nordabiznes.pl  |  Facebook

To powiadomienie zostało wysłane automatycznie.

''' def build_message_notification_email(sender_name: str, subject: str, content_preview: str, message_url: str, settings_url: str) -> tuple: """Build branded email for message notification. Returns (html, text).""" subject_html = f'

Temat: {subject}

' if subject else '' content_html = f'''

{sender_name} wysłał(a) Ci wiadomość na portalu Norda Biznes.

{subject_html}

{content_preview}

Przeczytaj wiadomość

Możesz wyłączyć powiadomienia e-mail w ustawieniach prywatności.

''' html = _email_v3_wrap('Nowa wiadomość', f'od {sender_name}', content_html) text = f'{sender_name} wysłał(a) Ci wiadomość na portalu Norda Biznes. Odczytaj: {message_url}' return html, text def send_password_reset_email(email: str, reset_url: str, admin_initiated: bool = False) -> bool: """Send password reset email.""" subject = "Reset hasła - Norda Biznes Partner" if admin_initiated: intro_text = "Administrator portalu Norda Biznes Partner wygenerował dla Ciebie link do ustawienia hasła." intro_html = "Administrator portalu wygenerował dla Ciebie link do ustawienia nowego hasła." validity = "24 godziny" else: intro_text = "Otrzymałeś ten email, ponieważ zażądano resetowania hasła dla Twojego konta Norda Biznes Partner." intro_html = "Otrzymałeś ten email, ponieważ zażądano resetowania hasła dla Twojego konta." validity = "1 godzinę" body_text = f"""{intro_text} Aby ustawić hasło, kliknij w poniższy link: {reset_url} Link będzie ważny przez {validity}. Jeśli nie spodziewałeś się tego emaila, skontaktuj się z administratorem. --- Norda Biznes Partner https://nordabiznes.pl """ content = f'''

{intro_html}

Kliknij poniższy przycisk, aby ustawić nowe hasło:

Zresetuj hasło

Ważność linku: {validity}

Po tym czasie konieczne będzie ponowne żądanie resetu hasła.

Jeśli przycisk nie działa, skopiuj i wklej ten link:

{reset_url}

Nie zażądałeś resetowania hasła?
Zignoruj ten email — Twoje hasło pozostanie bez zmian.

''' body_html = _email_v3_wrap('Reset hasła', 'Norda Biznes Partner', content) return send_email( to=[email], subject=subject, body_text=body_text, body_html=body_html, email_type='password_reset' ) def send_welcome_activation_email(email: str, name: str, reset_url: str) -> bool: """Send welcome activation email to users who never received credentials.""" subject = "Witamy w Norda Biznes Partner — Ustaw hasło do konta" body_text = f"""Witaj {name}! Twoje konto w portalu Norda Biznes Partner jest gotowe. Aby ustawić hasło i zalogować się, kliknij w poniższy link: {reset_url} Link jest ważny przez 72 godziny. Jeśli wygaśnie, wejdź na nordabiznes.pl/login i kliknij "Nie pamiętasz hasła?" — otrzymasz nowy link. Co znajdziesz w portalu: - Katalog firm członkowskich Izby - Forum dyskusyjne dla członków - Wiadomości prywatne i networking - Asystent AI do wyszukiwania usług Pozdrawiamy, Zespół Norda Biznes Partner https://nordabiznes.pl """ check = ( '
' '✓
' ) content = f'''

Witaj {name}!

Twoje konto w portalu Norda Biznes Partner jest gotowe. Kliknij poniższy przycisk, aby ustawić hasło i zalogować się po raz pierwszy.

Ustaw hasło i zaloguj się

Link jest ważny przez 72 godziny. Jeśli wygaśnie, wejdź na nordabiznes.pl/login i kliknij „Nie pamiętasz hasła?" — otrzymasz nowy link.

Co znajdziesz w portalu:

{check}Katalog firm członkowskich Izby
{check}Forum dyskusyjne dla członków
{check}Wiadomości prywatne i networking
{check}Asystent AI do wyszukiwania usług

Jeśli przycisk nie działa, skopiuj i wklej ten link:

{reset_url}

''' body_html = _email_v3_wrap('Witamy w portalu!', 'Norda Biznes Partner', content) return send_email( to=[email], subject=subject, body_text=body_text, body_html=body_html, email_type='welcome_activation', recipient_name=name ) def send_welcome_email(email: str, name: str, verification_url: str) -> bool: """Send welcome/verification email after registration.""" subject = "Witamy w Norda Biznes Partner — Potwierdź email" body_text = f"""Witaj {name}! Dziękujemy za rejestrację w Norda Biznes Partner. Aby aktywować konto, kliknij: {verification_url} Link ważny 24 godziny. Pozdrawiamy, Zespół Norda Biznes Partner https://nordabiznes.pl """ check = ( '
' '✓
' ) content = f'''

Witaj {name}!

Dziękujemy za rejestrację w Norda Biznes Partner — platformie networkingu Regionalnej Izby Przedsiębiorców.

Aby aktywować konto, potwierdź swój adres email:

Potwierdź email

Co zyskujesz:

{check}Katalog firm członkowskich Izby
{check}Asystent AI do wyszukiwania usług
{check}Networking i kontakty biznesowe

Link aktywacyjny jest ważny przez 24 godziny.

''' body_html = _email_v3_wrap('Witamy!', 'Norda Biznes Partner', content) return send_email( to=[email], subject=subject, body_text=body_text, body_html=body_html, email_type='welcome', recipient_name=name )