视频号商家电话采集器 微信小店电商导出工具使用说明

pip install beautifulsoup4

python export_video_shop_phones.py

import csv
import os
import re
from bs4 import BeautifulSoup

# 电话匹配规则
PHONE_PATTERNS = [
    ("手机", re.compile(r"(?<!\d)(?:\+?86)?1[3-9]\d{9}(?!\d)")),
    ("400", re.compile(r"(?<!\d)400[-\s]?\d{3}[-\s]?\d{4}(?!\d)")),
    ("座机", re.compile(r"(?<!\d)0\d{2,3}[-\s]?\d{7,8}(?!\d)")),
]

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


def clean_phone(phone: str) -> str:
    """清洗电话字符串"""
    phone = phone.strip()
    phone = re.sub(r"[\s\-()]", "", phone)
    if phone.startswith("+86"):
        phone = phone[3:]
    if phone.startswith("0086"):
        phone = phone[4:]
    return phone


def detect_type(phone: str) -> str:
    """判断电话类型"""
    if re.fullmatch(r"1[3-9]\d{9}", phone):
        return "手机"
    if re.fullmatch(r"400\d{7}", phone):
        return "400"
    if re.fullmatch(r"0\d{9,11}", phone):
        return "座机"
    return "其他"


def extract_phones_from_html(html: str):
    """从 HTML/文本中提取公开电话"""
    soup = BeautifulSoup(html, "html.parser")
    text = soup.get_text(separator=" ", strip=True)

    phones = []

    # 从正文文本提取
    for _, pattern in PHONE_PATTERNS:
        phones.extend(pattern.findall(text))

    # 从 tel: 链接提取
    for a in soup.find_all("a", href=True):
        href = a["href"].strip()
        if href.lower().startswith("tel:"):
            phones.append(href[4:].strip())

    result = set()
    for phone in phones:
        cleaned = clean_phone(phone)
        if cleaned:
            result.add(cleaned)

    return sorted(result)


def read_local_file(path: str) -> str:
    """读取本地公开页面或文本"""
    with open(path, "r", encoding="utf-8", errors="ignore") as f:
        return f.read()


def process_local(input_dir: str):
    """批量处理本地公开页面"""
    rows = []

    if not os.path.isdir(input_dir):
        print(f"[提示] 本地目录不存在：{input_dir}")
        print("请创建 input 文件夹，并放入公开页面 HTML 或 TXT 文件。")
        return rows

    for root, _, files in os.walk(input_dir):
        for name in files:
            if not name.lower().endswith((".html", ".htm", ".txt")):
                continue

            path = os.path.join(root, name)
            try:
                content = read_local_file(path)
            except Exception as e:
                print(f"[读取失败] {path}：{e}")
                continue

            for phone in extract_phones_from_html(content):
                rows.append([path, phone, detect_type(phone)])

    return rows


def deduplicate(rows):
    """按来源文件 + 电话去重"""
    seen = set()
    unique_rows = []

    for source, phone, phone_type in rows:
        key = (source, phone)
        if key not in seen:
            seen.add(key)
            unique_rows.append([source, phone, phone_type])

    return unique_rows


def export_csv(rows, output_csv: str):
    """导出 CSV"""
    out_dir = os.path.dirname(output_csv)
    if out_dir:
        os.makedirs(out_dir, exist_ok=True)

    with open(output_csv, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.writer(f)
        writer.writerow(["来源文件", "电话", "类型"])
        writer.writerows(rows)

    print(f"[完成] 共导出 {len(rows)} 条记录到 {output_csv}")


def main():
    rows = process_local(INPUT_DIR)
    rows = deduplicate(rows)
    export_csv(rows, OUTPUT_CSV)


if __name__ == "__main__":
    main()