freeleaps-service-hub/apps/notification/test_send_welcome_email.py

#!/usr/bin/env python3
"""
use magicleaps english version welcome template
I use email sender as freeleaps@freeleaps.com
I use tenant id as magicleaps
I use template id as welcome_email_tenant_magicleaps
I use region as 0
I use recipient emails as *@mathmast.com, *@163.com-> my own actual email address
I use subject properties as company_name, new_employee_name
I use body properties as new_employee_name, employee_id, position, department, email_address, start_date, company_name, company_logo, company_phone, company_address, manager_name, manager_email, hr_contact_name, hr_contact_email, it_support_email, initial_password, system_login_url, employee_handbook_url, company_policies_url, training_materials_url, onboarding_schedule, first_week_schedule
and I recieved the welcome email successfully
"""

import asyncio
import aiohttp
import json
import os
import sys
from datetime import datetime, timezone, timedelta
from jose import jwt

# import project config
sys.path.append('.')

# load environment variables
def load_env_file(file_path):
    """load environment variables from .env file"""
    if os.path.exists(file_path):
        with open(file_path, 'r') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#') and '=' in line:
                    if line.startswith('export '):
                        line = line[7:] 
                    key, value = line.split('=', 1)
                    
                    value = value.strip('"\'')
                    os.environ[key] = value

# load local.env file
load_env_file('local.env')

from common.config.app_settings import app_settings
from webapi.config.site_settings import site_settings

class WelcomeEmailSender:
    def __init__(self):
        # 配置
        host = 'localhost' if site_settings.SERVER_HOST == '0.0.0.0' else site_settings.SERVER_HOST
        port = site_settings.SERVER_PORT
        self.base_url = f"http://{host}:{port}/api/notification"
        self.tenant_id = "magicleaps"
        self.recipient_emails = ["*@mathmast.com", "*@163.com", *] # TODO: change to actual email addresses
        self.template_id = "welcome_email_tenant_magicleaps"  
        self.region = 0 
        
        self.tenant_token = None
        
        print(f"🔧 config: {self.base_url}")
        print(f"🎯 tenant: {self.tenant_id}")
        print(f"📧 recipient: {', '.join(self.recipient_emails)}")
        print(f"📝 template: {self.template_id} (English)")
    
    async def initialize(self):
        """get token"""
        self.tenant_token = await self.get_tenant_token()
        
    async def get_token_from_auth_service(self):
        """get token from auth service"""
        print("\n🔑 get token from auth service...")
        
        # in my local environment, the auth service is running on localhost:8103
        # TODO: change to actual auth service url
        auth_url = "http://localhost:8103/api/auth/token/generate-tokens" 
        auth_data = {
            "id": self.tenant_id, 
            "role": 2  # BUSINESS = 2
        }
        
        try:
            async with aiohttp.ClientSession() as session:
                async with session.post(auth_url, json=auth_data) as response:
                    print(f"auth service response status code: {response.status}")
                    
                    if response.status == 200:
                        result = await response.json()
                        access_token = result.get("access_token")
                        
                        print("✅ get token from auth service successfully")
                        print("=" * 60)
                        print(f"Access Token: {access_token}")
                        print("=" * 60)
                        
                        return access_token
                    else:
                        response_text = await response.text()
                        print(f"❌ get token from auth service failed: {response_text}")
                        return None
        except Exception as e:
            print(f"❌ connect to auth service failed: {e}")
            return None
    

    
    async def get_tenant_token(self):
        """get tenant token from auth service"""
        auth_token = await self.get_token_from_auth_service()
        
        if not auth_token:
            raise Exception("cannot get token from auth service, please ensure auth service is running")
        
        return auth_token
    
    async def verify_template_exists(self):
        """verify template exists"""
        print(f"\n🔍 verify template {self.template_id} exists...")
        
        headers = {"Authorization": f"Bearer {self.tenant_token}"}
        
        # print debug info
        print(f"request URL: {self.base_url}/tenant/templates/list?region={self.region}")
        print(f"request headers: {headers}")
        
        async with aiohttp.ClientSession() as session:
            async with session.get(
                f"{self.base_url}/tenant/templates/list?region={self.region}",
                headers=headers
            ) as response:
                print(f"response status code: {response.status}")
                response_text = await response.text()
                print(f"response: {response_text}")
                
                if response.status == 200:
                    data = json.loads(response_text)
                    templates = data.get('templates', [])
                    
                    print(f"found {len(templates)} templates:")
                    for template in templates:
                        print(f"  - {template.get('template_id')}: {template.get('subject')}")
                    
                    # find welcome email template
                    welcome_template = None
                    for template in templates:
                        if template.get('template_id') == self.template_id:
                            welcome_template = template
                            break
                    
                    if welcome_template:
                        print(f"✅ found template: {welcome_template.get('template_id')}")
                        print(f"    subject: {welcome_template.get('subject')}")
                        return True
                    else:
                        print(f"❌ not found template: {self.template_id}")
                        return False
                else:
                    print(f"❌ get template list failed")
                    return False
    
    async def verify_email_sender(self):
        """verify email sender is configured"""
        print(f"\n🔍 verify email sender is configured...")
        
        headers = {"Authorization": f"Bearer {self.tenant_token}"}
        
        async with aiohttp.ClientSession() as session:
            async with session.get(
                f"{self.base_url}/email_senders/get",
                headers=headers
            ) as response:
                print(f"response status code: {response.status}")
                response_text = await response.text()
                print(f"response: {response_text}")
                
                if response.status == 200:
                    data = json.loads(response_text)
                    email_senders = data.get('email_senders', [])
                    
                    if email_senders:
                        print(f"✅ email senders: {email_senders}")
                        return email_senders[0]  
                        print(f"❌ no email senders configured")
                        return None
                else:
                    print(f"❌ get email senders failed")
                    return None
    
    async def send_welcome_email(self):
        """send welcome email"""
        print(f"\n📧 send welcome email...")
        
        # prepare email data
        email_data = {
            "tenant_id": self.tenant_id,
            "template_id": self.template_id,
            "recipient_emails": self.recipient_emails,
            "subject_properties": {
                "company_name": "MagicLeaps",
                "new_employee_name": "Taniacao"
            },
            "body_properties": {
                # employee info
                "new_employee_name": "*", # TODO: change to test employee name
                "employee_id": "EMP001",
                "position": "Software Engineer",
                "department": "Engineering",
                "email_address": self.recipient_emails[0], 
                "start_date": "2024-01-15",
                
                # company info
                "company_name": "MagicLeaps",
                "company_logo": "https://magicleaps.com/logo.png",
                "company_phone": "+1-555-0123",
                "company_address": "123 Innovation Street, Tech City, TC 12345",
                
                # manager info
                "manager_name": "John Smith",
                "manager_email": "john.smith@magicleaps.com",
                
                # HR info
                "hr_contact_name": "Sarah Johnson",
                "hr_contact_email": "hr@magicleaps.com",
                
                # IT support
                "it_support_email": "it-support@magicleaps.com",
                "initial_password": "Welcome2024!",
                
                # system access
                "system_login_url": "https://portal.magicleaps.com",
                
                # document links
                "employee_handbook_url": "https://docs.magicleaps.com/handbook",
                "company_policies_url": "https://docs.magicleaps.com/policies",
                "training_materials_url": "https://training.magicleaps.com",
                
                # time schedule
                "onboarding_schedule": f"<strong>Week 1: </strong>Orientation<br><strong>Week 2: </strong>Training<br><strong>Week 3: </strong>Project Assignment",
                "first_week_schedule": f"<strong>Monday: </strong>Welcome Meeting<br><strong>Tuesday: </strong>System Setup<br><strong>Wednesday: </strong>Team Introduction"
            },
            "region": self.region,
            "sender_emails": ["freeleaps@freeleaps.com"],  
            "priority": "normal",
            "tracking_enabled": True
        }
        
        headers = {
            "Authorization": f"Bearer {self.tenant_token}",
            "Content-Type": "application/json"
        }
        
        print(f"📋 email data:")
        print(json.dumps(email_data, indent=2))
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/send_tenant_email",
                json=email_data,
                headers=headers
            ) as response:
                print(f"\nresponse status code: {response.status}")
                response_text = await response.text()
                print(f"response: {response_text}")
                
                if response.status == 200:
                    result = json.loads(response_text)
                    print(f"✅ email sending success!")
                    print(f"    message id: {result.get('message_id')}")
                    print(f"    email ids: {result.get('email_ids', [])}")
                    return result
                else:
                    print(f"❌ email sending failed")
                    return None
    
    async def check_email_status(self, email_id):
        """check email status"""
        print(f"\n📊 check email status: {email_id}")
        
        headers = {"Authorization": f"Bearer {self.tenant_token}"}
        
        async with aiohttp.ClientSession() as session:
            async with session.get(
                f"{self.base_url}/tenant_email_status/{self.tenant_id}?email_id={email_id}",
                headers=headers
            ) as response:
                print(f"response status code: {response.status}")
                response_text = await response.text()
                print(f"response: {response_text}")
                
                if response.status == 200:
                    status_data = json.loads(response_text)
                    print(f"✅ email status: {status_data}")
                    return status_data
                else:
                    print(f"❌ get email status failed")
                    return None
    
    async def run(self):
        """run full sending process"""
        print("🚀 start sending welcome email...")
        print("=" * 60)
        
        try:
            # 0. initialize (get token)
            await self.initialize()
            
            # 1. verify template exists
            template_exists = await self.verify_template_exists()
            if not template_exists:
                print("❌ template not found, cannot send email")
                return
            
            # 2. verify email sender is configured
            email_sender = await self.verify_email_sender()
            if not email_sender:
                print("❌ email sender not configured, cannot send email")
                return
            
            # 3. send welcome email
            result = await self.send_welcome_email()
            
            if result:
                # 4. wait for email to be processed
                print("⏳ wait for email to be processed...")
                await asyncio.sleep(3)
                
                # 5. check email status
                email_ids = result.get('email_ids', [])
                if email_ids:
                    for email_id in email_ids:
                        await self.check_email_status(email_id)
                
                print(f"\n✅ welcome email sending completed!")
                print(f"📧 recipient: {', '.join(self.recipient_emails)}")
                print(f"📝 template: {self.template_id}")
                print(f"🎯 tenant: {self.tenant_id}")
            else:
                print(f"\n❌ email sending failed")
                
        except Exception as e:
            print(f"❌ error during sending: {e}")
            import traceback
            traceback.print_exc()

async def main():
    """main function"""
    sender = WelcomeEmailSender()
    await sender.run()

if __name__ == "__main__":
    asyncio.run(main())