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

#!/usr/bin/env python3
import argparse
import json
import re
import sys
import zipfile
import xml.etree.ElementTree as ET
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
    sys.path.append(str(Path(__file__).resolve().parents[1]))
    from common.artifact_manager import resolve_artifact_path

NS = {
    "main": "http://schemas.openxmlformats.org/spreadsheetml/2006/main",
    "rel": "http://schemas.openxmlformats.org/officeDocument/2006/relationships",
}

ALIASES = {
    "dealer_name": "客户姓名/公司",
    "country": "国家",
    "city": "城市",
    "dealer_type": "客户类型",
    "page_url": "主页/链接",
    "contact_name": "联系人",
    "position": "职位",
    "phone": "电话/WhatsApp",
    "email": "邮箱",
    "main_business": "主营业务",
    "status": "建联状态",
    "next_follow_up": "下次跟进",
    "notes": "备注",
}

SENT_STATUS_MARKERS = [
    "已发送邮件",
    "已发邮件",
    "邮件已发送",
    "邮件发送成功",
    "email sent",
    "sent email",
    "sent",
    "success",
]


def already_sent(status):
    normalized = str(status or "").strip().casefold()
    if not normalized:
        return False
    explicit_email_markers = [
        "已发送邮件",
        "已发邮件",
        "邮件已发送",
        "邮件发送成功",
        "email sent",
        "sent email",
    ]
    if any(marker.casefold() in normalized for marker in explicit_email_markers):
        return True
    return normalized in {"已发送", "sent", "success"}


EMAIL_PATTERN = re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}")


def extract_email_addresses(*values):
    emails = []
    for value in values:
        for email in EMAIL_PATTERN.findall(str(value or "")):
            normalized = email.strip().strip(".,;，；")
            if normalized and normalized.casefold() not in {item.casefold() for item in emails}:
                emails.append(normalized)
    return emails


def col_to_index(ref):
    letters = "".join(ch for ch in ref if ch.isalpha())
    total = 0
    for ch in letters:
        total = total * 26 + (ord(ch.upper()) - ord("A") + 1)
    return total - 1


def read_shared_strings(zip_file):
    if "xl/sharedStrings.xml" not in zip_file.namelist():
        return []
    root = ET.fromstring(zip_file.read("xl/sharedStrings.xml"))
    strings = []
    for item in root.findall("main:si", NS):
        strings.append("".join((t.text or "") for t in item.iter(f"{{{NS['main']}}}t")))
    return strings


def cell_value(cell, shared_strings):
    cell_type = cell.attrib.get("t")
    if cell_type == "inlineStr":
        return "".join((t.text or "") for t in cell.iter(f"{{{NS['main']}}}t"))
    value = cell.find("main:v", NS)
    if value is None:
        return ""
    raw = value.text or ""
    if cell_type == "s":
        return shared_strings[int(raw)]
    return raw


def read_sheet_rows(xlsx_path, sheet_name):
    with zipfile.ZipFile(xlsx_path) as archive:
        shared_strings = read_shared_strings(archive)
        workbook = ET.fromstring(archive.read("xl/workbook.xml"))
        sheets = workbook.find("main:sheets", NS).findall("main:sheet", NS)
        sheet = next((s for s in sheets if s.attrib.get("name") == sheet_name), None)
        if sheet is None:
            available = ", ".join(s.attrib.get("name", "") for s in sheets)
            raise SystemExit(f"找不到工作表：{sheet_name}。可用工作表：{available}")

        rels = ET.fromstring(archive.read("xl/_rels/workbook.xml.rels"))
        rel_map = {rel.attrib["Id"]: rel.attrib["Target"] for rel in rels}
        rel_id = sheet.attrib[f"{{{NS['rel']}}}id"]
        target = rel_map[rel_id]
        target = target.lstrip("/")
        sheet_path = target if target.startswith("xl/") else "xl/" + target
        root = ET.fromstring(archive.read(sheet_path))

        parsed_rows = []
        for row in root.findall(".//main:sheetData/main:row", NS):
            values = {}
            for cell in row.findall("main:c", NS):
                values[col_to_index(cell.attrib.get("r", ""))] = cell_value(cell, shared_strings)
            if values:
                max_index = max(values)
                parsed_rows.append([values.get(i, "") for i in range(max_index + 1)])

    if not parsed_rows:
        return []
    headers = [str(value).strip() for value in parsed_rows[0]]
    records = []
    for row in parsed_rows[1:]:
        record = {}
        for index, header in enumerate(headers):
            if header:
                record[header] = str(row[index]).strip() if index < len(row) else ""
        if any(record.values()):
            records.append(record)
    return records


def parse_filters(filter_args):
    filters = []
    for item in filter_args:
        if "=" not in item:
            raise SystemExit(f"筛选条件格式错误，应为 字段=值：{item}")
        key, value = item.split("=", 1)
        filters.append((key.strip(), value.strip()))
    return filters


def matches(record, filters):
    for key, expected in filters:
        actual = record.get(key, "")
        if str(actual).strip() != expected:
            return False
    return True


def choose_capability_phrase(record):
    text = " ".join(str(record.get(key, "")) for key in (
        "客户类型",
        "主营业务",
        "公司主营业务",
        "备注",
        "客户属性",
        "官网链接",
        "主页/链接",
    )).casefold()

    if any(token in text for token in (
        "location", "rental", "rent", "leasing", "fleet", "flotte", "车队", "租赁", "政府", "corporate", "enterprise"
    )):
        return "corporate customer resources"
    if any(token in text for token in (
        "network", "réseau", "reseau", "dealer network", "downstream", "全国分销", "区域分销", "分销", "代理", "group", "groupe"
    )):
        return "a dealer network"
    if any(token in text for token in (
        "import", "importation", "importateur", "importer", "认证", "certification", "进口"
    )):
        return "vehicle import capabilities"
    if any(token in text for token in (
        "multi-brand", "multibrand", "multi brand", "多品牌", "showroom", "concessionnaire", "dealer", "汽车经销", "经销", "renault", "dacia", "audi", "volkswagen", "skoda", "škoda", "fiat", "jeep", "porsche", "kia"
    )):
        return "multi-brand distribution experience"
    return "local vehicle sales and customer resources"



def choose_platform_resource_phrase(record):
    text = " ".join(str(record.get(key, "")) for key in (
        "客户类型",
        "主营业务",
        "公司主营业务",
        "备注",
        "客户属性",
        "官网链接",
        "主页/链接",
    )).casefold()

    if any(token in text for token in (
        "交易平台", "marketplace", "platform", "annuaire", "portal", "site", "网站", "流量", "traffic", "leads", "线索"
    )):
        return "automotive industry traffic"
    if any(token in text for token in (
        "协会", "商会", "联盟", "federation", "association", "chamber", "alliance", "member", "会员", "network", "réseau", "reseau"
    )):
        return "dealer members and industry resources"
    if any(token in text for token in (
        "fleet", "flotte", "corporate", "enterprise", "政府", "institution", "机构", "企业客户", "车队", "采购"
    )):
        return "corporate customers and procurement resources"
    if any(token in text for token in (
        "dealer", "concessionnaire", "distributor", "importer", "importateur", "经销商", "进口商", "分销"
    )):
        return "dealer and importer resources"
    return "automotive industry resources"


def render_template(template, record, name_field="客户姓名/公司", sender_name=""):
    values = dict(record)
    for alias, header in ALIASES.items():
        values.setdefault(alias, record.get(header, ""))
    display_name = record.get(name_field, "").strip() or record.get("客户姓名/公司", "").strip()
    if not display_name:
        raise ValueError("[Name]/{{customer_name}} cannot be replaced because the customer name is empty.")
    values["dealer_name"] = display_name
    values["customer_name"] = display_name
    values["sender_name"] = sender_name.strip()
    values.setdefault("capability_phrase", choose_capability_phrase(record))
    values.setdefault("platform_resource_phrase", choose_platform_resource_phrase(record))
    rendered = template.replace("[Name]", display_name).replace("[name]", display_name)
    rendered = rendered.replace("[PICTURE]", "![Wuling Morocco Market Cooperation Evaluation](assets/wuling-channel-margin-comparison.png)")

    def replace(match):
        key = match.group(1).strip()
        return str(values.get(key, match.group(0)))

    rendered = re.sub(r"\{\{\s*([^}]+?)\s*\}\}", replace, rendered)
    unresolved_required = re.findall(r"\[(?:Name|name)\]|\{\{\s*(?:customer_name|dealer_name|sender_name)\s*\}\}", rendered)
    if unresolved_required:
        raise ValueError(f"Unresolved required placeholder(s): {', '.join(sorted(set(unresolved_required)))}")
    return rendered


def main():
    parser = argparse.ArgumentParser(description="从 Excel 生成经销商建联邮件预览，并替换 Dear [Name] 与模板占位符。")
    parser.add_argument("--excel", required=True, help="Excel .xlsx 文件路径")
    parser.add_argument("--sheet", required=True, help="工作表名称")
    parser.add_argument("--filter", action="append", default=[], help="筛选条件，格式为 字段=值，可重复传入")
    parser.add_argument("--template", required=True, help="邮件 .md 或 .txt 模板路径")
    parser.add_argument("--subject", required=True, help="邮件主题")
    parser.add_argument("--output", required=True, help="输出 JSON 预览文件")
    parser.add_argument("--run-id", default="", help="Run ID; bare output filenames are stored under runs/YYYYMMDD/<run_id>/")
    parser.add_argument("--email-field", default="邮箱", help="主收件邮箱字段名")
    parser.add_argument("--extra-email-field", action="append", default=[], help="补充收件邮箱字段名，可重复传入，例如 个人邮箱")
    parser.add_argument("--name-field", default="客户姓名/公司", help="客户名称/公司字段名")
    parser.add_argument("--sender-name", default="Chris Chen", help="Email sender display/body fallback. Email templates use fixed Chris Chen; social outreach uses per-account sender names separately.")
    parser.add_argument("--status-field", default="建联状态", help="建联状态字段名")
    parser.add_argument("--include-sent", action="store_true", help="包含已发送邮件的记录")
    args = parser.parse_args()

    excel_path = Path(args.excel)
    template_path = Path(args.template)
    if not excel_path.exists():
        raise SystemExit(f"Excel 文件不存在：{excel_path}")
    if not template_path.exists():
        raise SystemExit(f"邮件模板不存在：{template_path}")

    filters = parse_filters(args.filter)
    template = template_path.read_text(encoding="utf-8-sig")
    if re.search(r"\{\{\s*sender_name\s*\}\}", template) and not args.sender_name.strip():
        raise SystemExit("Missing --sender-name for a legacy email template. Current official email templates must use fixed Chris Chen; social outreach uses per-account sender names separately.")
    records = read_sheet_rows(excel_path, args.sheet)
    matched = [record for record in records if matches(record, filters)]

    emails = []
    skipped = []
    for index, record in enumerate(matched, 1):
        status_value = record.get(args.status_field, "").strip()
        email_fields = [args.email_field] + list(args.extra_email_field)
        recipients = extract_email_addresses(*(record.get(field, "") for field in email_fields))
        display_name = record.get(args.name_field, "").strip()
        if already_sent(status_value) and not args.include_sent:
            skipped.append({"index": index, "dealer_name": display_name, "reason": f"已发送过邮件：{status_value}", "record": record})
            continue
        if not display_name:
            skipped.append({"index": index, "dealer_name": "", "reason": f"缺少{args.name_field}", "record": record})
            continue
        if not recipients:
            skipped.append({"index": index, "dealer_name": display_name, "reason": f"缺少收件邮箱字段：{', '.join(email_fields)}", "record": record})
            continue
        for recipient in recipients:
            emails.append({
                "to": recipient,
                "dealer_name": display_name,
                "subject": render_template(args.subject, record, args.name_field, args.sender_name),
                "body": render_template(template, record, args.name_field, args.sender_name),
                "record": record,
            })

    result = {
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "source": {
            "excel": str(excel_path),
            "sheet": args.sheet,
            "filters": args.filter,
            "template": str(template_path),
            "email_field": args.email_field,
            "extra_email_field": args.extra_email_field,
            "name_field": args.name_field,
            "status_field": args.status_field,
            "include_sent": args.include_sent,
        },
        "summary": {
            "matched_records": len(matched),
            "ready_to_send": len(emails),
            "skipped": len(skipped),
        },
        "emails": emails,
        "skipped": skipped,
    }

    output_path = resolve_artifact_path(
        args.output,
        kind="email_preview",
        default_name="outreach_preview.json",
        run_id=args.run_id or None,
    )
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    json.dump(result["summary"], sys.stdout, ensure_ascii=False)
    print()


if __name__ == "__main__":
    main()