def build_mime_message( from_email: str, to_email: str, subject: str, html_body: str = "", text_body: str = "", attachments: Optional[list] = None, # list of {filename, content(bytes), mime_type} unsubscribe_url: str = "", unsubscribe_mailto: str = "", ) -> MIMEMultipart: """Build a real MIME multipart message. From header always equals the authenticated account.""" attachments = attachments or [] if html_body and text_body: msg = MIMEMultipart("alternative") else: msg = MIMEMultipart("mixed") msg["From"] = from_email # MUST be the authenticated Gmail account (never spoofed) msg["To"] = to_email msg["Subject"] = subject msg["Date"] = formatdate(localtime=True) msg["Message-ID"] = make_msgid() msg["X-Mailer"] = "BulkGmailMailer (permission-based sending)" # List-Unsubscribe headers for compliance if unsubscribe_url: parts = [f"<{unsubscribe_url}>"] if unsubscribe_mailto: parts.append(f"") msg["List-Unsubscribe"] = ", ".join(parts) msg["List-Unsubscribe-Post"] = "List-Unsubscribe=One-Click" # Text part first, then HTML (preferred last in alternative) if text_body: msg.attach(MIMEText(text_body, "plain", "utf-8")) if html_body: msg.attach(MIMEText(html_body, "html", "utf-8")) # Attachments for att in attachments: data = att["content"] ctype = att.get("mime_type") or "application/octet-stream" maintype, subtype = ctype.split("/", 1) if maintype == "text": part = MIMEText(data.decode("utf-8", errors="replace"), subtype, "utf-8") part.set_param("name", att["filename"], header="Content-Disposition") elif maintype == "image": part = MIMEImage(data, _subtype=subtype) elif maintype == "application" and subtype == "pdf": part = MIMEApplication(data, _subtype=subtype) else: part = MIMEBase(maintype, subtype) part.set_payload(data) part.add_header("Content-Disposition", "attachment", filename=att["filename"]) msg.attach(part) return msg def html_to_pdf(html: str) -> bytes: """Real HTML → PDF conversion via WeasyPrint.""" try: return WeasyHTML(string=html).write_pdf() except Exception as e: raise HtmlValidationError(f"PDF conversion failed: {e}") from e def sanitize_preview(html: str) -> str: """Sanitize HTML for safe rendering in the in-app editor/preview (server-side).""" import bleach allowed_tags = list(bleach.ALLOWED_TAGS) + ["table", "thead", "tbody", "tr", "td", "th", "img", "p", "span", "div", "h1", "h2", "h3", "strong", "em", "a", "ul", "ol", "li", "br", "hr", "font", "center"] allowed_attrs = {"a": ["href", "title"], "img": ["src", "alt", "width", "height"], "*": ["style", "class", "align", "bgcolor", "width", "border", "cellpadding", "cellspacing"]} protocols = ["http", "https", "mailto"] return bleach.clean(html or "", tags=allowed_tags, attributes=allowed_attrs, protocols=protocols, strip=False) def sanitize_html_for_email(html: str) -> str: """Remove scripts/forms from outgoing email HTML (keep email-safe markup).""" return sanitize_preview(html) def generate_attachment_filename(kind: str, invoice_number: str) -> str: """Unique, deterministic filenames like Invoice_INV-XXXX.pdf per campaign rules.""" inv = re.sub(r"[^A-Za-z0-9-]", "", invoice_number or "") if kind == "invoice": return f"Invoice_{inv or 'INV' + secrets.token_hex(3)}.pdf" if kind == "order": return f"Order_{inv or 'ORD' + secrets.token_hex(3)}.pdf" ---IMPORTS--- 9:import csv 10:import io 11:import json 12:import secrets 13:import uuid 14:from datetime import datetime, timedelta, timezone 16:import pytest 40: import redis as _r