抖音抖店卖家采集器 抖音精选联盟商家电话提取工具教程详解

pip install beautifulsoup4

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

from bs4 import BeautifulSoup

INPUT_DIR = "input"
OUTPUT_CSV = "output/contacts.csv"

# 宽松召回：手机、400、常见座机
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})"
)

CONTEXT_WORDS = ["联系", "电话", "客服", "咨询", "contact", "tel", "phone"]


@dataclass
class Contact:
    phone: str
    source: str
    channel: str          # 来源类型：text / tel_link / json_ld / inline_json
    context: str = ""
    score: int = 0


def normalize(raw: str) -> str:
    s = re.sub(r"[\s\-()]", "", raw)
    for prefix in ("+86", "0086"):
        if s.startswith(prefix):
            s = s[len(prefix):]
            break
    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 extract_from_text(text: str) -> Set[str]:
    return {m.group() for m in PHONE_CANDIDATE.finditer(text)}


def extract_from_tel_links(soup: BeautifulSoup) -> Set[str]:
    result = set()
    for a in soup.find_all("a", href=True):
        href = a["href"].strip()
        if href.lower().startswith("tel:"):
            result.add(href[4:].strip())
    return result


def extract_from_json_ld(soup: BeautifulSoup) -> Set[str]:
    result = set()
    for script in soup.find_all("script", type="application/ld+json"):
        try:
            data = json.loads(script.string or "")
        except (json.JSONDecodeError, TypeError):
            continue
        stack = [data]
        while stack:
            node = stack.pop()
            if isinstance(node, dict):
                for key, value in node.items():
                    if key.lower() in ("telephone", "phone", "tel") and isinstance(value, str):
                        result.add(value)
                    if isinstance(value, (dict, list)):
                        stack.append(value)
            elif isinstance(node, list):
                stack.extend(node)
    return result


def extract_from_inline_json(soup: BeautifulSoup) -> Set[str]:
    """从 script 标签中扫描内联 JSON 的 tel/phone 字段"""
    result = set()
    for script in soup.find_all("script"):
        if script.get("type") in ("application/ld+json",):
            continue
        raw = script.string or ""
        if not raw.strip().startswith(("{", "[")):
            continue
        try:
            data = json.loads(raw)
        except json.JSONDecodeError:
            continue
        stack = [data]
        while stack:
            node = stack.pop()
            if isinstance(node, dict):
                for key, value in node.items():
                    if key.lower() in ("tel", "phone", "telephone") and isinstance(value, str):
                        if PHONE_CANDIDATE.search(value):
                            result.add(value)
                    if isinstance(value, (dict, list)):
                        stack.append(value)
            elif isinstance(node, list):
                stack.extend(node)
    return result


def build_context(text: str, phone: str, window: int = 40) -> str:
    idx = text.find(phone)
    if idx == -1:
        return ""
    return text[max(0, idx - window): idx + len(phone) + window]


def score_contact(phone: str, context: str) -> int:
    score = 0
    ctx_lower = context.lower()
    score += sum(1 for w in CONTEXT_WORDS if w in ctx_lower)
    if re.fullmatch(r"1[3-9]\d{9}", phone):
        score += 2
    if phone.startswith("400"):
        score += 1
    return score


def process_file(path: str, source_name: str) -> List[Contact]:
    with open(path, "r", encoding="utf-8", errors="ignore") as f:
        html = f.read()

    soup = BeautifulSoup(html, "html.parser")
    text = soup.get_text(separator=" ", strip=True)

    raw_map = {}

    def add(phones: Set[str], channel: str):
        for p in phones:
            n = normalize(p)
            if n and is_plausible(n):
                raw_map.setdefault(n, channel)

    add(extract_from_text(text), "text")
    add(extract_from_tel_links(soup), "tel_link")
    add(extract_from_json_ld(soup), "json_ld")
    add(extract_from_inline_json(soup), "inline_json")

    contacts = []
    for phone, channel in raw_map.items():
        context = build_context(text, phone)
        contacts.append(Contact(
            phone=phone,
            source=source_name,
            channel=channel,
            context=context,
            score=score_contact(phone, context),
        ))

    contacts.sort(key=lambda c: (-c.score, c.phone))
    return contacts


def main():
    if not os.path.isdir(INPUT_DIR):
        print(f"[提示] 请创建 {INPUT_DIR}/ 并放入本地保存的公开页面 HTML 文件")
        return

    all_contacts = []
    for root, _, files in os.walk(INPUT_DIR):
        for name in files:
            if not name.lower().endswith((".html", ".htm")):
                continue
            path = os.path.join(root, name)
            try:
                all_contacts.extend(process_file(path, source_name=name))
            except Exception as e:
                print(f"[跳过] {name}：{e}")

    # 按“电话 + 来源”去重，保留首次出现
    seen = set()
    unique = []
    for c in all_contacts:
        key = (c.phone, c.source)
        if key not in seen:
            seen.add(key)
            unique.append(c)

    os.makedirs(os.path.dirname(OUTPUT_CSV), exist_ok=True)
    with open(OUTPUT_CSV, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.writer(f)
        writer.writerow(["电话", "来源文件", "来源通道", "上下文得分", "上下文片段"])
        for c in unique:
            writer.writerow([c.phone, c.source, c.channel, c.score, c.context])

    print(f"[完成] 有效记录 {len(unique)} 条 → {OUTPUT_CSV}")


if __name__ == "__main__":
    main()