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

本文涉及的产品
RDS MySQL DuckDB 分析主实例,集群系列 4核8GB
简介: 处理动态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的认证机制。
    相关文章
    |
    2月前
    |
    数据采集 Web App开发 数据安全/隐私保护
    实战:Python爬虫如何模拟登录与维持会话状态
    实战:Python爬虫如何模拟登录与维持会话状态
    |
    3月前
    |
    数据采集 Web App开发 自然语言处理
    新闻热点一目了然:Python爬虫数据可视化
    新闻热点一目了然:Python爬虫数据可视化
    |
    2月前
    |
    数据采集 监控 数据库
    Python异步编程实战:爬虫案例
    🌟 蒋星熠Jaxonic,代码为舟的星际旅人。从回调地狱到async/await协程天堂,亲历Python异步编程演进。分享高性能爬虫、数据库异步操作、限流监控等实战经验,助你驾驭并发,在二进制星河中谱写极客诗篇。
    Python异步编程实战:爬虫案例
    |
    3月前
    |
    数据采集 存储 XML
    Python爬虫技术:从基础到实战的完整教程
    最后强调: 父母法律法规限制下进行网络抓取活动; 不得侵犯他人版权隐私利益; 同时也要注意个人安全防止泄露敏感信息.
    764 19
    |
    2月前
    |
    数据采集 存储 JSON
    Python爬虫常见陷阱:Ajax动态生成内容的URL去重与数据拼接
    Python爬虫常见陷阱:Ajax动态生成内容的URL去重与数据拼接
    |
    2月前
    |
    数据采集 存储 JavaScript
    解析Python爬虫中的Cookies和Session管理
    Cookies与Session是Python爬虫中实现状态保持的核心。Cookies由服务器发送、客户端存储,用于标识用户;Session则通过唯一ID在服务端记录会话信息。二者协同实现登录模拟与数据持久化。
    |
    8月前
    |
    数据采集 测试技术 C++
    无headers爬虫 vs 带headers爬虫:Python性能对比
    无headers爬虫 vs 带headers爬虫:Python性能对比
    |
    数据采集 存储 JSON
    Python网络爬虫:Scrapy框架的实战应用与技巧分享
    【10月更文挑战第27天】本文介绍了Python网络爬虫Scrapy框架的实战应用与技巧。首先讲解了如何创建Scrapy项目、定义爬虫、处理JSON响应、设置User-Agent和代理,以及存储爬取的数据。通过具体示例,帮助读者掌握Scrapy的核心功能和使用方法,提升数据采集效率。
    566 6
    |
    8月前
    |
    数据采集 存储 监控
    Python 原生爬虫教程:网络爬虫的基本概念和认知
    网络爬虫是一种自动抓取互联网信息的程序,广泛应用于搜索引擎、数据采集、新闻聚合和价格监控等领域。其工作流程包括 URL 调度、HTTP 请求、页面下载、解析、数据存储及新 URL 发现。Python 因其丰富的库(如 requests、BeautifulSoup、Scrapy)和简洁语法成为爬虫开发的首选语言。然而,在使用爬虫时需注意法律与道德问题,例如遵守 robots.txt 规则、控制请求频率以及合法使用数据,以确保爬虫技术健康有序发展。
    1259 31
    |
    7月前
    |
    数据采集 存储 NoSQL
    分布式爬虫去重:Python + Redis实现高效URL去重
    分布式爬虫去重:Python + Redis实现高效URL去重

    推荐镜像

    更多