StructBERT模型Python爬虫数据清洗实战新闻内容聚合与去重你是不是也遇到过这样的烦恼用Python爬虫从各大新闻网站抓取信息辛辛苦苦攒了一大堆数据结果发现很多新闻内容其实大同小异只是标题换了个说法。手动筛选眼睛都要看花了。用简单的标题字符串匹配去重又很容易把真正不同的新闻给误删了。今天我们就来聊聊怎么用更聪明的方法解决这个问题。不靠简单的关键词而是让AI模型去“理解”新闻内容的真正含义从而实现精准的语义去重。我会带你走一遍完整的流程从用Python爬虫抓取原始数据到调用StructBERT模型进行深度语义分析最后得到一个干净、无重复的新闻内容池。整个过程都有可运行的代码你可以直接拿去用。1. 为什么简单的去重方法不够用在开始动手之前我们先搞清楚为什么传统的去重方法在新闻聚合场景下会“失灵”。想象一下你爬取了关于“某科技公司发布新产品”的新闻。A网站标题是“XX公司重磅推出新一代智能设备”B网站标题是“智能硬件领域迎来革新XX公司新品亮相”。如果只用字符串匹配这两个标题几乎没有重合的词会被当成两条不同的新闻。但实际上它们讲的是同一件事。这就是语义重复。它发生在两篇报道核心事实相同但表述方式、侧重点或行文结构不同的情况下。传统的解决方法比如计算标题的Jaccard相似度、或者用TF-IDF加余弦相似度主要依赖词汇的统计特征很难捕捉到这种深层的语义一致性。而我们要用的StructBERT模型是一种经过特殊训练的预训练语言模型。它不仅能理解单个词的意思还能理解词与词之间的结构关系比如主谓宾从而更好地把握句子的整体语义。用这个模型来计算新闻内容之间的相似度就像是请了一个专业的编辑来帮你判断两篇文章是不是在说同一件事准确度会高得多。2. 实战第一步用Python爬虫构建原始新闻池我们的目标是处理多源数据所以第一步就是搭建一个简单的、可扩展的多源新闻爬虫。这里我们以两个假设的新闻网站为例演示如何抓取并初步清洗数据。2.1 环境准备与基础爬虫搭建首先确保你的Python环境里安装了必要的库。打开终端执行以下命令pip install requests beautifulsoup4 pandas接下来我们写一个基础爬虫类。这个类的设计考虑了扩展性方便你后续添加更多的新闻源。import requests from bs4 import BeautifulSoup import pandas as pd import time import logging from typing import List, Dict, Optional logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) class BaseNewsCrawler: 新闻爬虫基类定义通用接口 def __init__(self, source_name: str, base_url: str): self.source_name source_name self.base_url base_url self.session requests.Session() self.session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 }) def fetch_page(self, url: str) - Optional[str]: 获取页面HTML try: resp self.session.get(url, timeout10) resp.raise_for_status() return resp.text except requests.RequestException as e: logging.error(f抓取 {url} 失败: {e}) return None def parse_list_page(self, html: str) - List[Dict]: 解析列表页提取文章链接和基本信息。子类必须实现。 raise NotImplementedError def parse_detail_page(self, html: str, link: str) - Optional[Dict]: 解析详情页提取文章标题和正文。子类必须实现。 raise NotImplementedError def crawl(self, list_url: str, max_articles: int 20) - List[Dict]: 执行爬取流程 articles [] html self.fetch_page(list_url) if not html: return articles links_info self.parse_list_page(html) logging.info(f从 {self.source_name} 列表页解析到 {len(links_info)} 个链接) for info in links_info[:max_articles]: detail_html self.fetch_page(info[url]) if not detail_html: continue article self.parse_detail_page(detail_html, info[url]) if article: article.update({source: self.source_name, list_url: list_url}) articles.append(article) time.sleep(1) # 礼貌性延迟避免给服务器造成压力 logging.info(f从 {self.source_name} 成功抓取 {len(articles)} 篇文章) return articles2.2 实现两个示例新闻源爬虫由于真实新闻网站结构复杂且经常变动这里我们创建两个模拟的爬虫类来演示如何适配不同网站结构。在实际应用中你需要根据目标网站的具体HTML结构来调整parse_list_page和parse_detail_page方法。class TechNewsSimulator(BaseNewsCrawler): 模拟科技新闻网站爬虫 def parse_list_page(self, html: str) - List[Dict]: soup BeautifulSoup(html, html.parser) # 假设列表页中文章链接在带有 news-item 类的div中的a标签里 articles [] for item in soup.find_all(div, class_news-item): link_tag item.find(a, hrefTrue) if link_tag: title link_tag.get_text(stripTrue) url link_tag[href] if not url.startswith(http): url self.base_url url articles.append({title: title, url: url}) return articles def parse_detail_page(self, html: str, link: str) - Optional[Dict]: soup BeautifulSoup(html, html.parser) # 假设标题在 h1 标签中正文在 article 标签中 title_tag soup.find(h1) content_tag soup.find(article) if not title_tag or not content_tag: return None title title_tag.get_text(stripTrue) # 清理正文移除脚本、样式等无关标签 for elem in content_tag([script, style, nav, footer]): elem.decompose() content content_tag.get_text(separator , stripTrue) return { title: title, content: content, url: link, crawl_time: pd.Timestamp.now() } class FinanceNewsSimulator(BaseNewsCrawler): 模拟财经新闻网站爬虫 def parse_list_page(self, html: str) - List[Dict]: soup BeautifulSoup(html, html.parser) articles [] # 假设财经网站结构不同链接在 li 标签中 for li in soup.find_all(li, class_report): link_tag li.find(a, hrefTrue) if link_tag: title link_tag.get_text(stripTrue) url link_tag[href] articles.append({title: title, url: url}) return articles def parse_detail_page(self, html: str, link: str) - Optional[Dict]: soup BeautifulSoup(html, html.parser) # 财经网站可能用不同的标签 title_tag soup.find(div, class_report-title) content_tag soup.find(div, class_report-content) if not title_tag or not content_tag: return None title title_tag.get_text(stripTrue) content content_tag.get_text(separator , stripTrue) return { title: title, content: content, url: link, crawl_time: pd.Timestamp.now() }2.3 执行爬取并保存原始数据现在让我们运行爬虫把数据抓取下来并保存到本地作为我们后续处理的原料。def run_crawlers_and_save(): 运行所有爬虫并保存结果 all_articles [] # 实例化爬虫这里用模拟的URL和源 tech_crawler TechNewsSimulator(source_nameTechSim, base_urlhttps://sim-tech-news.com) finance_crawler FinanceNewsSimulator(source_nameFinanceSim, base_urlhttps://sim-finance-news.com) # 模拟爬取实际使用时替换为真实的列表页URL tech_articles tech_crawler.crawl(https://sim-tech-news.com/latest, max_articles10) finance_articles finance_crawler.crawl(https://sim-finance-news.com/reports, max_articles10) all_articles.extend(tech_articles) all_articles.extend(finance_articles) # 转换为DataFrame并保存 df_raw pd.DataFrame(all_articles) if not df_raw.empty: df_raw.to_csv(raw_news_articles.csv, indexFalse, encodingutf-8-sig) logging.info(f原始数据已保存至 raw_news_articles.csv共 {len(df_raw)} 条记录。) print(df_raw[[source, title]].head()) # 预览一下 else: logging.warning(未抓取到任何文章。) return df_raw # 执行爬取 raw_news_df run_crawlers_and_save()跑完这段代码你会得到一个名为raw_news_articles.csv的文件里面包含了从不同网站抓来的新闻标题、正文、来源和抓取时间。数据可能很杂乱还有很多重复这就是我们接下来要处理的。3. 核心环节利用StructBERT模型进行语义去重原始数据准备好了重头戏来了。我们将使用StructBERT模型来计算新闻内容之间的语义相似度。这里我们假设通过一个API服务来调用该模型这比本地部署大型模型要方便得多。3.1 调用StructBERT语义相似度API首先你需要准备一个能够提供句子向量编码或相似度计算的服务。许多云服务商或开源项目都提供类似的API。下面的代码展示了一个通用的调用流程。import requests import numpy as np from typing import List import logging class StructBERTClient: StructBERT 语义相似度计算客户端示例 def __init__(self, api_endpoint: str, api_key: str None): 初始化客户端。 :param api_endpoint: 模型API的服务地址例如 https://your-api.com/encode :param api_key: 可选的API密钥 self.api_endpoint api_endpoint self.api_key api_key self.session requests.Session() if api_key: self.session.headers.update({Authorization: fBearer {api_key}}) def get_embedding(self, text: str) - Optional[np.ndarray]: 获取单个文本的语义向量嵌入。 向量可以代表这句话的深层含义。 payload {text: text} try: resp self.session.post(self.api_endpoint, jsonpayload, timeout30) resp.raise_for_status() result resp.json() # 假设API返回格式为 {embedding: [0.1, 0.2, ...]} embedding_list result.get(embedding) if embedding_list: return np.array(embedding_list) else: logging.error(fAPI响应中未找到 embedding 字段: {result}) return None except Exception as e: logging.error(f获取文本向量失败: {e}) return None def batch_get_embeddings(self, texts: List[str]) - List[Optional[np.ndarray]]: 批量获取文本向量效率更高 embeddings [] for text in texts: emb self.get_embedding(text) embeddings.append(emb) time.sleep(0.1) # 轻微延迟避免请求过快 return embeddings def calculate_similarity(self, vec1: np.ndarray, vec2: np.ndarray) - float: 计算两个向量之间的余弦相似度。 值越接近1表示语义越相似。 # 归一化向量 norm1 np.linalg.norm(vec1) norm2 np.linalg.norm(vec2) if norm1 0 or norm2 0: return 0.0 return np.dot(vec1, vec2) / (norm1 * norm2) # 示例初始化客户端请替换为你的实际API信息 # client StructBERTClient(api_endpointYOUR_MODEL_API_ENDPOINT, api_keyYOUR_API_KEY)重要提示在实际操作中你需要将YOUR_MODEL_API_ENDPOINT替换成真实的、可用的服务地址。你可以选择使用各大云平台提供的自然语言处理服务。自行部署开源的Sentence-BERT或类似模型的服务。使用其他提供句子向量功能的API。3.2 设计去重算法从语义向量到去重决策有了获取语义向量的能力我们就可以设计去重逻辑了。思路是为每篇新闻计算一个“语义指纹”即向量然后比较指纹的相似度。from sklearn.metrics.pairwise import cosine_similarity import hashlib class SemanticDeduplicator: 基于语义向量的新闻去重器 def __init__(self, similarity_threshold: float 0.85): :param similarity_threshold: 语义相似度阈值高于此值则认为内容重复。 通常设置在0.8-0.9之间可根据需求调整。 self.threshold similarity_threshold self.client None # 在实际使用时注入StructBERTClient实例 def set_client(self, client: StructBERTClient): self.client client def _generate_text_signature(self, title: str, content: str) - str: 生成一个用于快速初步筛选的文本签名。 结合标题和正文的前N个字符用于排除明显不同的文章。 # 取标题和正文前200字符可根据需要调整 content_preview content[:200] if len(content) 200 else content combined f{title}||{content_preview} # 用MD5生成一个固定长度的签名 return hashlib.md5(combined.encode(utf-8)).hexdigest() def deduplicate(self, articles_df: pd.DataFrame) - pd.DataFrame: 主去重函数。 返回一个去重后的DataFrame并保留一篇作为代表。 if self.client is None: raise ValueError(请先通过 set_client() 方法设置 StructBERTClient。) df articles_df.copy() logging.info(f开始对 {len(df)} 篇文章进行语义去重...) # 第一步生成快速签名进行粗略分组可选可加速处理 df[fast_signature] df.apply(lambda row: self._generate_text_signature(row[title], row[content]), axis1) # 注意此步骤仅为加速不能作为最终去重依据因为签名可能碰撞。 # 第二步获取每篇文章的语义向量 # 我们结合标题和正文的前512字符模型通常有长度限制作为输入文本 df[text_for_embedding] df.apply( lambda row: f{row[title]}。{row[content][:500]}, axis1 ) texts df[text_for_embedding].tolist() logging.info(正在获取文本语义向量...) embeddings self.client.batch_get_embeddings(texts) # 检查并处理获取失败的条目 valid_indices [i for i, emb in enumerate(embeddings) if emb is not None] valid_embeddings [embeddings[i] for i in valid_indices] valid_df df.iloc[valid_indices].copy() valid_df[embedding] valid_embeddings if len(valid_embeddings) 0: logging.warning(未能获取到任何有效的文本向量无法进行语义去重。) return df # 返回原始数据 # 第三步计算相似度矩阵 logging.info(计算文章间语义相似度...) embedding_matrix np.vstack(valid_embeddings) similarity_matrix cosine_similarity(embedding_matrix) # 第四步基于相似度进行聚类去重 to_keep_indices [] # 保留的文章在valid_df中的索引 visited set() for i in range(len(valid_df)): if i in visited: continue # 当前文章作为一类代表 to_keep_indices.append(i) visited.add(i) # 找出所有与当前文章高度相似的文章 similar_indices np.where(similarity_matrix[i] self.threshold)[0] for idx in similar_indices: if idx ! i and idx not in visited: visited.add(idx) # 可以选择在这里记录被合并的文章信息用于后续分析 # logging.debug(f文章 {valid_df.iloc[i][title][:30]}... 与 {valid_df.iloc[idx][title][:30]}... 相似度 {similarity_matrix[i][idx]:.2f}视为重复。) # 第五步提取去重后的数据 deduplicated_df valid_df.iloc[to_keep_indices].drop(columns[fast_signature, text_for_embedding, embedding]) logging.info(f语义去重完成。原始 {len(df)} 篇文章去重后剩余 {len(deduplicated_df)} 篇。) return deduplicated_df这个SemanticDeduplicator类是去重的核心。它做了几件事获取语义指纹为每篇新闻生成一个高维向量。计算相似度通过计算向量之间的余弦相似度得到0到1之间的分数。聚类去重把相似度超过阈值比如0.85的文章归为一组只保留其中一篇。3.3 整合流程从原始数据到纯净内容池现在我们把爬虫和去重器连接起来形成一个完整的处理流水线。def full_pipeline(crawler_configs: List[dict], api_endpoint: str, api_key: str None): 完整的新闻爬取与语义去重流水线。 :param crawler_configs: 爬虫配置列表每个配置是一个字典包含爬虫类、源名称、基础URL和列表页URL。 :param api_endpoint: StructBERT模型API地址。 :param api_key: API密钥。 # 1. 爬取数据 all_articles [] for config in crawler_configs: crawler_class config[crawler_class] crawler crawler_class(source_nameconfig[source_name], base_urlconfig[base_url]) articles crawler.crawl(config[list_url], max_articlesconfig.get(max_articles, 15)) all_articles.extend(articles) time.sleep(2) # 不同源之间稍作停顿 if not all_articles: logging.error(爬虫未获取到任何数据流程终止。) return None raw_df pd.DataFrame(all_articles) raw_df.to_csv(pipeline_raw_data.csv, indexFalse, encodingutf-8-sig) logging.info(f爬虫阶段完成共获取 {len(raw_df)} 条原始数据。) # 2. 语义去重 client StructBERTClient(api_endpointapi_endpoint, api_keyapi_key) deduplicator SemanticDeduplicator(similarity_threshold0.87) # 阈值可以微调 deduplicator.set_client(client) clean_df deduplicator.deduplicate(raw_df) # 3. 保存结果 if clean_df is not None and not clean_df.empty: clean_df.to_csv(deduplicated_news_pool.csv, indexFalse, encodingutf-8-sig) logging.info(f去重完成纯净内容池已保存共 {len(clean_df)} 篇文章。) # 简单统计一下 source_dist clean_df[source].value_counts() print(\n 去重后内容池统计 ) print(f总文章数: {len(clean_df)}) print(各来源文章分布:) for source, count in source_dist.items(): print(f - {source}: {count} 篇) return clean_df else: logging.warning(去重后未得到有效数据。) return None # 示例配置需要替换为真实爬虫类和URL example_configs [ { crawler_class: TechNewsSimulator, source_name: TechSourceA, base_url: https://tech-news-example-a.com, list_url: https://tech-news-example-a.com/tech, max_articles: 10 }, { crawler_class: FinanceNewsSimulator, source_name: FinanceSourceB, base_url: https://finance-news-example-b.com, list_url: https://finance-news-example-b.com/market, max_articles: 10 }, ] # 运行完整流水线请取消注释并填入真实的API信息 # clean_news_pool full_pipeline( # crawler_configsexample_configs, # api_endpointYOUR_REAL_API_ENDPOINT_HERE, # api_keyYOUR_REAL_API_KEY_HERE # 如果不需要则设为None # )运行这个流水线你最终会得到两个文件pipeline_raw_data.csv原始杂乱数据和deduplicated_news_pool.csv经过语义去重后的纯净数据。后者就是你进行后续分析、推荐或展示的优质内容池。4. 效果评估与优化建议做完这一切你可能会问效果到底怎么样比传统方法好在哪里这里提供几个简单的评估思路和优化方向。首先你可以进行人工抽检。从去重前后的数据中各随机抽样几十篇文章人工判断去重是否准确。计算一下准确率和召回率就能对模型效果有个直观感受。为了更量化地评估可以模拟一个测试集。比如手动创建几组“标题不同但内容相同”的新闻对以及“标题相似但内容不同”的新闻对看看模型能不能正确区分。在实际使用中有几点经验可以分享阈值是门艺术similarity_threshold这个参数很关键。设得太高如0.95可能会漏掉一些换了个说法的重复新闻设得太低如0.7又可能把不同主题但有些词相似的新闻误判为重复。建议从0.85开始根据你的数据特点微调。文本预处理很重要在计算向量前可以清洗一下文本比如移除URL、特殊字符、停用词等。对于长文章可以考虑截取开头、结尾和中间的关键段落进行综合向量化或者使用更擅长处理长文本的模型。结合多种方法语义去重虽然强大但计算成本高。可以把它和“快速签名”方法结合。先用简单的规则如标题关键词、发布机构、时间筛掉明显不重复的剩下的“嫌疑对象”再用StructBERT模型进行精细判断这样可以大大提高整体处理效率。关注源质量如果某个新闻源经常发布高度重复或洗稿的内容可以考虑在爬取阶段就进行限制或降权。5. 总结走完这一整套流程你会发现用Python爬虫结合StructBERT模型来做新闻去重其实并没有想象中那么复杂。核心思路就是让AI模型去理解内容的“意思”而不是机械地比较文字。这种方法最大的好处是精准。它能发现那些“换汤不换药”的重复内容帮你从信息洪流中提炼出真正独特、有价值的新闻。无论是做内容聚合、舆情监控还是竞争分析一个干净的数据池都是高质量分析的基础。当然世上没有完美的方案。语义模型需要计算资源处理速度可能不如规则匹配快。而且模型本身的理解能力也有边界对于非常专业或晦涩的内容判断也可能出现偏差。这就需要你在实践中不断调整和优化。代码都给你了你可以直接拿过去换上你自己的爬虫逻辑和可用的语义模型API跑起来看看效果。先从一两个新闻源开始试验调整好阈值等你看到那些烦人的重复内容被自动过滤掉时一定会觉得这点功夫没白费。数据处理就是这样前期多花点心思把基础打牢后面的分析工作就会顺畅得多。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。