#!/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 = ["客户信息汇总表", "客户信息汇总"] DEFAULT_TITLE = "五菱海外客户建联中台" 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"], } 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 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]], 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)] 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: 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], }) 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)), }, "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), }, "dashboard_rows": dashboard_rows, "attention_customers": attention_rows, "definitions": { "有效客户": "总表中未被备注或状态标记为已剔除、跳过、低优先级的客户。", "可建联客户": "有效客户中至少有邮箱、电话/WhatsApp、官网、Facebook、LinkedIn 或 Google Maps 入口之一。", "已建联客户": "状态或备注包含已发送邮件、已发私信、已关注并私信、等待回复、已回复、已建联或有意向，且未被标记退信/拒收。", "建联率": "已建联客户数 / 可建联客户数。", "邮箱覆盖率": "有个人邮箱或公共邮箱的有效客户数 / 有效客户数。", "退信率": "邮件退回、拒收、无法送达或域名/邮箱不存在客户数 / 有邮箱客户数。", }, } 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("--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) data = build_dashboard_data(rows, 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, "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())