突破验证码与IP封锁:Python爬虫应对复杂反爬机制的实战技巧

在高价值数据采集场景中,验证码(图文、滑动、点选)和IP封锁是最常见的“拦路虎”——简单的请求伪装已无法绕过,需针对性地采用“验证码自动识别”“IP池动态切换”“行为模拟优化”等组合策略。本文结合100+反爬场景实战经验,从验证码破解(全类型)、IP封锁突破、行为反爬规避、实战案例四个核心维度,拆解可直接落地的技术方案,帮你搞定复杂反爬网站。

一、核心认知:反爬机制的本质与应对逻辑

网站反爬的核心是“识别非人类行为”,主要通过三个维度判断:

身份标识:IP地址、设备指纹、Cookie/Session;行为特征:请求频率、操作路径、停留时间、鼠标轨迹;交互验证:验证码(区分人机)、短信验证、登录授权。

应对逻辑:“模拟真实用户+打破识别维度”——用IP池解决身份标识封锁,用行为模拟伪装操作特征,用验证码识别突破交互验证,三者结合形成闭环。

二、环境准备:核心工具库安装


# 验证码识别库
pip install ddddocr==1.5.0 pillow==10.3.0 opencv-python==4.9.0.80
# IP池与请求库
pip install requests==2.31.0 aiohttp==3.9.1 redis==5.0.1
# 行为模拟与自动化
pip install selenium==4.21.0 undetected-chromedriver==3.5.5 playwright==1.40.0
# 辅助工具(设备指纹、加密)
pip install fake-useragent==1.5.0 pycryptodome==3.20.0

关键工具说明:


ddddocr
:开源免费的验证码识别库,支持图文、滑动缺口识别,准确率高;
undetected-chromedriver
:规避Selenium特征检测,适合强反爬场景;
playwright
:微软开发的自动化工具,比Selenium更轻量,支持自动录制操作轨迹;
redis
:用于IP池存储与管理,支持分布式爬虫共享IP资源。

三、突破验证码:全类型自动识别与破解

验证码的核心目的是“区分人机”,常见类型包括图文验证码、滑动验证码、点选验证码,不同类型对应不同破解策略:

3.1 图文验证码(最基础):OCR自动识别

图文验证码(数字、字母、汉字组合)是最常见的类型,用
ddddocr
即可实现高准确率识别,无需复杂配置。

实战步骤:

抓取验证码图片(从网页元素或接口下载);用
ddddocr
加载对应模型(数字字母、汉字);识别图片内容,返回验证码字符串。

代码实现:

import requests
from ddddocr import DdddOcr
from PIL import Image
from io import BytesIO

def crack_text_captcha(captcha_url, headers):
    """
    破解图文验证码
    :param captcha_url: 验证码图片URL
    :param headers: 请求头
    :return: 识别后的验证码字符串
    """
    # 1. 下载验证码图片
    response = requests.get(captcha_url, headers=headers, timeout=10)
    img = Image.open(BytesIO(response.content))
    
    # 2. 初始化OCR模型(选择对应类型)
    # ocr = DdddOcr(show_ad=False)  # 默认模型(数字+字母)
    ocr = DdddOcr(show_ad=False, ocrType="cn")  # 中文模型(支持汉字)
    
    # 3. 识别验证码(需将PIL图片转为字节流)
    img_byte = BytesIO()
    img.save(img_byte, format="PNG")
    img_byte = img_byte.getvalue()
    
    result = ocr.classification(img_byte)
    print(f"图文验证码识别结果:{result}")
    return result

# 测试:破解某网站图文验证码
if __name__ == "__main__":
    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"
    }
    captcha_url = "https://example.com/captcha.jpg"  # 替换为真实验证码URL
    crack_text_captcha(captcha_url, headers)
优化技巧:

图片预处理:若验证码有干扰线、噪点,用
PIL

opencv
进行二值化、降噪处理,提升识别率:


import cv2
import numpy as np

def preprocess_img(img):
    """图片预处理:二值化+降噪"""
    img_cv = cv2.cvtColor(np.array(img), cv2.COLOR_RGB2GRAY)
    # 二值化(黑白分明)
    _, img_binary = cv2.threshold(img_cv, 127, 255, cv2.THRESH_BINARY_INV)
    # 降噪(去除小噪点)
    kernel = np.ones((2, 2), np.uint8)
    img_clean = cv2.morphologyEx(img_binary, cv2.MORPH_OPEN, kernel)
    return Image.fromarray(img_clean)

多模型验证:若识别失败,切换
ddddocr

beta
模型或第三方OCR(如百度OCR、阿里云OCR,付费但准确率更高)。

3.2 滑动验证码(最常用):缺口匹配+轨迹模拟

滑动验证码(拖动滑块填充缺口)是目前主流反爬手段,核心破解难点是“缺口定位”和“模拟人类滑动轨迹”(匀速滑动易被识别)。

核心原理:

缺口定位:对比“完整图片”和“带缺口图片”,用像素差异找到缺口位置;轨迹模拟:生成加速→匀速→减速的滑动轨迹(符合人类操作习惯)。

代码实现(基于Selenium+ddddocr):

import time
import random
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from ddddocr import DdddOcr
from PIL import Image
from io import BytesIO

def get_slide_distance(full_img, gap_img):
    """
    计算滑块需要滑动的距离(缺口定位)
    :param full_img: 完整图片(PIL对象)
    :param gap_img: 带缺口图片(PIL对象)
    :return: 缺口距离(像素)
    """
    # 转为灰度图
    full_gray = full_img.convert("L")
    gap_gray = gap_img.convert("L")
    
    # 计算像素差异(阈值设为20,差异超过则认为是缺口)
    threshold = 20
    for x in range(50, full_gray.width):  # 从左到右扫描(跳过左侧滑块区域)
        for y in range(full_gray.height):
            if abs(full_gray.getpixel((x, y)) - gap_gray.getpixel((x, y))) > threshold:
                return x  # 返回缺口左侧x坐标
    return 0

def generate_slide_track(distance):
    """
    生成人类滑动轨迹(加速→匀速→减速)
    :param distance: 滑动距离
    :return: 轨迹列表((x偏移, y偏移, 时间间隔))
    """
    track = []
    current_x = 0
    # 加速阶段(前30%距离)
    accelerate_distance = distance * 0.3
    while current_x < accelerate_distance:
        step = random.randint(2, 5)
        track.append((step, 0, random.randint(10, 20)))
        current_x += step
    # 匀速阶段(中间50%距离)
    uniform_distance = distance * 0.5
    while current_x < accelerate_distance + uniform_distance:
        step = random.randint(3, 4)
        track.append((step, 0, random.randint(15, 25)))
        current_x += step
    # 减速阶段(最后20%距离)
    while current_x < distance:
        step = random.randint(1, 3)
        track.append((step, 0, random.randint(20, 30)))
        current_x += step
    # 微调(避免超出目标位置)
    if current_x > distance:
        track.append((current_x - distance, 0, random.randint(10, 15)))
    return track

def crack_slide_captcha(driver, slide_btn_xpath, full_img_xpath, gap_img_xpath):
    """
    破解滑动验证码
    :param driver: Selenium驱动
    :param slide_btn_xpath: 滑块按钮XPath
    :param full_img_xpath: 完整图片XPath(部分网站需点击后加载)
    :param gap_img_xpath: 带缺口图片XPath
    """
    wait = WebDriverWait(driver, 10)
    
    # 1. 等待图片和滑块加载完成
    full_img_elem = wait.until(EC.presence_of_element_located((By.XPATH, full_img_xpath)))
    gap_img_elem = wait.until(EC.presence_of_element_located((By.XPATH, gap_img_xpath)))
    slide_btn = wait.until(EC.element_to_be_clickable((By.XPATH, slide_btn_xpath)))
    
    # 2. 下载完整图片和带缺口图片
    def get_img_from_elem(elem):
        img_src = elem.get_attribute("src")
        response = requests.get(img_src, timeout=10)
        return Image.open(BytesIO(response.content))
    
    full_img = get_img_from_elem(full_img_elem)
    gap_img = get_img_from_elem(gap_img_elem)
    
    # 3. 计算滑动距离
    distance = get_slide_distance(full_img, gap_img)
    print(f"缺口距离:{distance}像素")
    
    # 4. 生成滑动轨迹
    track = generate_slide_track(distance)
    
    # 5. 模拟滑动(按住滑块→按轨迹移动→松开)
    action = ActionChains(driver)
    action.click_and_hold(slide_btn).perform()
    time.sleep(0.2)  # 按住后停留0.2秒(模拟人类操作)
    
    for step_x, step_y, duration in track:
        action.move_by_offset(step_x, step_y).perform()
        time.sleep(duration / 1000)  # 时间间隔(毫秒转秒)
    
    action.release().perform()
    time.sleep(1)  # 等待验证结果

# 测试:破解某网站滑动验证码
if __name__ == "__main__":
    # 创建反检测驱动
    driver = webdriver.Chrome(options=undetected_chromedriver.ChromeOptions())
    driver.get("https://example.com/slide-captcha")  # 替换为带滑动验证码的页面
    
    # 破解滑动验证码(需根据实际页面调整XPath)
    crack_slide_captcha(
        driver=driver,
        slide_btn_xpath='//div[@class="slide-btn"]',
        full_img_xpath='//img[@class="full-img"]',
        gap_img_xpath='//img[@class="gap-img"]'
    )
    
    # 后续操作...
    driver.quit()
避坑要点:

图片缩放问题:若网页图片被缩放(如
width
属性修改),需根据缩放比例修正滑动距离;轨迹随机性:避免固定轨迹,每次滑动的步长、时间间隔需随机;滑块初始位置:部分网站滑块初始位置不在最左侧,需先获取初始X坐标再计算相对距离。

3.3 点选验证码(高难度):目标识别+坐标点击

点选验证码(如“按顺序点击汉字”“点击相同物体”)是更复杂的人机验证,需先识别目标位置,再模拟人类点击顺序和轨迹。

核心方案:

目标识别:用
ddddocr
的点选模型识别目标坐标,或用第三方API(如超级鹰);点击模拟:按识别结果的顺序,模拟人类点击(先慢后快、带微小偏移)。

代码实现(基于ddddocr点选模型):

import random
import time
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from ddddocr import DdddOcr
from PIL import Image
from io import BytesIO

def crack_click_captcha(driver, captcha_img_xpath, tip_text):
    """
    破解点选验证码(如“按顺序点击:我、爱、中、国”)
    :param driver: Selenium驱动
    :param captcha_img_xpath: 验证码图片XPath
    :param tip_text: 点选提示(如“我爱国中”)
    """
    wait = WebDriverWait(driver, 10)
    captcha_elem = wait.until(EC.presence_of_element_located((By.XPATH, captcha_img_xpath)))
    
    # 1. 下载验证码图片
    img_src = captcha_elem.get_attribute("src")
    response = requests.get(img_src, timeout=10)
    img = Image.open(BytesIO(response.content))
    
    # 2. 用ddddocr点选模型识别目标坐标(需提前准备训练数据,或用通用模型)
    ocr = DdddOcr(show_ad=False, det=True)  # det=True启用目标检测
    # 识别图片中的文字和坐标(返回格式:[{"text": "我", "box_2d": [x1, y1, x2, y2]}, ...])
    result = ocr.detection(img_bytes=img.tobytes())
    
    # 3. 按提示文本排序坐标(如提示“我爱国中”,则按顺序提取对应坐标)
    target_coords = []
    for item in result:
        text = item["text"]
        if text in tip_text:
            # 计算文字中心点坐标(相对图片)
            x1, y1, x2, y2 = item["box_2d"]
            center_x = (x1 + x2) / 2
            center_y = (y1 + y2) / 2
            target_coords.append((text, center_x, center_y))
    
    # 按提示顺序排序
    target_coords.sort(key=lambda x: tip_text.index(x[0]))
    print(f"点选顺序坐标:{[(t[0], t[1], t[2]) for t in target_coords]}")
    
    # 4. 模拟人类点击(带偏移、随机间隔)
    action = ActionChains(driver)
    # 获取验证码图片在页面中的绝对位置
    img_location = captcha_elem.location  # 左上角坐标
    img_size = captcha_elem.size  # 宽高
    
    for i, (text, x, y) in enumerate(target_coords):
        # 计算页面绝对坐标(添加±3像素偏移,模拟人类点击误差)
        abs_x = img_location["x"] + x + random.randint(-3, 3)
        abs_y = img_location["y"] + y + random.randint(-3, 3)
        
        # 移动到目标位置(带轨迹)
        action.move_by_offset(abs_x - action.w3c_actions.current_point["x"], 
                             abs_y - action.w3c_actions.current_point["y"])
        action.pause(random.uniform(0.3, 0.8))  # 点击间隔0.3-0.8秒
        action.click().perform()
        
        print(f"已点击:{text},坐标:({abs_x}, {abs_y})")
    
    time.sleep(1)  # 等待验证结果

# 测试:破解点选验证码
if __name__ == "__main__":
    driver = webdriver.Chrome(options=undetected_chromedriver.ChromeOptions())
    driver.get("https://example.com/click-captcha")  # 替换为带点选验证码的页面
    
    # 破解点选验证码(提示文本需从页面提取,此处简化为固定值)
    crack_click_captcha(
        driver=driver,
        captcha_img_xpath='//img[@class="click-captcha"]',
        tip_text="我爱国中"
    )
    
    driver.quit()
进阶方案:


ddddocr
识别率低,可使用付费验证码识别平台(如超级鹰、云打码),支持定制化训练,准确率可达95%+;点选顺序识别:部分网站提示文本是动态的(如随机生成“点击动物”),需先从页面提取提示文本,再进行目标识别。

3.4 验证码破解的终极方案:绕开验证(优先选择)

相比直接破解,以下方案更高效、稳定:

复用Cookie/Session:登录后保存已通过验证的Cookie,后续请求直接携带(避免重复验证);API接口绕过:分析网站是否有未验证的AJAX接口,直接请求接口获取数据;第三方登录:若网站支持微信、QQ登录,用Selenium模拟第三方登录(无需验证);人工打码平台:高难度验证码(如极验3/4、谷歌reCAPTCHA),可接入人工打码平台(如顶象、易盾),按次付费,效率高。

四、突破IP封锁:IP池构建与动态切换

IP封锁是网站最直接的反爬手段(单IP高频请求触发),核心解决方案是“构建高可用IP池+动态切换IP”,避免单一IP被封禁。

4.1 IP池核心架构:采集→验证→存储→调度

一个可用的IP池需包含四个模块,确保IP的“高可用”和“高匿名性”:


graph TD
    A[IP采集模块] --> B[IP验证模块]
    B --> C[IP存储模块(Redis)]
    C --> D[IP调度模块]
    D --> E[爬虫调用]
    E --> F[IP状态反馈(成功/失败)]
    F --> C[更新IP权重]

4.2 实战:构建分布式IP池(Redis+多源采集)

4.2.1 1. IP采集模块(多源获取代理IP)

从免费代理网站、付费代理平台(如阿布云、快代理)采集IP,支持HTTP/HTTPS/SOCKS5协议。


import requests
from bs4 import BeautifulSoup
import redis
from concurrent.futures import ThreadPoolExecutor

class IPPoolCollector:
    def __init__(self, redis_url="redis://localhost:6379/0"):
        self.redis_client = redis.Redis.from_url(redis_url)
        self.proxy_key = "proxy:ip_pool"  # Redis存储键名
        # 免费代理网站(可扩展更多来源)
        self.free_proxy_urls = [
            "https://www.kuaidaili.com/free/inha/",
            "https://www.xicidaili.com/nn/"
        ]
    
    def crawl_free_proxies(self, url):
        """爬取免费代理网站的IP"""
        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"
        }
        try:
            response = requests.get(url, headers=headers, timeout=10)
            soup = BeautifulSoup(response.text, "lxml")
            # 解析IP、端口、协议(根据目标网站结构调整)
            proxy_trs = soup.find_all("tr")[1:]  # 跳过表头
            for tr in proxy_trs:
                tds = tr.find_all("td")
                if len(tds) < 5:
                    continue
                ip = tds[0].text.strip()
                port = tds[1].text.strip()
                protocol = tds[3].text.strip().lower()  # HTTP/HTTPS
                proxy = f"{protocol}://{ip}:{port}"
                # 加入待验证队列
                self.redis_client.lpush("proxy:pending", proxy)
                print(f"采集到代理:{proxy}")
        except Exception as e:
            print(f"爬取代理失败:{url},原因:{e}")
    
    def collect_from_paid_platform(self):
        """从付费代理平台获取IP(以阿布云为例)"""
        # 付费平台通常提供API,直接调用获取高可用IP
        paid_api = "https://api.abuyun.com/proxy/get"
        params = {
            "appkey": "your_appkey",
            "secret": "your_secret",
            "num": 10,
            "type": 2  # 2=HTTPS
        }
        try:
            response = requests.get(paid_api, params=params, timeout=10)
            proxies = response.json()["data"]["proxies"]
            for proxy in proxies:
                self.redis_client.lpush("proxy:pending", f"https://{proxy}")
                print(f"获取付费代理:{proxy}")
        except Exception as e:
            print(f"获取付费代理失败:{e}")
    
    def run(self):
        """多线程采集IP"""
        with ThreadPoolExecutor(max_workers=5) as executor:
            # 爬取免费代理
            for url in self.free_proxy_urls:
                executor.submit(self.crawl_free_proxies, url)
            # 获取付费代理
            executor.submit(self.collect_from_paid_platform)
4.2.2 2. IP验证模块(确保IP可用)

验证IP的连通性、匿名性、响应速度,过滤无效IP。


import requests
import redis
import time
from concurrent.futures import ThreadPoolExecutor

class IPPoolValidator:
    def __init__(self, redis_url="redis://localhost:6379/0"):
        self.redis_client = redis.Redis.from_url(redis_url)
        self.test_url = "https://httpbin.org/ip"  # 用于验证IP的目标网站
        self.timeout = 5  # 验证超时时间
    
    def validate_proxy(self, proxy):
        """验证代理IP是否可用"""
        try:
            # 验证连通性
            start_time = time.time()
            response = requests.get(
                self.test_url,
                proxies={"http": proxy, "https": proxy},
                timeout=self.timeout,
                allow_redirects=False
            )
            response_time = time.time() - start_time
            
            # 验证匿名性(是否暴露真实IP)
            real_ip = requests.get(self.test_url, timeout=self.timeout).json()["origin"]
            proxy_ip = response.json()["origin"]
            if real_ip == proxy_ip:
                print(f"代理{proxy}为透明代理,丢弃")
                return
            
            # 可用IP存入正式池,设置权重(响应时间越短,权重越高)
            weight = int(1000 / (response_time + 0.1))  # 权重1-1000
            self.redis_client.zadd("proxy:available", {proxy: weight})
            print(f"验证通过:{proxy},响应时间:{response_time:.2f}秒,权重:{weight}")
        except Exception as e:
            print(f"验证失败:{proxy},原因:{e}")
    
    def run(self):
        """持续验证待验证队列中的IP"""
        while True:
            # 从待验证队列获取IP
            proxy = self.redis_client.rpop("proxy:pending")
            if not proxy:
                time.sleep(30)  # 无IP时休眠30秒
                continue
            proxy = proxy.decode("utf-8")
            # 单线程验证(避免高频请求目标网站)
            self.validate_proxy(proxy)
4.2.3 3. IP调度模块(爬虫动态获取IP)

爬虫从IP池获取高权重IP,使用失败后降低权重或移除,确保IP池动态更新。


import redis
import random

class IPPoolScheduler:
    def __init__(self, redis_url="redis://localhost:6379/0"):
        self.redis_client = redis.Redis.from_url(redis_url)
        self.available_key = "proxy:available"
    
    def get_proxy(self):
        """获取高权重代理IP(随机选择前20%高权重IP)"""
        # 获取IP总数
        proxy_count = self.redis_client.zcard(self.available_key)
        if proxy_count == 0:
            raise Exception("IP池无可用IP")
        
        # 选择前20%高权重IP,随机抽取一个
        start = 0
        end = int(proxy_count * 0.2) if proxy_count > 5 else proxy_count
        high_weight_proxies = self.redis_client.zrange(self.available_key, start, end, withscores=True)
        proxy, _ = random.choice(high_weight_proxies)
        return proxy.decode("utf-8")
    
    def update_proxy_status(self, proxy, success):
        """更新IP状态(成功→提升权重,失败→降低权重)"""
        current_weight = self.redis_client.zscore(self.available_key, proxy)
        if not current_weight:
            return
        
        if success:
            new_weight = current_weight + 10  # 成功+10权重
        else:
            new_weight = current_weight - 50  # 失败-50权重
        
        if new_weight <= 0:
            self.redis_client.zrem(self.available_key, proxy)
            print(f"IP{proxy}权重过低,移除IP池")
        else:
            self.redis_client.zadd(self.available_key, {proxy: new_weight})
            print(f"IP{proxy}权重更新为:{new_weight}")

4.3 IP池使用技巧:提升稳定性

混合代理类型:免费代理(低成本)+ 付费代理(高可用)结合,降低成本同时保证稳定性;定期清理无效IP:设置定时任务,移除超过24小时未使用的IP;IP切换策略:每请求5-10次切换一次IP,或当出现403/404/503时立即切换;地域匹配:选择与目标网站服务器地域相近的IP,降低响应时间;避免滥用:控制单IP请求频率(如每分钟不超过30次),避免IP被网站永久封禁。

4.4 应对IP封锁的进阶方案

动态拨号IP:家庭宽带/服务器拨号上网,每次拨号更换IP(适合大规模采集,成本低);云函数代理:用AWS Lambda、阿里云函数计算等服务,每次调用使用不同IP(隐蔽性强);CDN代理:通过CDN转发请求,隐藏真实IP(适合轻度反爬网站);IP质量监控:实时监控IP的响应时间、成功率,自动剔除低质量IP。

五、规避行为反爬:模拟真实用户操作

即使突破了验证码和IP封锁,网站仍可能通过“行为特征”识别爬虫(如匀速请求、无鼠标轨迹、操作路径固定),需针对性优化。

5.1 核心行为优化策略

5.1.1 1. 请求频率与间隔

随机延迟:用
time.sleep(random.uniform(1, 3))
替代固定延迟,模拟人类思考时间;梯度间隔:不同操作设置不同间隔(如登录后停留2-3秒,点击后停留0.5-1秒);避开高峰时段:在网站访问低峰期(如凌晨2-6点)采集,降低被检测概率。

5.1.2 2. 操作轨迹模拟

鼠标轨迹:用
ActionChains
模拟人类鼠标移动(如从页面顶部滚动到底部,带随机偏移);点击顺序:避免直接点击目标元素,先点击页面其他位置(如空白处、导航栏),再点击目标;输入模拟:用
send_keys
逐字符输入(带随机间隔),而非一次性输入:


def simulate_human_input(element, text):
    for char in text:
        element.send_keys(char)
        time.sleep(random.uniform(0.05, 0.2))  # 每个字符间隔0.05-0.2秒
5.1.3 3. 设备指纹伪装

网站通过
navigator.userAgent

canvas
指纹、
WebGL
指纹等识别设备,需针对性伪装:

随机User-Agent:用
fake-useragent
库随机生成真实浏览器UA:


from fake_useragent import UserAgent
ua = UserAgent()
headers["User-Agent"] = ua.random

清除Canvas指纹:用Selenium执行JS代码,修改Canvas渲染结果:


driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", {
    "source": """
        HTMLCanvasElement.prototype.toDataURL = function() {
            return "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==";
        }
    """
})
5.1.4 4. Cookie与Session管理

持久化Cookie:用
requests.Session()
或Selenium的
get_cookies()
保存Cookie,模拟长期登录状态;避免空Cookie请求:首次请求携带真实浏览器Cookie(从浏览器复制),后续自动更新;清理敏感Cookie:部分网站通过
__cfduid
等Cookie跟踪爬虫,需定期清理。

5.2 工具选择:Selenium vs Playwright

工具 行为模拟优势 反爬友好度 适用场景
Selenium 生态成熟、API丰富、支持全浏览器 易被检测(需配合undetected-chromedriver) 常规行为模拟、旧网站
Playwright 自动录制操作轨迹、内置指纹伪装、支持无头模式 反爬友好(默认隐藏自动化特征) 复杂行为模拟、现代SPA网站
Playwright行为模拟示例(自动录制轨迹):

from playwright.sync_api import sync_playwright

def simulate_human_operation():
    with sync_playwright() as p:
        # 启动浏览器(默认隐藏自动化特征)
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        
        # 模拟人类访问页面(带随机停留时间)
        page.goto("https://example.com", wait_until="networkidle")
        page.wait_for_timeout(random.randint(1000, 3000))  # 停留1-3秒
        
        # 模拟滚动页面
        page.mouse.wheel(0, random.randint(500, 1000))
        page.wait_for_timeout(random.randint(500, 1000))
        
        # 模拟点击按钮(带鼠标移动轨迹)
        button = page.locator('//button[contains(text(), "登录")]')
        button.scroll_into_view_if_needed()  # 滚动到按钮可见
        page.mouse.move(
            x=random.randint(100, 200), 
            y=random.randint(100, 200)
        )  # 先移动到随机位置
        page.mouse.move(
            x=button.bounding_box()["x"] + random.randint(5, 10),
            y=button.bounding_box()["y"] + random.randint(5, 10),
            steps=random.randint(5, 10)  # 分步移动,模拟轨迹
        )
        page.mouse.click()
        
        # 后续操作...
        browser.close()

六、实战案例:突破某电商平台反爬(验证码+IP封锁+行为检测)

需求:

目标网站:某电商平台(Vue开发,动态渲染);反爬机制:滑动验证码(登录时)、IP封锁(高频请求)、行为检测(操作轨迹);采集目标:商品列表+价格数据。

完整代码:


import random
import time
import requests
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import undetected_chromedriver as uc
from IPPoolScheduler import IPPoolScheduler  # 导入前面实现的IP池调度器
from crack_slide_captcha import crack_slide_captcha  # 导入滑动验证码破解函数

def main():
    # 1. 初始化IP池调度器
    ip_scheduler = IPPoolScheduler()
    
    # 2. 创建反检测浏览器(undetected-chromedriver)
    options = uc.ChromeOptions()
    options.add_argument("--headless=new")
    options.add_argument(f"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")
    driver = uc.Chrome(options=options)
    
    try:
        # 3. 访问登录页,破解滑动验证码
        driver.get("https://example.com/login")
        wait = WebDriverWait(driver, 15)
        
        # 模拟输入用户名密码(人类输入速度)
        username_input = wait.until(EC.presence_of_element_located((By.ID, "username")))
        password_input = wait.until(EC.presence_of_element_located((By.ID, "password")))
        
        simulate_human_input(username_input, "your_username")
        simulate_human_input(password_input, "your_password")
        
        # 点击登录,触发滑动验证码
        login_btn = wait.until(EC.element_to_be_clickable((By.XPATH, '//button[@type="submit"]')))
        login_btn.click()
        
        # 破解滑动验证码(根据实际页面调整XPath)
        crack_slide_captcha(
            driver=driver,
            slide_btn_xpath='//div[@class="slide-btn"]',
            full_img_xpath='//img[@class="full-img"]',
            gap_img_xpath='//img[@class="gap-img"]'
        )
        
        # 等待登录成功(跳转至首页)
        wait.until(EC.url_contains("home"))
        print("登录成功")
        
        # 4. 提取登录后的Cookie,用于Requests请求
        cookies = driver.get_cookies()
        cookie_dict = {cookie["name"]: cookie["value"] for cookie in cookies}
        
        # 5. 用Requests+IP池抓取商品数据(高效并发)
        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",
            "Referer": "https://example.com/home"
        }
        
        all_products = []
        page = 1
        
        while page <= 10:  # 采集前10页
            # 获取代理IP
            proxy = ip_scheduler.get_proxy()
            proxies = {"http": proxy, "https": proxy}
            
            # 构造商品列表AJAX接口(从开发者工具提取)
            api_url = f"https://example.com/api/products?page={page}&pageSize=20"
            
            try:
                response = requests.get(
                    api_url,
                    headers=headers,
                    cookies=cookie_dict,
                    proxies=proxies,
                    timeout=10
                )
                
                if response.status_code == 200:
                    data = response.json()
                    products = data["data"]["items"]
                    all_products.extend(products)
                    print(f"第{page}页采集成功,获取{len(products)}件商品")
                    ip_scheduler.update_proxy_status(proxy, success=True)
                else:
                    print(f"第{page}页采集失败,状态码:{response.status_code}")
                    ip_scheduler.update_proxy_status(proxy, success=False)
                    continue
            
            except Exception as e:
                print(f"第{page}页采集异常:{e}")
                ip_scheduler.update_proxy_status(proxy, success=False)
                continue
            
            page += 1
            time.sleep(random.uniform(1.5, 3))  # 随机延迟
        
        # 6. 保存数据
        print(f"共采集{len(all_products)}件商品")
        # save_to_database(all_products)  # 省略存储逻辑
    
    finally:
        driver.quit()

def simulate_human_input(element, text):
    """模拟人类输入"""
    for char in text:
        element.send_keys(char)
        time.sleep(random.uniform(0.05, 0.2))

if __name__ == "__main__":
    main()

七、避坑指南:10个高频反爬问题解决方案

验证码识别率低→ 图片预处理(降噪、二值化);切换更精准的OCR模型(如百度OCR);接入人工打码平台;IP池IP可用率低→ 增加付费代理来源;优化验证逻辑(延长超时时间、增加验证目标网站);定期清理无效IP;Selenium被检测→ 用
undetected-chromedriver

playwright
;执行JS隐藏
webdriver
特征;禁用自动化扩展;滑动验证码总是失败→ 优化轨迹模拟(增加加速/减速阶段);修正图片缩放比例;添加鼠标按下/松开的延迟;IP切换后仍被封锁→ 检查Cookie是否携带旧IP信息;确保IP是高匿名代理(避免暴露真实IP);增加IP切换频率;行为检测被拦截→ 模拟人类操作(随机滚动、停留时间、点击顺序);避免高频请求;用
playwright
录制真实操作轨迹;API接口返回403→ 携带完整请求头(
Referer

Origin
);更新Cookie/Session;切换IP;设备指纹被识别→ 随机User-Agent;修改Canvas/WebGL指纹;禁用浏览器插件;验证码频繁出现→ 降低请求频率;复用已验证的Cookie;避免同一IP短时间内多次登录;大规模采集时IP池耗尽→ 扩容IP池(增加代理来源);动态拨号IP补充;控制并发数。

八、总结与进阶方向

应对复杂反爬的核心是“组合策略”——验证码破解解决“人机区分”,IP池突破“身份封锁”,行为模拟规避“特征识别”,三者缺一不可。实际开发中需先分析网站反爬机制,再针对性选择方案,避免过度优化。

进阶方向:

自动化反爬绕过:用
mitmproxy
拦截请求,自动提取接口参数和加密逻辑;机器学习破解验证码:训练自定义模型识别复杂验证码(如极验4/5);分布式爬虫架构:多节点部署爬虫,共享IP池和Cookie池,提升采集效率;无代码反爬绕过:使用爬虫工具(如八爪鱼、神箭手),内置反爬策略,无需编写代码;法律合规采集:遵守网站
robots.txt
协议;避免采集敏感数据;控制采集频率,不影响网站正常运行。

如果在实战中遇到验证码破解、IP池优化、行为反爬拦截等问题,欢迎留言交流!

© 版权声明
THE END
如果内容对您有所帮助,就支持一下吧!
点赞0 分享
评论 抢沙发

请登录后发表评论

    暂无评论内容