# -*- coding: utf-8 -*- """Strict customer taxonomy and classification helpers for Wuling outreach. Source intentionally uses unicode escapes for Chinese labels so scripted edits do not depend on the Windows console code page. Runtime values are normal unicode. """ from __future__ import annotations import re from typing import Any, Mapping, Sequence, Tuple def zh(value: str) -> str: return value.encode("ascii").decode("unicode_escape") AUTO_CHANNEL = zh(r"\u6c7d\u8f66\u6e20\u9053\u5408\u4f5c\u4f19\u4f34") BULK_OPERATOR = zh(r"\u6279\u91cf\u91c7\u8d2d\u4e0e\u8fd0\u8425\u5ba2\u6237") USED_TRANSITION = zh(r"\u4e8c\u624b\u8f66\u8f6c\u578b\u5019\u9009") PLATFORM_CHANNEL = zh(r"\u5e73\u53f0\u4e0e\u884c\u4e1a\u6e20\u9053") ECOSYSTEM_SUPPORT = zh(r"\u751f\u6001\u652f\u6301\u8d44\u6e90") CAR_IMPORTER = zh(r"\u6c7d\u8f66\u8fdb\u53e3\u5546") NATIONAL_AGENT = zh(r"\u5168\u56fd\u4ee3\u7406\u5546") NATIONAL_DISTRIBUTOR = zh(r"\u5168\u56fd\u5206\u9500\u5546") REGIONAL_DISTRIBUTOR = zh(r"\u533a\u57df\u5206\u9500\u5546") MULTI_BRAND_DEALER = zh(r"\u591a\u54c1\u724c\u7ecf\u9500\u5546") COMMERCIAL_VEHICLE_CHANNEL = zh(r"\u5546\u7528\u8f66\u6e20\u9053\u5546") CHINA_BRAND_DEALER = zh(r"\u4e2d\u56fd\u54c1\u724c\u7ecf\u9500\u5546") NEV_SMALL_VEHICLE_CHANNEL = zh(r"\u65b0\u80fd\u6e90\u6216\u5c0f\u578b\u8f66\u6e20\u9053\u5546") CAR_RENTAL = zh(r"\u6c7d\u8f66\u79df\u8d41\u516c\u53f8") LONG_TERM_RENTAL = zh(r"\u957f\u671f\u79df\u8d41\u516c\u53f8") CORPORATE_FLEET = zh(r"\u4f01\u4e1a\u8f66\u961f") LOGISTICS_DELIVERY = zh(r"\u7269\u6d41\u914d\u9001\u4f01\u4e1a") GOV_PROCUREMENT = zh(r"\u653f\u5e9c\u6216\u673a\u6784\u91c7\u8d2d\u65b9") USED_CHAIN = zh(r"\u8fde\u9501\u4e8c\u624b\u8f66\u4f01\u4e1a") USED_PHYSICAL = zh(r"\u5b9e\u4f53\u4e8c\u624b\u8f66\u4f01\u4e1a") USED_IMPORTER = zh(r"\u8fdb\u53e3\u4e8c\u624b\u8f66\u4f01\u4e1a") NEW_USED_COMPREHENSIVE = zh(r"\u65b0\u8f66\u4e0e\u4e8c\u624b\u8f66\u7efc\u5408\u4f01\u4e1a") CAR_MARKETPLACE = zh(r"\u6c7d\u8f66\u4ea4\u6613\u5e73\u53f0") CAR_ASSOCIATION = zh(r"\u6c7d\u8f66\u534f\u4f1a") CHAMBER = zh(r"\u5546\u4f1a") DEALER_ALLIANCE = zh(r"\u8f66\u5546\u8054\u76df") DEALER_REFERRAL = zh(r"\u7ecf\u9500\u5546\u8d44\u6e90\u5f15\u8350\u673a\u6784") AFTER_SALES_NETWORK = zh(r"\u552e\u540e\u670d\u52a1\u7f51\u7edc") PARTS_WAREHOUSE = zh(r"\u5907\u4ef6\u4f9b\u5e94\u4e0e\u4ed3\u50a8\u4f01\u4e1a") IMPORT_CERTIFICATION = zh(r"\u8fdb\u53e3\u8ba4\u8bc1\u4e0e\u4e0a\u724c\u673a\u6784") FINANCE_INSURANCE = zh(r"\u91d1\u878d\u4fdd\u9669\u673a\u6784") VEHICLE_LOGISTICS = zh(r"\u8f66\u8f86\u7269\u6d41\u4f01\u4e1a") CUSTOMER_TAXONOMY = { AUTO_CHANNEL: [ CAR_IMPORTER, NATIONAL_AGENT, NATIONAL_DISTRIBUTOR, REGIONAL_DISTRIBUTOR, MULTI_BRAND_DEALER, COMMERCIAL_VEHICLE_CHANNEL, CHINA_BRAND_DEALER, NEV_SMALL_VEHICLE_CHANNEL, ], BULK_OPERATOR: [ CAR_RENTAL, LONG_TERM_RENTAL, CORPORATE_FLEET, LOGISTICS_DELIVERY, GOV_PROCUREMENT, ], USED_TRANSITION: [ USED_CHAIN, USED_PHYSICAL, USED_IMPORTER, NEW_USED_COMPREHENSIVE, ], PLATFORM_CHANNEL: [ CAR_MARKETPLACE, CAR_ASSOCIATION, CHAMBER, DEALER_ALLIANCE, DEALER_REFERRAL, ], ECOSYSTEM_SUPPORT: [ AFTER_SALES_NETWORK, PARTS_WAREHOUSE, IMPORT_CERTIFICATION, FINANCE_INSURANCE, VEHICLE_LOGISTICS, ], } VALID_ATTRIBUTES = set(CUSTOMER_TAXONOMY.keys()) VALID_CUSTOMER_TYPES = {item for items in CUSTOMER_TAXONOMY.values() for item in items} TYPE_TO_ATTRIBUTE = {item: attr for attr, items in CUSTOMER_TAXONOMY.items() for item in items} KEY_NAME = zh(r"\u516c\u53f8\u59d3\u540d") KEY_ATTRIBUTE = zh(r"\u5ba2\u6237\u5c5e\u6027") KEY_TYPE = zh(r"\u5ba2\u6237\u7c7b\u578b") KEY_BUSINESS = zh(r"\u4e3b\u8425\u4e1a\u52a1") KEY_NOTE = zh(r"\u5907\u6ce8") OLD_TO_NEW_TYPE = { zh(r"\u533a\u57df\u65b0\u8f66\u7ecf\u9500\u516c\u53f8"): REGIONAL_DISTRIBUTOR, zh(r"\u533a\u57df\u591a\u54c1\u724c\u6388\u6743\u7ecf\u9500\u516c\u53f8"): MULTI_BRAND_DEALER, zh(r"\u5168\u56fd\u591a\u54c1\u724c\u6c7d\u8f66\u5206\u9500\u96c6\u56e2"): NATIONAL_DISTRIBUTOR, zh(r"\u5168\u56fd\u591a\u54c1\u724c\u8fdb\u53e3\u4e0e\u5206\u9500\u96c6\u56e2"): NATIONAL_DISTRIBUTOR, zh(r"\u6c7d\u8f66\u8fdb\u53e3\u4e0e\u5206\u9500\u516c\u53f8"): CAR_IMPORTER, zh(r"\u6c7d\u8f66\u8fdb\u53e3\u4e0e\u5206\u9500\u8fd0\u8425\u516c\u53f8"): CAR_IMPORTER, zh(r"\u8f7b\u578b\u5546\u7528\u8f66\u8fdb\u53e3\u4e0e\u5206\u9500\u516c\u53f8"): COMMERCIAL_VEHICLE_CHANNEL, zh(r"\u591a\u54c1\u724c\u6c7d\u8f66\u4e0e\u5546\u7528\u8f66\u5206\u9500\u516c\u53f8"): COMMERCIAL_VEHICLE_CHANNEL, zh(r"\u6c7d\u8f66\u5bfc\u8d2d\u3001\u5185\u5bb9\u4e0e\u4ea4\u6613\u5e73\u53f0"): CAR_MARKETPLACE, zh(r"\u6c7d\u8f66\u4e0e\u5546\u7528\u8f66\u4ea4\u6613\u5e73\u53f0"): CAR_MARKETPLACE, zh(r"\u7efc\u5408\u6c7d\u8f66\u4ea4\u6613\u4e0e\u4e13\u4e1a\u76ee\u5f55\u5e73\u53f0"): CAR_MARKETPLACE, zh(r"\u7efc\u5408\u5206\u7c7b\u4fe1\u606f\u4e0e\u6c7d\u8f66\u4ea4\u6613\u5e73\u53f0"): CAR_MARKETPLACE, zh(r"\u8ba4\u8bc1\u4e8c\u624b\u8f66\u4ea4\u6613\u5e73\u53f0"): CAR_MARKETPLACE, zh(r"\u96c6\u56e2\u5316\u4e8c\u624b\u8f66\u4ea4\u6613\u4e0e\u7f6e\u6362\u5e73\u53f0"): CAR_MARKETPLACE, zh(r"\u6c7d\u8f66\u4ea4\u6613\u4e0e\u7ecf\u9500\u5546\u83b7\u5ba2\u5e73\u53f0"): CAR_MARKETPLACE, zh(r"\u8ba4\u8bc1\u4e8c\u624b\u8f66\u4e0e\u6c7d\u8f66\u4ea4\u6613\u5e73\u53f0"): CAR_MARKETPLACE, zh(r"\u6c7d\u8f66\u8fdb\u53e3\u5546\u884c\u4e1a\u534f\u4f1a"): CAR_ASSOCIATION, zh(r"\u6c7d\u8f66\u5de5\u4e1a\u4e0e\u884c\u4e1a\u534f\u4f1a"): CAR_ASSOCIATION, zh(r"\u65e0\u53f8\u673a\u6c7d\u8f66\u79df\u8d41\u884c\u4e1a\u8054\u5408\u4f1a"): CAR_ASSOCIATION, zh(r"\u884c\u4e1aB2B\u64ae\u5408\u4e0e\u4f1a\u8bae\u5e73\u53f0"): DEALER_REFERRAL, zh(r"\u6c7d\u8f66\u884c\u4e1a\u5c55\u4f1a\u4e0e\u4f01\u4e1a\u6d3b\u52a8\u5e73\u53f0"): DEALER_REFERRAL, zh(r"\u4e8c\u624b\u8f66\u5546"): USED_PHYSICAL, zh(r"\u4e8c\u624b\u8f66\u5546/\u4e8c\u624b\u8f66\u5e73\u53f0"): USED_PHYSICAL, zh(r"\u5546\u7528\u8f66/\u8f66\u961f\u6e20\u9053"): COMMERCIAL_VEHICLE_CHANNEL, zh(r"\u79df\u8d41/\u8f66\u961f\u516c\u53f8"): CAR_RENTAL, zh(r"\u8fdb\u53e3\u5546/\u6c7d\u8f66\u96c6\u56e2"): CAR_IMPORTER, zh(r"\u591a\u54c1\u724c\u7ecf\u9500\u5546/showroom"): MULTI_BRAND_DEALER, } QUESTION_MARK_RE = re.compile(r"\?{3,}") SPLIT_CLASS_RE = re.compile(r"[\n\r;/??,?]+") def clean(value: Any) -> str: if value is None: return "" return re.sub(r"\s+", " ", str(value).strip()) def contains_any(text: str, keywords: Sequence[str]) -> bool: lower = text.casefold() return any(keyword.casefold() in lower for keyword in keywords) def is_valid_pair(attribute: str, customer_type: str) -> bool: return clean(customer_type) in CUSTOMER_TAXONOMY.get(clean(attribute), []) def type_to_attribute(customer_type: str) -> str: return TYPE_TO_ATTRIBUTE.get(clean(customer_type), "") def _split_class_values(value: str) -> list[str]: return [item for item in (clean(part) for part in SPLIT_CLASS_RE.split(value or "")) if item] def _match_old_type(text: str) -> str: for old, new in OLD_TO_NEW_TYPE.items(): if old.casefold() in text.casefold(): return new return "" def _record_text(record: Mapping[str, Any], sheet_name: str = "", sources: Sequence[str] = ()) -> str: preferred = [ sheet_name, " ".join(str(item) for item in sources), str(record.get("name", "")), str(record.get("attribute", "")), str(record.get("type", "")), str(record.get("business", "")), str(record.get("note", "")), str(record.get("link", "")), str(record.get("website", "")), str(record.get(KEY_NAME, "")), str(record.get(KEY_ATTRIBUTE, "")), str(record.get(KEY_TYPE, "")), str(record.get(KEY_BUSINESS, "")), str(record.get(KEY_NOTE, "")), str(record.get("??", "")), ] return " ".join(item for item in preferred if item).casefold() def _platform_type(text: str) -> str: if contains_any(text, ["chamber", "commerce chamber", zh(r"\u5546\u4f1a")]): return CHAMBER if contains_any(text, ["association", "federation", "f?d?ration", "federation", "syndicat", zh(r"\u534f\u4f1a")]): return CAR_ASSOCIATION if contains_any(text, ["alliance", "union", zh(r"\u8054\u76df")]): return DEALER_ALLIANCE if contains_any(text, ["lead", "referral", "b2b", "event", "exhibition", "salon", zh(r"\u5f15\u8350"), zh(r"\u8d44\u6e90"), zh(r"\u64ae\u5408"), zh(r"\u5c55\u4f1a"), zh(r"\u83b7\u5ba2")]): return DEALER_REFERRAL return CAR_MARKETPLACE def _ecosystem_type(text: str) -> str: if contains_any(text, ["spare", "parts", "piece", "pi?ce", "warehouse", "storage", zh(r"\u5907\u4ef6"), zh(r"\u914d\u4ef6"), zh(r"\u4ed3\u50a8")]): return PARTS_WAREHOUSE if contains_any(text, ["homolog", "certification", "registration", "immatriculation", zh(r"\u8ba4\u8bc1"), zh(r"\u4e0a\u724c")]): return IMPORT_CERTIFICATION if contains_any(text, ["insurance", "assurance", "finance", "financement", "bank", zh(r"\u4fdd\u9669"), zh(r"\u91d1\u878d")]): return FINANCE_INSURANCE if contains_any(text, ["vehicle logistics", "transport automobile", "shipping", zh(r"\u8f66\u8f86\u7269\u6d41")]): return VEHICLE_LOGISTICS return AFTER_SALES_NETWORK def _bulk_type(text: str) -> str: if contains_any(text, ["government", "public agency", "tender", "procurement", zh(r"\u653f\u5e9c"), zh(r"\u673a\u6784"), zh(r"\u91c7\u8d2d")]): return GOV_PROCUREMENT if contains_any(text, ["logistics", "delivery", "courier", "express", zh(r"\u7269\u6d41"), zh(r"\u914d\u9001")]): return LOGISTICS_DELIVERY if contains_any(text, ["corporate fleet", "fleet", "flotte", zh(r"\u4f01\u4e1a\u8f66\u961f"), zh(r"\u8f66\u961f")]): return CORPORATE_FLEET if contains_any(text, ["long-term", "long term", "lld", "leasing longue", zh(r"\u957f\u671f\u79df\u8d41")]): return LONG_TERM_RENTAL return CAR_RENTAL def _used_type(text: str) -> str: if contains_any(text, ["chain", "network", "r?seau", "reseau", "branches", "succursale", "multi-city", zh(r"\u8fde\u9501"), zh(r"\u591a\u95e8\u5e97")]): return USED_CHAIN if contains_any(text, ["import", "importation", zh(r"\u8fdb\u53e3")]): return USED_IMPORTER if contains_any(text, ["new vehicle", "v?hicules neufs", "vehicules neufs", "neuf", zh(r"\u65b0\u8f66")]): return NEW_USED_COMPREHENSIVE return USED_PHYSICAL def _auto_channel_type(text: str) -> str: if contains_any(text, ["commercial vehicle", "utilitaire", "camion", "truck", "van", "bus", "mpv", zh(r"\u5546\u7528\u8f66"), zh(r"\u8d27\u8f66"), zh(r"\u5361\u8f66")]): return COMMERCIAL_VEHICLE_CHANNEL if contains_any(text, ["byd", "jac", "dfsk", "foton", "changan", "chery", "geely", "mg", "haval", "gwm", "leapmotor", zh(r"\u4e2d\u56fd\u54c1\u724c")]): return CHINA_BRAND_DEALER if contains_any(text, ["ev", "electric", "hybrid", "new energy", "mini", "micro", "small car", zh(r"\u65b0\u80fd\u6e90"), zh(r"\u5c0f\u578b"), zh(r"\u5fae\u578b")]): return NEV_SMALL_VEHICLE_CHANNEL if contains_any(text, ["national agent", "exclusive agent", "importateur exclusif", zh(r"\u5168\u56fd\u4ee3\u7406")]): return NATIONAL_AGENT if contains_any(text, ["national distributor", "nationwide", "dealer network", "r?seau", "reseau", "group", "groupe", "succursale", zh(r"\u5168\u56fd\u5206\u9500"), zh(r"\u5168\u56fd"), zh(r"\u96c6\u56e2"), zh(r"\u7f51\u7edc")]): return NATIONAL_DISTRIBUTOR if contains_any(text, ["importateur", "importer", "importation", zh(r"\u8fdb\u53e3")]): return CAR_IMPORTER if contains_any(text, ["distributeur", "distribution", zh(r"\u5206\u9500")]): return REGIONAL_DISTRIBUTOR return MULTI_BRAND_DEALER def classify_from_text(text: str) -> Tuple[str, str]: text = clean(text).casefold() old = _match_old_type(text) if old: return type_to_attribute(old), old # Strong non-dealer classes first, then dealership/channel evidence. if contains_any(text, ["association", "federation", "f?d?ration", "chamber", "marketplace", "annuaire", "directory", "classified", "portal", "b2b", "exhibition", zh(r"\u5e73\u53f0"), zh(r"\u534f\u4f1a"), zh(r"\u5546\u4f1a"), zh(r"\u8054\u76df"), zh(r"\u5f15\u8350"), zh(r"\u8d44\u6e90"), zh(r"\u76ee\u5f55")]): item = _platform_type(text) return PLATFORM_CHANNEL, item if contains_any(text, ["after-sales", "service network", "maintenance", "repair", "garage", "spare", "parts", "homolog", "certification", "registration", "insurance", "assurance", "finance", "vehicle logistics", zh(r"\u552e\u540e"), zh(r"\u7ef4\u4fee"), zh(r"\u5907\u4ef6"), zh(r"\u914d\u4ef6"), zh(r"\u8ba4\u8bc1"), zh(r"\u4e0a\u724c"), zh(r"\u91d1\u878d"), zh(r"\u4fdd\u9669"), zh(r"\u8f66\u8f86\u7269\u6d41")]): item = _ecosystem_type(text) return ECOSYSTEM_SUPPORT, item if contains_any(text, ["rent", "rental", "location", "leasing", "fleet", "flotte", "logistics", "delivery", "procurement", zh(r"\u79df\u8d41"), zh(r"\u79df\u8f66"), zh(r"\u8f66\u961f"), zh(r"\u7269\u6d41"), zh(r"\u914d\u9001"), zh(r"\u91c7\u8d2d")]): item = _bulk_type(text) return BULK_OPERATOR, item if contains_any(text, ["occasion", "used", "second hand", "second-hand", "pre-owned", "reprise", "parc occasion", "occaz", zh(r"\u4e8c\u624b"), zh(r"\u7f6e\u6362")]): item = _used_type(text) return USED_TRANSITION, item if contains_any(text, ["auto", "automobile", "cars", "motors", "concessionnaire", "dealer", "showroom", "vehicle", "v?hicule", "vehicule", "neuf", "import", "distributeur", "distribution", "multimarque", "multi-brand", "brand", "marque", zh(r"\u6c7d\u8f66"), zh(r"\u65b0\u8f66"), zh(r"\u5c55\u5385"), zh(r"\u7ecf\u9500"), zh(r"\u5206\u9500"), zh(r"\u8fdb\u53e3"), zh(r"\u591a\u54c1\u724c")]): item = _auto_channel_type(text) return AUTO_CHANNEL, item return "", "" def normalize_existing_classification(attribute: str, customer_type: str, evidence_text: str = "") -> Tuple[str, str]: attr = clean(attribute) typ = clean(customer_type) if is_valid_pair(attr, typ) and "\n" not in attr and "\n" not in typ: return attr, typ for part in _split_class_values(typ): old = OLD_TO_NEW_TYPE.get(part) if old: return type_to_attribute(old), old if part in VALID_CUSTOMER_TYPES: return type_to_attribute(part), part if attr in VALID_ATTRIBUTES and typ in CUSTOMER_TAXONOMY.get(attr, []): return attr, typ return classify_from_text(" ".join([attribute or "", customer_type or "", evidence_text or ""])) def classify_attribute_type(record: Mapping[str, Any], sheet_name: str = "", sources: Sequence[str] = ()) -> Tuple[str, str]: attr = clean(record.get("attribute") or record.get(KEY_ATTRIBUTE) or "") typ = clean(record.get("type") or record.get(KEY_TYPE) or "") text = _record_text(record, sheet_name, sources) normalized = normalize_existing_classification(attr, typ, text) if normalized != ("", ""): return normalized return classify_from_text(text)