小红书买手合作商家电话自动采集软件使用分享

pip install requests beautifulsoup4

python export_xhs_phones.py

import csv
import os
import re
import time
from urllib.parse import urlparse
from urllib.robotparser import RobotFileParser

import requests
from bs4 import BeautifulSoup

HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/124.0.0.0 Safari/537.36"
    ),
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}

# 匹配手机号、400 电话、常见座机号
PHONE_RE = re.compile(
    r"(?:(?:\+?86)?1[3-9]\d{9}|400[- ]?\d{3}[- ]?\d{4}|0\d{2,3}[- ]?\d{7,8})"
)


def can_fetch(url: str) -> bool:
    """检查 robots.txt 是否允许抓取"""
    parsed = urlparse(url)
    robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"
    rp = RobotFileParser()
    rp.set_url(robots_url)
    try:
        rp.read()
    except Exception as e:
        print(f"[robots] 读取失败：{e}")
        return False

    allowed = rp.can_fetch(HEADERS["User-Agent"], url)
    if not allowed:
        print(f"[robots] 不允许抓取：{url}")
    return allowed


def fetch_public_html(url: str):
    """获取公开页面 HTML，单线程、限速、超时"""
    if not can_fetch(url):
        return None

    try:
        resp = requests.get(url, headers=HEADERS, timeout=15)
        if resp.status_code != 200:
            print(f"[HTTP] 状态码 {resp.status_code}，URL：{url}")
            return None

        text = resp.text
        low = text.lower()

        # 遇到登录、验证码等页面立即停止，不尝试绕过
        if "captcha" in low or "验证码" in text or "登录" in text:
            print(f"[停止] 页面可能需要登录或验证，不继续处理：{url}")
            return None

        return text
    except Exception as e:
        print(f"[请求异常] {e}，URL：{url}")
        return None


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


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

    phones = PHONE_RE.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())

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

    return sorted(cleaned)


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}")
        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:
                html = read_local_file(path)
            except Exception as e:
                print(f"[读取失败] {path}：{e}")
                continue

            for phone in extract_phones(html):
                rows.append([path, phone])

    return rows


def process_urls(urls):
    """处理公开且 robots 允许的 URL，谨慎使用"""
    rows = []

    for url in urls:
        print(f"[处理] {url}")
        html = fetch_public_html(url)
        if not html:
            continue

        for phone in extract_phones(html):
            rows.append([url, phone])

        # 单线程限速，避免对平台造成压力
        time.sleep(8)

    return 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():
    # 推荐：手动保存公开页面到 input 目录
    local_rows = process_local("input")

    # 可选：公开且 robots.txt 允许的 URL。
    # 小红书多数页面可能禁止直接抓取，请谨慎使用，不要绕过登录和验证码。
    urls = [
        # "https://www.xiaohongshu.com/...",
    ]
    url_rows = process_urls(urls)

    all_rows = local_rows + url_rows

    # 去重
    seen = set()
    unique_rows = []
    for source, phone in all_rows:
        key = (source, phone)
        if key not in seen:
            seen.add(key)
            unique_rows.append([source, phone])

    export_csv(unique_rows, "output/xiaohongshu_public_phones.csv")


if __name__ == "__main__":
    main()

from bs4 import BeautifulSoup
import re

def extract_from_local(file_path):
    with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
        html = f.read()

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

    phones = re.findall(
        r"(?:(?:\+?86)?1[3-9]\d{9}|400[- ]?\d{3}[- ]?\d{4}|0\d{2,3}[- ]?\d{7,8})",
        text,
    )

    cleaned = set()
    for p in phones:
        p = re.sub(r"[\s\-()]", "", p)
        if p.startswith("+86"):
            p = p[3:]
        if p.startswith("0086"):
            p = p[4:]
        cleaned.add(p)

    return sorted(cleaned)

print(extract_from_local("public_page.html"))