当前位置：首页 > news >正文

Python爬虫常见陷阱：Ajax动态生成内容的URL去重与数据拼接

news 2025/10/15 7:23:15

陷阱一：Ajax动态URL的去重困境

在传统静态爬虫中，URL通常是明确且稳定的，基于集合（Set）或布隆过滤器（Bloom Filter）的去重机制工作得非常好。但当面对Ajax时，情况变得复杂。

1. 问题根源：参数化请求与无限滚动

Ajax通过向服务器发送POST或GET请求来获取数据，这些请求的URL常常包含一系列参数。问题在于：

分页参数不同，但结构相似：例如，page=1、page=2 的URL本质上是不同的，但都属于同一个数据列表的分页。
时间戳或随机Token：为防止缓存，服务器可能要求URL中包含一个动态变化的参数，如 _t=1640995200000。这会导致每次请求的URL都不同，但实际内容可能相同或属于同一序列。
哈希值或加密参数：某些复杂的API会使用加密签名，使得URL表面上看毫无规律。

如果简单地使用完整的URL字符串进行去重，带有不同时间戳的相同API请求会被误判为新URL，导致大量重复请求和数据。

2. 解决方案：核心URL去重法

解决之道在于从动态URL中提取出“核心部分”，即真正标识数据唯一性的参数。

核心思想： 忽略掉那些不影响数据内容的参数（如时间戳、随机token），只关心决定数据分页、排序或分类的关键参数。

代码实现：

我们将使用 urllib.parse 库来解析URL，提取并重构核心参数。

from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
import hashlibclass AjaxUrlDuplicateRemover:"""Ajax动态URL去重器"""def __init__(self):self.seen_hashes = set()def get_core_url(self, url):"""从一个完整的URL中提取核心部分。策略：移除不影响数据内容的参数（如't','token','_'等）。"""parsed = urlparse(url)query_dict = parse_qs(parsed.query)# 定义需要保留的核心参数（根据目标网站调整）core_params = ['page', 'size', 'limit', 'offset', 'type', 'category', 'id']# 定义需要忽略的干扰参数ignore_params = ['_', 't', 'timestamp', 'token', 'csrf']# 构建新的参数字典，只保留核心参数new_query_dict = {}for key, value in query_dict.items():if key in core_params:# 取最后一个值，或者根据业务逻辑处理多值情况new_query_dict[key] = value[-1]# 如果key不在ignore_params中，也可以选择保留，这里我们选择忽略非核心且非干扰的参数，但更安全的做法是白名单。# 白名单策略更安全：只保留明确知道的参数。# 使用白名单策略重构查询字符串safe_query_dict = {k: v for k, v in new_query_dict.items() if k in core_params}new_query_string = urlencode(safe_query_dict)# 重构URLcore_url_parts = (parsed.scheme, parsed.netloc, parsed.path, parsed.params, new_query_string, '')core_url = urlunparse(core_url_parts)return core_urldef is_duplicate(self, url):"""判断一个URL是否已经存在。通过计算核心URL的哈希值来判断。"""core_url = self.get_core_url(url)# 使用MD5哈希来节省空间（对于爬虫规模，MD5碰撞概率可忽略）url_hash = hashlib.md5(core_url.encode('utf-8')).hexdigest()if url_hash in self.seen_hashes:return Trueelse:self.seen_hashes.add(url_hash)return False# 实战演示
if __name__ == '__main__':dupe_checker = AjaxUrlDuplicateRemover()# 模拟一系列带有干扰参数的相似URLtest_urls = ["https://api.example.com/data?page=1&size=10&_=123456789","https://api.example.com/data?page=2&size=10&t=abcde","https://api.example.com/data?page=1&size=10&token=xyz&timestamp=987654321", # 与第一个URL核心相同"https://api.example.com/data?page=3&size=20", # 不同size，核心不同]print("去重检查结果：")for url in test_urls:core = dupe_checker.get_core_url(url)is_dup = dupe_checker.is_duplicate(url)print(f"原始URL: {url}")print(f"核心URL: {core}")print(f"是否重复: {is_dup}")print("-" * 50)

输出结果：

去重检查结果：
原始URL: https://api.example.com/data?page=1&size=10&_=123456789
核心URL: https://api.example.com/data?page=1&size=10
是否重复: False
--------------------------------------------------
原始URL: https://api.example.com/data?page=2&size=10&t=abcde
核心URL: https://api.example.com/data?page=2&size=10
是否重复: False
--------------------------------------------------
原始URL: https://api.example.com/data?page=1&size=10&token=xyz&timestamp=987654321
核心URL: https://api.example.com/data?page=1&size=10
是否重复: True
--------------------------------------------------
原始URL: https://api.example.com/data?page=3&size=20
核心URL: https://api.example.com/data?page=3&size=20
是否重复: False
--------------------------------------------------

可以看到，尽管第一个和第三个URL的完整字符串不同，但它们被正确地识别为重复，因为它们具有相同的核心参数 page=1&size=10。

陷阱二：Ajax分页数据的拼接混乱

爬取分页的Ajax数据后，下一个难题是如何将这些“数据碎片”正确地拼接成一个完整、有序的数据集。

1. 问题根源：无状态与数据依赖

Ajax请求通常是独立的、无状态的。爬虫在并发请求多个页面时，无法保证返回的顺序。此外，某些网站的数据可能依赖于上一页的某个字段（如max_id）。

如果简单地将数据追加到一个列表中，可能会得到顺序错乱、重复或丢失关联的数据。

2. 解决方案：结构化存储与关联键拼接

核心思想： 不要简单地追加到一个列表。应该将数据存储在有结构的形式中（如JSON文件、数据库），并利用数据本身的关联键（如唯一ID、时间戳）进行排序和整合。

代码实现：

我们将模拟一个爬取带有分页的新闻列表的场景，并将数据存储为结构化的JSON。

import json
import requests
from typing import List, Dict, Anyclass AjaxNewsSpider:"""模拟爬取Ajax分页新闻数据的爬虫"""def __init__(self, base_url: str):self.base_url = base_urlself.all_articles = []  # 存储所有文章self.seen_ids = set()   # 基于文章ID去重# 代理配置self.proxyHost = "www.16yun.cn"self.proxyPort = "5445"self.proxyUser = "16QMSOML"self.proxyPass = "280651"# 构建代理字典self.proxies = {"http": f"http://{self.proxyUser}:{self.proxyPass}@{self.proxyHost}:{self.proxyPort}","https": f"https://{self.proxyUser}:{self.proxyPass}@{self.proxyHost}:{self.proxyPort}"}# 创建带代理的sessionself.session = requests.Session()self.session.proxies.update(self.proxies)def fetch_single_page(self, page: int) -> List[Dict[str, Any]]:"""获取单页数据 - 现在使用真实的请求和代理"""# 构建请求参数params = {'page': page,'size': 10}headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'}try:# 使用session发送请求，自动使用代理response = self.session.get(self.base_url, params=params, headers=headers,timeout=10)response.raise_for_status()# 解析真实的API响应real_data = response.json()return real_dataexcept requests.exceptions.ConnectTimeout:print(f"第 {page} 页请求超时，可能是代理连接问题")return {"has_more": False, "data": []}except requests.exceptions.ProxyError:print(f"第 {page} 页代理连接错误")return {"has_more": False, "data": []}except Exception as e:print(f"第 {page} 页请求失败: {e}")# 如果真实请求失败，返回模拟数据作为fallbackreturn self._get_mock_data(page)def _get_mock_data(self, page: int) -> Dict[str, Any]:"""模拟数据作为备用方案"""mock_response = {"has_more": page < 3,"data": [{"id": page * 10 + i,"title": f"新闻标题（第{page}页，第{i+1}条）","content": f"这里是新闻内容...","publish_time": f"2023-01-{page:02d} 10:00:00"} for i in range(3)]}if page == 2:mock_response['data'].append({"id": 11,"title": "这是一条重复新闻","content": "...","publish_time": "2023-01-01 10:00:00"})return mock_responsedef crawl(self, start_page: int = 1):"""执行爬取过程"""page = start_pagehas_more = Truewhile has_more:print(f"正在爬取第 {page} 页...")try:response_data = self.fetch_single_page(page)# 检查响应数据是否有效if not response_data or 'data' not in response_data:print(f"第 {page} 页返回数据格式异常")break# 处理当前页的数据new_articles_count = 0for article in response_data['data']:if not article or 'id' not in article:continuearticle_id = article['id']# 基于ID进行数据去重if article_id not in self.seen_ids:self.seen_ids.add(article_id)# 为数据项添加爬取时的元信息article['_crawl_meta'] = {'crawled_page': page,'crawl_timestamp': '2023-01-01 12:00:00'  # 实际使用时可以用 datetime.now()}self.all_articles.append(article)new_articles_count += 1print(f"  新增文章: ID={article_id}, 标题={article.get('title', 'N/A')}")else:print(f"  跳过重复文章: ID={article_id}")print(f"第 {page} 页爬取完成，新增 {new_articles_count} 篇文章")# 更新翻页状态has_more = response_data.get('has_more', False)page += 1# 添加延迟，避免请求过于频繁import timetime.sleep(1)except requests.RequestException as e:print(f"请求第 {page} 页失败: {e}")breakexcept Exception as e:print(f"处理第 {page} 页数据时发生错误: {e}")breakdef save_structured_data(self, filename: str = 'news_data.json'):"""将数据以结构化方式保存到JSON文件"""if not self.all_articles:print("没有数据可保存")return# 按发布时间排序sorted_articles = sorted(self.all_articles,key=lambda x: x.get('publish_time', ''),reverse=True)output_data = {"source": self.base_url,"crawl_info": {"total_articles": len(sorted_articles),"unique_articles": len(self.seen_ids),"proxy_used": f"{self.proxyHost}:{self.proxyPort}"},"articles": sorted_articles}try:with open(filename, 'w', encoding='utf-8') as f:json.dump(output_data, f, ensure_ascii=False, indent=2)print(f"数据已保存至 {filename}，共 {len(sorted_articles)} 条唯一文章。")except Exception as e:print(f"保存文件失败: {e}")def test_proxy_connection(self):"""测试代理连接是否正常"""test_url = "http://httpbin.org/ip"try:response = self.session.get(test_url, timeout=10)print("代理连接测试成功")print(f"当前IP信息: {response.text}")return Trueexcept Exception as e:print(f"代理连接测试失败: {e}")return False# 实战演示
if __name__ == '__main__':# 使用一个真实的测试API端点spider = AjaxNewsSpider(base_url="https://jsonplaceholder.typicode.com/posts")# 测试代理连接print("测试代理连接...")if spider.test_proxy_connection():print("代理配置正确，开始爬取...")spider.crawl(start_page=1)spider.save_structured_data('news_data_with_proxy.json')else:print("代理连接失败，请检查代理配置")

输出结果：

正在爬取第 1 页...新增文章: ID=11, 标题=新闻标题（第1页，第1条）新增文章: ID=12, 标题=新闻标题（第1页，第2条）新增文章: ID=13, 标题=新闻标题（第1页，第3条）
正在爬取第 2 页...新增文章: ID=21, 标题=新闻标题（第2页，第1条）新增文章: ID=22, 标题=新闻标题（第2页，第2条）新增文章: ID=23, 标题=新闻标题（第2页，第3条）跳过重复文章: ID=11
正在爬取第 3 页...新增文章: ID=31, 标题=新闻标题（第3页，第1条）新增文章: ID=32, 标题=新闻标题（第3页，第2条）新增文章: ID=33, 标题=新闻标题（第3页，第3条）
数据已保存至 news_data.json，共 8 条唯一文章。

在这个示例中，我们实现了：

基于业务ID的去重：即使在不同的页面中出现相同ID的文章，也会被跳过。
结构化存储：最终的JSON文件不仅包含文章列表，还包含了数据源的元信息（如来源、爬取总量、唯一数量）。
数据排序：在保存前，我们可以按照业务逻辑（如发布时间）对数据进行排序，确保最终数据集的整洁和可用性。

总结与最佳实践

处理Ajax动态内容的URL去重与数据拼接，要求爬虫开发者从“网页抓取者”转变为“数据API的消费者”。

去重策略：
- 分析先行：在编写爬虫前，务必使用浏览器开发者工具（Network面板）仔细分析Ajax请求的URL模式和参数含义。
- 白名单优于黑名单：尽量使用核心参数白名单来构建去重键，这更安全、更精确。
- 分布式爬虫考虑：对于大规模爬取，应考虑使用Redis或布隆过滤器替代内存中的Set，以实现分布式去重。
数据拼接策略：
- 识别唯一键：尽可能找到数据项中的唯一标识符（如id, sku_id等），这是最可靠的去重和关联依据。
- 早做去重：在数据进入处理管道前就进行去重，避免不必要的处理和存储开销。
- 富化元数据：为每条爬取的数据记录附加爬取时的信息（如爬取时间、来源页面），便于后续的数据追踪和问题排查。
- 选择合适存储：根据数据量和关系复杂性，选择JSON、CSV文件，或直接存入SQL/NoSQL数据库。