【爬虫】05 - 爬虫攻防

爬虫05 - 爬虫攻防

一：随机User-Agent爬虫

1：fake-useragent

当爬虫请求头（User-Agent）暴露规律时，目标网站的反爬系统会在‌5秒内‌识别并封锁IP。2023年AlexTop百万网站统计显示，‌68.7%的反爬策略会检测User-Agent特征‌。

检测项	检测原理	典型案例
固定特征值	持续相同User-Agent触发阈值告警	某电商平台连续10次相同UA即封禁
非常用浏览器	识别非常规浏览器版本（如过时Chrome 85）	政府网站拒绝服务古董浏览器
设备类型冲突	移动端UA访问PC端网页触发异常	新闻APP接口校验设备一致性
协议完整性	缺失Accept-Encoding/Connection等标准头	金融数据接口强制校验完整协议头

pip install fake-useragent --upgrade # 添加upgrade是为了防止旧版数据源失效的问题

from fake_useragent import UserAgent
import requests# 创建UserAgent对象, 下面将使用ua.random 获取随机的 UserAgent
ua = UserAgent(browsers=['chrome', 'firefox', 'edge'], os=['windows', 'mac'])
header = {'User-Agent': ua.random, # 随机获取一个UserAgent'Accept-Encoding': 'gzip, deflate, br', # 告诉服务器，我们接受gzip压缩'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', # 告诉服务器，我们接受中文'Connection': 'keep-alive' # 告诉服务器，我们保持连接
}requests.get('https://www.baidu.com', headers=header)

可以封装设备的一致性

from fake_useragent import UserAgent
import requests# 可以封装设备一致性
def generate_user_agent(device_type="pc"):ua = UserAgent()base_headers = {'Accept-Encoding': 'gzip, deflate, br','Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8'}if device_type == 'mobile':return {**base_headers,'User-Agent': ua.firefox, # a'X-Requested-With': 'com.android.browser'}else:return {**base_headers,'User-Agent': ua.chrome, # b'Sec-CH-UA-Platform': '"Windows"'}for page in range(1, 11):headers = generate_user_agent('mobile' if page % 2 else 'pc')response = requests.get(f'https://www.zhipin.com/job_detail/?query=python&city=101010100&industry=&position=', headers=headers)while True:if response.status_code == 200:print(response.text)breakprint("=" * 20)

2：高级反反爬策略

方案一：动态版本更新‌（解决版本过时检测）

# 强制使用最新Chrome版本  
ua = UserAgent(min_version=120)  # Chrome 120+  
headers = {'User-Agent': ua.chrome}  

方案二：混合真实浏览器指纹‌

# 从真实浏览器捕获指纹注入  
real_fingerprint = {  'Sec-CH-UA': '"Chromium";v="118", "Google Chrome";v="118", "Not=A?Brand";v="8"',  'Sec-CH-UA-Mobile': '?0',  'Sec-CH-UA-Platform': '"Windows"'  
}  
headers = {‌**generate_context_headers(), **‌real_fingerprint}  

失败重试熔断机制

from tenacity import retry, stop_after_attempt, wait_exponential  @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1))  
def safe_request(url):  try:  return requests.get(url, headers=generate_context_headers())  except requests.exceptions.RequestException as e:  if e.response.status_code == 403:  # 触发UA刷新熔断  UserAgent().update()  raise  safe_request('https://target.com/api')  

3：生产环境建议

定时更新UA数据库

# 每天自动更新UA数据库  
0 3 * * * /usr/bin/python3 -c "from fake_useragent import UserAgent; UserAgent().update()"  

可以配置些监控和报警

# 当连续5次403错误时触发警报  
if error_count > 5:  send_alert(f"UA策略失效！当前拦截率：{error_count/request_count*100:.2f}%")  switch_to_backup_proxy()  

在生产环境中最好使用多库备用

# 当fake_useragent失效时切换至browser_useragent  
try:  from fake_useragent import UserAgent  
except ImportError:  from browswer_useragent import BrowserUserAgent as UserAgent

二：代理IP爬虫

当爬虫请求频率超过‌5次/秒‌时，目标网站的反爬系统将在‌10秒内‌封锁当前IP。据2024年全球反爬技术报告，‌83%的网站采用IP指纹检测‌作为核心防御手段

动态代理IP池‌，结合智能路由与熔断机制实现反爬突围，实测将IP封禁率从‌72%降至3%‌

检测维度	反爬策略	典型案例
请求频率阈值	单IP单位时间内请求次数超限触发封禁	某社交平台限制单IP每秒3次请求
IP黑名单库	识别代理服务器IP段并全局封禁	新闻网站屏蔽已知数据中心IP
地理位置异常	短时间跨国IP跳跃触发风控	电商平台拦截中美IP交替访问行为

1：获取代理IP

可以使用https://free-proxy-list.net/zh-cn/获取到免费的代理IP

import requests
import random
from bs4 import BeautifulSoup# 获取免费代理 -> https://free-proxy-list.net/zh-cn/
def scrape_free_proxies():url = "https://free-proxy-list.net/" # 获取免费的ip代理response = requests.get(url)soup = BeautifulSoup(response.text, 'html.parser')proxies = [] # 创建一个空列表, 存储获取到的代理for row in soup.select("table.table tbody tr"):cols = row.find_all('td')if len(cols) < 7:continueif cols[4].text == 'elite proxy' and cols[6].text == 'yes':proxies.append(f"{cols[0].text}:{cols[1].text}")return proxies# 请求, 如果失败了就换一个代理IP, 最多尝试5次
def do_request(proxy):for i in range(5):print(f"Trying proxy: {proxy}")try:response = requests.get('http://www.baidu.com', proxies=proxy)print(response.json())returnexcept:print(f"Failed to get IP, trying again...")proxy = {'http': f'http://{random.choice(proxy_list)}','https': f'http://{random.choice(proxy_list)}'}if __name__ == '__main__':proxy_list = scrape_free_proxies()print(proxy_list)# 随机选择代理current_proxy = {'http': f'http://{random.choice(proxy_list)}','https': f'http://{random.choice(proxy_list)}'}do_request(current_proxy)

还可以添加代理IP的智能容错机制

import requests
import random
from bs4 import BeautifulSoup
from tenacity import retry, stop_after_attempt, wait_fixed# 获取免费代理 -> https://free-proxy-list.net/zh-cn/
def scrape_free_proxies():url = "https://free-proxy-list.net/" # 获取免费的ip代理response = requests.get(url)soup = BeautifulSoup(response.text, 'html.parser')proxies = [] # 创建一个空列表, 存储获取到的代理for row in soup.select("table.table tbody tr"):cols = row.find_all('td')if len(cols) < 7:continueif cols[4].text == 'elite proxy' and cols[6].text == 'yes':proxies.append(f"{cols[0].text}:{cols[1].text}")return proxies# 代理IP的智能容错机制
@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def robust_request(url, proxy_pool):proxy = random.choice(proxy_pool)try:return requests.get(url,proxies={'http': f'http://{proxy}', 'https': f'http://{proxy}'},timeout=10)except (requests.ProxyError, requests.ConnectTimeout):proxy_pool.remove(proxy)  # 移除失效代理raiseif __name__ == '__main__':proxy_list = scrape_free_proxies()url = "http://www.baidu.com"response = robust_request(url, proxy_list)print(response.status_code)print(response.text)

2：高阶攻防

四类代理IP的选型

代理类型	优势	劣势	适用场景
数据中心代理	高速度、低延迟	易被识别封禁	快速抓取非敏感数据
住宅代理	高匿名性、真实用户IP	成本高、速度波动	对抗严格反爬（如Cloudflare）
移动4G代理	极高匿名性、IP池庞大	稳定性差、管理复杂	抓取APP接口数据
Socks5代理	支持UDP、加密传输	配置复杂、兼容性要求	需要深度匿名场景

由此演化出对应IP黑名单的三种策略：

策略一：协议混淆，将HTTP流量伪装成Socks5

import socks  
import socket  # 强制使用Socks5协议  
socks.set_default_proxy(socks.SOCKS5, "proxy_ip", 1080)  
socket.socket = socks.socksocket  # 发送请求（网站识别为普通Socks流量）  
requests.get("https://target.com")  

策略二：IP冷启动‌：新代理首次访问仅采集低风险页面

策略三：流量染色‌：在代理请求中注入真实浏览器指纹（如TLS指纹）

3：企业级的代理实战

redis自建代理池系统

import redis
import json
import requestsclass ProxyPool:def __init__(self):self.redis = redis.Redis(host='127.0.0.1', port=6379, db=0)def add_proxy(self, proxy:str, score:float=100):self.redis.zadd('proxies', {proxy: score})def get_best_proxy(self):return self.redis.zrange('proxies', 0, 0)[0].decode()def refresh_proxy(self, proxy:str, penalty:float=20):self.redis.zincrby('proxies', -penalty, proxy)if __name__ == '__main__':# 添加代理pool = ProxyPool()pool.add_proxy('127.0.0.1:8080')pool.add_proxy('127.0.0.1:8081')pool.add_proxy('127.0.0.1:8082')# 获取代理best_proxy = pool.get_best_proxy()try:# 请求, 如果失败了就换一个代理IP, 最多尝试5次requests.get("https://target.com", proxies={'http': best_proxy})except Exception:pool.refresh_proxy(best_proxy)  

商业代理集成

import hashlib  
import time  def gen_mogu_proxy():  # 生成动态签名  timestamp = str(int(time.time()))  secret = "your_api_secret"  sign = hashlib.md5(f"timestamp={timestamp}&secret={secret}".encode()).hexdigest()  # 获取独享代理（按需切换IP）  api_url = f"http://piping.mogumiao.com/proxy/api/get_ip?count=1&timestamp={timestamp}&sign={sign}"  result = requests.get(api_url).json()  return f"{result['msg'][0]['ip']}:{result['msg'][0]['port']}"  # 获取高匿名IP  
mogu_proxy = gen_mogu_proxy()  
requests.get("https://target.com", proxies={'http': f'http://{mogu_proxy}'})  

三：动态数据的抓取

当传统爬虫遭遇‌React/Vue单页应用‌时，‌83%的数据请求‌通过Ajax/WebSocket动态加载，直接获取HTML源码的成功率不足15%。

而如果结合‌逆向工程‌与‌无头浏览器控制技术‌，构建覆盖SPA（单页应用）、SSR（服务端渲染）、CSR（客户端渲染）的全场景解决方案，实现动态数据抓取成功率从‌12%到98%‌的技术跃迁

1：动态页面技术全景

技术类型	核心原理	典型场景
Ajax/XHR	XMLHttpRequest异步获取数据	电商商品分页加载
WebSocket	全双工通信实时更新	股票行情/在线聊天
SSR	服务端生成动态HTML（如Next.js）	新闻门户首屏渲染
CSR	客户端JS动态构建DOM（如React/Vue）	后台管理系统
JSONP	跨域数据获取（逐渐被CORS替代）	老旧天气预报接口

2：动态页面逆向工程

2.1：XHR请求追踪与解析

1. 打开‌Network面板‌并筛选XHR/Fetch请求
2. 定位目标数据的API端点（如/graphql）
3. 解析请求头认证参数（Authorization/X-API-Key）
4. 复制为Python代码（Copy as cURL → 转换为requests代码）

import requests
from urllib.parse import urlencode# 设置请求头，包含API版本和授权信息
headers = {'x-api-version': '3.2','authorization': 'Bearer eyJhbGciOiJIUzI1Ni...',
}# 定义请求参数，包括类别ID、排序方式、页码和平台信息
params = {'categoryId': 305,'sort': 'sales_desc','page': 1,'platform': 'web'
}# 直接请求数据接口
response = requests.get('https://api.shop.com/graphql',headers=headers,params=urlencode(params, doseq=True)
)# 解析JSON数据
products = response.json()['data']['products']

2.2：websocket实时数据捕获

import asyncio
import websockets
import jsonasync def fetch_danmu():uri = "wss://live-api.example.com/ws"  # 替换为实际的 WebSocket 地址while True:try:async with websockets.connect(uri) as websocket:print("成功连接到直播间！")# 可选：发送认证信息auth_message = json.dumps({"user": "test_user","token": "your_token"})await websocket.send(auth_message)print("认证信息已发送")while True:try:# 接收服务器发送的弹幕消息message = await websocket.recv()danmu_data = json.loads(message)print(f"收到弹幕: {danmu_data.get('content', '未知内容')}")except websockets.exceptions.ConnectionClosed:print("连接断开，重新连接中...")breakexcept Exception as e:print(f"发生错误: {e}")break# 运行异步任务
asyncio.get_event_loop().run_until_complete(fetch_danmu())

3：无头浏览器控制技术

无头浏览器(Headless Browser)是指没有图形用户界面的浏览器，可以通过编程方式控制，模拟用户操作，执行JavaScript渲染，是现代爬虫技术中的重要工具。

1. Puppeteer - Google开发的Node库，控制Chromium/Chrome
2. Playwright - Microsoft开发的多浏览器控制工具
3. Selenium - 传统的浏览器自动化框架(后面介绍)
4. Pyppeteer - Puppeteer的Python版本

3.1：Playwright详解

Playwright是由Microsoft开发的跨浏览器自动化测试工具，支持Chromium、WebKit和Firefox

Playwright有如下的特性：

• 多浏览器支持：Chromium (Chrome, Edge)、WebKit (Safari)、Firefox
• 跨平台能力：Windows、macOS、Linux全平台支持 & 可本地运行也可CI/CD集成
• 多语言绑定：JavaScript/TypeScript、Python、Java、.NET
• 现代化架构：基于WebSocket的通信协议、自动等待机制、强大的选择器引擎

pip install playwright
playwright install  # 安装浏览器

基本页面操作

from playwright.sync_api import sync_playwrightwith sync_playwright() as p:# 启动浏览器(无头模式)browser = p.chromium.launch(headless=False)# 创建新页面page = browser.new_page()# 导航到URLpage.goto("https://example.com")# 获取页面标题print(page.title())# 截图page.screenshot(path="example.png")# 关闭浏览器browser.close()

元素定位与交互 - Playwright提供多种强大的选择器：

# CSS选择器
page.click("button.submit")# 文本选择器
page.click("text=Login")# XPath
page.click("//button[@id='submit']")# 组合选择器
page.click("article:has-text('Playwright') >> button")

自动等待机制 -> Playwright内置智能等待，无需手动添加sleep

# 等待元素出现(最多10秒)
page.wait_for_selector("#dynamic-element", timeout=10000)# 等待导航完成
page.click("text=Navigate")
page.wait_for_url("**/new-page")# 等待网络请求完成
with page.expect_response("**/api/data") as response_info:page.click("button.load-data")
response = response_info.value

网络请求拦截

# 路由拦截
def handle_route(route):if "ads" in route.request.url:route.abort()  # 阻止广告请求else:route.continue_()page.route("**/*", handle_route)

文件下载处理

# 等待下载开始
with page.expect_download() as download_info:page.click("a#download-link")
download = download_info.value# 保存下载文件
path = download.path()
download.save_as("/path/to/save")

iframe处理

# 定位iframe
frame = page.frame(name="embedded")# 在iframe内操作
frame.fill("#username", "testuser")
frame.click("#submit")

爬虫实战应用：

动态内容抓取

async with async_playwright() as p:browser = await p.chromium.launch()page = await browser.new_page()await page.goto("https://dynamic-ecom-site.com")# 滚动加载所有商品while await page.locator("text=Load More").is_visible():await page.click("text=Load More")await page.wait_for_timeout(2000)  # 适当延迟# 提取所有商品数据products = await page.locator(".product").evaluate_all("""products => products.map(p => ({name: p.querySelector('.name').innerText,price: p.querySelector('.price').innerText}))""")print(products)await browser.close()

登录会话保持

# 保存登录状态
context = browser.new_context()
page = context.new_page()
page.goto("login_url")
page.fill("#username", "user")
page.fill("#password", "pass")
page.click("#login")# 保存cookies
context.storage_state(path="auth.json")# 后续使用保存的状态
context = browser.new_context(storage_state="auth.json")
page = context.new_page()

性能优化技巧

浏览器上下文复用：

context = browser.new_context()
page1 = context.new_page()
page2 = context.new_page()

请求过滤：

await page.route("**/*.{png,jpg,jpeg}", lambda route: route.abort())

并行处理：

async with asyncio.TaskGroup() as tg:tg.create_task(scrape_page(page1, url1))tg.create_task(scrape_page(page2, url2))

常见问题解决方案

检测规避：

# 修改WebGL供应商信息
await page.add_init_script("""const originalGetParameter = WebGLRenderingContext.prototype.getParameter;WebGLRenderingContext.prototype.getParameter = function(parameter) {if (parameter === 37445) return "Intel Open Source Technology Center";return originalGetParameter.call(this, parameter);};
""")

超时处理：

try:await page.wait_for_selector(".element", timeout=5000)
except TimeoutError:print("元素加载超时")

元素点击问题：

await page.locator("button").dispatch_event("click")  # 直接触发事件

与Puppeteer对比

特性	Playwright	Puppeteer
浏览器支持	多引擎	仅Chromium
语言支持	多种	主要JS
自动等待	更智能	基础
选择器引擎	更强大	标准
移动设备模拟	完善	有限
社区生态	快速增长	成熟稳定

3.2：反反爬虫策略应对

• 指纹伪装：修改浏览器指纹特征
• 行为模拟：模拟人类操作模式(鼠标移动、随机延迟)
• 代理轮换：结合代理IP池使用
• WebGL/Canvas指纹处理：定制化渲染参数

3.3：高级技术应用

分布式无头浏览器集群

智能渲染策略

• 按需渲染：根据目标网站特点定制渲染策略
• 资源加载控制：选择性加载CSS/JS/图片
• 预渲染缓存：对常见页面进行预渲染

性能优化技术

1. 浏览器实例复用：避免频繁启动关闭
2. 页面池管理：维护多个页面实例
3. 资源拦截：阻止不必要资源加载
4. CDN缓存利用：合理设置缓存策略

挑战	解决方案
资源消耗大	浏览器实例池化、资源限制
检测风险高	指纹伪装、行为模拟
稳定性问题	心跳检测、自动恢复
性能瓶颈	分布式架构、智能调度