网页内容提取与去重实战：从精准抓取到重复数据清零的全技巧

在网页数据采集场景中，“提取不精准”和“重复数据泛滥”是两个高频痛点——要么抓回一堆广告、导航等冗余内容，要么同一篇文章、同一个商品在数据集中反复出现，既浪费存储又影响后续分析。本文结合10万+网页采集实战经验，从精准提取（结构化/非结构化内容） 和分层去重（URL/内容/相似性） 两大核心维度，拆解可落地的工具、代码和避坑方案，帮你实现“抓得准、去得净”的采集目标。

一、网页内容提取：从“粗放抓取”到“精准定位”

网页内容提取的核心是“剥离噪音，保留核心”。根据内容结构，可分为“结构化内容”（如商品价格、新闻发布时间）和“非结构化内容”（如文章正文、评论），两者提取策略不同。

1.1 结构化内容提取：精准定位关键信息

结构化内容通常有明确的标签、类名或属性，适合用XPath/CSS选择器直接定位，关键是“选择器要健壮”，避免网站结构微调就失效。

1.1.1 核心工具：lxml（高效）vs BeautifulSoup（灵活）

lxml：基于C语言，解析速度快，支持XPath 1.0，适合大规模采集；BeautifulSoup：纯Python实现，API友好，适合小规模、快速开发。

1.1.2 实战技巧：写“抗变化”的选择器

避免依赖易变的div[class="col-3"]这类动态类名，优先用以下方式：

基于唯一ID：如//div[@id="product-price"]（ID通常全局唯一，极少变化）；基于内容特征：如提取“价格”时，先找包含“¥”的标签：//span[contains(text(), "¥")]/text()；基于父节点锚定：先定位稳定的父容器（如//div[@class="product-info"]），再找子节点：//div[@class="product-info"]//span[@class="price"]。

1.1.3 代码示例：提取电商商品结构化信息


from lxml import etree
import requests

def extract_product_structured(url):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
    }
    response = requests.get(url, headers=headers, timeout=10)
    html = etree.HTML(response.text)
    
    # 精准提取商品核心信息（结构化）
    product = {
        # 基于ID定位标题（稳定）
        "title": html.xpath('//h1[@id="product-title"]/text()')[0].strip() if html.xpath('//h1[@id="product-title"]/text()') else "",
        # 基于内容特征+父节点定位价格（抗变化）
        "price": html.xpath('//div[@class="price-container"]//span[contains(text(), "¥")]/text()')[0].replace("¥", "").strip() if html.xpath('//div[@class="price-container"]//span[contains(text(), "¥")]') else "",
        # 基于属性定位库存（灵活）
        "stock": html.xpath('//span[@data-key="stock"]/text()')[0].strip() if html.xpath('//span[@data-key="stock"]/text()') else "0",
        # 基于层级关系定位分类（避免抓错）
        "category": html.xpath('//div[@class="breadcrumb"]/a[last()]/text()')[0].strip() if html.xpath('//div[@class="breadcrumb"]/a[last()]') else ""
    }
    return product

# 测试：提取某电商商品信息
if __name__ == "__main__":
    product_url = "https://example.com/product/123"
    result = extract_product_structured(product_url)
    print("结构化提取结果：")
    for key, value in result.items():
        print(f"{key}: {value}")

1.2 非结构化内容提取：自动剥离噪音（正文/评论）

非结构化内容（如新闻正文、博客文章）通常夹杂广告、导航、推荐阅读等噪音，手动写选择器效率低且易失效，推荐用自动正文提取库，基于文本密度、标签层级等特征智能识别核心内容。

1.2.1 实战工具对比：newspaper3k vs goose3

工具	优势	劣势	适用场景
newspaper3k	开源免费、支持多语言、自动提取标题/正文/作者	对复杂HTML（如动态渲染）支持弱	静态新闻、博客页面
goose3	提取精度高、支持自定义规则、噪音过滤强	安装稍复杂、依赖较多	高质量正文提取（如学术文章）

1.2.2 代码示例：用newspaper3k自动提取新闻正文


from newspaper import Article
import requests

def extract_news_content(url):
    # 先获取页面HTML（避免newspaper3k的UA被封）
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
    }
    response = requests.get(url, headers=headers, timeout=10)
    response.encoding = response.apparent_encoding  # 自动识别编码
    
    # 用newspaper3k解析正文
    article = Article(url)
    article.set_html(response.text)  # 传入已获取的HTML，避免重复请求
    article.parse()  # 解析正文、标题、作者等
    
    return {
        "title": article.title,  # 自动提取标题
        "content": article.text,  # 自动提取正文（去广告、导航）
        "author": article.authors,  # 自动提取作者
        "publish_date": article.publish_date,  # 自动提取发布时间
        "keywords": article.keywords  # 自动提取关键词
    }

# 测试：提取新闻正文
if __name__ == "__main__":
    news_url = "https://example.com/news/456"
    result = extract_news_content(news_url)
    print(f"新闻标题：{result['title']}")
    print(f"发布时间：{result['publish_date']}")
    print(f"正文前500字：{result['content'][:500]}...")

1.2.3 进阶：动态内容提取（JavaScript渲染页面）

若页面用React/Vue动态渲染（如滚动加载正文），需先渲染页面再提取，推荐用Playwright（比Selenium更轻量、反爬友好）：


from playwright.sync_api import sync_playwright
from newspaper import Article

def extract_dynamic_content(url):
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)  # 无头模式
        page = browser.new_page()
        page.goto(url, wait_until="networkidle")  # 等待页面加载完成
        # 模拟滚动（加载更多正文内容）
        page.mouse.wheel(0, 2000)
        page.wait_for_timeout(1000)
        # 获取渲染后的HTML
        html = page.content()
        browser.close()
    
    # 用newspaper3k提取正文
    article = Article(url)
    article.set_html(html)
    article.parse()
    return article.text

# 测试：提取动态渲染的文章正文
if __name__ == "__main__":
    dynamic_url = "https://example.com/dynamic-article/789"
    content = extract_dynamic_content(dynamic_url)
    print(f"动态页面正文前500字：{content[:500]}...")

1.3 提取后清洗：统一格式，去除冗余

提取到的内容可能包含HTML标签、特殊字符、多余空格，需进一步清洗：


import re

def clean_extracted_content(content):
    if not content:
        return ""
    # 1. 去除HTML标签（若有残留）
    content = re.sub(r"<[^>]+>", "", content)
    # 2. 去除多余空格、换行符
    content = re.sub(r"s+", " ", content).strip()
    # 3. 去除特殊字符（如emoji、不可见字符）
    content = re.sub(r"[^u4e00-u9fa5a-zA-Z0-9.,!?；。，！？]", "", content)
    return content

# 测试清洗
dirty_content = "<p>  这是一篇测试文章！
包含HTML标签和多余空格  </p>"
clean_content = clean_extracted_content(dirty_content)
print(f"清洗后：{clean_content}")  # 输出：这是一篇测试文章！包含HTML标签和多余空格

二、网页内容去重：从“完全重复”到“相似去重”

去重需分“分层处理”，从简单到复杂，兼顾效率和精度。核心逻辑是：先过滤完全重复（高效），再处理相似重复（精准）。

2.1 第一层：URL去重（抓取前过滤，最高效）

URL是最直观的去重依据，同一URL通常对应同一内容，抓取前先判断URL是否已爬过，避免无效请求。

2.1.1 工具选择：本地集合（小规模）vs Redis（大规模/分布式）

本地集合：适合万级URL，用Python的set存储已爬URL，判断url in set即可；Redis：适合百万/亿级URL或分布式爬虫，用Redis Set存储，支持多节点共享。

2.1.2 代码示例：Redis分布式URL去重


import redis

class URLDupeFilter:
    def __init__(self, redis_url="redis://localhost:6379/0", key="crawler:urls"):
        self.redis_client = redis.Redis.from_url(redis_url)
        self.key = key  # Redis Set的键名

    def is_duplicate(self, url):
        """判断URL是否已爬过：已存在返回True，不存在返回False并加入Set"""
        # Redis Set的sadd操作：若元素不存在则添加，返回1；存在则不添加，返回0
        return not self.redis_client.sadd(self.key, url)

    def get_dupe_count(self):
        """获取已去重的URL总数"""
        return self.redis_client.scard(self.key)

# 测试：URL去重
if __name__ == "__main__":
    dupe_filter = URLDupeFilter()
    test_urls = [
        "https://example.com/news/1",
        "https://example.com/news/1",  # 重复URL
        "https://example.com/news/2"
    ]
    for url in test_urls:
        if dupe_filter.is_duplicate(url):
            print(f"URL已重复，跳过：{url}")
        else:
            print(f"URL新，可抓取：{url}")
    print(f"已去重URL总数：{dupe_filter.get_dupe_count()}")  # 输出：2

2.1.3 进阶：URL归一化（避免“同内容不同URL”误判）

同一内容可能对应不同URL（如参数顺序变化、带无关参数），需先“归一化”处理：


from urllib.parse import urlparse, parse_qs, urlunparse

def normalize_url(url):
    """URL归一化：去除无关参数、统一参数顺序、小写域名"""
    parsed = urlparse(url)
    # 1. 小写域名（如Example.com → example.com）
    netloc = parsed.netloc.lower()
    # 2. 去除无关参数（如utm_source、session_id）
    query_params = parse_qs(parsed.query)
    irrelevant_params = ["utm_source", "utm_medium", "utm_campaign", "session_id", "timestamp"]
    filtered_params = {k: v for k, v in query_params.items() if k not in irrelevant_params}
    # 3. 统一参数顺序（按键名排序）
    sorted_query = "&".join([f"{k}={v[0]}" for k, v in sorted(filtered_params.items())])
    # 4. 重组URL
    normalized = urlunparse((
        parsed.scheme, netloc, parsed.path, parsed.params, sorted_query, parsed.fragment
    ))
    return normalized

# 测试：归一化不同参数顺序的URL
url1 = "https://Example.com/news?id=1&utm_source=wechat"
url2 = "https://example.com/news?utm_source=baidu&id=1"
print(normalize_url(url1))  # 输出：https://example.com/news?id=1
print(normalize_url(url2))  # 输出：https://example.com/news?id=1（与url1归一化后相同）

2.2 第二层：内容完全重复去重（抓取后过滤）

即使URL不同，也可能存在完全相同的内容（如镜像网站、转载未修改），此时需基于内容哈希去重。

2.2.1 核心方法：MD5/SHA-1哈希

将清洗后的内容（如正文）计算哈希值，哈希相同则视为完全重复，效率高（计算哈希比对比全文快100倍+）。

2.2.2 代码示例：内容哈希去重


import hashlib
import redis

class ContentDupeFilter:
    def __init__(self, redis_url="redis://localhost:6379/0", key="crawler:content_hashes"):
        self.redis_client = redis.Redis.from_url(redis_url)
        self.key = key

    def calculate_content_hash(self, content):
        """计算内容的MD5哈希（内容过长时取前1000字+后1000字，避免计算量过大）"""
        if len(content) > 2000:
            content = content[:1000] + content[-1000:]  # 截取首尾，保留核心特征
        return hashlib.md5(content.encode("utf-8")).hexdigest()

    def is_duplicate(self, content):
        """判断内容是否完全重复"""
        content_hash = self.calculate_content_hash(content)
        return not self.redis_client.sadd(self.key, content_hash)

# 测试：内容完全重复去重
if __name__ == "__main__":
    content_filter = ContentDupeFilter()
    content1 = "这是一篇完全相同的文章内容，可能来自不同URL。"
    content2 = "这是一篇完全相同的文章内容，可能来自不同URL。"  # 完全重复
    content3 = "这是一篇不同的文章内容。"
    
    for i, content in enumerate([content1, content2, content3]):
        if content_filter.is_duplicate(content):
            print(f"内容{i+1}完全重复，过滤")
        else:
            print(f"内容{i+1}为新内容，保留")

2.3 第三层：相似内容去重（解决“改标题、改分段”问题）

最棘手的是“相似重复”——如同一篇文章改了标题、调整了段落顺序、添加了少量注释，此时哈希去重失效，需用文本相似度算法或指纹算法。

2.3.1 核心算法对比：SimHash（高效）vs 余弦相似度（精准）

算法	原理	优势	劣势	适用场景
SimHash	将文本转为64/128位指纹，汉明距离≤3视为相似	计算快、内存占用小，支持百万级数据	对短文本（<100字）精度低	长文本（文章、报告）去重
余弦相似度	基于TF-IDF将文本转为向量，计算向量夹角	精度高，支持短文本	计算慢、内存占用大，不适合亿级数据	短文本（评论、商品描述）去重

2.3.2 实战：用SimHash实现长文本相似去重


import jieba
import numpy as np
from redis import Redis

class SimHashDupeFilter:
    def __init__(self, redis_url="redis://localhost:6379/0", key="crawler:simhashes", hash_bits=64):
        self.redis_client = Redis.from_url(redis_url)
        self.key = key
        self.hash_bits = hash_bits  # SimHash位数（64/128，位数越高精度越高）

    def _tokenize(self, content):
        """中文分词（用jieba），过滤停用词"""
        stop_words = {"的", "了", "是", "在", "和", "及", "等"}  # 简单停用词表
        words = jieba.lcut(content)
        return [word for word in words if word not in stop_words and len(word) > 1]

    def calculate_simhash(self, content):
        """计算文本的SimHash指纹"""
        if len(content) < 100:
            return None  # 短文本不适合SimHash
        
        # 1. 分词并计算词频（简化版：用出现次数作为权重）
        words = self._tokenize(content)
        word_freq = {}
        for word in words:
            word_freq[word] = word_freq.get(word, 0) + 1
        
        # 2. 初始化SimHash向量
        simhash_vec = np.zeros(self.hash_bits, dtype=int)
        for word, freq in word_freq.items():
            # 3. 计算每个词的哈希（MD5）
            word_hash = hashlib.md5(word.encode("utf-8")).hexdigest()
            # 4. 生成词的二进制向量（根据哈希值，1取1，0取-1）
            word_vec = []
            for c in word_hash[:self.hash_bits//4]:  # MD5每字符4位，取前hash_bits//4个字符
                bin_str = bin(int(c, 16))[2:].zfill(4)  # 转为4位二进制
                for bit in bin_str:
                    word_vec.append(1 if bit == "1" else -1)
            # 5. 按词频加权，更新SimHash向量
            for i in range(self.hash_bits):
                simhash_vec[i] += word_vec[i] * freq
        
        # 6. 生成最终SimHash指纹（向量>0取1，否则取0）
        simhash = 0
        for i in range(self.hash_bits):
            if simhash_vec[i] > 0:
                simhash |= 1 << (self.hash_bits - 1 - i)
        return simhash

    def hamming_distance(self, hash1, hash2):
        """计算两个SimHash的汉明距离（不同位的数量）"""
        xor = hash1 ^ hash2
        return bin(xor).count("1")

    def is_similar(self, content, threshold=3):
        """判断内容是否相似：汉明距离≤threshold视为相似"""
        current_simhash = self.calculate_simhash(content)
        if not current_simhash:
            return False  # 短文本跳过相似判断
        
        # 1. 遍历已存储的SimHash，计算汉明距离
        stored_simhashes = self.redis_client.smembers(self.key)
        for stored_hash in stored_simhashes:
            stored_simhash = int(stored_hash.decode("utf-8"))
            distance = self.hamming_distance(current_simhash, stored_simhash)
            if distance <= threshold:
                return True  # 相似，返回重复
        
        # 2. 新内容，存储SimHash
        self.redis_client.sadd(self.key, current_simhash)
        return False

# 测试：相似内容去重（改标题、调整分段）
if __name__ == "__main__":
    simhash_filter = SimHashDupeFilter()
    # 相似内容1：原内容
    content1 = """
    人工智能（AI）是当前科技领域的热点。随着算力的提升和算法的优化，AI在医疗、教育、工业等领域的应用越来越广泛。
    例如，在医疗领域，AI可以辅助医生进行疾病诊断，提高诊断效率；在教育领域，AI可以实现个性化教学，满足不同学生的需求。
    """
    # 相似内容2：改标题+调整分段
    content2 = """
    科技热点：人工智能的广泛应用
    人工智能（AI）作为当前科技领域的热点，其发展离不开算力的提升和算法的优化。在多个领域，AI都发挥着重要作用：
    医疗领域中，AI辅助医生诊断疾病，提升诊断效率；教育领域中，AI实现个性化教学，满足不同学生需求。
    """
    # 不同内容
    content3 = """
    区块链技术是一种去中心化的分布式账本技术，具有不可篡改、透明等特点。其在金融、供应链管理等领域有重要应用，
    例如，在金融领域，区块链可以用于跨境支付，降低支付成本和时间；在供应链管理中，区块链可以实现商品溯源，确保商品质量。
    """
    
    for i, content in enumerate([content1, content2, content3]):
        if simhash_filter.is_similar(content):
            print(f"内容{i+1}与已有内容相似，过滤")
        else:
            print(f"内容{i+1}为新内容，保留")
    # 输出：内容1保留，内容2过滤（相似），内容3保留

2.3.3 进阶：大规模相似去重优化

当数据量超过100万时，遍历所有SimHash计算汉明距离会变慢，可通过“分桶策略”优化：

将64位SimHash分成4段，每段16位；新SimHash计算后，将4段中的3段作为“桶键”，存入Redis的Hash结构；相似的SimHash至少有3段相同，只需在对应桶中查找，减少对比次数。

三、完整流程：从抓取到去重的闭环

将“内容提取”和“去重”结合，形成完整的采集流程：


def crawl_and_deduplicate(url):
    # 1. URL归一化
    normalized_url = normalize_url(url)
    
    # 2. URL去重（抓取前）
    url_filter = URLDupeFilter()
    if url_filter.is_duplicate(normalized_url):
        print(f"URL已爬过，跳过：{url}")
        return
    
    # 3. 抓取页面并提取内容
    try:
        content = extract_news_content(url)["content"]  # 提取正文
        clean_content = clean_extracted_content(content)  # 清洗内容
    except Exception as e:
        print(f"抓取/提取失败：{e}")
        return
    
    # 4. 内容完全重复去重（抓取后）
    content_filter = ContentDupeFilter()
    if content_filter.is_duplicate(clean_content):
        print(f"内容完全重复，过滤：{url}")
        return
    
    # 5. 相似内容去重（抓取后）
    simhash_filter = SimHashDupeFilter()
    if simhash_filter.is_similar(clean_content):
        print(f"内容相似，过滤：{url}")
        return
    
    # 6. 存储有效内容（此处省略存储逻辑，如存入MongoDB）
    print(f"内容有效，存储：{url}")
    # save_to_database(clean_content)

# 测试完整流程
if __name__ == "__main__":
    test_urls = [
        "https://example.com/news/1?utm_source=wechat",
        "https://example.com/news/1?utm_source=baidu",  # URL归一化后重复
        "https://example.com/news/1-copy",  # 内容完全重复
        "https://example.com/news/1-similar",  # 内容相似
        "https://example.com/news/2"  # 新内容
    ]
    for url in test_urls:
        crawl_and_deduplicate(url)

四、避坑指南：10个高频问题解决方案

提取正文时抓到广告→ 用newspaper3k/goose3自动过滤，或手动添加广告标签黑名单（如//div[contains(@class, "ad-")]）；XPath选择器频繁失效→ 避免依赖动态类名，用ID、数据属性（data-*）或内容特征定位；动态内容提取为空→ 用Playwright等待networkidle或domcontentloaded，必要时模拟滚动；URL归一化不彻底→ 除了过滤参数，还要处理域名大小写、结尾斜杠（如/news和/news/）；SimHash对短文本精度低→ 短文本改用“编辑距离”或“余弦相似度”，如difflib.SequenceMatcher；Redis去重内存溢出→ 完全重复用Redis Set，相似去重用Redis Hash分桶，或用布隆过滤器（redisbloom）；中文分词效果差→ 用jieba加载自定义词典（如行业术语），提升分词精度；内容哈希计算慢→ 长文本截取首尾各1000字计算哈希，避免全量计算；相似去重阈值难确定→ 测试不同阈值（汉明距离2-4），根据业务需求调整（精度优先用2，召回优先用4）；分布式去重数据不一致→ 用Redis集群存储去重数据，确保所有爬虫节点连接同一Redis实例。

五、总结与进阶方向

网页内容提取与去重的核心是“分层处理，精准匹配”：提取时区分结构化/非结构化内容，用合适的工具提高精度；去重时分URL、完全重复、相似重复三层，兼顾效率和效果。

进阶方向：

多模态内容提取：结合OCR（如ddddocr）提取图片中的文字，避免图文混排内容丢失；智能去重模型：用BERT等预训练模型生成文本嵌入（Embedding），替代SimHash，提升相似去重精度；实时去重服务：将去重逻辑封装为API服务，支持多爬虫节点调用，统一管理去重数据；内容质量过滤：在去重基础上，添加质量评分（如正文长度、关键词密度），过滤低质量内容。

如果在实战中遇到具体问题（如动态内容提取、相似去重阈值调整、大规模去重优化），欢迎留言交流！