Python爬虫常见陷阱：Ajax动态生成内容的URL去重与数据拼接
原创
Python爬虫常见陷阱：Ajax动态生成内容的URL去重与数据拼接
原创
小白学大数据
发布于 2025-10-14 16:43:44
发布于 2025-10-14 16:43:44
代码可运行
运行总次数：0
代码可运行
陷阱一：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 hashlib
class 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)
# 重构URL
core_url_parts = (parsed.scheme, parsed.netloc, parsed.path, parsed.params, new_query_string, '')
core_url = urlunparse(core_url_parts)
return core_url
def 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 True
else:
self.seen_hashes.add(url_hash)
return False
# 实战演示
if __name__ == '__main__':
dupe_checker = AjaxUrlDuplicateRemover()
# 模拟一系列带有干扰参数的相似URL
test_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×tamp=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×tamp=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, Any
class AjaxNewsSpider:
"""
模拟爬取Ajax分页新闻数据的爬虫
"""
def __init__(self, base_url: str):
self.base_url = base_url
self.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}"
}
# 创建带代理的session
self.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_data
except 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}")
# 如果真实请求失败，返回模拟数据作为fallback
return 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_response
def crawl(self, start_page: int = 1):
"""
执行爬取过程
"""
page = start_page
has_more = True
while 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 = 0
for article in response_data['data']:
if not article or 'id' not in article:
continue
article_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 += 1
print(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 time
time.sleep(1)
except requests.RequestException as e:
print(f"请求第 {page} 页失败: {e}")
break
except Exception as e:
print(f"处理第 {page} 页数据时发生错误: {e}")
break
def 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 True
except 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数据库。
- 识别唯一键：尽可能找到数据项中的唯一标识符（如
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
评论
登录后参与评论
推荐阅读
目录
腾讯云开发者
