"""
Norda Biznes - Email Service
=============================
Sends emails via SMTP (OVH Zimbra).
Author: Norda Biznes Development Team
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
) -> bool:
"""
Send email via SMTP
Args:
to: List of recipient email addresses
subject: Email subject
body_text: Plain text email body
body_html: HTML email body (optional)
from_address: Sender email (optional, defaults to configured mail_from)
bcc: List of BCC recipient email addresses (optional)
Returns:
True if sent successfully, False otherwise
"""
return self._send_via_smtp(subject, body_html or body_text, to, bcc=bcc)
def _send_via_smtp(self, subject, body, to, sender_name=None, bcc=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
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
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
) -> 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 — admin gets a copy of every email
if bcc is None:
default_bcc = os.getenv('MAIL_BCC', 'maciej.pienczyn@inpi.pl')
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]
result = _email_service.send_mail(to, subject, body_text, body_html, from_address, bcc=bcc)
# 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'''
{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}
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:
|
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.
|
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:
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
)