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

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

# ============ 数据模型 ============

@dataclass
class ShopContact:
    nick: str
    title: str
    raw_tel: str
    phones: List[str] = field(default_factory=list)
    sid: str = ""
    status: str = ""

# ============ 清洗规则 ============

# 宽松召回：从 tel 字段中拆出可能的号码
PHONE_CANDIDATE = 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})"
)

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_plausible(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 clean_tel(raw: str) -> List[str]:
    """从 tel 字段中提取并清洗号码"""
    if not raw or not isinstance(raw, str):
        return []
    candidates = PHONE_CANDIDATE.findall(raw)
    # 如果正则没命中但字段本身像单个号码，也纳入
    if not candidates:
        stripped = normalize(raw)
        if stripped:
            candidates = [stripped]
    result = []
    for c in candidates:
        n = normalize(c)
        if n and is_plausible(n):
            result.append(n)
    # 去重保序
    seen = set()
    return [p for p in result if not (p in seen or seen.add(p))]

# ============ 响应解析 ============

def parse_shop_response(data: dict) -> Optional[ShopContact]:
    """
    解析 taobao.shop.get 风格的 JSON 响应。
    兼容 shop 对象嵌套在 shop_get_response 下的常见结构。
    """
    shop = None
    if isinstance(data, dict):
        shop = data.get("shop")
        if shop is None:
            resp = data.get("shop_get_response") or {}
            shop = resp.get("shop") if isinstance(resp, dict) else None
    if not isinstance(shop, dict):
        return None

    nick = shop.get("nick", "")
    title = shop.get("title", "")
    raw_tel = shop.get("tel", "")
    sid = str(shop.get("sid", ""))

    phones = clean_tel(raw_tel)

    return ShopContact(
        nick=nick,
        title=title,
        raw_tel=raw_tel,
        phones=phones,
        sid=sid,
        status="ok" if phones else "no_phone",
    )

# ============ 批量处理与导出 ============

def process_responses(responses: List[dict]) -> List[ShopContact]:
    results = []
    for data in responses:
        contact = parse_shop_response(data)
        if contact is not None:
            results.append(contact)
    return results

def export_csv(contacts: List[ShopContact], path: str):
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.writer(f)
        writer.writerow(["卖家昵称", "店铺标题", "店铺ID", "原始tel字段", "清洗后电话", "状态"])
        for c in contacts:
            writer.writerow([
                c.nick, c.title, c.sid,
                c.raw_tel,
                ";".join(c.phones),
                c.status,
            ])
    total = len(contacts)
    with_phone = sum(1 for c in contacts if c.phones)
    print(f"[完成] 共 {total} 个店铺，{with_phone} 个含有效电话 → {path}")

# ============ 演示入口 ============

if __name__ == "__main__":
    # 本地模拟响应，仅用于演示解析逻辑
    demo = [
        {
            "shop_get_response": {
                "shop": {
                    "sid": 44750468,
                    "nick": "demo_shop_a",
                    "title": "示例店铺A",
                    "tel": "0571-88886666",
                }
            }
        },
        {
            "shop": {
                "sid": 44750469,
                "nick": "demo_shop_b",
                "title": "示例店铺B",
                "tel": "",
            }
        },
        {
            "shop": {
                "sid": 44750470,
                "nick": "demo_shop_c",
                "title": "示例店铺C",
                "tel": "13800138000;021-12345678",
            }
        },
    ]

    contacts = process_responses(demo)
    export_csv(contacts, "output/taobao_shop_contacts.csv")

[完成] 共 3 个店铺，2 个含有效电话 → output/taobao_shop_contacts.csv