AI训练数据采集代理IP

-x 指定代理，-I 只取响应头，用来确认连通和认证都没问题。socks5h:// 走 SOCKS5 并由代理解析域名；IP 白名单认证时去掉 USER:PASS@。

# Check connectivity and auth, headers only
curl -I -x http://USER:PASS@GATEWAY:PORT https://example.com/

# SOCKS5, hostnames resolved by the proxy
curl -I -x socks5h://USER:PASS@GATEWAY:PORT https://example.com/

AsyncClient 用 proxy 参数走代理（HTTPX 0.26+）；Semaphore 控制同时在途的请求数，按目标站点的承受能力调。User-Agent 写明爬虫名称和联系方式，方便站长识别。

import asyncio

import httpx

PROXY = "http://USER:PASS@GATEWAY:PORT"
UA = "ExampleCorpusBot/1.0 (+https://example.com/bot)"
URLS = ["https://example.com/a", "https://example.com/b"]


async def fetch(client, sem, url):
    async with sem:
        r = await client.get(url, timeout=30)
        return url, r.status_code, len(r.text)


async def main():
    sem = asyncio.Semaphore(8)
    client = httpx.AsyncClient(proxy=PROXY, headers={"User-Agent": UA})
    async with client:
        jobs = [fetch(client, sem, u) for u in URLS]
        for row in await asyncio.gather(*jobs):
            print(row)


asyncio.run(main())

跟随站内链接抓正文，每个请求都在 meta 里带上 proxy。ROBOTSTXT_OBEY 保持开启，AutoThrottle 自动调速，DEPTH_LIMIT 控制抓取深度；async def start() 需要 Scrapy 2.13+。

import scrapy

PROXY = "http://USER:PASS@GATEWAY:PORT"


class CorpusSpider(scrapy.Spider):
    name = "corpus"
    allowed_domains = ["example.com"]
    custom_settings = {
        "ROBOTSTXT_OBEY": True,
        "AUTOTHROTTLE_ENABLED": True,
        "DEPTH_LIMIT": 3,
    }

    async def start(self):
        url = "https://example.com/"
        yield scrapy.Request(url, meta={"proxy": PROXY})

    def parse(self, response):
        text = " ".join(response.css("p::text").getall())
        yield {"url": response.url, "text": text}
        for href in response.css("a::attr(href)").getall():
            yield response.follow(href, meta={"proxy": PROXY})