一、动态Token:爬虫的新挑战
动态Token是一种由服务器生成并下发给客户端的凭证,客户端在后续请求(如AJAX分页、数据提交)中必须携带该凭证以供验证。其核心特点是一次一性或有时效性,常见形式包括:
- CSRF Token: 常用于表单提交,通常隐藏在HTML的
当爬虫遇到这类机制时,直接复制浏览器地址栏的URL或简单模仿GET请求往往会失败,并返回403 Forbidden或401 Unauthorized错误。破解之道在于清晰地拆解Web客户端(浏览器)与服务器的交互流程,并用Python代码完整地复现这一流程。
二、核心策略:拆解与模拟
这是最关键的一步。打开浏览器的“开发者工具”(F12),切换到“网络”(Network)面板,勾选“保留日志”(Preserve log)。然后执行触发AJAX请求的操作(如点击翻页)。
○ 请求头 (Headers): 仔细查看Request Headers,注意是否有Authorization, X-CSRFToken, X-Requested-With等非常规字段。
○ 负载 (Payload): 如果是POST请求,查看Form Data或Payload,寻找可能存在的token, csrf_token等参数。
○ 查询参数 (Query String Parameters): 如果是GET请求,查看URL参数中是否包含了Token。
找到数据请求中的Token后,下一步是找出这个Token是从哪里来的。
● 来源一:初始HTML页面:在最早获取的HTML文档中搜索该Token。它可能存在于一个
● 来源二:之前的AJAX响应:Token也可能来自一个先前的API响应。例如,访问/api/get_token可能会返回一个JSON对象:{"token": "abcde12345"}。这种情况下,你需要先模拟这个获取Token的请求。
在Python中,我们使用requests.Session()对象来维持一个会话,自动处理Cookies,这是模拟登录状态的关键。
三、实战代码:模拟CSRF Token的AJAX翻页
假设我们要爬取一个网站的用户列表,该列表通过AJAX分页加载,且每个POST请求都需要一个从初始页面获取的CSRF Token。
目标分析:
● 第一页数据在初始HTML中。
● “下一页”按钮会触发一个AJAX POST请求。
● 该请求需要携带一个名为csrf_token的表单数据,该Token存在于初始页面的
Python实现代码:
```import requests
from lxml import html
import time
import random
from urllib.parse import urljoin # 用于处理相对URL
========== 代理配置 ==========
proxyHost = "www.16yun.cn"
proxyPort = "5445"
proxyUser = "16QMSOML"
proxyPass = "280651"
proxyMeta = f"http://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}"
proxies = {
"http": proxyMeta,
"https": proxyMeta,
}
========== 爬虫配置 ==========
BASE_DOMAIN = "example.com"
BASE_URL = f"https://{BASE_DOMAIN}/users"
AJAX_URL = f"https://{BASE_DOMAIN}/api/get_users"
更加真实的浏览器 User-Agent 列表
USER_AGENTS = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/121.0'
]
def create_session():
"""创建并配置会话"""
session = requests.Session()
session.proxies.update(proxies)
session.headers.update({
'User-Agent': random.choice(USER_AGENTS),
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,/;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
})
return session
def extract_csrf_token(html_content):
"""从HTML内容中提取CSRF Token,支持多种可能的定位方式"""
tree = html.fromstring(html_content)
# 尝试多种常见的CSRF Token存放位置
selectors = [
'//meta[@name="csrf-token"]/@content',
'//meta[@name="_token"]/@content',
'//input[@name="csrf_token"]/@value',
'//input[@name="_token"]/@value',
'//input[@name="csrf-token"]/@value',
]
for selector in selectors:
tokens = tree.xpath(selector)
if tokens:
return tokens[0]
raise ValueError("CSRF Token not found in the HTML")
def make_request_with_retry(session, url, method='get', max_retries=3, kwargs):
"""带重试机制的请求函数"""
for attempt in range(max_retries):
try:
if method.lower() == 'get':
response = session.get(url, timeout=15, kwargs)
elif method.lower() == 'post':
response = session.post(url, timeout=15, **kwargs)
else:
raise ValueError(f"Unsupported HTTP method: {method}")
response.raise_for_status()
return response
except (requests.exceptions.RequestException, requests.exceptions.Timeout) as e:
if attempt == max_retries - 1:
raise e
print(f"Request failed (attempt {attempt + 1}/{max_retries}): {e}")
time.sleep(2 ** attempt) # 指数退避策略
def scrape_ajax_with_token_enhanced():
"""增强版的爬虫函数,包含更好的错误处理和重试机制"""
session = create_session()
try:
# 1. 首次请求获取初始页面和CSRF Token
print("🔍 正在通过代理请求初始页面...")
response = make_request_with_retry(session, BASE_URL)
# 2. 提取CSRF Token
csrf_token = extract_csrf_token(response.text)
print(f"✅ CSRF Token 获取成功: {csrf_token[:20]}...") # 只显示部分Token
# 3. 设置AJAX请求的公共头部
ajax_headers = {
'X-Requested-With': 'XMLHttpRequest',
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'Origin': f'https://{BASE_DOMAIN}',
'Referer': BASE_URL,
}
total_pages = 5
successful_pages = 0
for page in range(2, total_pages + 1):
print(f"\n📄 正在请求第 {page} 页数据...")
payload = {
'page': page,
'size': 20, # 通常分页API会有size参数
'csrf_token': csrf_token
}
try:
# 4. 发送AJAX请求
ajax_response = make_request_with_retry(
session, AJAX_URL, method='post',
data=payload, headers=ajax_headers
)
# 5. 解析响应数据
data = ajax_response.json()
# 更健壮的数据提取
users = data.get('data', {}).get('list', [])
if not users:
users = data.get('list', [])
if users:
print(f"✅ 第 {page} 页获取成功,共 {len(users)} 条数据")
successful_pages += 1
# 数据处理逻辑
process_users(users, page)
else:
print(f"⚠️ 第 {page} 页无数据,可能已到末页")
break
# 6. 随机延迟,模拟人类行为
time.sleep(random.uniform(1, 3))
except ValueError as e:
print(f"❌ 第 {page} 页JSON解析失败: {e}")
break
except Exception as e:
print(f"❌ 第 {page} 页请求失败: {e}")
# 可以选择继续尝试下一页或跳出循环
continue
print(f"\n🎉 爬取完成!成功获取 {successful_pages} 页数据")
except requests.exceptions.ProxyError as e:
print(f"❌ 代理连接失败: {e}")
print("请检查代理配置或联系代理服务商")
except requests.exceptions.SSLError as e:
print(f"❌ SSL证书错误: {e}")
except Exception as e:
print(f"❌ 爬虫执行失败: {e}")
finally:
session.close()
print("会话已关闭")
def process_users(users, page_num):
"""处理获取到的用户数据"""
# 这里实现您的具体业务逻辑
for i, user in enumerate(users, 1):
# 示例:打印用户信息
user_id = user.get('id', 'N/A')
user_name = user.get('name', 'N/A')
# print(f" 用户 {i}: ID={user_id}, Name={user_name}")
# 实际应用中,您可能会:
# 1. 保存到数据库
# 2. 写入CSV或JSON文件
# 3. 进行数据清洗和转换
pass
if name == 'main':
start_time = time.time()
scrape_ajax_with_token_enhanced()
end_time = time.time()
print(f"⏱️ 总耗时: {end_time - start_time:.2f} 秒")
```
代码关键点解释:
● 会话管理:requests.Session() 是核心,它确保了在第一次请求base_url时获得的Cookies(可能包含会话ID)在后续的POST请求中被自动带上。
● Token提取:使用lxml.html的XPath语法可以高效地从HTML文档中定位并提取所需的Token值。
● Token放置:根据抓包分析的结果,我们将Token以表单数据(data=payload)的形式发送。如果分析发现Token在请求头中,则应修改为headers['X-CSRFToken'] = csrf_token。
● 错误处理:使用response.raise_for_status()可以在请求失败时抛出异常,便于调试。
四、更复杂的情况与进阶建议
- Token有时效性:某些Token可能一次有效或短期有效。解决方案是:每次请求数据前,都重新获取一次Token。这意味着你的爬虫逻辑需要先请求Token生成接口,再请求数据接口。
- Token经过加密或混淆:有时前端JavaScript会对Token或参数进行二次处理。这时单纯的静态分析可能不够,需要用到如selenium、playwright等浏览器自动化工具来执行JS代码,或者使用pyexecjs库执行特定的JS函数来生成参数。但这会大幅增加复杂性和资源消耗。
- JWT处理:JWT通常通过登录接口获取。策略是先模拟登录请求,从响应中获取JWT,然后在后续所有请求的Authorization头中带上它:headers['Authorization'] = f'Bearer {jwt_token}'。
- 频率限制:即使正确处理了Token,过于频繁的请求也会触发服务器的风控。合理设置请求间隔(time.sleep())、使用代理IP池是走向工业级可靠爬虫的必经之路。
结论
处理动态Token的爬虫不再是简单的数据抓取,而是一场对Web应用逻辑的深度复盘。成功的关键在于精细的抓包分析、对HTTP会话的理解以及精准的代码模拟。通过requests.Session保持状态、使用lxml或BeautifulSoup解析HTML提取Token、并最终将其注入到AJAX请求中,这一套组合拳可以攻克大部分基于动态Token的认证机制。