天猫电商采集器 天猫店商家电话导出工具使用说明

import csv
import os
import re
from dataclasses import dataclass, field
from typing import List

# ============ 字段形态枚举 ============

FORM_FULL = "full"          # 完整可辨识号码
FORM_MASKED = "masked"      # 明显脱敏/截断
FORM_EMPTY = "empty"        # 字段缺失或空字符串
FORM_MULTI = "multi"        # 包含多个号码
FORM_AMBIGUOUS = "ambiguous"  # 无法归类为以上任何形态

PHONE_RE = re.compile(
    r"(?:(?:\+|00)86[\s-]?)?"
    r"(?:1[3-9]\d[\s-]?\d{4}[\s-]?\d{4}"
    r"|400[\s-]?\d{3}[\s-]?\d{4}"
    r"|0\d{2,3}[\s-]?\d{7,8})"
)

SPLIT_RE = re.compile(r"[;,，；/|]")


@dataclass
class TelField:
    raw: str
    form: str = FORM_EMPTY
    numbers: List[str] = field(default_factory=list)
    note: str = ""


def normalize(raw: str) -> str:
    s = re.sub(r"[\s\-()]", "", raw)
    for p in ("+86", "0086"):
        if s.startswith(p):
            return s[len(p):]
    return s


def is_full(phone: str) -> bool:
    if re.fullmatch(r"1[3-9]\d{9}", phone):
        return phone[1] not in "012"
    if re.fullmatch(r"400\d{7}", phone):
        return True
    if re.fullmatch(r"0\d{9,11}", phone):
        return True
    return False


def parse_tel(raw) -> TelField:
    if raw is None or not isinstance(raw, str) or not raw.strip():
        return TelField(raw=str(raw) if raw is not None else "", form=FORM_EMPTY,
                        note="字段缺失或为空")

    text = raw.strip()

    # 先尝试按分隔符拆分，判断是否多值
    parts = [p.strip() for p in SPLIT_RE.split(text) if p.strip()]
    if len(parts) > 1:
        numbers = []
        for p in parts:
            n = normalize(p)
            if n and is_full(n):
                numbers.append(n)
        if numbers:
            return TelField(raw=text, form=FORM_MULTI, numbers=numbers,
                            note=f"拆分出 {len(numbers)} 个完整号码")

    # 单值：尝试完整号码匹配
    candidates = PHONE_RE.findall(text)
    full_numbers = []
    for c in candidates:
        n = normalize(c)
        if n and is_full(n):
            full_numbers.append(n)

    if full_numbers:
        return TelField(raw=text, form=FORM_FULL, numbers=full_numbers,
                        note="完整号码")

    # 没有完整号码，检查是否为疑似脱敏/截断
    digits_only = re.sub(r"\D", "", text)
    if 5 <= len(digits_only) <= 11 and not is_full(normalize(text)):
        return TelField(raw=text, form=FORM_MASKED, numbers=[],
                        note=f"疑似脱敏或截断（原始数字长度 {len(digits_only)}）")

    return TelField(raw=text, form=FORM_AMBIGUOUS, numbers=[],
                    note="无法归类，需人工复核")

def process_single_shop(data: dict, output_csv: str):
    """
    data: taobao.seller.info 风格的本地模拟响应。
    严格限定单店铺，不聚合。
    """
    info = data.get("seller_info") or data.get("shop") or {}
    if not info:
        print("[跳过] 响应中无 seller_info 节点")
        return

    shop_id = info.get("shop_id", "")
    shop_name = info.get("shop_name", "")

    tel = parse_tel(info.get("tel"))

    rows = [{
        "shop_id": shop_id,
        "shop_name": shop_name,
        "tel_raw": tel.raw,
        "tel_form": tel.form,
        "tel_numbers": ";".join(tel.numbers),
        "tel_note": tel.note,
    }]

    os.makedirs(os.path.dirname(output_csv) or ".", exist_ok=True)
    with open(output_csv, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=rows[0].keys())
        writer.writeheader()
        writer.writerows(rows)

    print(f"[完成] shop_id={shop_id}，tel形态={tel.form}，完整号码={len(tel.numbers)} 个")


if __name__ == "__main__":
    # 本地模拟响应，覆盖多种 tel 形态
    demo_responses = [
        {"seller_info": {"shop_id": "S001", "shop_name": "示例旗舰店",
                         "tel": "1891226351"}},          # 脱敏/截断
        {"seller_info": {"shop_id": "S002", "shop_name": "示例专营店",
                         "tel": ""}},                     # 空
        {"seller_info": {"shop_id": "S003", "shop_name": "示例专卖店",
                         "tel": "400-123-4567;0571-88886666"}},  # 多值
        {"seller_info": {"shop_id": "S004", "shop_name": "示例企业店",
                         "tel": "13800138000"}},          # 完整
    ]

    # 单店铺独立处理，不合并
    for i, resp in enumerate(demo_responses):
        process_single_shop(resp, f"output/shop_{i}.csv")