精选联盟商家电话导出实践：授权链路下的增量同步与幂等导出

检查凭证态 → 拉取增量数据（模拟） → 字段清洗与校验 → 幂等落盘 → 导出 CSV

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_phones.json"
OUTPUT_CSV = "output/selected_league_contacts.csv"

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})"
)


@dataclass
class ShopPhone:
    shop_id: str
    shop_name: str
    raw_tel: str
    normalized_phones: List[str] = field(default_factory=list)


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) -> List[str]:
    if not raw or not isinstance(raw, str):
        return []
    candidates = PHONE_RE.findall(raw)
    if not candidates:
        n = normalize(raw)
        if n:
            candidates = [n]
    result, seen = [], set()
    for c in candidates:
        n = normalize(c)
        if n and is_plausible(n) and n not in seen:
            seen.add(n)
            result.append(n)
    return result


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:
    """
    模拟凭证门卫。
    真实场景中：判断 expires_in 是否进入刷新窗口，
    调用 refresh_token 接口；刷新失败则标记需重新授权。
    """
    if not state.get("token_valid", False):
        print("[停止] 凭证态无效，需刷新或重新授权")
        return False
    return True


def fetch_increment(cursor: int):
    """
    本地模拟增量响应，替代真实接口调用。
    真实场景中携带 access_token + shop_id 请求官方接口。
    """
    demo_pages = {
        0: [
            {"shop_id": "S1001", "shop_name": "示例店铺A", "tel": "13800138000"},
            {"shop_id": "S1002", "shop_name": "示例店铺B", "tel": ""},
            {"shop_id": "S1003", "shop_name": "示例店铺C",
             "tel": "400-123-4567;021-87654321"},
        ],
        1: [
            {"shop_id": "S1001", "shop_name": "示例店铺A", "tel": "13800138000"},
            {"shop_id": "S1004", "shop_name": "示例店铺D", "tel": "0571-88886666"},
        ],
    }
    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 idempotent_append(records: List[ShopPhone], seen: Set[str]) -> List[ShopPhone]:
    new_records = []
    for r in records:
        for phone in r.normalized_phones:
            key = f"{r.shop_id}:{phone}"
            if key not in seen:
                seen.add(key)
                new_records.append(ShopPhone(
                    shop_id=r.shop_id,
                    shop_name=r.shop_name,
                    raw_tel=r.raw_tel,
                    normalized_phones=[phone],
                ))
    return new_records


def export_csv(records: List[ShopPhone], 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(["shop_id", "店铺名称", "原始tel字段", "规范化电话", "幂等键"])
        for r in records:
            for phone in r.normalized_phones:
                writer.writerow([r.shop_id, r.shop_name, r.raw_tel, phone,
                                 f"{r.shop_id}:{phone}"])
    print(f"[导出] {len(records)} 条新增记录 → {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)

    parsed = []
    for item in raw_batch:
        phones = clean_tel(item.get("tel", ""))
        parsed.append(ShopPhone(
            shop_id=str(item.get("shop_id", "")),
            shop_name=item.get("shop_name", ""),
            raw_tel=item.get("tel", ""),
            normalized_phones=phones,
        ))

    new_records = idempotent_append(parsed, seen)

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

    export_csv(new_records, OUTPUT_CSV)

    no_phone = sum(1 for r in parsed if not r.normalized_phones)
    print(f"[统计] 本批 {len(parsed)} 个店铺，{no_phone} 个无有效公开电话")


if __name__ == "__main__":
    main()