# -*- coding: utf-8 -*- """Build the consolidated customer summary sheet for the outreach workbook.""" from __future__ import annotations import argparse import json import re from collections import Counter, defaultdict from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse from openpyxl import load_workbook from openpyxl.styles import Font, PatternFill from openpyxl.utils import get_column_letter try: from .artifact_manager import create_backup_once, resolve_artifact_path from .workbook_resolver import resolve_workbook_path from .customer_taxonomy import ( CUSTOMER_TAXONOMY, VALID_ATTRIBUTES, VALID_CUSTOMER_TYPES, classify_attribute_type, is_valid_pair, normalize_existing_classification, ) except ImportError: # pragma: no cover - supports direct CLI execution from artifact_manager import create_backup_once, resolve_artifact_path from workbook_resolver import resolve_workbook_path from customer_taxonomy import ( CUSTOMER_TAXONOMY, VALID_ATTRIBUTES, VALID_CUSTOMER_TYPES, classify_attribute_type, is_valid_pair, normalize_existing_classification, ) SUMMARY_SHEET = "客户信息汇总表" DEFAULT_COUNTRY = "摩洛哥" DEFAULT_STATUS = "未联系" PLATFORM_SHEETS = [ "Facebook", "Google Maps", "LinkedIn", "TikTok", "协会商会", "本地汽车网站", "Sheet11", "汽车网站精选线索", ] # Keep this header exactly aligned with the user's reference workbook: # 摩洛哥重点客户50家-汽车渠道与平台行业渠道.xlsx SUMMARY_HEADERS = [ "公司姓名", "国家", "城市", "客户类型", "官网链接", "联系人", "职位", "个人邮箱", "联系人电话", "Facebook主页链接", "linkined主页链接", "google map链接", "公共电话/WhatsApp", "公共邮箱", "客户属性", "建联状态", "下次跟进", "备注", ] HEADER_ALIASES = { "name": ["公司姓名", "客户姓名/公司", "公司名称", "客户名称", "Name", "Company"], "country": ["国家", "Country"], "city": ["城市", "City"], "attribute": ["客户属性", "客户大类", "大类", "Customer Attribute"], "type": ["客户类型", "细分客户类型", "类型", "Customer Type"], "link": ["主页/链接", "公司链接", "链接", "Link", "URL"], "facebook_link": ["Facebook主页链接", "Facebook链接", "facebook链接", "主页/链接"], "linkedin_link": ["linkined主页链接", "LinkedIn主页链接", "LinkedIn链接", "linkin链接", "linkin连接"], "google_maps_link": ["google map链接", "Google Maps链接", "Google Map链接", "地图链接"], "website": ["官网链接", "公司官网", "官网", "官方网站", "Website", "Company Website"], "source_site": ["来源网站", "来源", "Source"], "contact": ["联系人", "姓名", "Contact"], "position": ["职位", "职务", "Position", "Title"], "public_phone": ["公共电话/WhatsApp", "电话/WhatsApp", "公司公共电话", "电话", "WhatsApp", "Phone"], "contact_phone": ["联系人电话", "个人电话", "联系电话"], "public_email": ["公共邮箱", "邮箱", "公司公共邮箱（任一有效即可）", "公司公共邮箱", "Email"], "personal_email": ["个人邮箱", "个人邮箱（不一定有效）"], "business": ["主营业务", "公司主营业务", "业务", "Business"], "status": ["建联状态", "建联情况", "状态", "Status"], "next_followup": ["下次跟进", "下次跟进时间", "Next Follow-up"], "note": ["备注", "说明", "Notes"], } AUTO_NAME_KEYWORDS = ["auto", "autos", "automobile", "automobiles", "cars", "car center", "motors", "garage"] BRAND_CONTEXT_KEYWORDS = ["renault", "dacia", "seat", "cupra", "honda", "peugeot", "citroen", "citroën", "hyundai", "kia", "toyota", "nissan", "ford", "fiat", "volkswagen", "vw", "bmw", "mercedes", "audi", "opel", "skoda", "suzuki", "mazda", "jeep", "chery", "geely", "dfsk", "jac", "byd", "mg", "haval", "foton", "changan", "alfa romeo", "škoda", "porsche", "jaguar", "land rover", "ds", "gwm", "mini", "stellantis", "leapmotor", "sopriam"] SOURCE_MAP = { "Facebook": "Facebook", "LinkedIn": "LinkedIn", "Google Maps": "Google Maps", "TikTok": "TikTok", "协会商会": "平台与行业渠道", "本地汽车网站": "当地汽车网站", "Sheet11": "Moteur.ma", } GENERIC_NAME_WORDS = { "sarl", "sa", "sas", "ltd", "llc", "inc", "co", "company", "groupe", "group", "maroc", "morocco", "officiel", "official", } # Strict five-attribute taxonomy is imported from customer_taxonomy.py. EMAIL_RE = re.compile(r"[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}", re.I) PHONE_RE = re.compile(r"(?:\+?\d[\d\s()./-]{6,}\d)") SPLIT_RE = re.compile(r"[；;,，\n]+") @dataclass class Record: idx: int sheet: str row: int values: Dict[str, str] sources: List[str] = field(default_factory=list) source_links: List[str] = field(default_factory=list) @dataclass class DSU: parent: Dict[int, int] = field(default_factory=dict) def find(self, value: int) -> int: self.parent.setdefault(value, value) if self.parent[value] != value: self.parent[value] = self.find(self.parent[value]) return self.parent[value] def union(self, left: int, right: int) -> None: root_left = self.find(left) root_right = self.find(right) if root_left != root_right: self.parent[root_right] = root_left def clean(value: Any) -> str: if value is None: return "" return re.sub(r"\s+", " ", str(value).strip()) def split_values(text: str) -> List[str]: output: List[str] = [] for part in SPLIT_RE.split(clean(text)): item = part.strip() if item and item not in output: output.append(item) return output def first_non_empty(*values: str, default: str = "") -> str: for value in values: if clean(value): return clean(value) return default STATUS_PRIORITY = { "\u90ae\u4ef6\u9000\u56de": 90, "\u90e8\u5206\u90ae\u4ef6\u9000\u56de": 85, "\u5df2\u53d1\u9001\u90ae\u4ef6": 80, "\u5df2\u53d1\u90ae\u4ef6": 80, "\u5df2\u5efa\u8054": 70, "\u65e0\u6cd5\u786e\u8ba4": 30, "\u5f85\u53d1\u9001": 20, "\u672a\u53d1\u9001": 10, "\u672a\u8054\u7cfb": 5, "\u672a\u5efa\u8054": 5, } def normalize_status(value: str) -> str: value = clean(value) if value == "\u5df2\u53d1\u90ae\u4ef6": return "\u5df2\u53d1\u9001\u90ae\u4ef6" if value == "\u672a\u5efa\u8054": return "\u672a\u8054\u7cfb" return value def preferred_status(values: Sequence[str], default: str = DEFAULT_STATUS) -> str: statuses = [normalize_status(value) for value in values if clean(value)] if not statuses: return default return max(statuses, key=lambda value: STATUS_PRIORITY.get(value, 40)) def preferred_name(values: Sequence[str]) -> str: cleaned = dedupe_keep_order(values) if not cleaned: return "" normalized_counts = Counter(normalize_name(value) for value in values if normalize_name(value)) if not normalized_counts: return cleaned[0] best_key, _ = normalized_counts.most_common(1)[0] for value in cleaned: if normalize_name(value) == best_key: return value return cleaned[0] def dedupe_keep_order(values: Iterable[str]) -> List[str]: seen = set() output: List[str] = [] for value in values: item = clean(value) key = item.casefold() if item and key not in seen: seen.add(key) output.append(item) return output def joined(values: Iterable[str]) -> str: return "；".join(dedupe_keep_order(values)) def normalize_name(name: str) -> str: lowered = clean(name).casefold() lowered = re.sub(r"[^\w\s]+", " ", lowered, flags=re.U) words = [word for word in lowered.split() if word not in GENERIC_NAME_WORDS] return " ".join(words).strip() def normalize_phone(value: str) -> List[str]: phones: List[str] = [] for match in PHONE_RE.findall(value or ""): digits = re.sub(r"\D+", "", match) if len(digits) >= 7 and digits not in phones: phones.append(digits) return phones def normalize_emails(value: str) -> List[str]: emails: List[str] = [] for match in EMAIL_RE.findall(value or ""): email = match.casefold() if email not in emails: emails.append(email) return emails def normalize_url(url: str) -> str: text = clean(url) if not text: return "" if not re.match(r"^[a-z]+://", text, re.I): text = "https://" + text parsed = urlparse(text) host = parsed.netloc.casefold().removeprefix("www.") path = re.sub(r"/+$", "", parsed.path or "") query = "" if "google." in host and path.startswith("/maps"): pairs = [(k, v) for k, v in parse_qsl(parsed.query, keep_blank_values=False) if k in {"q", "query", "cid", "place_id"}] query = urlencode(pairs) if path in {"/maps/search", "/maps"} and not query: return "" return urlunparse(("https", host, path, "", query, "")) def header_map(ws) -> Dict[str, int]: raw_headers = {clean(cell.value): idx for idx, cell in enumerate(ws[1], start=1) if clean(cell.value)} mapped: Dict[str, int] = {} for target, aliases in HEADER_ALIASES.items(): for alias in aliases: if alias in raw_headers: mapped[target] = raw_headers[alias] break return mapped def row_value(ws, row: int, columns: Dict[str, int], key: str) -> str: col = columns.get(key) if not col: return "" return clean(ws.cell(row=row, column=col).value) def is_blank_reserved(record: Dict[str, str]) -> bool: evidence_keys = ["name", "link", "facebook_link", "linkedin_link", "google_maps_link", "website", "contact", "public_phone", "contact_phone", "public_email", "personal_email", "business", "note"] return not any(clean(record.get(key, "")) for key in evidence_keys) def source_values(sheet_name: str, record: Dict[str, str]) -> List[str]: if sheet_name == "汽车网站精选线索": return split_values(record.get("source_site", "")) or ["汽车网站精选线索"] return [SOURCE_MAP.get(sheet_name, sheet_name)] def all_record_links(values: Dict[str, str]) -> List[str]: links: List[str] = [] for key in ["link", "facebook_link", "linkedin_link", "google_maps_link", "website"]: links.extend(split_values(values.get(key, ""))) return dedupe_keep_order(links) def platform_links(record: Record) -> Dict[str, List[str]]: values = record.values result = {"website": [], "facebook": [], "linkedin": [], "google_maps": []} result["website"].extend(split_values(values.get("website", ""))) result["facebook"].extend(split_values(values.get("facebook_link", ""))) result["linkedin"].extend(split_values(values.get("linkedin_link", ""))) result["google_maps"].extend(split_values(values.get("google_maps_link", ""))) generic_links = split_values(values.get("link", "")) for link in generic_links: lower = link.casefold() if "facebook.com" in lower or record.sheet == "Facebook": result["facebook"].append(link) elif "linkedin.com" in lower or record.sheet == "LinkedIn": result["linkedin"].append(link) elif "google." in lower and "/maps" in lower or record.sheet == "Google Maps": result["google_maps"].append(link) else: result["website"].append(link) return {key: dedupe_keep_order(value) for key, value in result.items()} def combined_text(values: Dict[str, str], sheet_name: str, sources: Sequence[str]) -> str: return " ".join([ sheet_name, " ".join(sources), values.get("name", ""), values.get("attribute", ""), values.get("type", ""), values.get("business", ""), values.get("note", ""), values.get("link", ""), values.get("website", ""), ]).casefold() def contains_any(text: str, keywords: Sequence[str]) -> bool: return any(keyword.casefold() in text for keyword in keywords) def classify_existing_type(value: str) -> Optional[Tuple[str, str]]: mapped = normalize_existing_classification("", value, value) return mapped if mapped != ("", "") else None def classify_attribute_and_type(values: Dict[str, str], sheet_name: str, sources: Sequence[str]) -> Tuple[str, str]: return classify_attribute_type(values, sheet_name=sheet_name, sources=sources) def read_records(wb) -> Tuple[List[Record], Dict[str, int], int]: records: List[Record] = [] rows_by_sheet: Dict[str, int] = {} blank_rows = 0 idx = 0 for sheet_name in PLATFORM_SHEETS: if sheet_name not in wb.sheetnames: continue ws = wb[sheet_name] columns = header_map(ws) if "name" not in columns and "link" not in columns and "website" not in columns: continue for row in range(2, ws.max_row + 1): values = {key: row_value(ws, row, columns, key) for key in HEADER_ALIASES} if is_blank_reserved(values) or not any(values.get(key) for key in ["name", "link", "facebook_link", "linkedin_link", "google_maps_link", "website", "public_phone", "contact_phone", "public_email", "personal_email"]): blank_rows += 1 continue idx += 1 sources = source_values(sheet_name, values) values["attribute"], values["type"] = classify_attribute_and_type(values, sheet_name, sources) if not values.get("business"): values["business"] = "汽车渠道线索，需人工确认" links = all_record_links(values) records.append(Record(idx=idx, sheet=sheet_name, row=row, values=values, sources=sources, source_links=links)) rows_by_sheet[sheet_name] = rows_by_sheet.get(sheet_name, 0) + 1 return records, rows_by_sheet, blank_rows def choose_group_classification(group: Sequence[Record]) -> Tuple[str, str]: for item in group: attr = clean(item.values.get("attribute", "")) typ = clean(item.values.get("type", "")) if is_valid_pair(attr, typ): return attr, typ merged_values = { "attribute": joined(item.values.get("attribute", "") for item in group), "type": joined(item.values.get("type", "") for item in group), "name": joined(item.values.get("name", "") for item in group), "business": joined(item.values.get("business", "") for item in group), "note": joined(item.values.get("note", "") for item in group), "link": joined(item.values.get("link", "") for item in group), "website": joined(item.values.get("website", "") for item in group), } sheets = [item.sheet for item in group] sources = [source for item in group for source in item.sources] return classify_attribute_and_type(merged_values, joined(sheets), sources) def merge_records(records: Sequence[Record]) -> List[Dict[str, Any]]: dsu = DSU() buckets: Dict[str, int] = {} name_bucket: Dict[str, int] = {} for record in records: dsu.find(record.idx) strong_keys: List[str] = [] for email in normalize_emails(joined([record.values.get("public_email", ""), record.values.get("personal_email", "")])): strong_keys.append("email:" + email) for phone in normalize_phone(joined([record.values.get("public_phone", ""), record.values.get("contact_phone", "")])): strong_keys.append("phone:" + phone) # Only platform/detail URLs are strong duplicate keys. Merchant homepages are # kept as evidence, but not used alone because group websites can represent # many brands, branches, or dealer pages. strong_link_values: List[str] = [] for key_name in ["facebook_link", "linkedin_link", "google_maps_link", "link"]: for link in split_values(record.values.get(key_name, "")): lower = link.casefold() is_platform_detail = ( "facebook.com" in lower or "linkedin.com" in lower or ("google." in lower and "/maps" in lower) ) if is_platform_detail: strong_link_values.append(link) for link in strong_link_values: normalized = normalize_url(link) if normalized: strong_keys.append("url:" + normalized) for key in strong_keys: if key in buckets: dsu.union(record.idx, buckets[key]) else: buckets[key] = record.idx for record in records: name_key = normalize_name(record.values.get("name", "")) if len(name_key) < 4: continue key = "name:" + name_key if key in name_bucket: dsu.union(record.idx, name_bucket[key]) else: name_bucket[key] = record.idx grouped: Dict[int, List[Record]] = defaultdict(list) for record in records: grouped[dsu.find(record.idx)].append(record) output: List[Dict[str, Any]] = [] for group in sorted(grouped.values(), key=lambda items: min(item.idx for item in items)): link_sets = {"website": [], "facebook": [], "linkedin": [], "google_maps": []} for item in group: item_links = platform_links(item) for key in link_sets: link_sets[key].extend(item_links[key]) names = [item.values.get("name", "") for item in group] countries = [item.values.get("country", "") for item in group] cities = [item.values.get("city", "") for item in group] summary_attribute, summary_type = choose_group_classification(group) contacts = [item.values.get("contact", "") for item in group] positions = [item.values.get("position", "") for item in group] personal_emails = [value for item in group for value in split_values(item.values.get("personal_email", ""))] public_emails = [value for item in group for value in split_values(item.values.get("public_email", ""))] contact_phones = [value for item in group for value in split_values(item.values.get("contact_phone", ""))] public_phones = [value for item in group for value in split_values(item.values.get("public_phone", ""))] businesses = [item.values.get("business", "") for item in group] statuses = [item.values.get("status", "") for item in group] next_followups = [item.values.get("next_followup", "") for item in group] sources = [source for item in group for source in item.sources] sheets = [item.sheet for item in group] notes = [item.values.get("note", "") for item in group] source_evidence = [f"来自 {item.sheet} row {item.row}" for item in group] merged_count_note = f"合并来源数量：{len(group)}" if len(group) > 1 else "合并来源数量：1" source_note = f"客户来源：{joined(sources)}；来源Sheet：{joined(sheets)}" business_note = f"主营业务汇总：{joined(businesses)}" if joined(businesses) else "" note = joined([merged_count_note, source_note, business_note, *source_evidence, *notes]) row = { "公司姓名": preferred_name(names), "国家": first_non_empty(*countries, default=DEFAULT_COUNTRY), "城市": joined(cities), "客户类型": summary_type, "官网链接": joined(link_sets["website"]), "联系人": joined(contacts), "职位": joined(positions), "个人邮箱": joined(personal_emails), "联系人电话": joined(contact_phones), "Facebook主页链接": joined(link_sets["facebook"]), "linkined主页链接": joined(link_sets["linkedin"]), "google map链接": joined(link_sets["google_maps"]), "公共电话/WhatsApp": joined(public_phones), "公共邮箱": joined(public_emails), "客户属性": summary_attribute, "建联状态": first_non_empty(*statuses, default=DEFAULT_STATUS), "下次跟进": joined(next_followups), "备注": note, "_merge_count": len(group), "_sources": joined(sources), } output.append(row) return output def write_summary_sheet(wb, rows: Sequence[Dict[str, Any]], sheet_name: str) -> None: if sheet_name in wb.sheetnames: del wb[sheet_name] ws = wb.create_sheet(sheet_name, 0) ws.append(SUMMARY_HEADERS) for row in rows: ws.append([row.get(header, "") for header in SUMMARY_HEADERS]) ws.freeze_panes = "A2" ws.auto_filter.ref = ws.dimensions header_fill = PatternFill(fill_type="solid", fgColor="D9EAF7") for cell in ws[1]: cell.font = Font(bold=True) cell.fill = header_fill widths = [28, 12, 18, 22, 36, 18, 18, 28, 24, 36, 36, 36, 24, 28, 20, 14, 18, 76] for idx, width in enumerate(widths, start=1): ws.column_dimensions[get_column_letter(idx)].width = width def lock_file_for(path: Path) -> Path: return path.with_name("~$" + path.name) def build_report(workbook_path: Path, backup_path: Optional[Path], records: Sequence[Record], rows: Sequence[Dict[str, Any]], rows_by_sheet: Dict[str, int], blank_rows: int, dry_run: bool, sheet_name: str) -> Dict[str, Any]: source_counts = Counter(source for record in records for source in record.sources) attribute_counts = Counter(row.get("客户属性", "") for row in rows if clean(row.get("客户属性", ""))) type_counts = Counter(row.get("客户类型", "") for row in rows if clean(row.get("客户类型", ""))) missing = { "客户属性": sum(1 for row in rows if not clean(row.get("客户属性", ""))), "客户类型": sum(1 for row in rows if not clean(row.get("客户类型", ""))), "主营业务": sum(1 for row in rows if "主营业务" in SUMMARY_HEADERS and not clean(row.get("主营业务", ""))), "备注": sum(1 for row in rows if not clean(row.get("备注", ""))), } invalid_category_rows = [ {"name": row.get("公司姓名", ""), "attribute": row.get("客户属性", ""), "type": row.get("客户类型", ""), "sources": row.get("_sources", "")} for row in rows if not is_valid_pair(clean(row.get("客户属性", "")), clean(row.get("客户类型", ""))) ] samples = [] for row in rows: merge_count = int(row.get("_merge_count", 0) or 0) if merge_count > 1: samples.append({"name": row.get("公司姓名", ""), "attribute": row.get("客户属性", ""), "type": row.get("客户类型", ""), "sources": row.get("_sources", ""), "merged_source_rows": merge_count}) if len(samples) >= 5: break return { "workbook": str(workbook_path), "summary_sheet": sheet_name, "summary_headers": SUMMARY_HEADERS, "dry_run": dry_run, "backup": str(backup_path) if backup_path else "", "input_valid_rows": len(records), "summary_rows": len(rows), "merged_duplicates": len(records) - len(rows), "rows_by_sheet": rows_by_sheet, "source_counts": dict(source_counts), "attribute_counts": dict(attribute_counts), "type_counts": dict(type_counts), "missing_counts": missing, "invalid_category_count": len(invalid_category_rows), "invalid_category_rows": invalid_category_rows[:50], "skipped_blank_or_reserved_rows": blank_rows, "merge_samples": samples, } def save_json_report(report: Dict[str, Any], output: str) -> None: if not output: return path = resolve_artifact_path(output, kind="summary_report", default_name="report.json") path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Build or preview the consolidated customer summary sheet.") parser.add_argument("--excel", default="", help="Workbook path. If omitted, resolve the project workbook by skill rules.") parser.add_argument("--summary-sheet", default=SUMMARY_SHEET, help="Summary sheet name.") parser.add_argument("--write-summary", action="store_true", help="Write or overwrite the summary sheet. Omit for preview only.") parser.add_argument("--dry-run", action="store_true", help="Preview only; never saves the workbook.") parser.add_argument("--no-backup", action="store_true", help="Skip backup when writing.") parser.add_argument("--output", default="", help="Optional JSON report path.") parser.add_argument("--run-id", default="", help="Run ID used for artifact and backup paths.") return parser.parse_args(argv) def main(argv: Optional[Sequence[str]] = None) -> int: args = parse_args(argv) 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) sheet_name = clean(args.summary_sheet) or SUMMARY_SHEET should_write = bool(args.write_summary and not args.dry_run) if should_write and lock_file_for(workbook_path).exists(): raise PermissionError(f"Workbook appears to be open in Excel: {lock_file_for(workbook_path)}") wb = load_workbook(workbook_path) records, rows_by_sheet, blank_rows = read_records(wb) rows = merge_records(records) backup_path: Optional[Path] = None if should_write: if not args.no_backup: backup_path = create_backup_once(workbook_path, purpose="summary", run_id=args.run_id or None) write_summary_sheet(wb, rows, sheet_name) wb.save(workbook_path) report = build_report(workbook_path, backup_path, records, rows, rows_by_sheet, blank_rows, dry_run=not should_write, sheet_name=sheet_name) save_json_report(report, args.output) print(json.dumps(report, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())