""" Norda Biznes - Email Service ============================= Sends emails via Microsoft Graph API using Application permissions. Based on mtbtracker implementation. 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__) # Check if msal is available try: import msal import requests MSAL_AVAILABLE = True except ImportError: MSAL_AVAILABLE = False logger.warning("msal package not installed. Email service will be disabled.") class EmailService: """Service for sending emails via Microsoft Graph API""" def __init__(self, tenant_id: str, client_id: str, client_secret: str, mail_from: str): """ Initialize Email Service Args: tenant_id: Azure AD Tenant ID client_id: Application (client) ID client_secret: Client secret value mail_from: Default sender email address (e.g., noreply@inpi.pl) """ self.tenant_id = tenant_id self.client_id = client_id self.client_secret = client_secret self.mail_from = mail_from self.authority = f"https://login.microsoftonline.com/{tenant_id}" self.scope = ["https://graph.microsoft.com/.default"] self.graph_endpoint = "https://graph.microsoft.com/v1.0" def _get_access_token(self) -> Optional[str]: """ Acquire access token using client credentials flow Returns: Access token string or None if failed """ if not MSAL_AVAILABLE: logger.error("msal package not available") return None try: app = msal.ConfidentialClientApplication( self.client_id, authority=self.authority, client_credential=self.client_secret, ) result = app.acquire_token_silent(self.scope, account=None) if not result: logger.info("No token in cache, acquiring new token") result = app.acquire_token_for_client(scopes=self.scope) if "access_token" in result: return result["access_token"] else: logger.error(f"Failed to acquire token: {result.get('error')}: {result.get('error_description')}") return None except Exception as e: logger.error(f"Exception during token acquisition: {e}", exc_info=True) return None def send_mail( self, to: List[str], subject: str, body_text: str, body_html: Optional[str] = None, from_address: Optional[str] = None ) -> bool: """ Send email via Microsoft Graph API 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) Returns: True if sent successfully, False otherwise """ if not MSAL_AVAILABLE: logger.error("msal package not available - cannot send email") return False try: # Get access token token = self._get_access_token() if not token: logger.error("Failed to acquire access token") return False # Use default sender if not specified sender = from_address or self.mail_from # Prepare recipients recipients = [{"emailAddress": {"address": email}} for email in to] # Prepare message body if body_html: content_type = "HTML" content = body_html else: content_type = "Text" content = body_text # Build email message # Display name for sender - shown in email clients sender_display_name = "Norda Biznes Hub" email_msg = { "message": { "subject": subject, "body": { "contentType": content_type, "content": content }, "toRecipients": recipients, "from": { "emailAddress": { "address": sender, "name": sender_display_name } } }, "saveToSentItems": "false" } # Send email via Graph API url = f"{self.graph_endpoint}/users/{sender}/sendMail" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json" } response = requests.post(url, headers=headers, json=email_msg, timeout=30) if response.status_code == 202: logger.info(f"Email sent successfully to {to}") return True else: logger.error(f"Failed to send email. Status: {response.status_code}, Response: {response.text}") return False except Exception as e: logger.error(f"Exception during email sending: {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: AZURE_TENANT_ID: Azure AD Tenant ID AZURE_CLIENT_ID: Application (client) ID AZURE_CLIENT_SECRET: Client secret value MAIL_FROM: Default sender email address """ global _email_service tenant_id = os.getenv('AZURE_TENANT_ID') client_id = os.getenv('AZURE_CLIENT_ID') client_secret = os.getenv('AZURE_CLIENT_SECRET') mail_from = os.getenv('MAIL_FROM', 'noreply@inpi.pl') if tenant_id and client_id and client_secret: _email_service = EmailService(tenant_id, client_id, client_secret, mail_from) logger.info(f"Email Service initialized with sender: {mail_from}") return True else: logger.warning("Email Service not configured - missing Azure credentials") 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 ) -> 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) 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] result = _email_service.send_mail(to, subject, body_text, body_html, from_address) # 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 # ============================================================ def send_password_reset_email(email: str, reset_url: str) -> bool: """ Send password reset email Args: email: Recipient email address reset_url: Password reset URL with token Returns: True if sent successfully, False otherwise """ subject = "Reset hasła - Norda Biznes Hub" body_text = f"""Otrzymałeś ten email, ponieważ zażądano resetowania hasła dla Twojego konta Norda Biznes Hub. Aby zresetować hasło, kliknij w poniższy link: {reset_url} Link będzie ważny przez 1 godzinę. Jeśli nie zażądałeś resetowania hasła, zignoruj ten email. --- Norda Biznes Hub - Platforma Networkingu Regionalnej Izby Przedsiębiorców https://nordabiznes.pl """ body_html = f"""

Reset hasła

Otrzymałeś ten email, ponieważ zażądano resetowania hasła dla Twojego konta Norda Biznes Hub.

Aby zresetować hasło, kliknij w poniższy przycisk:

Zresetuj hasło
Ważność linku: 1 godzina

Jeśli przycisk nie działa, skopiuj i wklej poniższy link do przeglądarki:

{reset_url}

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

""" return send_email( to=[email], subject=subject, body_text=body_text, body_html=body_html, email_type='password_reset' ) def send_welcome_email(email: str, name: str, verification_url: str) -> bool: """ Send welcome/verification email after registration Args: email: Recipient email address name: User's name verification_url: Email verification URL with token Returns: True if sent successfully, False otherwise """ subject = "Witamy w Norda Biznes Hub - Potwierdź email" body_text = f"""Witaj {name}! Dziękujemy za rejestrację w Norda Biznes Hub - platformie networkingu Regionalnej Izby Przedsiębiorców Norda Biznes. Aby aktywować swoje konto, potwierdź adres email klikając w poniższy link: {verification_url} Link będzie ważny przez 24 godziny. Po potwierdzeniu email będziesz mógł: - Przeglądać profile firm członkowskich Izby - Korzystać z asystenta AI do wyszukiwania usług - Nawiązywać kontakty biznesowe Pozdrawiamy, Zespół Norda Biznes Hub --- Norda Biznes Hub - Platforma Networkingu Regionalnej Izby Przedsiębiorców https://nordabiznes.pl """ body_html = f"""

Witamy w Norda Biznes Hub!

Witaj {name}!

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

Aby aktywować swoje konto, potwierdź adres email:

Potwierdź email
Po potwierdzeniu email będziesz mógł:
  • Przeglądać profile firm członkowskich Izby
  • Korzystać z asystenta AI do wyszukiwania usług
  • Nawiązywać kontakty biznesowe

Link będzie ważny przez 24 godziny.

""" return send_email( to=[email], subject=subject, body_text=body_text, body_html=body_html, email_type='welcome', recipient_name=name )