"""Outbound email for the chatbot. Mailchimp Transactional (Mandrill) is the only delivery path. A tenant must have an active Mailchimp connection (app/services/integrations/mailchimp.py) to send email at all -- there is no shared-SMTP fallback. That fallback used to send every tenant's mail from one global address regardless of which tenant it was for, which is exactly the identity mix-up this module now avoids by requiring each tenant to send from their own connected account. A tenant with no connection, or whose connection fails (revoked key, Mandrill outage), gets that failure back rather than a silent send under a different identity. Email bodies are always composed by the caller from knowledge-base content -- this module only formats and delivers. """ import logging import re from datetime import datetime from app.core.config import settings from app.core.email_branding import brand_for_tenant from app.core.email_template import ( CTA_BLOCK, DISCLAIMER_BLOCK, EMAIL_TEMPLATE, IMAGE_SHELL, LINKS_BLOCK, LOGO_CELL, UNSUBSCRIBE_BLOCK, ) from app.services.messaging.email_image import generate_body_image, image_generation_available from app.services.chat.tools_settings import get_email_settings, render_placeholders from app.services.integrations.mailchimp import get_mailchimp_connection, send_via_mandrill logger = logging.getLogger(__name__) # Deliberately permissive: reject the obviously-wrong, let the mail server be # the real authority on deliverability. EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[a-zA-Z]{2,}$") # Used when no tenant is given (mirrors tools_settings.EMAIL_COPY_DEFAULTS). DEFAULT_COPY = { "header": "Thanks for chatting with {{brandName}}", "footer": "© {{currentYear}} {{brandName}}. All rights reserved.", "aiDisclaimer": ("Responses are powered by AI and may contain errors or omissions. " "Please verify important information before relying on it."), } class EmailNotConfigured(Exception): """Raised when email is requested but no transport is configured.""" def is_valid_email(address: str) -> bool: return bool(address and EMAIL_RE.match(address.strip())) def _mailchimp_connection_for(tenant_id: str): """This tenant's active Mailchimp connection, or None.""" if not tenant_id: return None return get_mailchimp_connection(tenant_id) def email_configured(tenant_id: str = None) -> bool: return bool(_mailchimp_connection_for(tenant_id)) def _render_html(subject: str, body_html: str, first_name: str = None, tenant_id: str = None, settings_override: dict = None) -> str: """Fill the branded template using this tenant's branding and copy.""" brand = brand_for_tenant(tenant_id) copy = settings_override or get_email_settings(tenant_id) if tenant_id else DEFAULT_COPY year = str(datetime.now().year) def fill(text): return _escape(render_placeholders(text, brand.name, year)) disclaimer = "" if (copy.get("aiDisclaimer") or "").strip(): disclaimer = DISCLAIMER_BLOCK.replace("{disclaimerText}", fill(copy["aiDisclaimer"])) cta = CTA_BLOCK.replace("{chatbotUrl}", brand.chatbot_url) if ( brand.show_cta and brand.chatbot_url) else "" unsubscribe = UNSUBSCRIBE_BLOCK.replace("{unsubscribeUrl}", brand.unsubscribe_url) if ( brand.show_unsubscribe and brand.unsubscribe_url) else "" # A tenant with no logo gets its name alone. Showing GalaxiQ's logo on # another company's email would be worse than showing none. logo_cell = "" if brand.logo_url: logo_cell = LOGO_CELL.replace("{logoUrl}", brand.logo_url).replace( "{brandName}", _escape(brand.name)) # "Hi there," reads better than "Hi ," when we never learned their name. greeting = _escape((first_name or "").strip().split(" ")[0]) or "there" html = EMAIL_TEMPLATE for key, value in { "{subject}": _escape(subject), "{brandName}": _escape(brand.name), "{firstName}": greeting, "{generatedAnswer}": body_html, "{logoCell}": logo_cell, "{brandCellPad}": ' style="padding-left: 12px;"' if logo_cell else "", "{headerText}": fill(copy.get("header") or ""), "{footerText}": fill(copy.get("footer") or ""), "{disclaimerBlock}": disclaimer, "{ctaBlock}": cta, "{unsubscribeBlock}": unsubscribe, }.items(): html = html.replace(key, value) return html def _render_image_shell(subject, image_url, brand, copy, first_name, links): """HTML around a generated body image. Links stay real HTML so they remain clickable, and the footer keeps enough text that the mail is not image-only.""" year = str(datetime.now().year) def fill(text): return _escape(render_placeholders(text or "", brand.name, year)) logo_cell = "" if brand.logo_url: logo_cell = LOGO_CELL.replace("{logoUrl}", brand.logo_url).replace( "{brandName}", _escape(brand.name)) link_items = "" if links: link_items = "  ·  ".join( f'{_escape(u.replace("https://", "").replace("http://", ""))}' for u in links[:6] ) html = IMAGE_SHELL for key, value in { "{subject}": _escape(subject), "{brandName}": _escape(brand.name), "{firstName}": _escape((first_name or "").strip().split(" ")[0]) or "there", "{headerText}": fill(copy.get("header")), "{footerText}": fill(copy.get("footer")), "{disclaimerText}": fill(copy.get("aiDisclaimer")), "{bodyImageUrl}": image_url, "{imageAlt}": _escape(subject), "{primaryColor}": brand.primary_color or "#4B4B8F", "{headerTint}": (brand.primary_color or "#4B4B8F") + "14", "{logoCell}": logo_cell, "{brandCellPad}": ' style="padding-left:12px;"' if logo_cell else "", "{linksBlock}": LINKS_BLOCK.replace("{linkItems}", link_items) if link_items else "", }.items(): html = html.replace(key, value) return html def _text_to_html(text: str) -> str: """Turn the model's plain-text body into simple paragraphs and bullets.""" html_parts = [] bullets = [] def flush_bullets(): if bullets: items = "".join(f"
  • {b}
    • " for b in bullets) html_parts.append(f"") bullets.clear() for raw in (text or "").split("\n"): line = raw.strip() if not line: flush_bullets() continue if line.startswith(("- ", "* ", "• ")): bullets.append(_escape(line[2:].strip())) else: flush_bullets() html_parts.append(f"

    {_escape(line)}

    ") flush_bullets() return "".join(html_parts) def _escape(s: str) -> str: return (s.replace("&", "&").replace("<", "<").replace(">", ">")) def send_email(to_address: str, subject: str, body_text: str, first_name: str = None, tenant_id: str = None) -> dict: """Send one email. Returns a result dict; never raises for delivery failure. body_text is plain text (as written by the model from knowledge-base content); it is rendered to simple HTML and also sent as the text part. """ if not email_configured(tenant_id): logger.error(f"send_email called for {tenant_id} but no Mailchimp connection is active.") return {"status": "error", "detail": "Connect Mailchimp to send emails."} # Every outbound email passes through here, so the tenant's switch is enforced # at this choke point rather than only in the chat tool. Any other caller -- # a ticket confirmation, a scheduled job, a script -- would otherwise send # mail for a tenant that has emailing turned off. if tenant_id and not get_email_settings(tenant_id).get("enabled", True): logger.warning(f"Blocked outbound email for {tenant_id}: emailing is disabled.") return {"status": "error", "detail": "Emailing is turned off for this business."} to_address = (to_address or "").strip() if not is_valid_email(to_address): return {"status": "error", "detail": f"'{to_address}' is not a valid email address."} subject = (subject or "").strip() or f"Information from {settings.EMAIL_FROM_NAME}" brand = brand_for_tenant(tenant_id) # The model designs a layout around this tenant's brand; the static template # is the fallback when generation fails or the output is not email-safe. copy = get_email_settings(tenant_id) if tenant_id else DEFAULT_COPY html = None # The body is a generated image: model-written HTML varied too much between # sends. Links and footer stay real HTML around it so they remain clickable. if image_generation_available(): image = generate_body_image(brand=brand, subject=subject, body_text=body_text or "", tenant_id=tenant_id) if image: html = _render_image_shell(subject, image["url"], brand, copy, first_name, list(brand.reference_links or [])) # Static template only as a safety net: an email must still go out if image # generation or hosting fails. if not html: logger.warning("Body image unavailable; falling back to the static template.") html = _render_html(subject, _text_to_html(body_text), first_name, tenant_id, copy) # email_configured() already confirmed this exists; re-fetching here # (rather than threading it through) keeps this function's only # dependency on tenant state at the top, where the guard is. mailchimp = _mailchimp_connection_for(tenant_id) result = send_via_mandrill(mailchimp, to_address, subject, body_text or "", html, reply_to=brand.reply_to) if result["status"] != "sent": return result # Audit trail: what actually left the system, not just that something did. preview = " ".join((body_text or "").split())[:300] logger.info( f"EMAIL SENT | to={to_address} | subject={subject!r} | " f"chars={len(body_text or '')} | body={preview!r}" ) return {"status": "sent", "to": to_address, "subject": subject}