pip install crawl4ai playwright
playwright install

import random  
import os  

class ProxyManager:  
    def __init__(self, proxy_type="residential", protocol="http", rotate=True):  
        """  
        初始化代理管理器。  
        :param proxy_type: 类型 ('residential', 'mobile', 'datacenter')  
        :param protocol: 协议 ('http' 或 'socks5') — socks5 仅支持无身份验证！  
        :param rotate: 开启/关闭轮换  
        """  
        self.proxy_type = proxy_type.lower()  
        self.protocol = protocol.lower()  
        self.rotate = rotate  
        self.proxies = self._load_proxies()  # 从文件加载  
        if self.protocol == "socks5" and self.proxies and '@' in self.proxies[0]:  # 检查验证信息  
            raise ValueError("Playwright 中的 SOCKS5 不支持身份验证！请从文件中删除 login:password。")  
        self.current_proxy = None  

    def _load_proxies(self):  
        """从文件加载代理。"""  
        file_map = {  
            "residential": "proxies_residential.txt",  
            "mobile": "proxies_mobile.txt",  
            "datacenter": "proxies_datacenter.txt"  
        }  
        file_name = file_map.get(self.proxy_type)  
        if not file_name:  
            raise ValueError(f"未知类型: {self.proxy_type}。")  

        file_path = os.path.join(os.path.dirname(__file__), file_name)  
        if not os.path.exists(file_path):  
            raise FileNotFoundError(f"文件 {file_name} 未找到。")  

        with open(file_path, 'r') as f:  
            lines = f.read().strip().splitlines()  
            # 添加协议: protocol://line (line = user:pass@host:port 或 host:port)  
            proxies = [f"{self.protocol://{line.strip()" for line in lines if line.strip()]  

        if not proxies:  
            raise ValueError(f"文件 {file_name 为空。")  

        return proxies  

    def get_proxy(self):  
        """返回代理。"""  
        if not self.proxies:  
            raise ValueError("没有代理！")  
        if self.rotate:  
            self.current_proxy = random.choice(self.proxies)  
        else:  
            self.current_proxy = self.proxies[0]  
        return self.current_proxy  

    def get_proxy_config(self):  
        """返回 Crawl4AI 专用的 ProxyConfig。"""  
        if not self.current_proxy:  
            self.get_proxy()  
        # 分离: protocol://user:pass@host:port 或 protocol://host:port  
        if '@' in self.current_proxy:  
            parts = self.current_proxy.split('@')  
            auth = parts[0].split('//')[1]  # user:pass  
            host_port = parts[1]  # host:port  
            user, password = auth.split(':')  
        else:  
            host_port = self.current_proxy.split('//')[1]  # host:port 无验证模式  
            user, password = None, None  
        from crawl4ai.async_configs import ProxyConfig  
        return ProxyConfig(  
            server=f"{self.protocol://{host_port",  
            username=user,  
            password=password  
        )  

    def add_proxy(self, proxy_url):  
        self.proxies.append(proxy_url)  

import asyncio  
from crawl4ai import AsyncWebCrawler  
from crawl4ai.async_configs import BrowserConfig  
from proxy_module import ProxyManager  

# 切换类型 (residential, mobile, datacenter)  
PROXY_TYPE = "residential"  
# 切换协议 (http 或 socks5 — socks5 仅限无验证模式！)  
PROXY_PROTOCOL = "http"  # 建议用于验证；对于 SOCKS5 请删除文件中的 login:password  

# 初始化管理器  
proxy_manager = ProxyManager(proxy_type=PROXY_TYPE, protocol=PROXY_PROTOCOL, rotate=True)  

async def main():  
    try:  
        proxy_config = proxy_manager.get_proxy_config()  
        browser_config = BrowserConfig(proxy_config=proxy_config)  
        async with AsyncWebCrawler(verbose=True, config=browser_config) as crawler:  
            current_proxy = proxy_manager.get_proxy()  
            print(f"🚀 正在通过 {PROXY_PROTOCOL.upper() 代理启动...")  
            print(f"代理地址: {current_proxy}")  

            result = await crawler.arun(  
                url="https://ipinfo.io/json",  
                bypass_cache=True  
            )  

            if result.success:  
                print("\n✅ 成功！响应内容:")  
                print(result.markdown)  
            else:  
                print(f"\n❌ 错误: {result.error_message}")  
    except Exception as e:  
        print(f"❌ 错误: {str(e)}")  

if __name__ == "__main__":  
    asyncio.run(main())  

python scraper.py

for url in urls_list:  
    proxy_config = proxy_manager.get_proxy_config()  # 用于轮换的新配置  
    browser_config = BrowserConfig(proxy_config=proxy_config)  
    # 如有必要，请使用新配置重新初始化爬虫  
    result = await crawler.arun(url=url, bypass_cache=True)  

import time  
time.sleep(random.uniform(1, 5))  # 请求前随机暂停 1-5 秒