Hugging Face's logo

Spaces:
Seth0330
/
EZOFISOCR
Running

App Files Community
EZOFISOCR / backend /app /brevo_service.py
Seth
Update
8e8c6a4
raw
history blame contribute delete
19.9 kB
"""
Brevo (formerly Sendinblue) email service for sending transactional emails.
Reference: https://developers.brevo.com/reference/sendtransacemail
"""
import os
import httpx
from typing import Optional, Dict, Any
from difflib import SequenceMatcher
BREVO_API_KEY = os.environ.get("BREVO_API_KEY", "")
BREVO_API_URL = "https://api.brevo.com/v3/smtp/email"
BREVO_SENDER_EMAIL = os.environ.get("BREVO_SENDER_EMAIL", "noreply@yourdomain.com")
BREVO_SENDER_NAME = os.environ.get("BREVO_SENDER_NAME", "EZOFIS AI")
BREVO_TRIAL_LIST_ID = int(os.environ.get("BREVO_TRIAL_LIST_ID", "5")) # Default to 5 for "VRP Trials"
# Brevo standard attribute names mapping
BREVO_ATTRIBUTE_MAP = {
 "first_name": "FIRSTNAME",
 "last_name": "LASTNAME",
 "organization_name": "COMPANY",
 "phone_number": "SMS",
 "linkedin_url": "LINKEDIN",
 "title": "JOB_TITLE",
 "headline": "HEADLINE",
 "organization_website": "WEBSITE",
 "organization_address": "ADDRESS",
 # Common variations
 "firstname": "FIRSTNAME",
 "fname": "FIRSTNAME",
 "given_name": "FIRSTNAME",
 "lastname": "LASTNAME",
 "lname": "LASTNAME",
 "surname": "LASTNAME",
 "family_name": "LASTNAME",
 "company": "COMPANY",
 "org": "COMPANY",
 "organization": "COMPANY",
 "phone": "SMS",
 "mobile": "SMS",
 "telephone": "SMS",
 "linkedin": "LINKEDIN",
 "linkedin_profile": "LINKEDIN",
 "job_title": "JOB_TITLE",
 "position": "JOB_TITLE",
 "role": "JOB_TITLE",
 "website": "WEBSITE",
 "url": "WEBSITE",
 "web": "WEBSITE",
 "address": "ADDRESS",
 "location": "ADDRESS",
}
def _get_brevo_attribute_name(field_name: str) -> Optional[str]:
 """
 Get Brevo attribute name for a given field name using semantic matching.
Args:
 field_name: Field name (e.g., "first_name", "email")
Returns:
 Brevo attribute name (e.g., "FIRSTNAME") or None if not found
 """
 # Normalize field name
 normalized = field_name.lower().replace("_", "").replace("-", "")
# Direct lookup first
 if field_name.lower() in BREVO_ATTRIBUTE_MAP:
 return BREVO_ATTRIBUTE_MAP[field_name.lower()]
if normalized in BREVO_ATTRIBUTE_MAP:
 return BREVO_ATTRIBUTE_MAP[normalized]
# Semantic matching using similarity
 best_match = None
 best_score = 0.0
for key, value in BREVO_ATTRIBUTE_MAP.items():
 score = SequenceMatcher(None, normalized, key.lower()).ratio()
 if score > best_score:
 best_score = score
 best_match = value
# Only return if similarity is high enough
 if best_score >= 0.6:
 return best_match
return None
async def send_otp_email(email: str, otp: str) -> bool:
 """
 Send OTP email using Brevo transactional email API.
Args:
 email: Recipient email address
 otp: One-time password code
Returns:
 True if email sent successfully
Raises:
 ValueError: If BREVO_API_KEY is not set
 Exception: If email sending fails
 """
 if not BREVO_API_KEY:
 raise ValueError("BREVO_API_KEY environment variable is not set")
# Brevo API payload structure
 payload = {
 "sender": {
 "name": BREVO_SENDER_NAME,
 "email": BREVO_SENDER_EMAIL
 },
 "to": [
 {
 "email": email
 }
 ],
 "subject": "Your OTP Code for EZOFIS AI",
 "htmlContent": f"""
 <!DOCTYPE html>
 <html>
 <head>
 <meta charset="utf-8">
 <meta name="viewport" content="width=device-width, initial-scale=1.0">
 <style>
 body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, 'Helvetica Neue', Arial, sans-serif; line-height: 1.6; color: #333; margin: 0; padding: 0; background-color: #f4f4f4; }}
 .container {{ max-width: 600px; margin: 20px auto; background: white; border-radius: 10px; overflow: hidden; box-shadow: 0 2px 10px rgba(0,0,0,0.1); }}
 .content {{ padding: 40px 30px; }}
 .content p {{ margin: 0 0 15px 0; color: #555; }}
 .otp-box {{ background: #f8f9fa; border: 2px dashed #667eea; padding: 30px; text-align: center; margin: 30px 0; border-radius: 8px; }}
 .otp-label {{ font-size: 14px; color: #666; margin-bottom: 10px; }}
 .otp-code {{ font-size: 36px; font-weight: bold; color: #667eea; letter-spacing: 8px; font-family: 'Courier New', monospace; }}
 .expiry {{ color: #888; font-size: 14px; margin-top: 20px; }}
 .footer {{ text-align: center; margin-top: 30px; padding-top: 20px; border-top: 1px solid #eee; color: #999; font-size: 12px; }}
 .warning {{ background: #fff3cd; border-left: 4px solid #ffc107; padding: 15px; margin: 20px 0; border-radius: 4px; font-size: 14px; color: #856404; }}
 </style>
 </head>
 <body>
 <div class="container">
 <div class="content">
 <p>Hello,</p>
 <p>You requested a one-time password (OTP) to sign in to your EZOFIS account.</p>
 <div class="otp-box">
 <div class="otp-label">Your OTP code is:</div>
 <div class="otp-code">{otp}</div>
 </div>
 <p class="expiry">This code will expire in <strong>10 minutes</strong>.</p>
 <div class="warning">
 <strong>⚠️ Security Notice:</strong> If you didn't request this code, please ignore this email. Do not share this code with anyone.
 </div>
 <div class="footer">
 <p>© EZOFIS - Agentic Intelligence Platform</p>
 <p>This is an automated message, please do not reply.</p>
 </div>
 </div>
 </div>
 </body>
 </html>
 """,
 "textContent": f"""
Your OTP Code for EZOFIS AI
Hello,
You requested a one-time password (OTP) to sign in to your EZOFIS account.
Your OTP code is: {otp}
This code will expire in 10 minutes.
⚠️ Security Notice: If you didn't request this code, please ignore this email. Do not share this code with anyone.
© EZOFIS - Agentic Intelligence Platform
This is an automated message, please do not reply.
 """
 }
headers = {
 "accept": "application/json",
 "api-key": BREVO_API_KEY,
 "content-type": "application/json"
 }
try:
 async with httpx.AsyncClient(timeout=30.0) as client:
 response = await client.post(BREVO_API_URL, json=payload, headers=headers)
 response.raise_for_status()
result = response.json()
 message_id = result.get('messageId', 'N/A')
 print(f"[INFO] Brevo email sent successfully to {email}. Message ID: {message_id}")
 return True
 except httpx.HTTPStatusError as e:
 error_detail = {}
 try:
 error_detail = e.response.json() if e.response else {}
 except:
 error_detail = {"message": str(e)}
error_msg = error_detail.get('message', f'HTTP {e.response.status_code}' if e.response else 'Unknown error')
 print(f"[ERROR] Brevo API error: {e.response.status_code if e.response else 'N/A'} - {error_msg}")
 raise Exception(f"Failed to send email via Brevo: {error_msg}")
 except httpx.TimeoutException:
 print(f"[ERROR] Brevo API request timed out")
 raise Exception("Email service timeout. Please try again.")
 except Exception as e:
 print(f"[ERROR] Brevo email sending failed: {str(e)}")
 raise Exception(f"Failed to send email: {str(e)}")
async def send_share_email(recipient_email: str, sender_email: str, share_link: str, sender_name: str = None) -> bool:
 """
 Send share email using Brevo transactional email API.
Args:
 recipient_email: Recipient email address
 sender_email: Sender email address
 share_link: Share link URL
 sender_name: Sender's display name (optional, falls back to email if not provided)
Returns:
 True if email sent successfully
Raises:
 ValueError: If BREVO_API_KEY is not set
 Exception: If email sending fails
 """
 if not BREVO_API_KEY:
 raise ValueError("BREVO_API_KEY environment variable is not set")
# Get base URL from environment or use default
 base_url = os.environ.get("VITE_API_BASE_URL", "https://seth0330-ezofisocr.hf.space")
# Determine sender display name: use sender_name if available, otherwise extract from email
 # This is the logged-in user's name, NOT the email sender name (BREVO_SENDER_NAME)
 # BREVO_SENDER_NAME is only used for the "From" field, not the email body
 if sender_name and sender_name.strip():
 # Use the actual logged-in user's name
 sender_display = sender_name.strip()
 print(f"[INFO] Using user's name from database: {sender_display}")
 else:
 # Extract name from email (part before @) and format it nicely
 email_name = sender_email.split('@')[0]
 # Handle cases like "seth.smith" -> "Seth Smith" or "seth_smith" -> "Seth Smith"
 if '.' in email_name:
 parts = email_name.split('.')
 sender_display = ' '.join(part.capitalize() for part in parts)
 elif '_' in email_name:
 parts = email_name.split('_')
 sender_display = ' '.join(part.capitalize() for part in parts)
 else:
 # Simple case: "seth" -> "Seth"
 sender_display = email_name.capitalize()
 print(f"[INFO] Extracted name from email: {sender_display} (from {sender_email})")
# Brevo API payload structure
 # Note: BREVO_SENDER_NAME is used only for the "From" field in the email header
 # The email body uses sender_display (the logged-in user's name)
 payload = {
 "sender": {
 "name": BREVO_SENDER_NAME,
 "email": BREVO_SENDER_EMAIL
 },
 "to": [
 {
 "email": recipient_email
 }
 ],
 "subject": f"{sender_display} shared a document extraction with you",
 "htmlContent": f"""
 <!DOCTYPE html>
 <html>
 <head>
 <meta charset="utf-8">
 <meta name="viewport" content="width=device-width, initial-scale=1.0">
 <style>
 body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, 'Helvetica Neue', Arial, sans-serif; line-height: 1.6; color: #333; margin: 0; padding: 0; background-color: #f4f4f4; }}
 .container {{ max-width: 600px; margin: 20px auto; background: white; border-radius: 10px; overflow: hidden; box-shadow: 0 2px 10px rgba(0,0,0,0.1); }}
 .content {{ padding: 40px 30px; }}
 .content p {{ margin: 0 0 15px 0; color: #555; }}
 .share-box {{ background: #f8f9fa; border: 2px solid #667eea; padding: 30px; text-align: center; margin: 30px 0; border-radius: 8px; }}
 .share-button {{ display: inline-block; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: #ffffff !important; padding: 15px 30px; text-decoration: none; border-radius: 8px; font-weight: 600; margin-top: 20px; font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, 'Helvetica Neue', Arial, sans-serif; }}
 .share-button:hover {{ color: #ffffff !important; }}
 .footer {{ text-align: center; margin-top: 30px; padding-top: 20px; border-top: 1px solid #eee; color: #999; font-size: 12px; }}
 </style>
 </head>
 <body>
 <div class="container">
 <div class="content">
 <p>Hello,</p>
 <p><strong>{sender_display}</strong> wants you to take a look at a document extraction output.</p>
 <div class="share-box">
 <p style="margin-bottom: 20px; color: #666;">Click the button below to view the shared extraction:</p>
 <a href="{share_link}" class="share-button">View Shared Extraction</a>
 </div>
 <p style="color: #888; font-size: 14px;">You'll need to sign in to your EZOFIS account to view this extraction. If you don't have an account, you can create one using the link above.</p>
 <div class="footer">
 <p>© EZOFIS - Agentic Intelligence Platform</p>
 <p>This is an automated message, please do not reply.</p>
 </div>
 </div>
 </div>
 </body>
 </html>
 """,
 "textContent": f"""
{sender_display} shared a document extraction with you
Hello,
{sender_display} wants you to take a look at a document extraction output.
View the shared extraction: {share_link}
You'll need to sign in to your EZOFIS account to view this extraction. If you don't have an account, you can create one using the link above.
© EZOFIS - Agentic Intelligence Platform
This is an automated message, please do not reply.
 """
 }
headers = {
 "accept": "application/json",
 "api-key": BREVO_API_KEY,
 "content-type": "application/json"
 }
try:
 async with httpx.AsyncClient(timeout=30.0) as client:
 response = await client.post(BREVO_API_URL, json=payload, headers=headers)
 response.raise_for_status()
result = response.json()
 message_id = result.get('messageId', 'N/A')
 print(f"[INFO] Brevo share email sent successfully to {recipient_email}. Message ID: {message_id}")
 return True
 except httpx.HTTPStatusError as e:
 error_detail = {}
 try:
 error_detail = e.response.json() if e.response else {}
 except:
 error_detail = {"message": str(e)}
error_msg = error_detail.get('message', f'HTTP {e.response.status_code}' if e.response else 'Unknown error')
 print(f"[ERROR] Brevo API error: {e.response.status_code if e.response else 'N/A'} - {error_msg}")
 raise Exception(f"Failed to send email via Brevo: {error_msg}")
 except httpx.TimeoutException:
 print(f"[ERROR] Brevo API request timed out")
 raise Exception("Email service timeout. Please try again.")
 except Exception as e:
 print(f"[ERROR] Brevo email sending failed: {str(e)}")
 raise Exception(f"Failed to send email: {str(e)}")
async def create_brevo_contact(
 email: str,
 first_name: Optional[str] = None,
 last_name: Optional[str] = None,
 organization_name: Optional[str] = None,
 phone_number: Optional[str] = None,
 linkedin_url: Optional[str] = None,
 title: Optional[str] = None,
 headline: Optional[str] = None,
 organization_website: Optional[str] = None,
 organization_address: Optional[str] = None,
 list_id: Optional[int] = None
) -> bool:
 """
 Create a contact in Brevo and optionally add to a list.
Args:
 email: Contact email address (required)
 first_name: Contact first name
 last_name: Contact last name
 organization_name: Organization name
 phone_number: Phone number
 linkedin_url: LinkedIn profile URL
 title: Job title
 headline: Professional headline
 organization_website: Company website
 organization_address: Company address
 list_id: ID of the list to add contact to (e.g., 5 for "VRP Trials")
Returns:
 True if contact created successfully, False otherwise
 """
 if not BREVO_API_KEY:
 print("[WARNING] BREVO_API_KEY not set, skipping Brevo contact creation")
 return False
# Prepare contact attributes using automatic field mapping
 attributes = {}
# Map all fields automatically
 field_mappings = {
 "first_name": first_name,
 "last_name": last_name,
 "organization_name": organization_name,
 "phone_number": phone_number,
 "linkedin_url": linkedin_url,
 "title": title,
 "headline": headline,
 "organization_website": organization_website,
 "organization_address": organization_address,
 }
for field_name, field_value in field_mappings.items():
 if field_value:
 brevo_attr = _get_brevo_attribute_name(field_name)
 if brevo_attr:
 attributes[brevo_attr] = str(field_value).strip() # Ensure it's a string and trimmed
 print(f"[DEBUG] Mapped '{field_name}' ({field_value}) to Brevo attribute '{brevo_attr}'")
 else:
 print(f"[DEBUG] No Brevo attribute mapping found for '{field_name}'")
 else:
 print(f"[DEBUG] Skipping '{field_name}' - value is empty/None")
print(f"[DEBUG] Final Brevo attributes to send: {attributes}")
# Prepare contact data
 contact_data = {
 "email": email.lower(),
 "updateEnabled": True # Update existing contact if email already exists
 }
if attributes:
 contact_data["attributes"] = attributes
# Add to list if list_id is provided
 if list_id:
 contact_data["listIds"] = [list_id]
headers = {
 "accept": "application/json",
 "api-key": BREVO_API_KEY,
 "content-type": "application/json"
 }
try:
 async with httpx.AsyncClient(timeout=30.0) as client:
 response = await client.post(
 "https://api.brevo.com/v3/contacts",
 json=contact_data,
 headers=headers
 )
if response.status_code in [200, 201, 204]:
 print(f"[INFO] Successfully created Brevo contact: {email}" +
(f" and added to list {list_id}" if list_id else ""))
 return True
 elif response.status_code == 400:
 # Contact might already exist, try to update it
 try:
 error_data = response.json()
 if "already exists" in str(error_data).lower():
 print(f"[INFO] Contact {email} already exists in Brevo, updating...")
 # Use PUT to update existing contact
 update_response = await client.put(
 f"https://api.brevo.com/v3/contacts/{email.lower()}",
 json=contact_data,
 headers=headers
 )
 if update_response.status_code in [200, 204]:
 print(f"[INFO] Successfully updated Brevo contact: {email}" +
(f" and added to list {list_id}" if list_id else ""))
 return True
 except:
 pass
error_data = response.text
 print(f"[ERROR] Failed to create Brevo contact: {response.status_code} - {error_data}")
 return False
 else:
 error_data = response.text
 print(f"[ERROR] Failed to create Brevo contact: {response.status_code} - {error_data}")
 return False
except httpx.HTTPStatusError as e:
 print(f"[ERROR] Brevo API HTTP error: {e.response.status_code} - {e.response.text}")
 return False
 except Exception as e:
 print(f"[ERROR] Failed to create Brevo contact: {str(e)}")
 return False