处理动态Token:Python爬虫应对AJAX授权请求的策略

本文涉及的产品
云数据库 Tair(兼容Redis),内存型 2GB
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
简介: 处理动态Token:Python爬虫应对AJAX授权请求的策略

一、动态Token:爬虫的新挑战
动态Token是一种由服务器生成并下发给客户端的凭证,客户端在后续请求(如AJAX分页、数据提交)中必须携带该凭证以供验证。其核心特点是一次一性或有时效性,常见形式包括:

  1. CSRF Token: 常用于表单提交,通常隐藏在HTML的
标签或表单的字段中,用于验证请求来源的合法性。
  • JWT (JSON Web Tokens): 常存在于用户登录后的API请求头(Authorization)中,是一种包含签名信息的编码字符串,用于维持用户会话状态。
  • 自定义认证令牌: 由服务器端算法生成,可能通过特定的接口获取,并在后续请求中以Query参数或请求头的形式发送。
    当爬虫遇到这类机制时,直接复制浏览器地址栏的URL或简单模仿GET请求往往会失败,并返回403 Forbidden或401 Unauthorized错误。破解之道在于清晰地拆解Web客户端(浏览器)与服务器的交互流程,并用Python代码完整地复现这一流程。
    二、核心策略:拆解与模拟
  • 人工分析:使用开发者工具
    这是最关键的一步。打开浏览器的“开发者工具”(F12),切换到“网络”(Network)面板,勾选“保留日志”(Preserve log)。然后执行触发AJAX请求的操作(如点击翻页)。
  • 寻找数据请求:在请求列表中找到返回实际数据的那个XHR或Fetch请求。
  • 检查请求细节:
    ○ 请求头 (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后,下一步是找出这个Token是从哪里来的。
    ● 来源一:初始HTML页面:在最早获取的HTML文档中搜索该Token。它可能存在于一个
  • 标签中:,或者在一个隐藏的表单字段里:。
    ● 来源二:之前的AJAX响应:Token也可能来自一个先前的API响应。例如,访问/api/get_token可能会返回一个JSON对象:{"token": "abcde12345"}。这种情况下,你需要先模拟这个获取Token的请求。
  • 代码实现:保持会话与自动获取
    在Python中,我们使用requests.Session()对象来维持一个会话,自动处理Cookies,这是模拟登录状态的关键。
  • 从HTML中提取Token:通常使用lxml.html或BeautifulSoup来解析。
  • 从JSON API中提取Token:直接解析响应的JSON数据。
  • 将Token注入后续请求:根据服务器要求,将其放入请求头、表单数据或URL参数中。
    三、实战代码:模拟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()可以在请求失败时抛出异常,便于调试。
    四、更复杂的情况与进阶建议

    1. Token有时效性:某些Token可能一次有效或短期有效。解决方案是:每次请求数据前,都重新获取一次Token。这意味着你的爬虫逻辑需要先请求Token生成接口,再请求数据接口。
    2. Token经过加密或混淆:有时前端JavaScript会对Token或参数进行二次处理。这时单纯的静态分析可能不够,需要用到如selenium、playwright等浏览器自动化工具来执行JS代码,或者使用pyexecjs库执行特定的JS函数来生成参数。但这会大幅增加复杂性和资源消耗。
    3. JWT处理:JWT通常通过登录接口获取。策略是先模拟登录请求,从响应中获取JWT,然后在后续所有请求的Authorization头中带上它:headers['Authorization'] = f'Bearer {jwt_token}'。
    4. 频率限制:即使正确处理了Token,过于频繁的请求也会触发服务器的风控。合理设置请求间隔(time.sleep())、使用代理IP池是走向工业级可靠爬虫的必经之路。
      结论
      处理动态Token的爬虫不再是简单的数据抓取,而是一场对Web应用逻辑的深度复盘。成功的关键在于精细的抓包分析、对HTTP会话的理解以及精准的代码模拟。通过requests.Session保持状态、使用lxml或BeautifulSoup解析HTML提取Token、并最终将其注入到AJAX请求中,这一套组合拳可以攻克大部分基于动态Token的认证机制。
    相关文章
    |
    8天前
    |
    人工智能 运维 安全
    |
    6天前
    |
    人工智能 异构计算
    敬请锁定《C位面对面》,洞察通用计算如何在AI时代持续赋能企业创新,助力业务发展!
    敬请锁定《C位面对面》,洞察通用计算如何在AI时代持续赋能企业创新,助力业务发展!
    |
    7天前
    |
    机器学习/深度学习 人工智能 自然语言处理
    B站开源IndexTTS2,用极致表现力颠覆听觉体验
    在语音合成技术不断演进的背景下,早期版本的IndexTTS虽然在多场景应用中展现出良好的表现,但在情感表达的细腻度与时长控制的精准性方面仍存在提升空间。为了解决这些问题,并进一步推动零样本语音合成在实际场景中的落地能力,B站语音团队对模型架构与训练策略进行了深度优化,推出了全新一代语音合成模型——IndexTTS2 。
    639 22
    |
    7天前
    |
    人工智能 测试技术 API
    智能体(AI Agent)搭建全攻略:从概念到实践的终极指南
    在人工智能浪潮中,智能体(AI Agent)正成为变革性技术。它们具备自主决策、环境感知、任务执行等能力,广泛应用于日常任务与商业流程。本文详解智能体概念、架构及七步搭建指南,助你打造专属智能体,迎接智能自动化新时代。
    |
    13天前
    |
    人工智能 JavaScript 测试技术
    Qwen3-Coder入门教程|10分钟搞定安装配置
    Qwen3-Coder 挑战赛简介:无论你是编程小白还是办公达人,都能通过本教程快速上手 Qwen-Code CLI,利用 AI 轻松实现代码编写、文档处理等任务。内容涵盖 API 配置、CLI 安装及多种实用案例,助你提升效率,体验智能编码的乐趣。
    1036 110
    人工智能 数据可视化 数据挖掘
    229 0