Email Infrastructure for Cannabis Operators

import multimail

client = multimail.Client(api_key="mm_live_...")

def send_order_confirmation(order: dict, customer: dict) -> dict:
    "cm"># Only send to states where operator holds a license
    licensed_states = {"CA", "CO", "NV", "OR", "WA", "MI", "IL"}
    customer_state = customer.get("state", "").upper()

    if customer_state not in licensed_states:
        return {"status": "skipped", "reason": "recipient_state_not_licensed"}

    return client.send_email(
        from_address="[email protected]",
        to=customer["email"],
        subject=f"Order "cm">#{order['id']} confirmed — pickup ready",
        body=(
            f"Hi {customer[&"cm">#039;first_name']},\n\n"
            f"Your order "cm">#{order['id']} is confirmed for pickup at "
            f"{order[&"cm">#039;location_name']}.\n\n"
            f"Items: {&"cm">#039;, '.join(order['items'])}\n"
            f"Pickup window: {order[&"cm">#039;pickup_window']}\n\n"
            "This message contains no promotional content."
        ),
        tags=["transactional", "order-confirmation"],
    )

import multimail

client = multimail.Client(api_key="mm_live_...")

def queue_promotional_email(campaign: dict, recipient_segment: list[str]) -> dict:
    """
    Queues a promotional draft for human compliance review.
    No send occurs until approved via decide_email.
    """
    "cm"># Compose the draft — agent writes, human approves
    draft = client.send_email(
        from_address="[email protected]",
        to=recipient_segment,
        subject=campaign["subject"],
        body=campaign["body"],
        tags=["promotional", f"campaign:{campaign[&"cm">#039;id']}"],
        hold=True,  # gated_send: hold for human approval
        metadata={
            "campaign_id": campaign["id"],
            "target_states": campaign["states"],
            "age_verified_list": campaign["age_verified"],
        },
    )

    # Returns message_id for the reviewer to act on
    return {
        "message_id": draft["message_id"],
        "status": "pending_review",
        "review_url": f"https://app.multimail.dev/pending/{draft[&"cm">#039;message_id']}",
    }


def compliance_approve(message_id: str, reviewer_id: str) -> dict:
    """Called by the human reviewer after checking state restrictions."""
    return client.decide_email(
        message_id=message_id,
        decision="approve",
        reviewer=reviewer_id,
    )

import multimail
from datetime import datetime

client = multimail.Client(api_key="mm_live_...")

def send_loyalty_update(member: dict) -> dict:
    if not member.get("age_verified"):
        return {"status": "skipped", "reason": "age_not_verified"}

    result = client.send_email(
        from_address="[email protected]",
        to=member["email"],
        subject=f"You have {member[&"cm">#039;points']} points — here's what's available",
        body=(
            f"Hi {member[&"cm">#039;first_name']},\n\n"
            f"Your current balance: {member[&"cm">#039;points']} points.\n"
            f"Eligible rewards are available in-store at your next visit.\n\n"
            "Offers are subject to local availability and licensing.\n\n"
            "To stop receiving loyalty emails, reply STOP or "
            "visit dispensary.multimail.dev/unsubscribe."
        ),
        tags=["loyalty", "transactional"],
    )

    # Tag with age-verification audit trail
    client.tag_email(
        message_id=result["message_id"],
        tags=[
            f"age_verified:{member[&"cm">#039;age_verified_date']}",
            f"state:{member[&"cm">#039;state']}",
        ],
    )

    return result

import re
import multimail

client = multimail.Client(api_key="mm_live_...")

"cm"># FTC-flagged claim patterns for CBD/cannabis
PROHIBITED_PATTERNS = [
    r"\bcures\b",
    r"\btreats\b",
    r"\bheals\b",
    r"FDA[- ]approved",
    r"clinically proven",
    r"reduces (?:anxiety|inflammation|pain)",
    r"medical(?:ally)? (?:proven|tested|verified)",
]

def scan_sent_folder_for_violations(mailbox: str, days: int = 7) -> list[dict]:
    """Read-only scan — no mutations, no sends."""
    messages = client.check_inbox(
        mailbox=mailbox,
        folder="sent",
        since_days=days,
    )

    violations = []
    for msg_summary in messages["emails"]:
        msg = client.read_email(message_id=msg_summary["message_id"])
        body = msg["body"].lower()

        for pattern in PROHIBITED_PATTERNS:
            if re.search(pattern, body, re.IGNORECASE):
                violations.append({
                    "message_id": msg["message_id"],
                    "subject": msg["subject"],
                    "sent_at": msg["date"],
                    "matched_pattern": pattern,
                })
                break

    return violations