构建稳定爬虫：为番茄小说爬虫添加IP代理与请求头伪装

原创

小白学大数据

发布于 2025-11-20 16:53:39

540

一、引言：为何我们的爬虫会被“封杀”？当我们兴致勃勃地编写好一个爬虫脚本，初期运行顺畅，但很快便会遭遇 403 Forbidden、429 Too Many Requests，甚至IP被直接封禁的窘境。这背后，是网站防御系统对我们发起的挑战：频率特征：同一IP在短时间内发起大量请求，行为明显异于正常人类用户。指纹特征：使用默认的requests或urllib库的User-Agent，服务器可轻易识别出这是爬虫程序。行为模式：缺乏有效的Cookie、Referer等头部信息，访问路径单一。要解决这些问题，我们的核心策略是：让爬虫尽可能地模拟真实用户的访问行为。这其中的两大基石便是IP代理与请求头伪装。二、核心技术一：请求头伪装 - 成为“浏览器”而非“脚本”请求头是HTTP请求的“身份证”，它向服务器传递了客户端的详细信息。我们的首要任务就是为爬虫伪造一张合法的“身份证”。1. 基础伪装：设置User-AgentUser-Agent是标识浏览器类型和版本的最关键字段。

import requests

# 一个简单的伪装示例（不推荐用于生产）
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36'
}
response = requests.get('https://fanqienovel.com', headers=headers)

2. 高级伪装：构建完整的浏览器指纹一个真实的浏览器请求会携带数十个头部字段。仅设置User-Agent是远远不够的。

def create_headers():
    """
    创建一个完整的、随机的请求头字典
    """
    # 准备多个常见的User-Agent，实现随机轮换
    user_agents = [
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36',
        'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15',
        'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36 Edg/119.0.0.0'
    ]
    import random
    chosen_ua = random.choice(user_agents)
    
    headers = {
        'User-Agent': chosen_ua,
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
        'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', # 语言偏好
        'Accept-Encoding': 'gzip, deflate, br',
        'Referer': 'https://www.douyin.com/', # 模拟从其他网站跳转而来
        'Sec-Ch-Ua': '"Google Chrome";v="119", "Chromium";v="119", "Not?A_Brand";v="24"', # 客户端提示
        'Sec-Ch-Ua-Mobile': '?0',
        'Sec-Ch-Ua-Platform': '"Windows"',
        'Sec-Fetch-Dest': 'document',
        'Sec-Fetch-Mode': 'navigate',
        'Sec-Fetch-Site': 'none',
        'Upgrade-Insecure-Requests': '1',
        'Connection': 'keep-alive',
        'Cache-Control': 'max-age=0',
    }
    return headers

# 使用自定义头部发起请求
headers = create_headers()
response = requests.get('https://fanqienovel.com/page/example', headers=headers)

关键点：Referer字段尤为重要，它告诉服务器当前请求是从哪个页面链接过来的，模拟用户的连续浏览行为。

三、核心技术二：IP代理 - 打造“千面”爬虫

即使请求头伪装得再好，来自单一IP的高频请求依然会暴露。使用IP代理池，让请求来自全球各地的不同机器，是突破IP封锁的根本手段。

1. 获取代理IP

代理IP的来源主要有以下几类：

免费代理：缺点是稳定性差、可用率低，仅适用于测试。
付费代理服务：提供高匿、稳定的HTTP/HTTPS/SOCKS5代理，通常通过API接口获取IP列表，是商业项目的首选。
自建代理池：通过扫描公网或云服务器自建，技术门槛和成本较高，但控制力最强。

2. 构建IP代理池管理器

我们将创建一个类来管理代理IP，包括获取、验证、轮换和剔除失效IP。

import requests
import random
import time

class ProxyPoolManager:
    def __init__(self, proxy_source_url=None):
        self.proxy_source_url = proxy_source_url # 付费代理服务商提供的API地址
        self.proxies_pool = [] # 代理池
        self.last_fetch_time = 0
        self.fetch_interval = 600 # 10分钟更新一次代理池
        
    def fetch_proxies(self):
        """从代理服务商API获取代理IP列表"""
        if time.time() - self.last_fetch_time < self.fetch_interval and self.proxies_pool:
            return # 未到更新时间且池子不为空，则不更新
        
        try:
            # 示例：调用代理服务商API
            resp = requests.get(self.proxy_source_url, timeout=10)
            if resp.status_code == 200:
                # 假设API返回JSON格式： [{"ip": "1.2.3.4", "port": "8080"}, ...]
                new_proxies = resp.json()
                # 格式化为requests库需要的格式
                formatted_proxies = [f"http://{p['ip']}:{p['port']}" for p in new_proxies]
                self.proxies_pool = formatted_proxies
                self.last_fetch_time = time.time()
                print(f"成功更新代理池，当前共有 {len(self.proxies_pool)} 个代理IP。")
        except Exception as e:
            print(f"获取代理IP失败: {e}")
            
    def validate_proxy(self, proxy_url, test_url='https://httpbin.org/ip', timeout=5):
        """验证单个代理IP是否有效"""
        try:
            proxies = {'http': proxy_url, 'https': proxy_url}
            response = requests.get(test_url, proxies=proxies, timeout=timeout)
            if response.status_code == 200:
                # 检查返回的IP是否确实是代理IP
                returned_ip = response.json().get('origin')
                proxy_ip = proxy_url.split('//')[1].split(':')[0]
                if returned_ip == proxy_ip:
                    print(f"代理 {proxy_url} 验证有效。")
                    return True
        except Exception as e:
            # 代理无效，静默失败或记录日志
            pass
        print(f"代理 {proxy_url} 验证无效，已剔除。")
        return False
    
    def get_random_valid_proxy(self):
        """获取一个随机的、经过验证的有效代理"""
        self.fetch_proxies() # 确保代理池是最新的
        
        if not self.proxies_pool:
            raise Exception("代理池为空，无法获取代理。")
            
        # 随机选择一个代理
        proxy = random.choice(self.proxies_pool)
        
        # 验证该代理
        if self.validate_proxy(proxy):
            return proxy
        else:
            # 如果无效，从池中移除并重试
            self.proxies_pool.remove(proxy)
            return self.get_random_valid_proxy() # 递归调用，直到找到有效的

3. 集成IP代理与请求头伪装

现在，我们将两大技术整合到番茄小说爬虫的核心请求函数中

import requests
import time
import random

# 代理配置信息
proxyHost = "www.16yun.cn"
proxyPort = "5445"
proxyUser = "16QMSOML"
proxyPass = "280651"

def create_headers():
    """
    创建一个完整的、随机的请求头字典
    """
    user_agents = [
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36',
        'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15',
        'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36 Edg/119.0.0.0'
    ]
    chosen_ua = random.choice(user_agents)
    
    headers = {
        'User-Agent': chosen_ua,
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
        'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
        'Accept-Encoding': 'gzip, deflate, br',
        'Referer': 'https://www.douyin.com/',
        'Sec-Ch-Ua': '"Google Chrome";v="119", "Chromium";v="119", "Not?A_Brand";v="24"',
        'Sec-Ch-Ua-Mobile': '?0',
        'Sec-Ch-Ua-Platform': '"Windows"',
        'Sec-Fetch-Dest': 'document',
        'Sec-Fetch-Mode': 'navigate',
        'Sec-Fetch-Site': 'none',
        'Upgrade-Insecure-Requests': '1',
        'Connection': 'keep-alive',
        'Cache-Control': 'max-age=0',
    }
    return headers

class ProxyPoolManager:
    def __init__(self):
        self.proxies_pool = []
        self.last_fetch_time = 0
        self.fetch_interval = 600
        
    def get_proxy_with_auth(self):
        """
        构建带认证信息的代理URL
        """
        proxy_url = f"http://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}"
        return proxy_url
    
    def validate_proxy(self, proxy_url, test_url='https://httpbin.org/ip', timeout=5):
        """验证代理是否有效"""
        try:
            proxies = {
                'http': proxy_url,
                'https': proxy_url
            }
            response = requests.get(test_url, proxies=proxies, timeout=timeout)
            if response.status_code == 200:
                print(f"代理验证有效: {proxyHost}:{proxyPort}")
                return True
        except Exception as e:
            print(f"代理验证失败: {e}")
        return False
    
    def get_valid_proxy(self):
        """获取有效的代理"""
        proxy_url = self.get_proxy_with_auth()
        
        # 验证代理
        if self.validate_proxy(proxy_url):
            return proxy_url
        else:
            raise Exception("代理验证失败，无法获取有效代理")

def create_robust_session(proxy_pool_manager):
    """创建一个配置了随机请求头和代理的Session对象"""
    session = requests.Session()
    
    def _make_request(url, method='GET', **kwargs):
        # 1. 设置随机请求头
        headers = create_headers()
        if 'headers' in kwargs:
            headers.update(kwargs['headers'])
        kwargs['headers'] = headers
        
        # 2. 设置带认证的代理
        try:
            proxy_url = proxy_pool_manager.get_valid_proxy()
            kwargs['proxies'] = {
                'http': proxy_url,
                'https': proxy_url
            }
            print(f"使用代理: {proxyHost}:{proxyPort}")
        except Exception as e:
            print(f"获取代理失败，将使用本机IP: {e}")
            # 即使代理失败，也继续请求，但风险自负
        
        # 3. 添加超时和重试逻辑
        if 'timeout' not in kwargs:
            kwargs['timeout'] = 15  # 稍微增加超时时间，因为代理可能较慢
            
        max_retries = 3
        for attempt in range(max_retries):
            try:
                print(f"第 {attempt + 1} 次尝试请求: {url}")
                response = session.request(method, url, **kwargs)
                
                # 检查响应状态和内容
                if response.status_code == 200:
                    if "验证码" not in response.text and "access denied" not in response.text.lower():
                        print("请求成功！")
                        return response
                    else:
                        print("请求可能被反爬系统拦截，检测到验证码页面")
                else:
                    print(f"请求失败，状态码: {response.status_code}")
                    
            except (requests.exceptions.ProxyError, 
                    requests.exceptions.ConnectTimeout, 
                    requests.exceptions.ReadTimeout,
                    requests.exceptions.ConnectionError) as e:
                print(f"第 {attempt + 1} 次请求失败: {e}")
                if attempt < max_retries - 1:
                    wait_time = 2 ** attempt  # 指数退避策略
                    print(f"等待 {wait_time} 秒后重试...")
                    time.sleep(wait_time)
                    continue
                else:
                    print("所有重试均失败")
                    raise e
                    
            # 如果不是网络错误，但请求被拦截，也进行重试
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt
                print(f"请求可能被拦截，等待 {wait_time} 秒后重试...")
                time.sleep(wait_time)
        
        return None
    
    # 为session绑定新的方法
    session.robust_request = _make_request
    return session

# 主爬虫函数
def crawl_fanqie_novel(book_id):
    """爬取番茄小说内容"""
    # 初始化代理池管理器
    proxy_manager = ProxyPoolManager()
    
    # 创建稳健的Session
    session = create_robust_session(proxy_manager)
    
    # 构造目标URL
    url = f'https://fanqienovel.com/reader/{book_id}'
    print(f"开始爬取: {url}")
    
    try:
        # 使用我们自定义的稳健请求方法
        response = session.robust_request(url)
        
        if response and response.status_code == 200:
            # 此处接续你的HTML解析逻辑
            print("成功获取页面内容！")
            # 可以在这里添加内容解析代码
            # soup = BeautifulSoup(response.text, 'html.parser')
            # ... 解析逻辑
            return response.text
        else:
            print("爬取失败，请检查网络或反爬策略。")
            return None
            
    except Exception as e:
        print(f"爬虫执行过程中发生错误: {e}")
        return None

# 使用示例
if __name__ == "__main__":
    # 测试爬虫
    book_id = '123456789012345'
    result = crawl_fanqie_novel(book_id)
    
    if result:
        print("爬取成功！")
        # 处理爬取到的内容
    else:
        print("爬取失败！")

四、高级策略与最佳实践请求频率控制：在关键请求之间使用time.sleep(random.uniform(1, 3))，模拟人类阅读的随机间隔。会话保持：对于需要登录或保持状态的任务，使用requests.Session()对象，它会自动处理Cookies。CAPTCHA处理：当遭遇验证码时，可以：a) 使用第三方打码平台；b) 临时切换更高匿名度的代理；c) 暂停爬虫一段时间。监控与日志：记录每个请求使用的代理、耗时、状态码，便于分析代理质量和网站反爬策略的变化。五、总结通过系统地集成请求头伪装与IP代理池，我们的番茄小说爬虫成功地从一只容易被拍死的“苍蝇”，进化成了一个拥有无数伪装身份的“特工”。这不仅是技术的叠加，更是一种工程思维的体现：将爬虫系统视为一个需要持续对抗、适应和演化的有机体。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python

动态代理

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python

动态代理

登录后参与评论

0 条评论

热度