wuling-overseas-dealer-expansion-skill/scripts/social/prepare_facebook_outreach.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Generate English Facebook outreach previews from the customer workbook."""
from __future__ import annotations

import argparse
import json
import random
import re
import sys
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence, Tuple

from openpyxl import load_workbook

SCRIPT_DIR = Path(__file__).resolve().parent
SKILL_ROOT = SCRIPT_DIR.parents[1]
sys.path.insert(0, str(SKILL_ROOT / "scripts" / "common"))
try:
    from workbook_resolver import resolve_workbook_path
except Exception:  # pragma: no cover
    resolve_workbook_path = None

DEFAULT_SHEET = "Facebook"
DEFAULT_STATUS = "未联系"

HEADER_ALIASES = {
    "index": ["序号", "Index", "No."],
    "name": ["客户姓名/公司", "公司名称", "客户名称", "Name", "Company"],
    "country": ["国家", "Country"],
    "city": ["城市", "City"],
    "type": ["客户类型", "类型", "Type"],
    "link": ["主页/链接", "Facebook链接", "主页", "链接", "Link", "URL"],
    "website": ["公司官网", "官网", "Website", "Official Website"],
    "contact": ["联系人", "姓名", "Contact"],
    "position": ["职位", "职务", "Position", "Title"],
    "phone": ["电话/WhatsApp", "电话", "WhatsApp", "Phone"],
    "email": ["邮箱", "Email"],
    "business": ["主营业务", "公司主营业务", "业务", "Business"],
    "status": ["建联状态", "建联情况", "状态", "Status"],
    "next_followup": ["下次跟进", "Next Follow-up"],
    "note": ["备注", "说明", "Notes"],
}

CONTACTED_MARKERS = [
    "已发送邮件", "已发邮件", "邮件已发送", "邮件发送成功", "已加好友", "已发送好友请求",
    "已发私信", "已发送私信", "已联系", "email sent", "sent", "success",
]

OEM_BRANDS = [
    "byd", "bmw", "jac", "mercedes", "toyota", "renault", "dacia", "kia", "hyundai",
    "volkswagen", "peugeot", "citroen", "citroën", "ford", "chery", "geely", "audi",
    "honda", "nissan", "suzuki", "mazda", "fiat", "opel", "skoda", "seat", "mg",
    "changan", "foton", "sitrak", "dfsk", "great wall", "haval", "dongfeng", "jetour", "baic",
    "gac", "maxus", "wuling",
]
OEM_BRANCH_HINTS = [
    "official", "officiel", "page officielle", "maroc", "morocco", "branch", "subsidiary",
    "官方", "分公司", "当地分公司", "国家页",
]

SCENARIOS = {
    "used_car_dealer": {
        "label_cn": "二手车商/occasion渠道",
        "judgment_cn": "该客户有二手车或 occasion 客户基础。值得建联的原因是其客户通常对总购车成本敏感，可能愿意评估低成本新车作为二手车库存补充；批量潜力取决于库存周转和本地客源规模。",
        "angle_cn": "从二手车客户升级到低成本新车的利润机会切入，强调先看小批量和价格区间，不压库存。",
        "signal_label": "used-car/showroom activity",
        "customer_base": "used-car buyers",
        "fit_context": "buyers who want a new vehicle but still care most about total cost",
    },
    "multibrand_dealer": {
        "label_cn": "多品牌经销商/展厅",
        "judgment_cn": "该客户像多品牌经销商或 showroom，已有汽车销售场景和客户流量，可能具备消化一批高性价比车型的能力。",
        "angle_cn": "从补充现有品牌和库存缺口切入，突出低成本新车线和小批量测试带来的走量可能。",
        "signal_label": "multi-brand/showroom activity",
        "customer_base": "showroom buyers",
        "fit_context": "buyers comparing practical new vehicles across brands",
    },
    "commercial_vehicle_channel": {
        "label_cn": "商用车/车队/实用车型渠道",
        "judgment_cn": "该客户涉及商用车、车队、配送或实用车型。五菱的经济实用定位适合小企业、配送和家商两用需求，有机会形成批量采购或渠道分销。",
        "angle_cn": "围绕小企业、配送、工具车和家商两用需求，测试实用低成本车型的批量消化能力。",
        "signal_label": "commercial/practical vehicle activity",
        "customer_base": "SME and practical-vehicle buyers",
        "fit_context": "customers watching purchase cost, uptime, and practical daily use",
    },
    "local_showroom": {
        "label_cn": "本地展厅/区域车商",
        "judgment_cn": "该客户像本地展厅或区域车商，直接接触本地终端客户，适合用小批量方式测试价格敏感市场的接受度。",
        "angle_cn": "从本地客户对价格和周转敏感切入，强调低压力首批试单。",
        "signal_label": "local auto sales activity",
        "customer_base": "local showroom buyers",
        "fit_context": "local buyers who compare total cost before choosing a vehicle",
    },
    "importer_group": {
        "label_cn": "进口商/集团/分销渠道",
        "judgment_cn": "该客户可能具备进口、集团、分销或区域渠道能力，可能不只消化零售订单，也可能评估持续批量供货和区域分销。",
        "angle_cn": "从进口/分销能力切入，先判断是否愿意评估首批试单和后续批量潜力。",
        "signal_label": "import/distribution activity",
        "customer_base": "regional dealer or importer networks",
        "fit_context": "channels that can evaluate a first batch and possible later volume",
    },
    "rental_fleet": {
        "label_cn": "租赁/车队客户",
        "judgment_cn": "该客户经营租赁或车队。车队客户对购置成本、维护成本和车辆周转敏感，可能通过小批量先验证五菱车型适配度。",
        "angle_cn": "围绕车队更新成本和车辆使用成本切入，先验证小批量车辆是否适合租赁/车队场景。",
        "signal_label": "rental/fleet activity",
        "customer_base": "rental or fleet buyers",
        "fit_context": "fleet operators trying to lower renewal and operating cost",
    },
    "unknown_auto_channel": {
        "label_cn": "信息不足的汽车相关渠道",
        "judgment_cn": "该客户看起来与汽车业务相关，但职责和渠道能力不明确。可以轻量建联，但重点是先确认其是否涉及采购、销售、进口或分销，不应直接强推。",
        "angle_cn": "信息不足时降低推销强度，先确认对方是否负责车辆采购、销售、进口或分销。",
        "signal_label": "auto-sector activity",
        "customer_base": "local auto-sector contacts",
        "fit_context": "channels that may handle vehicle sourcing, sales, import, or distribution",
    },
}

SIGNAL_PATTERNS = [
    ("used_car", ["used", "occasion", "second hand", "pre-owned", "reprise", "二手", "置换"], "二手车/occasion", "used-car or occasion activity"),
    ("showroom", ["showroom", "concessionnaire", "dealer", "multimarque", "multi-brand", "展厅", "经销", "多品牌"], "showroom/经销", "showroom or dealer activity"),
    ("stock", ["stock", "inventory", "parc auto", "annonce", "库存", "车源", "车辆较多"], "库存/车源", "visible stock or vehicle listings"),
    ("rental_fleet", ["rental", "rentcar", "location", "flotte", "fleet", "租赁", "租车", "车队"], "location/fleet", "rental or fleet activity"),
    ("import_distribution", ["import", "importation", "importateur", "distributeur", "distribution", "group", "groupe", "进口", "分销", "集团"], "importation/distribution", "import or distribution activity"),
    ("commercial", ["commercial", "utilitaire", "truck", "camion", "van", "delivery", "mpv", "商用", "货车", "卡车", "配送", "微型车"], "商用/实用车型", "commercial or practical-vehicle activity"),
    ("website", ["官网", "website", "http", ".ma", ".com", ".net"], "官网/正式页面", "official website or public business page"),
    ("contactable", ["whatsapp", "phone", "email", "电话", "邮箱", "公开联系方式", "+212"], "公开联系方式", "public WhatsApp, phone, or email"),
    ("active_page", ["recent", "post", "active", "近期", "发帖", "活跃", "followers", "粉丝"], "主页活跃信号", "recent page activity"),
]


def clean(value: Any) -> str:
    if value is None:
        return ""
    return re.sub(r"\s+", " ", str(value).strip())


def header_map(headers: Sequence[Any]) -> Dict[str, int]:
    raw = {clean(header): idx for idx, header in enumerate(headers) if clean(header)}
    mapped: Dict[str, int] = {}
    for key, aliases in HEADER_ALIASES.items():
        for alias in aliases:
            if alias in raw:
                mapped[key] = raw[alias]
                break
    return mapped


def read_records(excel_path: Path, sheet_name: str) -> List[Dict[str, str]]:
    wb = load_workbook(excel_path, read_only=True, data_only=True)
    if sheet_name not in wb.sheetnames:
        raise KeyError(f"Sheet not found: {sheet_name}")
    ws = wb[sheet_name]
    rows = list(ws.iter_rows(values_only=True))
    if not rows:
        return []
    columns = header_map(rows[0])
    records: List[Dict[str, str]] = []
    for row_number, raw in enumerate(rows[1:], start=2):
        record = {key: clean(raw[idx]) if idx < len(raw) else "" for key, idx in columns.items()}
        record["_row_number"] = str(row_number)
        if any(record.get(key) for key in ["name", "link", "phone", "email", "website", "business", "note"]):
            records.append(record)
    return records


def is_contacted(status: str) -> bool:
    lowered = clean(status).casefold()
    return bool(lowered) and any(marker.casefold() in lowered for marker in CONTACTED_MARKERS)


def is_oem_branch(record: Dict[str, str]) -> bool:
    name = record.get("name", "").casefold()
    text = " ".join(record.get(key, "") for key in ["name", "type", "business", "note", "link"]).casefold()
    has_brand = any(brand in name or brand in text for brand in OEM_BRANDS)
    has_branch_hint = any(hint in name or hint in text for hint in OEM_BRANCH_HINTS)
    return has_brand and has_branch_hint


def classify_channel(record: Dict[str, str]) -> str:
    combined = " ".join(record.get(key, "") for key in ["type", "business", "note", "name"]).casefold()
    if any(k in combined for k in ["rental", "rentcar", "location", "flotte", "fleet", "租赁", "租车", "车队"]):
        return "rental_fleet"
    if any(k in combined for k in ["import", "importateur", "distributeur", "distribution", "group", "groupe", "集团", "进口", "分销"]):
        return "importer_group"
    if any(k in combined for k in ["commercial", "utilitaire", "truck", "camion", "delivery", "fleet", "商用", "货车", "卡车", "配送"]):
        return "commercial_vehicle_channel"
    if any(k in combined for k in ["used", "occasion", "second hand", "pre-owned", "二手"]):
        return "used_car_dealer"
    if any(k in combined for k in ["multi-brand", "multibrand", "multimarque", "concessionnaire", "dealer", "showroom", "多品牌", "展厅", "经销"]):
        return "multibrand_dealer"
    if any(k in combined for k in ["local", "showroom", "本地", "区域"]):
        return "local_showroom"
    return "unknown_auto_channel"


def apply_filters(record: Dict[str, str], filters: Sequence[Tuple[str, str]]) -> bool:
    for key, expected in filters:
        value = record.get(key, record.get(key.lower(), ""))
        if clean(value) != expected:
            return False
    return True


COUNTRY_NAME_MAP = {
    "摩洛哥": "Morocco",
    "埃及": "Egypt",
    "阿联酋": "the UAE",
    "沙特": "Saudi Arabia",
    "沙特阿拉伯": "Saudi Arabia",
    "智利": "Chile",
    "秘鲁": "Peru",
    "墨西哥": "Mexico",
    "哥伦比亚": "Colombia",
    "阿尔及利亚": "Algeria",
    "突尼斯": "Tunisia",
    "南非": "South Africa",
}
GENERIC_CITY_MARKERS = {
    "", "多城市", "多个城市", "全国", "全国范围", "全境", "多地区", "多个地区", "各地",
    "morocco", "maroc", "national", "nationwide", "multiple cities", "multi-city", "all cities",
}


def market_name(country: str) -> str:
    value = clean(country)
    if not value:
        return "the target market"
    return COUNTRY_NAME_MAP.get(value, value)


def is_generic_city(city: str) -> bool:
    value = clean(city)
    lowered = value.casefold()
    if lowered in GENERIC_CITY_MARKERS or value in GENERIC_CITY_MARKERS:
        return True
    return any(ord(ch) > 127 for ch in value)


def city_phrase(city: str, country: str) -> str:
    market = market_name(country)
    if is_generic_city(city):
        return f"in {market}"
    return f"in {clean(city)}"


def sentence_join(parts: Sequence[str]) -> str:
    return "; ".join(dict.fromkeys(clean(part) for part in parts if clean(part)))


def english_signal_phrase(parts: Sequence[str], fallback: str) -> str:
    unique = list(dict.fromkeys(clean(part) for part in parts if clean(part)))
    if not unique:
        return fallback
    selected = unique[:2]
    if len(selected) == 1:
        return selected[0]
    return f"{selected[0]} and {selected[1]}"


def has_signal_text(record: Dict[str, str]) -> str:
    return " ".join(record.get(key, "") for key in ["name", "type", "business", "note", "website", "phone", "email", "link"]).casefold()


def extract_customer_signals(record: Dict[str, str], scenario_key: str) -> Dict[str, Any]:
    text = has_signal_text(record)
    hits_cn: List[str] = []
    hits_en: List[str] = []
    hit_keys: List[str] = []
    for key, keywords, label_cn, label_en in SIGNAL_PATTERNS:
        if any(keyword.casefold() in text for keyword in keywords):
            hits_cn.append(label_cn)
            hits_en.append(label_en)
            hit_keys.append(key)
    scenario = SCENARIOS[scenario_key]
    market = market_name(record.get("country", ""))
    location = city_phrase(record.get("city", ""), record.get("country", ""))
    observed_signal_cn = sentence_join(hits_cn[:4]) or scenario["signal_label"]
    observed_signal_en = sentence_join(hits_en[:3]) or scenario["signal_label"]
    customer_signal_en = english_signal_phrase(hits_en, scenario["signal_label"])
    strong_signal = len(hit_keys) >= 1 and scenario_key != "unknown_auto_channel"
    signal_quality = "strong" if len(hit_keys) >= 2 else "medium" if strong_signal else "weak"

    if scenario_key == "unknown_auto_channel" or signal_quality == "weak":
        business_hypothesis = "对方与汽车行业相关，但采购、销售、进口或分销职责不清，首轮应先确认角色。"
        profit_angle = "低压力确认是否负责车辆采购/分销，不直接强推。"
        light_offer = "a short model and price-range overview"
        reply_question = "Does your team handle vehicle purchasing or distribution?"
    elif scenario_key == "rental_fleet":
        business_hypothesis = "租赁/车队业务对购置成本、维护成本和周转敏感，可能关注低成本车队更新方案。"
        profit_angle = "用低采购成本和小批量 fleet fit check 切入。"
        light_offer = "a small fleet-fit and price-range overview"
        reply_question = "Should I send a short fleet-fit and price-range overview?"
    elif scenario_key == "importer_group":
        business_hypothesis = "对方可能具备进口、集团或区域分销能力，适合验证首批试单和后续批量潜力。"
        profit_angle = "用 import/distribution 能力和可能的 volume potential 切入。"
        light_offer = "a short first-batch fit check"
        reply_question = "Would a short first-batch fit check be useful for your team?"
    elif scenario_key == "commercial_vehicle_channel":
        business_hypothesis = "对方客户可能重视实用车型、配送、小企业和家商两用需求。"
        profit_angle = "用低成本实用新车补充商用/工具车需求。"
        light_offer = "a practical-vehicle model and price-range overview"
        reply_question = "Should I send a short practical-vehicle overview for your team to judge fit?"
    else:
        business_hypothesis = "对方已有汽车销售或 showroom 客户基础，可能接触价格敏感买家。"
        profit_angle = "用 affordable new-vehicle line 补充现有库存，先小批量判断周转潜力。"
        light_offer = "a short model and price-range overview"
        reply_question = "Should I send a short model and price-range overview?"

    return {
        "observed_signal": observed_signal_en,
        "observed_signal_cn": observed_signal_cn,
        "customer_signal": customer_signal_en,
        "observed_signal_keys": hit_keys,
        "signal_quality": signal_quality,
        "business_hypothesis": business_hypothesis,
        "profit_angle": profit_angle,
        "light_offer": light_offer,
        "reply_question": reply_question,
        "market_name": market,
        "location_phrase": location,
        "risk_reason": "信息不足，话术已降级为低压确认型。" if signal_quality == "weak" else "需人工确认页面真实性和客户是否负责采购/分销。",
    }


def compact_english(text: str) -> str:
    text = re.sub(r"\s+([.,;:!?])", r"\1", text)
    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n ", "\n", text)
    return text.strip()


def build_connect_variants(record: Dict[str, str], scenario: Dict[str, str], signals: Dict[str, Any]) -> Dict[str, str]:
    contact = record.get("contact") or "there"
    name = record.get("name") or "your company"
    location = signals["location_phrase"]
    signal = signals["customer_signal"]
    if signals["signal_quality"] == "weak":
        return {
            "direct_profit_hook": compact_english(f"Hi {contact}, I am mapping auto channels {location} that may handle vehicle sourcing or distribution. {name} seemed relevant. Thought it would be useful to connect."),
            "stock_gap_hook": compact_english(f"Hi {contact}, your page looks connected to the auto sector {location}. I am checking who reviews affordable new-vehicle options for local buyers. Useful to connect?"),
            "soft_research_hook": compact_english(f"Hi {contact}, I am learning which auto channels {location} handle sourcing or distribution. {name} came up as relevant, so I thought I’d connect."),
        }
    return {
        "direct_profit_hook": compact_english(f"Hi {contact}, your page shows {signal}, so {name} may reach price-sensitive buyers. I am checking whether a low-cost new-vehicle line is worth a small first-batch review."),
        "stock_gap_hook": compact_english(f"Hi {contact}, {name} looks close to {scenario['customer_base']} {location}. A low-cost new-vehicle option may complement current stock. Thought it would be useful to connect."),
        "soft_research_hook": compact_english(f"Hi {contact}, I am looking at auto channels {location} serving practical, cost-sensitive buyers. {name} stood out from its {signal}. Thought it would be useful to connect."),
    }


def build_dm_variants(record: Dict[str, str], scenario_key: str, scenario: Dict[str, str], signals: Dict[str, Any]) -> Dict[str, str]:
    market = signals["market_name"]
    signal = signals["customer_signal"]
    question = signals["reply_question"]
    light_offer = signals["light_offer"]
    fit_context = scenario["fit_context"]

    if signals["signal_quality"] == "weak" or scenario_key == "unknown_auto_channel":
        return {
            "direct_profit_hook": compact_english("Your page appears connected to the auto sector, but I am not sure whether your team handles sourcing, sales, import, or distribution. We support Wuling export and are checking if affordable new-vehicle options are relevant locally.\n\nDoes your team handle vehicle purchasing or distribution?"),
            "stock_gap_hook": compact_english("I am mapping auto businesses that may review affordable new-vehicle lines for price-sensitive buyers. I don’t want to assume your role from limited public info.\n\nAre you the right team to review vehicle sourcing or distribution opportunities?"),
            "soft_research_hook": compact_english(f"I am looking at auto channels in {market} and your page seemed relevant, though the public info is limited. The aim is simply to see whether a small Wuling first-batch review fits the right local channels.\n\nDoes your team usually evaluate vehicle sourcing opportunities?"),
        }

    return {
        "direct_profit_hook": compact_english(f"Your page shows {signal}, which looks close to {fit_context}. A low-cost new-vehicle line could be worth testing with limited stock pressure; Huatu Overseas can support Wuling export for a small first-batch fit check.\n\n{question}"),
        "stock_gap_hook": compact_english(f"If some of your buyers want newer vehicles but still decide mainly on total cost, there may be a gap between used stock and higher-priced brands. Wuling could be reviewed as an affordable line to complement your current offer.\n\nWould {light_offer} help your team judge fit?"),
        "soft_research_hook": compact_english(f"I am looking at auto channels in {market} where affordable practical vehicles could match local demand. Your {signal} stood out, so I am checking whether Wuling is worth a low-pressure first-batch review.\n\nIs this something your team would normally evaluate?"),
    }


def recommended_variant_key(scenario_key: str, signals: Dict[str, Any]) -> str:
    if signals["signal_quality"] == "weak" or scenario_key == "unknown_auto_channel":
        return "soft_research_hook"
    if scenario_key in {"used_car_dealer", "rental_fleet", "commercial_vehicle_channel"}:
        return "direct_profit_hook"
    return "stock_gap_hook"


def build_message_variants(record: Dict[str, str], scenario_key: str) -> Dict[str, Any]:
    scenario = SCENARIOS[scenario_key]
    signals = extract_customer_signals(record, scenario_key)
    connects = build_connect_variants(record, scenario, signals)
    dms = build_dm_variants(record, scenario_key, scenario, signals)
    alternatives = {
        key: {
            "label_cn": {
                "direct_profit_hook": "直接利润机会",
                "stock_gap_hook": "库存/车型补充机会",
                "soft_research_hook": "低压行业交流",
            }[key],
            "english_connect": connects[key],
            "english_first_dm": dms[key],
        }
        for key in ["direct_profit_hook", "stock_gap_hook", "soft_research_hook"]
    }
    recommended_key = recommended_variant_key(scenario_key, signals)
    return {
        "recommended_key": recommended_key,
        "recommended_message": alternatives[recommended_key]["english_first_dm"],
        "recommended_connect": alternatives[recommended_key]["english_connect"],
        "alternatives": alternatives,
        "signals": signals,
    }



def first_sentence(text: str) -> str:
    value = clean(text)
    match = re.search(r"^(.+?[.!?])(?:\s|$)", value)
    return clean(match.group(1) if match else value[:120]).casefold()


def apply_variant_to_item(item: Dict[str, Any], variant_key: str) -> None:
    variant = item.get("alternatives", {}).get(variant_key, {})
    if not variant:
        return
    connect = variant.get("english_connect", "")
    dm = variant.get("english_first_dm", "")
    item["recommended_variant"] = variant_key
    item["recommended_message"] = dm
    item["english_connect"] = connect
    item["english_first_dm"] = dm
    item["messages"]["connect"]["en"] = connect
    item["messages"]["dm"]["en"] = dm
    item["formatted_preview"]["英文加好友话术"] = connect
    item["formatted_preview"]["英文首轮私信"] = dm


def ensure_unique_message_openers(items: List[Dict[str, Any]]) -> None:
    seen: set[str] = set()
    variant_order = ["direct_profit_hook", "stock_gap_hook", "soft_research_hook"]
    for item in items:
        current = item.get("recommended_variant") or ""
        candidates = [current] + [key for key in variant_order if key != current]
        selected = current
        for key in candidates:
            message = item.get("alternatives", {}).get(key, {}).get("english_first_dm", "")
            opener = first_sentence(message)
            if opener and opener not in seen:
                selected = key
                break
        apply_variant_to_item(item, selected)
        opener = first_sentence(item.get("recommended_message", ""))
        if opener:
            seen.add(opener)


def suggested_action(status: str) -> str:
    if not clean(status) or clean(status) == DEFAULT_STATUS:
        return "follow_and_first_dm"
    if is_contacted(status):
        return "skip_or_follow_up"
    return "first_dm"


def parse_filters(raw_filters: Sequence[str]) -> List[Tuple[str, str]]:
    filters: List[Tuple[str, str]] = []
    for item in raw_filters:
        if "=" not in item:
            raise ValueError(f"Invalid filter, expected field=value: {item}")
        key, value = item.split("=", 1)
        filters.append((key.strip(), value.strip()))
    return filters


def resolve_excel(excel: str) -> Path:
    if excel:
        path = Path(excel).expanduser()
        return path if path.is_absolute() else (Path.cwd() / path).resolve()
    if resolve_workbook_path:
        resolved = resolve_workbook_path("", create_from_template=False)
        if resolved.get("path"):
            return Path(resolved["path"])
    raise FileNotFoundError("No workbook found. Pass --excel.")


def build_preview(records: Sequence[Dict[str, str]], filters: Sequence[Tuple[str, str]], include_sent: bool) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
    items: List[Dict[str, Any]] = []
    skipped: List[Dict[str, Any]] = []
    for ordinal, record in enumerate(records, start=1):
        name = record.get("name", "")
        link = record.get("link", "")
        status = record.get("status", "")
        if not name or not link:
            skipped.append({"row_number": record.get("_row_number"), "dealer_name": name or "(blank)", "reason": "missing name or Facebook page link"})
            continue
        if filters and not apply_filters(record, filters):
            continue
        if is_contacted(status) and not include_sent:
            skipped.append({"row_number": record.get("_row_number"), "dealer_name": name, "reason": f"already contacted: {status}"})
            continue
        if is_oem_branch(record):
            skipped.append({"row_number": record.get("_row_number"), "dealer_name": name, "reason": "建议跳过：疑似官方品牌当地页，不生成发送话术"})
            continue

        scenario_key = classify_channel(record)
        scenario = SCENARIOS[scenario_key]
        variant_pack = build_message_variants(record, scenario_key)
        signals = variant_pack["signals"]
        connect_en = variant_pack["recommended_connect"]
        dm_en = variant_pack["recommended_message"]
        signal_line = f"真实信号：{signals['observed_signal_cn']}；商业假设：{signals['business_hypothesis']}"
        risk_note = f"{signals['risk_reason']} 预览用途。真实执行前必须先在聊天框展示整批预览，并由用户一次性确认。"
        item = {
            "index": record.get("index") or str(ordinal),
            "row_number": record.get("_row_number"),
            "dealer_name": name,
            "city": record.get("city", ""),
            "dealer_type": record.get("type", ""),
            "main_business": record.get("business", ""),
            "page_url": link,
            "company_website": record.get("website", ""),
            "status": status or DEFAULT_STATUS,
            "channel": scenario_key,
            "customer_judgment_cn": f"{scenario['judgment_cn']} {signal_line}",
            "recommended_angle_cn": f"{scenario['angle_cn']} 推荐使用“{variant_pack['alternatives'][variant_pack['recommended_key']]['label_cn']}”版本。",
            "observed_signal": signals["observed_signal"],
            "observed_signal_cn": signals["observed_signal_cn"],
            "signal_quality": signals["signal_quality"],
            "business_hypothesis": signals["business_hypothesis"],
            "profit_angle": signals["profit_angle"],
            "light_offer": signals["light_offer"],
            "reply_question": signals["reply_question"],
            "suggested_action": suggested_action(status),
            "recommended_variant": variant_pack["recommended_key"],
            "recommended_message": dm_en,
            "english_connect": connect_en,
            "english_first_dm": dm_en,
            "alternatives": variant_pack["alternatives"],
            "message_variants": variant_pack["alternatives"],
            "messages": {
                "connect": {"客户判断": f"{scenario['judgment_cn']} {signal_line}", "推荐切入点": scenario["angle_cn"], "en": connect_en},
                "dm": {"客户判断": f"{scenario['judgment_cn']} {signal_line}", "推荐切入点": scenario["angle_cn"], "en": dm_en},
            },
            "formatted_preview": {
                "客户判断": f"{scenario['judgment_cn']} {signal_line}",
                "推荐切入点": f"{scenario['angle_cn']} 推荐使用“{variant_pack['alternatives'][variant_pack['recommended_key']]['label_cn']}”版本。",
                "英文加好友话术": connect_en,
                "英文首轮私信": dm_en,
                "备选话术": variant_pack["alternatives"],
                "风险提示": risk_note,
            },
            "risk_note": risk_note,
            "risk_note_cn": risk_note,
            "record": record,
        }
        items.append(item)
    return items, skipped


def render_human_preview(result: Dict[str, Any], max_chars: int = 16000) -> str:
    """Render a chat-friendly preview for one batch before any send action."""
    summary = result.get("summary", {})
    lines: List[str] = []
    lines.append("# Facebook 建联话术预览")
    lines.append(f"准备发送：{summary.get('ready_to_send', 0)} 条；跳过：{summary.get('skipped', 0)} 条")
    lines.append("确认后将按本批预览批量执行 Follow + Messenger DM，不再逐条确认。")
    lines.append("")
    for idx, item in enumerate(result.get("items", []), start=1):
        fp = item.get("formatted_preview", {}) or {}
        lines.append(f"## {idx}. {item.get('dealer_name', '')}")
        lines.append(f"主页：{item.get('page_url', '')}")
        lines.append(f"客户判断：{fp.get('客户判断') or item.get('customer_judgment_cn', '')}")
        lines.append(f"推荐切入点：{fp.get('推荐切入点') or item.get('outreach_angle_cn', '')}")
        lines.append("英文首轮私信：")
        lines.append(item.get("recommended_message") or item.get("english_first_dm", ""))
        lines.append(f"风险提示：{fp.get('风险提示') or item.get('risk_note', '')}")
        lines.append("")
    rendered = "\n".join(lines).strip()
    if len(rendered) > max_chars:
        rendered = rendered[:max_chars] + "\n\n[预览过长，已截断；完整 JSON 见输出文件]"
    return rendered


def main(argv: Optional[Sequence[str]] = None) -> int:
    parser = argparse.ArgumentParser(description="Generate English Facebook outreach preview JSON.")
    parser.add_argument("--excel", default="", help="Customer outreach workbook path. If omitted, use workbook resolver.")
    parser.add_argument("--sheet", default=DEFAULT_SHEET, help="Source sheet name.")
    parser.add_argument("--filter", action="append", default=[], help="Filter condition, field=value. Can repeat.")
    parser.add_argument("--sample", type=int, default=0, help="Randomly sample N matched records. 0 means all.")
    parser.add_argument("--include-sent", action="store_true", help="Include already-contacted records for follow-up preview.")
    parser.add_argument("--seed", type=int, default=None, help="Random seed for sampling.")
    parser.add_argument("--output", required=True, help="Output JSON preview path.")
    args = parser.parse_args(argv)

    excel_path = resolve_excel(args.excel)
    if not excel_path.exists():
        raise FileNotFoundError(f"Workbook not found: {excel_path}")
    filters = parse_filters(args.filter)
    records = read_records(excel_path, args.sheet)
    items, skipped = build_preview(records, filters, args.include_sent)

    if args.sample and args.sample < len(items):
        if args.seed is not None:
            random.seed(args.seed)
        items = random.sample(items, args.sample)
    ensure_unique_message_openers(items)

    result = {
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "language_policy": {"customer_facing": "English", "internal_review": "Chinese", "facebook_default_uses_french": False},
        "message_strategy": {
            "framework": "specific signal -> commercial hypothesis -> light offer -> one question",
            "variants": ["direct_profit_hook", "stock_gap_hook", "soft_research_hook"],
            "recommended_field": "recommended_message",
        },
        "source": {"excel": str(excel_path), "sheet": args.sheet, "filters": args.filter, "sample": args.sample, "seed": args.seed, "include_sent": args.include_sent},
        "summary": {"total_records": len(records), "ready_to_send": len(items), "matched_records": len(items), "skipped": len(skipped)},
        "items": items,
        "skipped": skipped,
    }
    result["human_preview"] = render_human_preview(result)

    output_path = Path(args.output).expanduser()
    if not output_path.is_absolute():
        output_path = Path.cwd() / output_path
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(result["summary"], ensure_ascii=False))
    print("\n" + result["human_preview"] + "\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())