中国制造英文站采集器 跨境电商电话导出教程

pip install requests beautifulsoup4

python export_mic_contacts.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": "en-US,en;q=0.9,zh-CN;q=0.8",
}

EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")

# 国际电话简单匹配：支持 +86、0086、1-xxx、44 20 xxxx 等
PHONE_RE = re.compile(
    r"(?:(?:\+|00)\d{1,3}[\s\-]?)?(?:\(?\d{2,4}\)?[\s\-]?)?\d{3,4}[\s\-]?\d{4}"
)


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 "sign in" in low:
            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 = phone.strip()
    phone = re.sub(r"[\s\-()]", "", phone)
    if phone.startswith("00"):
        phone = "+" + phone[2:]
    return phone


def extract_contacts(html: str):
    """从公开 HTML 中提取公司名、邮箱、电话"""
    soup = BeautifulSoup(html, "html.parser")
    text = soup.get_text(separator=" ", strip=True)

    # 公司名 / 页面标题
    company = ""
    if soup.title:
        company = soup.title.get_text(strip=True)
    h1 = soup.find("h1")
    if h1:
        company = h1.get_text(" ", strip=True)

    emails = set(EMAIL_RE.findall(text))
    phones = set(PHONE_RE.findall(text))

    # 提取 mailto: 和 tel: 链接
    for a in soup.find_all("a", href=True):
        href = a["href"].strip()
        if href.lower().startswith("mailto:"):
            emails.add(href[7:].split("?")[0])
        if href.lower().startswith("tel:"):
            phones.add(href[4:])

    emails = {e.strip().lower() for e in emails if e.strip()}
    phones = {clean_phone(p) for p in phones if clean_phone(p)}

    return company, sorted(emails), sorted(phones)


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

            company, emails, phones = extract_contacts(html)

            if not emails and not phones:
                rows.append([path, company, "", ""])
            else:
                max_len = max(len(emails), len(phones), 1)
                for i in range(max_len):
                    rows.append([
                        path,
                        company,
                        emails[i] if i < len(emails) else "",
                        phones[i] if i < len(phones) else "",
                    ])

    return rows


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

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

        company, emails, phones = extract_contacts(html)

        if not emails and not phones:
            rows.append([url, company, "", ""])
        else:
            max_len = max(len(emails), len(phones), 1)
            for i in range(max_len):
                rows.append([
                    url,
                    company,
                    emails[i] if i < len(emails) else "",
                    phones[i] if i < len(phones) else "",
                ])

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

    return rows


def deduplicate(rows):
    """按来源 + 公司 + 邮箱 + 电话去重"""
    seen = set()
    unique_rows = []

    for row in rows:
        key = tuple(row)
        if key not in seen:
            seen.add(key)
            unique_rows.append(row)

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

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

    all_rows = local_rows + url_rows
    unique_rows = deduplicate(all_rows)

    export_csv(unique_rows, "output/made_in_china_public_contacts.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)

    emails = set(re.findall(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}", text))
    phones = set(re.findall(
        r"(?:(?:\+|00)\d{1,3}[\s\-]?)?(?:\(?\d{2,4}\)?[\s\-]?)?\d{3,4}[\s\-]?\d{4}",
        text,
    ))

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

    return {
        "emails": sorted(emails),
        "phones": sorted(cleaned),
    }

print(extract_from_local("public_page.html"))