|
|
@@ -1,16 +1,12 @@
|
|
|
-# -*- coding: utf-8 -*-
|
|
|
-"""Build the consolidated customer summary sheet for the outreach workbook."""
|
|
|
+'''Build or repair the single customer summary sheet.'''
|
|
|
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 collections import Counter
|
|
|
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
|
|
|
@@ -19,495 +15,97 @@ 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 .direct_summary import SUMMARY_COLUMNS, SUMMARY_SHEET, identity_keys, merge_record, normalize_summary_record
|
|
|
+except ImportError: # pragma: no cover
|
|
|
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"],
|
|
|
+ from direct_summary import SUMMARY_COLUMNS, SUMMARY_SHEET, identity_keys, merge_record, normalize_summary_record
|
|
|
+
|
|
|
+CONVERSATION_SHEET = 'Facebook\u5bf9\u8bdd\u8bb0\u5f55'
|
|
|
+SKIP_SHEETS = {'\u586b\u5199\u8bf4\u660e', '\u9644\u4ef6', CONVERSATION_SHEET}
|
|
|
+LEGACY_SOURCE_SHEETS = {
|
|
|
+ 'Facebook', 'LinkedIn', 'Google Maps', 'TikTok', '\u534f\u4f1a\u5546\u4f1a',
|
|
|
+ '\u672c\u5730\u6c7d\u8f66\u7f51\u7ad9', '\u6c7d\u8f66\u7f51\u7ad9\u7cbe\u9009\u7ebf\u7d22',
|
|
|
+ 'Sheet11', '\u653f\u5e9c\u91c7\u8d2d\u6295\u6807', '\u6d4b\u8bc4\u535a\u4e3b',
|
|
|
}
|
|
|
-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)
|
|
|
+ return ''
|
|
|
+ return re.sub(r'\s+', ' ', str(value).strip())
|
|
|
|
|
|
|
|
|
-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 headers_for(ws) -> Dict[str, int]:
|
|
|
+ return {clean(cell.value): idx for idx, cell in enumerate(ws[1], start=1) if clean(cell.value)}
|
|
|
|
|
|
|
|
|
-def contains_any(text: str, keywords: Sequence[str]) -> bool:
|
|
|
- return any(keyword.casefold() in text for keyword in keywords)
|
|
|
+def row_to_record(ws, row_idx: int, headers: Dict[str, int]) -> Dict[str, Any]:
|
|
|
+ return {header: ws.cell(row_idx, col_idx).value for header, col_idx in headers.items()}
|
|
|
|
|
|
|
|
|
-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 has_customer_evidence(record: Dict[str, Any]) -> bool:
|
|
|
+ evidence = [
|
|
|
+ '\u516c\u53f8\u59d3\u540d', '\u5ba2\u6237\u59d3\u540d/\u516c\u53f8', '\u516c\u53f8\u540d\u79f0',
|
|
|
+ '\u4e3b\u9875/\u94fe\u63a5', '\u5b98\u7f51\u94fe\u63a5', '\u516c\u53f8\u5b98\u7f51',
|
|
|
+ 'Facebook\u4e3b\u9875\u94fe\u63a5', 'linkined\u4e3b\u9875\u94fe\u63a5', 'google map\u94fe\u63a5',
|
|
|
+ '\u516c\u5171\u90ae\u7bb1', '\u4e2a\u4eba\u90ae\u7bb1', '\u90ae\u7bb1',
|
|
|
+ '\u516c\u5171\u7535\u8bdd/WhatsApp', '\u7535\u8bdd/WhatsApp', '\u8054\u7cfb\u4eba\u7535\u8bdd',
|
|
|
+ ]
|
|
|
+ return any(clean(record.get(key)) for key in evidence)
|
|
|
|
|
|
|
|
|
-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:
|
|
|
+def source_sheets(wb) -> List[str]:
|
|
|
+ ordered: List[str] = []
|
|
|
+ if SUMMARY_SHEET in wb.sheetnames:
|
|
|
+ ordered.append(SUMMARY_SHEET)
|
|
|
+ for sheet in wb.sheetnames:
|
|
|
+ if sheet == SUMMARY_SHEET or sheet in SKIP_SHEETS:
|
|
|
continue
|
|
|
- ws = wb[sheet_name]
|
|
|
- columns = header_map(ws)
|
|
|
- if "name" not in columns and "link" not in columns and "website" not in columns:
|
|
|
+ if sheet in LEGACY_SOURCE_SHEETS:
|
|
|
+ ordered.append(sheet)
|
|
|
+ return ordered
|
|
|
+
|
|
|
+
|
|
|
+def read_all_customer_records(wb) -> Tuple[List[Dict[str, Any]], Counter, int]:
|
|
|
+ records: List[Dict[str, Any]] = []
|
|
|
+ source_counts: Counter = Counter()
|
|
|
+ skipped_blank = 0
|
|
|
+ for sheet in source_sheets(wb):
|
|
|
+ ws = wb[sheet]
|
|
|
+ headers = headers_for(ws)
|
|
|
+ if not headers:
|
|
|
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
|
|
|
+ for row_idx in range(2, ws.max_row + 1):
|
|
|
+ raw = row_to_record(ws, row_idx, headers)
|
|
|
+ if not has_customer_evidence(raw):
|
|
|
+ skipped_blank += 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] = {}
|
|
|
+ normalized = normalize_summary_record(raw, sheet)
|
|
|
+ if not has_customer_evidence(normalized):
|
|
|
+ skipped_blank += 1
|
|
|
+ continue
|
|
|
+ records.append(normalized)
|
|
|
+ source_counts[sheet] += 1
|
|
|
+ return records, source_counts, skipped_blank
|
|
|
|
|
|
- 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
|
|
|
|
|
|
+def merge_records(records: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
|
+ output: List[Dict[str, Any]] = []
|
|
|
+ key_to_index: Dict[Tuple[str, str], int] = {}
|
|
|
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])
|
|
|
+ match_idx: Optional[int] = None
|
|
|
+ for key in identity_keys(record):
|
|
|
+ if key in key_to_index:
|
|
|
+ match_idx = key_to_index[key]
|
|
|
+ break
|
|
|
+ if match_idx is None:
|
|
|
+ output.append(record)
|
|
|
+ match_idx = len(output) - 1
|
|
|
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)
|
|
|
+ output[match_idx] = merge_record(output[match_idx], record)
|
|
|
+ for key in identity_keys(output[match_idx]):
|
|
|
+ key_to_index.setdefault(key, match_idx)
|
|
|
return output
|
|
|
|
|
|
|
|
|
@@ -515,117 +113,97 @@ def write_summary_sheet(wb, rows: Sequence[Dict[str, Any]], sheet_name: str) ->
|
|
|
if sheet_name in wb.sheetnames:
|
|
|
del wb[sheet_name]
|
|
|
ws = wb.create_sheet(sheet_name, 0)
|
|
|
- ws.append(SUMMARY_HEADERS)
|
|
|
+ ws.append(SUMMARY_COLUMNS)
|
|
|
for row in rows:
|
|
|
- ws.append([row.get(header, "") for header in SUMMARY_HEADERS])
|
|
|
- ws.freeze_panes = "A2"
|
|
|
+ ws.append([row.get(header, '') for header in SUMMARY_COLUMNS])
|
|
|
+ ws.freeze_panes = 'A2'
|
|
|
ws.auto_filter.ref = ws.dimensions
|
|
|
- header_fill = PatternFill(fill_type="solid", fgColor="D9EAF7")
|
|
|
+ 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]
|
|
|
+ widths = [28, 12, 16, 22, 34, 18, 18, 28, 24, 34, 34, 34, 24, 28, 20, 12, 20, 18, 18, 80]
|
|
|
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 path.with_name('~$' + path.name)
|
|
|
+
|
|
|
+
|
|
|
+def build_report(workbook_path: Path, backup_path: Optional[Path], rows: Sequence[Dict[str, Any]], source_counts: Counter, skipped_blank: int, dry_run: bool, sheet_name: str) -> Dict[str, Any]:
|
|
|
+ grade_col = '\u7ebf\u7d22\u7b49\u7ea7'
|
|
|
+ manual_col = '\u9700\u4eba\u5de5\u786e\u8ba4'
|
|
|
+ attr_col = '\u5ba2\u6237\u5c5e\u6027'
|
|
|
+ type_col = '\u5ba2\u6237\u7c7b\u578b'
|
|
|
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,
|
|
|
+ 'workbook': str(workbook_path),
|
|
|
+ 'summary_sheet': sheet_name,
|
|
|
+ 'summary_headers': SUMMARY_COLUMNS,
|
|
|
+ 'dry_run': dry_run,
|
|
|
+ 'backup': str(backup_path) if backup_path else '',
|
|
|
+ 'summary_rows': len(rows),
|
|
|
+ 'source_rows': dict(source_counts),
|
|
|
+ 'skipped_blank_or_reserved_rows': skipped_blank,
|
|
|
+ 'attribute_counts': dict(Counter(clean(row.get(attr_col)) for row in rows if clean(row.get(attr_col)))),
|
|
|
+ 'type_counts': dict(Counter(clean(row.get(type_col)) for row in rows if clean(row.get(type_col)))),
|
|
|
+ 'lead_grade_counts': dict(Counter(clean(row.get(grade_col)) for row in rows if clean(row.get(grade_col)))),
|
|
|
+ 'manual_review_counts': dict(Counter(clean(row.get(manual_col)) for row in rows if clean(row.get(manual_col)))),
|
|
|
+ 'missing_counts': {
|
|
|
+ '\u5ba2\u6237\u5c5e\u6027': sum(1 for row in rows if not clean(row.get(attr_col))),
|
|
|
+ '\u5ba2\u6237\u7c7b\u578b': sum(1 for row in rows if not clean(row.get(type_col))),
|
|
|
+ '\u7ebf\u7d22\u7b49\u7ea7': sum(1 for row in rows if not clean(row.get(grade_col))),
|
|
|
+ '\u9700\u4eba\u5de5\u786e\u8ba4': sum(1 for row in rows if not clean(row.get(manual_col))),
|
|
|
+ '\u5907\u6ce8': sum(1 for row in rows if not clean(row.get('\u5907\u6ce8'))),
|
|
|
+ },
|
|
|
}
|
|
|
|
|
|
|
|
|
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 = 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")
|
|
|
+ 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.")
|
|
|
+ parser = argparse.ArgumentParser(description='Build or preview the single 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")
|
|
|
+ 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.")
|
|
|
+ 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)}")
|
|
|
-
|
|
|
+ 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)
|
|
|
+ records, source_counts, skipped_blank = read_all_customer_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)
|
|
|
+ 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)
|
|
|
+ report = build_report(workbook_path, backup_path, rows, source_counts, skipped_blank, 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__":
|
|
|
+if __name__ == '__main__':
|
|
|
raise SystemExit(main())
|
|
|
-
|