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 = ["\u5ba2\u6237\u4fe1\u606f\u6c47\u603b\u8868", "\u5ba2\u6237\u4fe1\u606f\u6c47\u603b"]
CONVERSATION_SHEET = "Facebook\u5bf9\u8bdd\u8bb0\u5f55"
DEFAULT_TITLE = "\u4e94\u83f1\u6d77\u5916\u5ba2\u6237\u5efa\u8054\u4e2d\u53f0"

HEADER_ALIASES = {
    "company": ["\u516c\u53f8\u59d3\u540d", "\u5ba2\u6237\u59d3\u540d/\u516c\u53f8", "\u516c\u53f8\u540d\u79f0", "\u5ba2\u6237\u540d\u79f0", "\u516c\u53f8/\u5ba2\u6237", "Name", "Company"],
    "country": ["\u56fd\u5bb6", "Country"],
    "city": ["\u57ce\u5e02", "City"],
    "customer_type": ["\u5ba2\u6237\u7c7b\u578b", "\u7ec6\u5206\u5ba2\u6237\u7c7b\u578b", "Customer Type"],
    "website": ["\u5b98\u7f51\u94fe\u63a5", "\u516c\u53f8\u5b98\u7f51", "\u5b98\u7f51", "\u5b98\u65b9\u7f51\u7ad9", "Website"],
    "contact": ["\u8054\u7cfb\u4eba", "\u59d3\u540d", "Contact"],
    "position": ["\u804c\u4f4d", "\u804c\u52a1", "Position", "Title"],
    "personal_email": ["\u4e2a\u4eba\u90ae\u7bb1", "\u4e2a\u4eba\u90ae\u7bb1\uff08\u4e0d\u4e00\u5b9a\u6709\u6548\uff09"],
    "contact_phone": ["\u8054\u7cfb\u4eba\u7535\u8bdd", "\u4e2a\u4eba\u7535\u8bdd", "\u8054\u7cfb\u7535\u8bdd"],
    "facebook_link": ["Facebook\u4e3b\u9875\u94fe\u63a5", "Facebook\u94fe\u63a5", "facebook\u94fe\u63a5", "\u4e3b\u9875/\u94fe\u63a5"],
    "linkedin_link": ["linkined\u4e3b\u9875\u94fe\u63a5", "LinkedIn\u4e3b\u9875\u94fe\u63a5", "LinkedIn\u94fe\u63a5", "linkin\u94fe\u63a5", "linkin\u8fde\u63a5"],
    "google_maps_link": ["google map\u94fe\u63a5", "Google Maps\u94fe\u63a5", "Google Map\u94fe\u63a5", "\u5730\u56fe\u94fe\u63a5"],
    "public_phone": ["\u516c\u5171\u7535\u8bdd/WhatsApp", "\u7535\u8bdd/WhatsApp", "\u516c\u53f8\u516c\u5171\u7535\u8bdd", "\u7535\u8bdd", "WhatsApp"],
    "public_email": ["\u516c\u5171\u90ae\u7bb1", "\u90ae\u7bb1", "\u516c\u53f8\u516c\u5171\u90ae\u7bb1\uff08\u4efb\u4e00\u6709\u6548\u5373\u53ef\uff09", "\u516c\u53f8\u516c\u5171\u90ae\u7bb1", "Email"],
    "attribute": ["\u5ba2\u6237\u5c5e\u6027", "\u5ba2\u6237\u5927\u7c7b", "Customer Attribute"],
    "lead_grade": ["\u7ebf\u7d22\u7b49\u7ea7", "Lead Grade", "Grade"],
    "manual_review": ["\u9700\u4eba\u5de5\u786e\u8ba4", "Manual Review"],
    "status": ["\u5efa\u8054\u72b6\u6001", "\u5efa\u8054\u60c5\u51b5", "\u72b6\u6001", "Status"],
    "next_followup": ["\u4e0b\u6b21\u8ddf\u8fdb", "\u4e0b\u6b21\u8ddf\u8fdb\u65f6\u95f4", "Next Follow-up"],
    "note": ["\u5907\u6ce8", "\u8bf4\u660e", "Notes"],
}

CONVERSATION_HEADER_ALIASES = {
    "record_id": ["\u8bb0\u5f55ID"],
    "customer_index": ["\u5ba2\u6237\u5e8f\u53f7"],
    "company": ["\u5ba2\u6237\u59d3\u540d/\u516c\u53f8", "\u516c\u53f8\u59d3\u540d", "\u516c\u53f8\u540d\u79f0"],
    "facebook_link": ["Facebook\u4e3b\u9875\u94fe\u63a5", "\u4e3b\u9875/\u94fe\u63a5"],
    "message_time": ["\u6d88\u606f\u65f6\u95f4"],
    "direction": ["\u6d88\u606f\u65b9\u5411"],
    "translation": ["\u4e2d\u6587\u7ffb\u8bd1"],
    "effective": ["\u662f\u5426\u6709\u6548\u5ba2\u6237\u56de\u590d"],
    "intent": ["\u5408\u4f5c\u610f\u5411"],
    "intent_reason": ["\u610f\u5411\u5224\u65ad\u4f9d\u636e"],
    "next_action": ["\u4e0b\u4e00\u6b65\u5efa\u8bae"],
}

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=False)
    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=False)
    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") == "\u6211\u65b9\u53d1\u9001":
            sent_customers.add(key)
        if row.get("direction") != "\u5ba2\u6237\u56de\u590d" or row.get("effective") != "\u662f":
            continue
        replied_customers.add(key)
        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
    reply_customers = [
        {
            "company": row.get("company", ""),
            "latest_reply": row.get("translation", "") or row.get("original", ""),
            "next_action": row.get("next_action", "") or "\u8bf7\u4eba\u5de5\u67e5\u770b\u5ba2\u6237\u56de\u590d\uff0c\u5e76\u5224\u65ad\u662f\u5426\u9700\u8981\u7ee7\u7eed\u8ddf\u8fdb\u3002",
            "message_time": row.get("message_time", ""),
        }
        for row in latest_by_customer.values()
    ]
    pending = [item for item in reply_customers 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,
        "reply_status_counter": Counter({"\u5df2\u56de\u590d": len(replied_customers), "\u672a\u56de\u590d": max(len(sent_customers) - len(replied_customers), 0)}),
        "reply_trend": trend,
        "reply_customers": sorted(reply_customers, key=lambda item: item.get("message_time", ""), reverse=True),
        "pending_review": sorted(pending, key=lambda item: item.get("message_time", ""), reverse=True),
    }

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 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 counter_to_items(counter: Counter) -> 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()]


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

def is_excluded(row: Dict[str, str]) -> bool:
    return contains_any(" ".join([row.get("status", ""), row.get("note", "")]), EXCLUDED_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 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 is_bounced(row: Dict[str, str]) -> bool:
    return contains_any(" ".join([row.get("status", ""), row.get("note", "")]), BOUNCED_KEYWORDS)

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 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 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_replies = 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)
    grade_counter = Counter(row.get("lead_grade") 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_replies.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", ""),
            "lead_grade": row.get("lead_grade", ""),
            "manual_review": row.get("manual_review", ""),
            "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_replied": bool(reply),
            "latest_reply": (reply.get("translation", "") or reply.get("original", ""))[:160],
            "next_action": (reply.get("next_action", "") or "\u8bf7\u4eba\u5de5\u67e5\u770b\u5ba2\u6237\u56de\u590d\uff0c\u5e76\u5224\u65ad\u662f\u5426\u9700\u8981\u7ee7\u7eed\u8ddf\u8fdb\u3002")[:160] if reply else "",
        })

    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_pending_review_customers": len(conversation["pending_review"]),
        },
        "charts": {
            "attribute": top_counter(attribute_counter, top_n),
            "customer_type": top_counter(type_counter, top_n),
            "lead_grade": counter_to_items(grade_counter),
            "status": top_counter(status_counter, top_n),
            "source": top_counter(source_counter, top_n),
            "city": top_counter(city_counter, top_n),
            "facebook_reply_status": counter_to_items(conversation["reply_status_counter"]),
            "facebook_reply_trend": conversation["reply_trend"],
        },
        "dashboard_rows": dashboard_rows,
        "attention_customers": attention_rows,
        "facebook_reply_customers_detail": conversation["reply_customers"][:top_n],
        "facebook_pending_review_customers": conversation["pending_review"][: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())