wuling-overseas-dealer-expansion-skill/scripts/common/customer_taxonomy.py

# -*- 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)