from typing import Dict, Tuple, Any, List # ========================================== # 0. Internal Helpers # ========================================== def _format_slot_summary(availability: Any) -> str: if not isinstance(availability, list) or not availability: return "No specific time slots recorded." items = [] for day in availability[:4]: date = day.get("date", "") # Minimalist text for anti-spam items.append(f"- {date}") return "\n".join(items) def _format_currency(amount_in_cents: Any) -> str: """将分转换为元,保留两位小数""" try: # 确保是数字类型 value = float(amount_in_cents) / 100.0 return f"{value:.2f}" except (ValueError, TypeError): return str(amount_in_cents) def _get_status_meta(status: str) -> Tuple[str, str, str]: if status == 'Available': return "READY", "#16a34a", "Available" if status == 'Waitlist': return "WAIT", "#ca8a04", "Waitlist Only" return "NONE", "#dc2626", "No Slots" # ========================================== # 1. EMAIL TEMPLATES (Anti-Spam Optimized) # ========================================== def _email_base(content_html: str, app_name: str = "Visafly") -> str: """ Ultra-clean HTML structure. Uses standard fonts and minimal CSS to pass through ESP filters. """ return f""" Notification
{content_html}

Regards,
{app_name} Team

""" def template_email_verification_bind(payload: Dict) -> str: # Use neutral language: "Verification code" instead of "URGENT ACTION" html = f"""

Email Verification

You requested to link this email address to your {payload.get('app_name', 'Visafly')} account.

Your verification code is:

{payload.get('code')}

This code will expire in {payload.get('expiration_time', '10 minutes')}.

If you did not make this request, you can safely ignore this email.

""" return _email_base(html, payload.get('app_name', 'Visafly')) def template_email_password_reset(payload: Dict) -> str: html = f"""

Password Reset Code

We received a request to reset your password. Please enter the following code to continue:

{payload.get('code')}

Valid for {payload.get('expiration_time', '10 minutes')}. For security reasons, do not share this code.

""" return _email_base(html, payload.get('app_name', 'Visafly')) def template_email_login_credentials(payload: Dict) -> str: html = f"""

Your Account Credentials

Your account has been successfully created. Here are your temporary login details:

User:{payload.get('username')}
Pass:{payload.get('password')}

Log in to your account

""" return _email_base(html, payload.get('app_name', 'Visafly')) def template_email_slot_subscription(payload: Dict) -> str: # Change "Alert" to "Update" to sound less like spam html = f"""

Visa Slot Update

This is an automated update regarding your visa slot subscription.

Location: {payload.get('country')}, {payload.get('city')}
Visa Type: {payload.get('visa_type')}
Earliest Date: {payload.get('earliest_date')}

Check Official Website

""" return _email_base(html) def template_email_appointment_confirmation(payload: Dict) -> str: """预约成功确认邮件 (Anti-Spam Optimized)""" html = f"""

Appointment Confirmed

Dear {payload.get('username', 'Customer')},

We are pleased to inform you that your visa appointment has been successfully booked.

Order ID: {payload.get('order_id', 'N/A')}
Location: {payload.get('country', '')}, {payload.get('city', '')}
Visa Type: {payload.get('visa_type', 'N/A')}
Appointment Date: {payload.get('appointment_date', 'N/A')}
Account Email: {payload.get('user_email', 'N/A')}

Please ensure you prepare all required documents well in advance of your appointment date. If you have any questions, feel free to contact our support team.

""" return _email_base(html, payload.get('app_name', 'Visafly')) def template_email_order_event_update(payload: Dict) -> str: """ Template for visa progress updates. Focuses on neutral language to ensure inbox delivery. """ order_no = payload.get('order_no', 'N/A') summary = payload.get('summary', 'No summary available.') # Neutral subject line used in get_email_meta: "Notification: Visa Appointment Update" html = f"""

Visa Appointment Update

We have received a new update regarding your visa process for Order #{order_no}.

Update Summary:

{summary}

For your security and privacy, please log in to your dashboard to view the full message and any relevant documents.

Access Dashboard

""" return _email_base(html, "Visafly") # ========================================== # 2. WECHAT / TELEGRAM / WHATSAPP (Keeping the styles) # ========================================== # [Note: These channels don't have spam filters based on HTML structure, # so we keep the rich formatting we designed previously.] def template_wechat_payment_confirmed(payload: Dict) -> str: """支付成功通知模板""" amount_decimal = _format_currency(payload.get('amount')) return ( f"# ✅ Payment Success\n" f"--- \n" f"💰 **Amount:** {amount_decimal} {payload.get('currency')}\n" f"🆔 **Order ID:** `{payload.get('order_id')}`\n" f"📧 **User Account:** {payload.get('user_email')}\n" f"🕒 **Time:** {payload.get('confirmed_at')}\n\n" f"Your service has been activated. Please stay tuned for further updates." ) def template_wechat_appointment_confirmation(payload: Dict) -> str: """预约成功微信通知 (Markdown)""" return ( f"# 🎉 Appointment Confirmed\n" f"--- \n" f"👤 **User:** {payload.get('username', 'Customer')}\n" f"📍 **Location:** {payload.get('country')}, {payload.get('city')}\n" f"🛂 **Visa Type:** {payload.get('visa_type')}\n" f"📅 **Date:** `{payload.get('appointment_date')}`\n" f"🆔 **Order ID:** `{payload.get('order_id')}`\n\n" f"Your appointment is successfully booked. Please prepare your documents." ) def template_wechat_slot_snapshot(payload: Dict) -> str: emoji, _, status_text = _get_status_meta(payload.get("availability_status")) summary = _format_slot_summary(payload.get("availability")) return ( f"# {emoji} {payload.get('country')} {payload.get('city')}\n" f"> {status_text} | {payload.get('visa_type')}\n\n" f"📅 **Earliest Date:** `{payload.get('earliest_date', 'N/A')}`\n\n" f"🔍 **Slots Overview:**\n{summary}\n\n" f"🔗 [Book Now]({payload.get('website', '#')})" ) def template_telegram_slot_snapshot(payload: Dict) -> str: emoji, _, status_text = _get_status_meta(payload.get("availability_status")) summary = _format_slot_summary(payload.get("availability")) return ( f"{emoji} {payload.get('country')}, {payload.get('city')}\n" f"{status_text} · {payload.get('visa_type')}\n" f"──────────────────\n" f"📅 Earliest Date: {payload.get('earliest_date', 'N/A')}\n\n" f"📊 Availability:\n{summary}\n" f"──────────────────\n" f"🔗 Official Website ➜\n\n" ) def template_telegram_appointment_confirmation(payload: Dict) -> str: """预约成功 TG 通知 (HTML)""" return ( f"🎉 Appointment Confirmed\n" f"──────────────────\n" f"👤 Name: {payload.get('username', 'Customer')}\n" f"📍 Location: {payload.get('country')}, {payload.get('city')}\n" f"🛂 Visa Type: {payload.get('visa_type')}\n" f"📅 Date: {payload.get('appointment_date')}\n" f"──────────────────\n" f"🆔 Order: {payload.get('order_id')}\n" f"📧 Email: {payload.get('user_email')}" ) def template_telegram_order_event_update(payload: Dict) -> str: summary = payload.get('summary', 'New status update received.') return ( f"🔔 Visa Progress Update\n" f"──────────────────\n" f"🆔 Order: #{payload.get('order_no')}\n" f"📊 Summary:\n{summary}\n" f"──────────────────\n" f"🔗 View Dashboard ➜" ) def template_whatsapp_slot_snapshot(payload: Dict) -> str: emoji, _, _ = _get_status_meta(payload.get("availability_status")) return ( f"{emoji} *VISA SLOT UPDATE*\n\n" f"📍 *Location:* {payload.get('country')}, {payload.get('city')}\n" f"🛂 *Type:* {payload.get('visa_type')}\n" f"📅 *Earliest:* {payload.get('earliest_date')}\n\n" f"🌐 *Book Now:* {payload.get('website')}" ) def template_whatsapp_appointment_confirmation(payload: Dict) -> str: """预约成功 WhatsApp 通知 (Clean Text)""" return ( f"✅ *APPOINTMENT CONFIRMED*\n\n" f"Dear {payload.get('username', 'Customer')},\n" f"Your visa appointment has been successfully booked.\n\n" f"📍 *Location:* {payload.get('country')}, {payload.get('city')}\n" f"🛂 *Visa Type:* {payload.get('visa_type')}\n" f"📅 *Date:* {payload.get('appointment_date')}\n" f"🆔 *Order ID:* {payload.get('order_id')}\n\n" f"Please ensure all required documents are prepared. Thank you for choosing Visafly." ) def template_whatsapp_order_event_update(payload: Dict) -> str: """ WhatsApp template for progress updates. """ order_no = payload.get('order_no', 'N/A') # If the service pre-generated a message, use it, otherwise build one message = payload.get('message') if not message: summary = payload.get('summary', 'New status update received.') message = ( f"🔔 *VISA PROGRESS UPDATE*\n\n" f"Order: #{order_no}\n\n" f"Latest Update:\n" f"_{summary}_\n\n" f"Please log in to your dashboard or check your email for full details." ) return message # ========================================== # Dispatchers (保持不变) # ========================================== EMAIL_MAP = { "email_verification_for_bind": (template_email_verification_bind, "Verification code for your account"), "email_verification_for_reset": (template_email_password_reset, "Password reset code"), "login_credentials": (template_email_login_credentials, "Your login credentials"), "slot_subscription": (template_email_slot_subscription, "Update: Visa availability change"), "appointment_confirmation": (template_email_appointment_confirmation, "Your Visa Appointment Confirmation"), "order_event_update": (template_email_order_event_update, "Notification: Visa Appointment Update"), } WECHAT_MAP = { "slot_snapshot": template_wechat_slot_snapshot, "payment_user_confirmed": template_wechat_payment_confirmed, "appointment_confirmation": template_wechat_appointment_confirmation, } WHATSAPP_MAP = { "slot_snapshot": template_whatsapp_slot_snapshot, "appointment_confirmation": template_whatsapp_appointment_confirmation, "order_event_update": template_whatsapp_order_event_update, } TELEGRAM_MAP = { "slot_snapshot": template_telegram_slot_snapshot, "appointment_confirmation": template_telegram_appointment_confirmation, "order_event_update": template_telegram_order_event_update, } def render_email(tid: str, payload: Dict) -> str: return EMAIL_MAP[tid][0](payload) if tid in EMAIL_MAP else "Notification from Visafly" def get_email_meta(tid: str) -> Tuple[str, str]: sender = "donotreply@visafly.top" return (sender, EMAIL_MAP[tid][1]) if tid in EMAIL_MAP else (sender, "Notification") def render_wechat_markdown(tid: str, payload: Dict) -> str: return WECHAT_MAP[tid](payload) if tid in WECHAT_MAP else str(payload) def render_whatsapp_text(tid: str, payload: Dict) -> str: return WHATSAPP_MAP[tid](payload) if tid in WHATSAPP_MAP else str(payload) def render_telegram_html(tid: str, payload: Dict) -> str: return TELEGRAM_MAP[tid](payload) if tid in TELEGRAM_MAP else str(payload)