wuling-overseas-dealer-expansion-skill/scripts/dashboard/build_dashboard.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Build a visual HTML dashboard from the consolidated outreach workbook."""
from __future__ import annotations

import argparse
import json
import re
import shutil
import sys
from collections import Counter
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]
COMMON_DIR = SKILL_ROOT / "scripts" / "common"
if str(COMMON_DIR) not in sys.path:
    sys.path.insert(0, str(COMMON_DIR))

from artifact_manager import new_run_id, resolve_artifact_path  # type: ignore  # noqa: E402
from workbook_resolver import resolve_workbook_path  # type: ignore  # noqa: E402

SUMMARY_SHEET_ALIASES = ["客户信息汇总表", "客户信息汇总"]
CONVERSATION_SHEET = "Facebook对话记录"
DEFAULT_TITLE = "五菱海外客户建联中台"
FACEBOOK_INTENTS = ["明确有意向", "潜在意向", "需澄清", "暂不考虑", "明确拒绝"]

HEADER_ALIASES = {
    "company": ["公司姓名", "客户姓名/公司", "公司名称", "客户名称", "公司/客户", "Name", "Company"],
    "country": ["国家", "Country"],
    "city": ["城市", "City"],
    "customer_type": ["客户类型", "细分客户类型", "Customer Type"],
    "website": ["官网链接", "公司官网", "官网", "官方网站", "Website"],
    "contact": ["联系人", "姓名", "Contact"],
    "position": ["职位", "职务", "Position", "Title"],
    "personal_email": ["个人邮箱", "个人邮箱（不一定有效）"],
    "contact_phone": ["联系人电话", "个人电话", "联系电话"],
    "facebook_link": ["Facebook主页链接", "Facebook链接", "facebook链接", "主页/链接"],
    "linkedin_link": ["linkined主页链接", "LinkedIn主页链接", "LinkedIn链接", "linkin链接", "linkin连接"],
    "google_maps_link": ["google map链接", "Google Maps链接", "Google Map链接", "地图链接"],
    "public_phone": ["公共电话/WhatsApp", "电话/WhatsApp", "公司公共电话", "电话", "WhatsApp"],
    "public_email": ["公共邮箱", "邮箱", "公司公共邮箱（任一有效即可）", "公司公共邮箱", "Email"],
    "attribute": ["客户属性", "客户大类", "Customer Attribute"],
    "status": ["建联状态", "建联情况", "状态", "Status"],
    "next_followup": ["下次跟进", "下次跟进时间", "Next Follow-up"],
    "note": ["备注", "说明", "Notes"],
}

CONVERSATION_HEADER_ALIASES = {
    "record_id": ["记录ID"],
    "customer_index": ["客户序号"],
    "company": ["客户姓名/公司", "公司姓名", "公司名称"],
    "facebook_link": ["Facebook主页链接", "主页/链接"],
    "message_time": ["消息时间"],
    "direction": ["消息方向"],
    "translation": ["中文翻译"],
    "effective": ["是否有效客户回复"],
    "intent": ["合作意向"],
    "intent_reason": ["意向判断依据"],
    "next_action": ["下一步建议"],
}

CONTACTED_KEYWORDS = ["已发送邮件", "已发邮件", "邮件已发送", "邮件发送成功", "已发私信", "已关注并私信", "已建联", "等待回复", "已回复", "有意向"]
BOUNCED_KEYWORDS = ["邮件退回", "退回", "拒收", "无法送达", "域名不存在", "邮箱不存在", "发送失败"]
EXCLUDED_KEYWORDS = ["已剔除", "剔除", "跳过", "低优先级", "skip_"]
REVIEW_KEYWORDS = ["待确认", "人工复核", "需人工", "需确认", "信息不足"]

SOURCE_PATTERNS = [
    ("Facebook", re.compile(r"facebook", re.I)),
    ("LinkedIn", re.compile(r"linkedin|linkin", re.I)),
    ("Google Maps", re.compile(r"google\s*maps|google map|maps", re.I)),
    ("Moteur.ma", re.compile(r"moteur", re.I)),
    ("当地汽车网站", re.compile(r"当地汽车网站|OtoMoto|Wandaloo|Kerix|Kompass|Maroc Annuaire|Telecontact", re.I)),
    ("平台与行业渠道", re.compile(r"协会商会|平台与行业渠道|商会|协会")),
]


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


def find_sheet_name(wb, requested: str) -> str:
    if requested and requested in wb.sheetnames:
        return requested
    for candidate in SUMMARY_SHEET_ALIASES:
        if candidate in wb.sheetnames:
            return candidate
    raise ValueError(f"未找到客户信息汇总 Sheet，可用 Sheet：{', '.join(wb.sheetnames)}")


def header_map(ws) -> Dict[str, int]:
    raw = {clean(cell.value): idx for idx, cell in enumerate(ws[1], start=1) if clean(cell.value)}
    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 row_value(ws, row: int, headers: Dict[str, int], key: str) -> str:
    col = headers.get(key)
    if not col:
        return ""
    return clean(ws.cell(row=row, column=col).value)


def is_effective_row(row: Dict[str, str]) -> bool:
    evidence = ["company", "website", "facebook_link", "linkedin_link", "google_maps_link", "public_email", "personal_email", "public_phone", "contact_phone", "note"]
    return any(row.get(key) for key in evidence)


def read_summary_rows(workbook_path: Path, sheet_name: str) -> Tuple[str, List[Dict[str, str]]]:
    wb = load_workbook(workbook_path, data_only=True, read_only=True)
    actual_sheet = find_sheet_name(wb, sheet_name)
    ws = wb[actual_sheet]
    headers = header_map(ws)
    rows: List[Dict[str, str]] = []
    for row_idx in range(2, ws.max_row + 1):
        row = {key: row_value(ws, row_idx, headers, key) for key in HEADER_ALIASES}
        row["_excel_row"] = str(row_idx)
        if is_effective_row(row):
            rows.append(row)
    return actual_sheet, rows


def normalized_link(value: str) -> str:
    value = clean(value).casefold().split("?", 1)[0].rstrip("/")
    return value.removeprefix("https://").removeprefix("http://").removeprefix("www.")


def customer_key(company: str, facebook_link: str, customer_index: str = "") -> str:
    link = normalized_link(facebook_link)
    if link:
        return "link:" + link
    name = re.sub(r"[^0-9a-z\u00c0-\u024f\u0600-\u06ff\u4e00-\u9fff]+", "", clean(company).casefold())
    return "name:" + name if name else "index:" + clean(customer_index)


def read_conversation_rows(workbook_path: Path, sheet_name: str) -> List[Dict[str, str]]:
    wb = load_workbook(workbook_path, data_only=True, read_only=True)
    if sheet_name not in wb.sheetnames:
        return []
    ws = wb[sheet_name]
    raw_headers = {clean(cell.value): idx for idx, cell in enumerate(ws[1], start=1) if clean(cell.value)}
    mapped: Dict[str, int] = {}
    for key, aliases in CONVERSATION_HEADER_ALIASES.items():
        for alias in aliases:
            if alias in raw_headers:
                mapped[key] = raw_headers[alias]
                break
    if "record_id" not in mapped:
        return []
    rows: List[Dict[str, str]] = []
    for row_idx in range(2, ws.max_row + 1):
        row = {
            key: clean(ws.cell(row=row_idx, column=column).value)
            for key, column in mapped.items()
        }
        if row.get("record_id"):
            rows.append(row)
    return rows


def message_datetime(value: str) -> datetime:
    value = clean(value)
    if not value:
        return datetime.min
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00")).replace(tzinfo=None)
    except ValueError:
        match = re.search(r"(20\d{2})[-/](\d{1,2})[-/](\d{1,2})", value)
        if match:
            return datetime(int(match.group(1)), int(match.group(2)), int(match.group(3)))
    return datetime.min


def build_conversation_insights(rows: Sequence[Dict[str, str]]) -> Dict[str, Any]:
    sent_customers = set()
    replied_customers = set()
    latest_by_customer: Dict[str, Dict[str, str]] = {}
    reply_trend: Counter = Counter()
    for row in sorted(rows, key=lambda item: message_datetime(item.get("message_time", ""))):
        key = customer_key(row.get("company", ""), row.get("facebook_link", ""), row.get("customer_index", ""))
        if row.get("direction") == "我方发送":
            sent_customers.add(key)
        if row.get("direction") != "客户回复" or row.get("effective") != "是":
            continue
        replied_customers.add(key)
        if row.get("intent") in FACEBOOK_INTENTS:
            latest_by_customer[key] = row
        parsed = message_datetime(row.get("message_time", ""))
        if parsed != datetime.min and (datetime.now() - parsed).days <= 30:
            reply_trend[parsed.strftime("%Y-%m-%d")] += 1
    intent_counter = Counter(
        row.get("intent") or "未分析" for row in latest_by_customer.values()
    )
    high_intent = [
        {
            "company": row.get("company", ""),
            "intent": row.get("intent", ""),
            "latest_reply": row.get("translation", ""),
            "reason": row.get("intent_reason", ""),
            "next_action": row.get("next_action", ""),
            "message_time": row.get("message_time", ""),
        }
        for row in latest_by_customer.values()
        if row.get("intent") in {"明确有意向", "潜在意向"}
    ]
    pending = [
        item for item in high_intent
        if item.get("next_action")
    ]
    trend = [
        {"name": day, "count": count}
        for day, count in sorted(reply_trend.items())
    ]
    return {
        "sent_customers": sent_customers,
        "replied_customers": replied_customers,
        "latest_by_customer": latest_by_customer,
        "intent_counter": intent_counter,
        "reply_trend": trend,
        "high_intent": sorted(high_intent, key=lambda item: item.get("message_time", ""), reverse=True),
        "pending_followups": sorted(pending, key=lambda item: item.get("message_time", ""), reverse=True),
    }


def contains_any(text: str, keywords: Sequence[str]) -> bool:
    lowered = text.casefold()
    return any(keyword.casefold() in lowered for keyword in keywords)


def has_email(row: Dict[str, str]) -> bool:
    return bool(row.get("public_email") or row.get("personal_email"))


def has_phone(row: Dict[str, str]) -> bool:
    return bool(row.get("public_phone") or row.get("contact_phone"))


def has_any_link(row: Dict[str, str]) -> bool:
    return bool(row.get("website") or row.get("facebook_link") or row.get("linkedin_link") or row.get("google_maps_link"))


def is_excluded(row: Dict[str, str]) -> bool:
    return contains_any(" ".join([row.get("status", ""), row.get("note", "")]), EXCLUDED_KEYWORDS)


def is_bounced(row: Dict[str, str]) -> bool:
    return contains_any(" ".join([row.get("status", ""), row.get("note", "")]), BOUNCED_KEYWORDS)


def is_contacted(row: Dict[str, str]) -> bool:
    text = " ".join([row.get("status", ""), row.get("note", "")])
    return contains_any(text, CONTACTED_KEYWORDS) and not is_bounced(row)


def needs_review(row: Dict[str, str]) -> bool:
    text = " ".join([row.get("customer_type", ""), row.get("attribute", ""), row.get("note", "")])
    return contains_any(text, REVIEW_KEYWORDS)


def is_contactable(row: Dict[str, str]) -> bool:
    return (has_email(row) or has_phone(row) or has_any_link(row)) and not is_excluded(row)


def row_sources(row: Dict[str, str]) -> List[str]:
    sources: List[str] = []
    if row.get("facebook_link"):
        sources.append("Facebook")
    if row.get("linkedin_link"):
        sources.append("LinkedIn")
    if row.get("google_maps_link"):
        sources.append("Google Maps")
    note = " ".join([row.get("note", ""), row.get("website", "")])
    for label, pattern in SOURCE_PATTERNS:
        if pattern.search(note) and label not in sources:
            sources.append(label)
    return sources or ["未标明来源"]


def pct(part: int, total: int) -> float:
    return round(part * 100 / total, 1) if total else 0.0


def top_counter(counter: Counter, limit: int) -> List[Dict[str, Any]]:
    total = sum(counter.values())
    return [{"name": name or "未填写", "count": count, "rate": pct(count, total)} for name, count in counter.most_common(limit)]


def display_contact(row: Dict[str, str]) -> str:
    values = [row.get("public_email"), row.get("personal_email"), row.get("public_phone"), row.get("contact_phone")]
    return next((value for value in values if value), "有主页/链接" if has_any_link(row) else "")


def build_dashboard_data(
    rows: Sequence[Dict[str, str]],
    conversations: Sequence[Dict[str, str]],
    workbook_path: Path,
    sheet_name: str,
    run_id: str,
    top_n: int,
) -> Dict[str, Any]:
    valid_rows = [row for row in rows if not is_excluded(row)]
    contactable_rows = [row for row in valid_rows if is_contactable(row)]
    contacted_rows = [row for row in valid_rows if is_contacted(row)]
    bounced_rows = [row for row in rows if is_bounced(row)]
    review_rows = [row for row in valid_rows if needs_review(row)]
    email_rows = [row for row in valid_rows if has_email(row)]
    phone_rows = [row for row in valid_rows if has_phone(row)]
    conversation = build_conversation_insights(conversations)
    facebook_sent = len(conversation["sent_customers"])
    facebook_replied = len(conversation["replied_customers"])
    latest_intents = conversation["latest_by_customer"]

    attribute_counter = Counter(row.get("attribute") or "未填写" for row in valid_rows)
    type_counter = Counter(row.get("customer_type") or "未填写" for row in valid_rows)
    status_counter = Counter(row.get("status") or "未填写" for row in rows)
    city_counter = Counter(row.get("city") or "未填写" for row in valid_rows)
    source_counter: Counter = Counter()
    for row in valid_rows:
        for source in row_sources(row):
            source_counter[source] += 1

    dashboard_rows = []
    for row in valid_rows:
        reply = latest_intents.get(customer_key(row.get("company", ""), row.get("facebook_link", "")), {})
        dashboard_rows.append({
            "row": row.get("_excel_row", ""),
            "company": row.get("company", ""),
            "city": row.get("city", ""),
            "attribute": row.get("attribute", ""),
            "type": row.get("customer_type", ""),
            "source": "；".join(row_sources(row)),
            "contact": display_contact(row),
            "status": row.get("status", ""),
            "has_email": has_email(row),
            "has_phone": has_phone(row),
            "contacted": is_contacted(row),
            "bounced": is_bounced(row),
            "review": needs_review(row),
            "note": row.get("note", "")[:220],
            "facebook_intent": reply.get("intent", ""),
            "latest_reply": reply.get("translation", "")[:160],
            "next_action": reply.get("next_action", "")[:160],
        })

    attention_rows = [row for row in dashboard_rows if row["contact"] and not row["contacted"] and not row["bounced"]][:top_n]
    return {
        "title": DEFAULT_TITLE,
        "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "run_id": run_id,
        "workbook": str(workbook_path),
        "sheet_name": sheet_name,
        "metrics": {
            "total_rows": len(rows),
            "valid_customers": len(valid_rows),
            "excluded_customers": len(rows) - len(valid_rows),
            "contactable_customers": len(contactable_rows),
            "contacted_customers": len(contacted_rows),
            "outreach_rate": pct(len(contacted_rows), len(contactable_rows)),
            "email_customers": len(email_rows),
            "email_coverage_rate": pct(len(email_rows), len(valid_rows)),
            "phone_customers": len(phone_rows),
            "phone_coverage_rate": pct(len(phone_rows), len(valid_rows)),
            "bounced_customers": len(bounced_rows),
            "bounce_rate": pct(len(bounced_rows), len(email_rows)),
            "needs_review_customers": len(review_rows),
            "review_rate": pct(len(review_rows), len(valid_rows)),
            "facebook_dm_customers": facebook_sent,
            "facebook_reply_customers": facebook_replied,
            "facebook_reply_rate": pct(facebook_replied, facebook_sent),
            "facebook_clear_intent_customers": sum(
                1 for row in latest_intents.values() if row.get("intent") == "明确有意向"
            ),
            "facebook_potential_intent_customers": sum(
                1 for row in latest_intents.values() if row.get("intent") == "潜在意向"
            ),
            "facebook_pending_followups": len(conversation["pending_followups"]),
        },
        "charts": {
            "attribute": top_counter(attribute_counter, top_n),
            "customer_type": top_counter(type_counter, top_n),
            "status": top_counter(status_counter, top_n),
            "source": top_counter(source_counter, top_n),
            "city": top_counter(city_counter, top_n),
            "facebook_intent": top_counter(conversation["intent_counter"], len(FACEBOOK_INTENTS)),
            "facebook_reply_trend": conversation["reply_trend"],
        },
        "dashboard_rows": dashboard_rows,
        "attention_customers": attention_rows,
        "facebook_high_intent_customers": conversation["high_intent"][:top_n],
        "facebook_pending_followups": conversation["pending_followups"][:top_n],
        "definitions": {
            "有效客户": "总表中未被备注或状态标记为已剔除、跳过、低优先级的客户。",
            "可建联客户": "有效客户中至少有邮箱、电话/WhatsApp、官网、Facebook、LinkedIn 或 Google Maps 入口之一。",
            "已建联客户": "状态或备注包含已发送邮件、已发私信、已关注并私信、等待回复、已回复、已建联或有意向，且未被标记退信/拒收。",
            "建联率": "已建联客户数 / 可建联客户数。",
            "邮箱覆盖率": "有个人邮箱或公共邮箱的有效客户数 / 有效客户数。",
            "退信率": "邮件退回、拒收、无法送达或域名/邮箱不存在客户数 / 有邮箱客户数。",
            "Facebook回复率": "Facebook对话记录中有有效客户回复的唯一客户数 / 已发送Facebook私信的唯一客户数；系统消息和自动回复不计入。",
            "合作意向": "按每个客户最新一条有效回复计算五级意向，同一客户多条消息只计一次。",
        },
    }


def render_html(data: Dict[str, Any]) -> str:
    template = (SKILL_ROOT / "assets" / "dashboard_template.html").read_text(encoding="utf-8")
    payload = json.dumps(data, ensure_ascii=False).replace("<", "\\u003c")
    return template.replace("{{dashboard_json}}", payload)


def write_outputs(data: Dict[str, Any], html_text: str, html_path: Path, json_path: Path, latest_dir: str = "") -> Dict[str, str]:
    html_path.parent.mkdir(parents=True, exist_ok=True)
    json_path.parent.mkdir(parents=True, exist_ok=True)
    html_path.write_text(html_text, encoding="utf-8")
    json_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
    copied_latest = ""
    if latest_dir:
        target_dir = Path(latest_dir).expanduser()
        if not target_dir.is_absolute():
            target_dir = Path.cwd() / target_dir
        target_dir.mkdir(parents=True, exist_ok=True)
        latest_html = target_dir / "customer_dashboard.html"
        latest_json = target_dir / "dashboard_data.json"
        shutil.copy2(html_path, latest_html)
        shutil.copy2(json_path, latest_json)
        copied_latest = str(latest_html)
    return {"html": str(html_path), "json": str(json_path), "latest_html": copied_latest}


def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Generate a Wuling outreach dashboard from 客户信息汇总表.")
    parser.add_argument("--excel", default="", help="Workbook path. If omitted, resolve the project workbook by skill rules.")
    parser.add_argument("--summary-sheet", default="客户信息汇总表", help="Summary sheet name.")
    parser.add_argument("--conversation-sheet", default=CONVERSATION_SHEET, help="Facebook conversation sheet name.")
    parser.add_argument("--output-html", default="", help="Optional HTML output path. Bare filename goes to runs/.")
    parser.add_argument("--output-json", default="", help="Optional JSON output path. Bare filename goes to runs/.")
    parser.add_argument("--run-id", default="", help="Run ID used for artifacts.")
    parser.add_argument("--top-n", type=int, default=12, help="Top N categories and attention rows.")
    parser.add_argument("--latest-dir", default="", help="Optional stable directory for a copy of the latest dashboard.")
    return parser.parse_args(argv)


def main(argv: Optional[Sequence[str]] = None) -> int:
    args = parse_args(argv)
    run_id = args.run_id or new_run_id("dashboard")
    resolved = resolve_workbook_path(args.excel, create_from_template=False)
    workbook_path = resolved.get("path")
    if not workbook_path:
        raise FileNotFoundError("No outreach workbook found. Pass --excel or create one from the skill blank template in write-enabled workflows.")
    workbook_path = Path(workbook_path)
    actual_sheet, rows = read_summary_rows(workbook_path, args.summary_sheet)
    conversations = read_conversation_rows(workbook_path, args.conversation_sheet)
    data = build_dashboard_data(rows, conversations, workbook_path, actual_sheet, run_id, args.top_n)
    html_text = render_html(data)
    html_path = resolve_artifact_path(args.output_html, kind="dashboard", default_name="customer_dashboard.html", run_id=run_id)
    json_path = resolve_artifact_path(args.output_json, kind="dashboard", default_name="dashboard_data.json", run_id=run_id)
    outputs = write_outputs(data, html_text, html_path, json_path, args.latest_dir)
    report = {
        "workbook": str(workbook_path),
        "summary_sheet": actual_sheet,
        "conversation_sheet": args.conversation_sheet if conversations else "",
        "conversation_rows": len(conversations),
        "run_id": run_id,
        "outputs": outputs,
        "metrics": data["metrics"],
    }
    print(json.dumps(report, ensure_ascii=False, indent=2))
    return 0


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