wuling-overseas-dealer-expansion-skill/scripts/email_outreach/send_outreach_emails.py

#!/usr/bin/env python3
import argparse
import html
import json
import mimetypes
import re
import smtplib
import ssl
import uuid
from email.message import EmailMessage
from email.utils import formatdate, make_msgid
from datetime import datetime
from pathlib import Path

try:
    from common.artifact_manager import resolve_artifact_path
except ImportError:  # pragma: no cover - supports direct CLI execution
    import sys
    sys.path.append(str(Path(__file__).resolve().parents[1]))
    from common.artifact_manager import resolve_artifact_path

IMAGE_PATTERN = re.compile(r"!\[([^\]]*)\]\(([^)]+)\)")
STRONG_PHRASES = [
    "Wuling Overseas Business Department",
    "SGMW / Wuling",
    "30 million vehicles",
    "1 million vehicles",
    "60+ countries",
    "200+ overseas sales and service outlets",
    "BYD",
    "Chery",
    "more practical vehicle models, broader customer scenarios, and clearer visible profit margins for dealers",
    "20-30 minute online meeting",
    "20–30 minute online meeting",
]


def extract_inline_images(markdown_body, base_dirs=None):
    images = []
    base_dirs = [Path(item) for item in (base_dirs or []) if item]

    def replace(match):
        alt = match.group(1).strip() or "image"
        raw_path = match.group(2).strip().strip('"')
        path = Path(raw_path)
        if not path.is_absolute():
            for base_dir in base_dirs:
                candidate = (base_dir / raw_path).resolve()
                if candidate.exists():
                    path = candidate
                    break
        cid = uuid.uuid4().hex
        images.append({"alt": alt, "path": path, "cid": cid})
        return f"[[INLINE_IMAGE:{len(images) - 1}]]"

    return IMAGE_PATTERN.sub(replace, markdown_body), images


def render_inline_markdown(text):
    escaped = html.escape(text)
    escaped = re.sub(r"\*\*(.+?)\*\*", r"<strong>\1</strong>", escaped)
    escaped = re.sub(
        r"\[([^\]]+)\]\((mailto:[^)]+)\)",
        r'<a href="\2">\1</a>',
        escaped,
    )
    escaped = re.sub(
        r"\[([^\]]+)\]\((https?://[^)]+)\)",
        r'<a href="\2">\1</a>',
        escaped,
    )
    escaped = re.sub(
        r"(Email:\s*)(?!<a)([A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,})",
        r'\1<a href="mailto:\2">\2</a>',
        escaped,
    )
    escaped = re.sub(
        r"(Website:\s*)(?!<a)(https?://[^\s<]+)",
        r'\1<a href="\2">\2</a>',
        escaped,
    )
    return escaped


def markdownish_to_html(body, base_dirs=None):
    body_without_images, images = extract_inline_images(body, base_dirs=base_dirs)
    lines = [line.rstrip() for line in body_without_images.replace("\r\n", "\n").split("\n")]
    html_blocks = []
    in_list = False

    def close_list():
        nonlocal in_list
        if in_list:
            html_blocks.append("</ul>")
            in_list = False

    for line in lines:
        stripped = line.strip()
        if not stripped:
            close_list()
            continue

        if stripped.startswith("[[INLINE_IMAGE:"):
            close_list()
            index = int(re.search(r"\d+", stripped).group(0))
            image = images[index]
            alt = html.escape(image["alt"])
            html_blocks.append(
                f'<div class="image-wrap" style="margin:10px 0 16px 0;">'
                f'<img src="cid:{image["cid"]}" alt="{alt}" width="300" '
                f'style="display:block;width:300px;max-width:300px;height:auto;border:0;outline:none;text-decoration:none;">'
                f'</div>'
            )
            continue

        if stripped.startswith("- "):
            if not in_list:
                html_blocks.append('<ul style="margin:4px 0 14px 23px;padding:0;">')
                in_list = True
            html_blocks.append(
                f'<li style="margin:0 0 7px 0;padding-left:3px;">{render_inline_markdown(stripped[2:].strip())}</li>'
            )
            continue

        close_list()
        css_class = "signature" if stripped in {"Senior Business Development Manager", "Wuling Overseas Business Department"} or stripped.startswith(("WhatsApp", "Email:", "Website:")) else ""
        margin = "0 0 3px 0" if css_class else "0 0 10px 0"
        if stripped.startswith("We can offer our partners"):
            margin = "24px 0 10px 0"
        if stripped == "Best regards,":
            margin = "24px 0 10px 0"
        html_blocks.append(
            f'<p class="{css_class}" style="margin:{margin};">{render_inline_markdown(stripped)}</p>'
        )

    close_list()

    styles = """
.email-shell {
  margin: 0;
  padding: 0;
  color: #1f2937;
  font-family: Arial, Helvetica, sans-serif;
  font-size: 12px;
  line-height: 1.55;
}
.email-shell p {
  margin: 0 0 10px 0;
}
.email-shell strong {
  font-weight: 700;
  color: #111827;
}
.email-shell .signature {
  margin: 0 0 3px 0;
}
.email-shell .image-wrap {
  margin: 10px 0 16px 0;
}
.email-shell img {
  display: block;
  width: 300px;
  max-width: 300px;
  height: auto;
  border: 0;
  outline: none;
  text-decoration: none;
}
.email-shell a {
  color: #2563eb;
  text-decoration: underline;
}
"""
    html_body = "\n".join(html_blocks)
    return f"""<!doctype html>
<html>
  <body>
    <style>{styles}</style>
    <div class="email-shell" style="margin:0;padding:0;color:#1f2937;font-family:Arial,Helvetica,sans-serif;font-size:12px;line-height:1.55;">
      {html_body}
    </div>
  </body>
</html>""", images


def plain_text_without_markdown_images(body):
    text = IMAGE_PATTERN.sub("", body)
    text = re.sub(r"\*\*(.+?)\*\*", r"\1", text)
    text = re.sub(r"\[([^\]]+)\]\((?:mailto:)?([^\)]+)\)", r"\1", text)
    return text.strip() + "\n"


def attach_inline_images(message, images):
    html_part = message.get_payload()[1]
    for image in images:
        path = image["path"]
        if not path.exists():
            raise FileNotFoundError(f"图片文件不存在：{path}")
        content_type, _ = mimetypes.guess_type(str(path))
        if not content_type or not content_type.startswith("image/"):
            content_type = "image/png"
        maintype, subtype = content_type.split("/", 1)
        html_part.add_related(path.read_bytes(), maintype=maintype, subtype=subtype, cid=f"<{image['cid']}>", disposition="inline")


def build_message(sender, sender_name, to, subject, body, base_dirs=None):
    message = EmailMessage()
    from_value = sender
    if sender_name:
        from_value = f"{sender_name} <{sender}>"
    message["From"] = from_value
    message["To"] = to
    message["Subject"] = subject
    message["Date"] = formatdate(localtime=True)
    message["Message-ID"] = make_msgid(domain="huatu.hk")
    message["Reply-To"] = sender
    message.set_content(plain_text_without_markdown_images(body))
    html_body, images = markdownish_to_html(body, base_dirs=base_dirs)
    message.add_alternative(html_body, subtype="html")
    attach_inline_images(message, images)
    return message


def main():
    parser = argparse.ArgumentParser(description="发送已预览确认的经销商建联 HTML 邮件。")
    parser.add_argument("--input", required=True, help="prepare 脚本生成的 JSON 预览文件")
    parser.add_argument("--smtp-host", required=True, help="SMTP 主机，例如 smtp.qq.com")
    parser.add_argument("--smtp-port", type=int, default=465, help="SMTP SSL 端口，默认 465")
    parser.add_argument("--sender", required=True, help="发件邮箱")
    parser.add_argument("--auth-code", required=True, help="SMTP 授权码；只用于本次连接，不写入日志")
    parser.add_argument("--sender-name", default="Chris Chen", help="Email From display name. Official email identity is fixed to Chris Chen.")
    parser.add_argument("--confirm-send", action="store_true", help="必须显式提供才会发送")
    parser.add_argument("--sent-log", default="", help="发送日志 JSONL 路径；默认写到预览文件旁边，不写入 skill 包")
    parser.add_argument("--run-id", default="", help="Run ID; default send log is stored under runs/YYYYMMDD/<run_id>/")
    args = parser.parse_args()

    if not args.confirm_send:
        raise SystemExit("未提供 --confirm-send，已停止发送。")

    input_path = Path(args.input)
    preview = json.loads(input_path.read_text(encoding="utf-8"))
    emails = preview.get("emails", [])
    if not emails:
        raise SystemExit("预览文件中没有可发送邮件。")

    template_value = preview.get("source", {}).get("template", "")
    template_path = Path(template_value) if template_value else None
    skill_root = Path(__file__).resolve().parents[2]
    base_dirs = [input_path.parent, Path.cwd(), skill_root]
    if template_path:
        if template_path.is_absolute():
            base_dirs.append(template_path.parent)
        else:
            base_dirs.append((Path.cwd() / template_path).parent)
            base_dirs.append((skill_root / template_path).parent)
    if args.sent_log:
        sent_log_path = Path(args.sent_log).expanduser()
        if not sent_log_path.is_absolute():
            sent_log_path = Path.cwd() / sent_log_path
    else:
        if "runs" in input_path.parts:
            sent_log_path = input_path.with_name("send-log.jsonl")
        else:
            sent_log_path = resolve_artifact_path(
                "",
                kind="email_send",
                default_name="send-log.jsonl",
                run_id=args.run_id or input_path.stem,
            )

    context = ssl.create_default_context()
    results = []
    with smtplib.SMTP_SSL(args.smtp_host, args.smtp_port, context=context) as server:
        server.login(args.sender, args.auth_code)
        for item in emails:
            try:
                message = build_message(
                    sender=args.sender,
                    sender_name=args.sender_name,
                    to=item["to"],
                    subject=item["subject"],
                    body=item["body"],
                    base_dirs=base_dirs,
                )
                server.send_message(message)
                sent_result = {"to": item["to"], "dealer_name": item.get("dealer_name", ""), "status": "sent"}
                results.append(sent_result)
                log_entry = {
                    "sent_at": datetime.now().isoformat(timespec="seconds"),
                    "to": item["to"],
                    "dealer_name": item.get("dealer_name", ""),
                    "subject": item.get("subject", ""),
                    "sender": args.sender,
                }
                sent_log_path.parent.mkdir(parents=True, exist_ok=True)
                with sent_log_path.open("a", encoding="utf-8") as log_file:
                    log_file.write(json.dumps(log_entry, ensure_ascii=False) + "\n")
            except Exception as exc:
                results.append({
                    "to": item["to"],
                    "dealer_name": item.get("dealer_name", ""),
                    "status": "failed",
                    "error": str(exc),
                })

    print(json.dumps({"sent": sum(1 for r in results if r["status"] == "sent"), "results": results}, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()