微信小店商家采集器 批量导出视频号卖家电话的方案

GET https://api.weixin.qq.com/channels/ec/basics/info/get?access_token=ACCESS_TOKEN

凭证检查 → 拉取增量（模拟） → 字段清洗与边界处理 → 幂等落盘 → 导出

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

STATE_FILE = "sync_state.json"
SEEN_FILE = "seen_shops.json"
OUTPUT_CSV = "output/wechat_shop_public_info.csv"


@dataclass
class ShopInfo:
    username: str
    nickname: str
    subject_type: str
    status: str
    raw_fields: dict = field(default_factory=dict)


def clean_text(value) -> str:
    if not isinstance(value, str):
        return ""
    return re.sub(r"\s+", " ", value).strip()


def load_state() -> dict:
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    return {"cursor": 0, "token_valid": True}


def save_state(state: dict):
    with open(STATE_FILE, "w", encoding="utf-8") as f:
        json.dump(state, f, ensure_ascii=False, indent=2)


def ensure_token(state: dict) -> bool:
    if not state.get("token_valid", False):
        print("[停止] 凭证态无效，需刷新或重新授权")
        return False
    return True


def fetch_increment(cursor: int):
    """
    本地模拟增量响应，替代真实接口调用。
    真实场景中携带 access_token / authorizer_access_token 请求官方接口。
    返回字段严格对应 /channels/ec/basics/info/get 的 info 对象。
    """
    demo_pages = {
        0: [
            {"username": "shop_gh_001", "nickname": "示例店铺A",
             "subject_type": "企业", "status": "opening"},
            {"username": "shop_gh_002", "nickname": " 示例店铺B ",
             "subject_type": "个体工商户", "status": "open_finished"},
            {"username": "shop_gh_003", "nickname": "",
             "subject_type": "企业", "status": "opening"},
        ],
        1: [
            {"username": "shop_gh_001", "nickname": "示例店铺A",
             "subject_type": "企业", "status": "opening"},
            {"username": "shop_gh_004", "nickname": "示例店铺D",
             "subject_type": "企业", "status": "closing"},
        ],
    }
    data = demo_pages.get(cursor, [])
    next_cursor = cursor + 1 if cursor + 1 in demo_pages else cursor
    return data, next_cursor


def load_seen() -> Set[str]:
    if os.path.exists(SEEN_FILE):
        with open(SEEN_FILE, "r", encoding="utf-8") as f:
            return set(json.load(f))
    return set()


def save_seen(seen: Set[str]):
    with open(SEEN_FILE, "w", encoding="utf-8") as f:
        json.dump(sorted(seen), f, ensure_ascii=False, indent=2)


def parse_shop(raw: dict) -> ShopInfo:
    return ShopInfo(
        username=clean_text(raw.get("username", "")),
        nickname=clean_text(raw.get("nickname", "")),
        subject_type=clean_text(raw.get("subject_type", "")),
        status=clean_text(raw.get("status", "")),
        raw_fields=raw,
    )


def idempotent_filter(shops: List[ShopInfo], seen: Set[str]) -> List[ShopInfo]:
    """以 username（店铺原始ID）为幂等键"""
    new_shops = []
    for shop in shops:
        if not shop.username:
            continue
        if shop.username not in seen:
            seen.add(shop.username)
            new_shops.append(shop)
    return new_shops


def export_csv(shops: List[ShopInfo], 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", "店铺名称", "店铺类型", "店铺状态"])
        for s in shops:
            writer.writerow([s.username, s.nickname, s.subject_type, s.status])
    print(f"[导出] {len(shops)} 条新增店铺信息 → {path}")


def main():
    state = load_state()
    if not ensure_token(state):
        return

    seen = load_seen()
    cursor = state.get("cursor", 0)

    raw_batch, next_cursor = fetch_increment(cursor)
    shops = [parse_shop(item) for item in raw_batch]

    missing_nickname = sum(1 for s in shops if not s.nickname)
    new_shops = idempotent_filter(shops, seen)

    save_seen(seen)
    state["cursor"] = next_cursor
    save_state(state)

    export_csv(new_shops, OUTPUT_CSV)

    print(f"[统计] 本批 {len(shops)} 条，{missing_nickname} 条缺少店铺名称，"
          f"{len(new_shops)} 条为新增")


if __name__ == "__main__":
    main()