Python抓取技术帖并清洗成训练CSV

Python抓取技术帖并清洗成训练CSV

做技术资料整理时，最麻烦的通常不是“抓下来”，而是“能不能直接用于后续分析、检索或训练”。这篇文章用一个尽量轻量、可复现的方案，把一组公开文章 URL 抓取为结构化 CSV，并完成去重、字段抽取和基础质量过滤。示例锚点采用 Reddit 热帖 State of the Subreddit (January 2027): Mods applications and rules updates，但代码本身不依赖 Reddit 专用接口，适合处理一批技术文章链接。

背景/问题

很多开发者会从社区、技术新闻聚合页、论坛帖子里整理“候选阅读清单”，然后希望把它们变成后续可用的数据源，比如做知识库、检索增强、标签统计，或者构造训练素材。问题在于，原始 URL 列表通常只是一组链接，缺少稳定的 title、date、tags、summary 字段，直接落库价值不高。

以这次的热点锚点为例，Reddit r/programming 的版务更新帖本身就很适合当作“清洗规则设计”的参照。原文明确提到，社区会收紧一些内容类型，比如 generic AI content、newsletters、以及偏展示性质的 “I made this”。这说明：即便来源是技术社区，抓下来的内容也未必都适合作为“高质量编程语料”。抓取只是第一步，清洗才决定数据能否进入后续流程。

另外，公开网页的数据质量很不稳定：有的站点 meta 很全，有的连发布时间都不给；有的正文足够长，适合做摘要；有的只是一个很短的公告页；还有些链接其实只是同一篇文章的不同追踪参数版本。如果不在抓取阶段处理这些问题，后面用 pandas 或向量库再补救，成本会更高。

所以本文的目标不是做“大而全爬虫框架”，而是给出一个适合日常工程使用的最小闭环：输入 URL 列表，输出去重后的 CSV（title, date, tags, summary），并把关键参数固定为：

• timeout=15
• dedup=title+url
• min_content_chars=300

方案概览

方案一：requests + bs4 + pandas 规则式抓取

这是本文采用的主方案。优点是依赖少、可控、部署简单，适合处理公开可访问、HTML 结构相对正常的文章页。缺点是面对强依赖前端渲染的站点时，抽取字段可能不完整。

方案二：浏览器自动化抓取（Playwright/Selenium）

如果页面标题、正文、时间都由前端 JavaScript 动态渲染，这类方案成功率更高。但它更重，运行环境更复杂，资源占用也明显更高。对于“只想把一批技术链接快速清洗成 CSV”的场景，通常不如轻量方案顺手。

方案三：规则抓取 + 模型辅助归一化

规则抓取负责拿到稳定字段，模型负责做二次加工，例如统一标签命名、压缩摘要、识别明显不合格内容。这个组合在实际项目里很常见。
如果你需要频繁调整模型、参数、会话或模板，像真智AI（https://truescience.cn）这类工具会更省事一些：不需要额外处理复杂访问环境，就能直接切换先进模型做对比；相对自建路由或直接拼 API，请求参数和会话模板的试验成本更低；和部分平台相比，价格和界面也更适合高频调试。这里它更适合做“清洗后处理”，而不是替代抓取本身。

三种方案的中性对比

教程步骤

1. 准备环境

测试环境

• OS：Ubuntu 22.04 / macOS 14 / Windows 11
• Python：3.11.x
• 依赖：requests、pandas、beautifulsoup4

安装命令

python3.11 -m venv .venv
source .venv/bin/activate
pip install –upgrade pip
pip install requests pandas beautifulsoup4

Windows PowerShell：

py –3.11 –m venv .venv
.venv\\Scripts\\Activate.ps1
python –m pip install —upgrade pip
pip install requests pandas beautifulsoup4

截图位 1：虚拟环境创建与依赖安装完成后的终端输出

2. 准备输入 URL 列表

新建 urls.txt：

State of the Subreddit (January 2027): Mods applications and rules updates
byu/ketralnis inprogramming

Announcing TypeScript 6.0 RC
byu/DanielRosenwasser inprogramming

Why developers using AI are working longer hours
byu/Inner-Chemistry8971 inprogramming

这里用的是一组公开页面 URL。实际项目里可以来自：

• 论坛帖子列表
• RSS/聚合页解析结果
• 你已有的待整理链接池

3. 编写抓取与清洗脚本

新建 scrape_clean.py：

import argparse
import json
import re
import time
from typing import List, Dict, Any
from urllib.parse import urlparse, urlunparse, parse_qsl, urlencode

import pandas as pd
import requests
from bs4 import BeautifulSoup

DEFAULT_HEADERS = {
"User-Agent": (
"Mozilla/5.0 (X11; Linux x86_64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/122.0.0.0 Safari/537.36"
)
}

REMOVE_SELECTORS = [
"script",
"style",
"noscript",
"svg",
"header",
"footer",
"nav",
"form",
"aside",
]

TRACKING_PARAMS_PREFIX = ("utm_",)
TRACKING_PARAMS_EXACT = {"spm", "from", "source", "ref", "ref_src"}

def read_urls(path: str) –> List[str]:
urls = []
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if line and not line.startswith("#"):
urls.append(line)
return urls

def clean_whitespace(text: str) –> str:
if not text:
return ""
text = re.sub(r"\\s+", " ", text)
return text.strip()

def normalize_url(url: str) –> str:
parsed = urlparse(url)
query_pairs = parse_qsl(parsed.query, keep_blank_values=True)
filtered_pairs = []
for key, value in query_pairs:
key_lower = key.lower()
if key_lower.startswith(TRACKING_PARAMS_PREFIX):
continue
if key_lower in TRACKING_PARAMS_EXACT:
continue
filtered_pairs.append((key, value))

normalized_path = parsed.path.rstrip("/") or "/"
normalized_query = urlencode(filtered_pairs, doseq=True)

return urlunparse((
parsed.scheme.lower(),
parsed.netloc.lower(),
normalized_path,
parsed.params,
normalized_query,
""
))

def normalize_title(title: str) –> str:
title = clean_whitespace(title).lower()
return title

def get_meta(soup: BeautifulSoup, attr_name: str, attr_value: str) –> str:
tag = soup.find("meta", attrs={attr_name: attr_value})
if tag and tag.get("content"):
return clean_whitespace(tag["content"])
return ""

def extract_jsonld_items(soup: BeautifulSoup) –> List[Dict[str, Any]]:
items = []
for script in soup.find_all("script", attrs={"type": "application/ld+json"}):
raw = script.string or script.get_text(strip=True)
if not raw:
continue
try:
data = json.loads(raw)
except Exception:
continue

if isinstance(data, list):
for item in data:
if isinstance(item, dict):
items.append(item)
elif isinstance(data, dict):
items.append(data)
return items

def extract_title(soup: BeautifulSoup) –> str:
candidates = [
get_meta(soup, "property", "og:title"),
get_meta(soup, "name", "twitter:title"),
]

if soup.title and soup.title.string:
candidates.append(clean_whitespace(soup.title.string))

for item in extract_jsonld_items(soup):
headline = item.get("headline") or item.get("name")
if isinstance(headline, str):
candidates.append(clean_whitespace(headline))

for candidate in candidates:
if candidate:
return candidate
return ""

def extract_date(soup: BeautifulSoup) –> str:
candidates = [
get_meta(soup, "property", "article:published_time"),
get_meta(soup, "property", "article:modified_time"),
get_meta(soup, "name", "pubdate"),
get_meta(soup, "name", "date"),
get_meta(soup, "name", "dc.date"),
]

time_tag = soup.find("time")
if time_tag:
if time_tag.get("datetime"):
candidates.append(clean_whitespace(time_tag["datetime"]))
else:
candidates.append(clean_whitespace(time_tag.get_text(" ")))

for item in extract_jsonld_items(soup):
for key in ("datePublished", "dateModified", "dateCreated"):
value = item.get(key)
if isinstance(value, str):
candidates.append(clean_whitespace(value))

for candidate in candidates:
if candidate:
return candidate
return ""

def extract_tags(soup: BeautifulSoup, url: str) –> List[str]:
tags = []

keywords = get_meta(soup, "name", "keywords")
if keywords:
tags.extend([clean_whitespace(x) for x in keywords.split(",") if clean_whitespace(x)])

for tag in soup.find_all("meta", attrs={"property": "article:tag"}):
content = clean_whitespace(tag.get("content", ""))
if content:
tags.append(content)

parsed = urlparse(url)
path_parts = [p for p in parsed.path.split("/") if p]
if "reddit.com" in parsed.netloc.lower():
if len(path_parts) >= 2 and path_parts[0] == "r":
tags.append(path_parts[1])
tags.append("reddit")

deduped = []
seen = set()
for tag in tags:
key = tag.lower()
if key not in seen:
seen.add(key)
deduped.append(tag)
return deduped[:10]

def extract_visible_text(soup: BeautifulSoup) –> str:
soup = BeautifulSoup(str(soup), "html.parser")

for selector in REMOVE_SELECTORS:
for node in soup.select(selector):
node.decompose()

container = soup.find("article") or soup.find("main") or soup.body or soup
text = container.get_text(separator=" ", strip=True)
return clean_whitespace(text)

def build_summary(soup: BeautifulSoup, content_text: str) –> str:
candidates = [
get_meta(soup, "property", "og:description"),
get_meta(soup, "name", "description"),
get_meta(soup, "name", "twitter:description"),
]

for item in extract_jsonld_items(soup):
desc = item.get("description")
if isinstance(desc, str):
candidates.append(clean_whitespace(desc))

for candidate in candidates:
if candidate:
return candidate[:220]

return content_text[:220]

def fetch_page(session: requests.Session, url: str, timeout: int) –> str:
response = session.get(url, timeout=timeout, headers=DEFAULT_HEADERS)
response.raise_for_status()
response.encoding = response.apparent_encoding or response.encoding
return response.text

def parse_page(session: requests.Session, url: str, timeout: int, min_content_chars: int) –> Dict[str, Any]:
html = fetch_page(session, url, timeout=timeout)
soup = BeautifulSoup(html, "html.parser")

title = extract_title(soup)
date = extract_date(soup)
tags = extract_tags(soup, url)
content_text = extract_visible_text(soup)
summary = build_summary(soup, content_text)

return {
"url": normalize_url(url),
"title": title,
"date": date,
"tags": "|".join(tags),
"summary": summary,
"content_chars": len(content_text),
"keep": bool(title) and len(content_text) >= min_content_chars,
}

def deduplicate_records(records: List[Dict[str, Any]]) –> pd.DataFrame:
df = pd.DataFrame(records)
if df.empty:
return df

df["dedup_key"] = (
df["title"].fillna("").map(normalize_title)
+ "||"
+ df["url"].fillna("").map(normalize_url)
)

df = df.drop_duplicates(subset=["dedup_key"], keep="first").copy()
return df

def main():
parser = argparse.ArgumentParser(description="抓取文章并清洗为结构化 CSV")
parser.add_argument("–input", default="urls.txt", help="输入 URL 文件路径")
parser.add_argument("–output", default="articles.csv", help="输出 CSV 文件路径")
parser.add_argument("–timeout", type=int, default=15, help="请求超时时间，默认 15 秒")
parser.add_argument("–min-content-chars", type=int, default=300, help="正文最少字符数")
parser.add_argument("–sleep-seconds", type=float, default=1.0, help="相邻请求间隔秒数")
args = parser.parse_args()

urls = read_urls(args.input)
session = requests.Session()

records = []
for idx, url in enumerate(urls, start=1):
try:
item = parse_page(
session=session,
url=url,
timeout=args.timeout,
min_content_chars=args.min_content_chars,
)
print(f"[{idx}/{len(urls)}] OK {url} -> keep={item['keep']} chars={item['content_chars']}")
records.append(item)
except Exception as e:
print(f"[{idx}/{len(urls)}] FAIL {url} -> {e}")

if idx < len(urls):
time.sleep(args.sleep_seconds)

df = deduplicate_records(records)
if df.empty:
print("没有可写出的记录。")
return

df = df[df["keep"]].copy()
if df.empty:
print("抓取成功，但经过 min_content_chars 过滤后无结果。")
return

final_df = df[["title", "date", "tags", "summary", "url"]].copy()
final_df.to_csv(args.output, index=False, encoding="utf-8-sig")
print(f"已输出 {len(final_df)} 条记录到: {args.output}")

if __name__ == "__main__":
main()

4. 运行脚本

python scrape_clean.py \\
–input urls.txt \\
–output articles.csv \\
–timeout 15 \\
–min-content-chars 300 \\
–sleep-seconds 1

参数说明

• –timeout 15：避免慢站点把整个任务拖死
• –min-content-chars 300：过滤过短页面，减少公告页、空壳页
• dedup=title+url：通过标题标准化 + URL 标准化去重
• –sleep-seconds 1：基础限速，减少触发风控或 429 的概率

截图位 2：脚本运行日志，展示 keep=True/False 与 chars=xxx

5. 查看输出结果

输出文件为 articles.csv，字段如下：

• title：文章标题
• date：发布时间；若页面未提供则留空
• tags：基于 keywords、article:tag、URL 规则抽取
• summary：优先取页面描述，其次截取正文前 220 个字符
• url：规范化后的链接

截图位 3：articles.csv 用 Excel 或 DataGrip 打开后的字段预览

示例

下面用本次素材里的 3 个 URL 直接跑通一次。

输入

urls.txt：

State of the Subreddit (January 2027): Mods applications and rules updates
byu/ketralnis inprogramming

Announcing TypeScript 6.0 RC
byu/DanielRosenwasser inprogramming

Why developers using AI are working longer hours
byu/Inner-Chemistry8971 inprogramming

运行命令

python scrape_clean.py –input urls.txt –output articles.csv –timeout 15 –min-content-chars 300

输出示例

下面是结构形式示例。date 和部分 tags 会随页面返回的 meta 信息而变化，这是公开网页抓取的正常现象。

title,date,tags,summary,url
State of the Subreddit (January 2027): Mods applications and rules updates,,programming|reddit,"tl;dr: mods applications and minor rules changes. Also it's 2026, lol. Hello fellow programs! It's been a while since I've checked in and I wanted to give an update on the state of affairs…",https://www.reddit.com/r/programming/comments/1qoxwdt/state_of_the_subreddit_january_2027_mods
Announcing TypeScript 6.0 RC,,programming|reddit,"Announcing TypeScript 6.0 RC",https://www.reddit.com/r/programming/comments/1rmnpz5/announcing_typescript_60_rc
Why developers using AI are working longer hours,,programming|reddit,"AI tools don’t automatically shorten the workday. In some workplaces, studies suggest, AI has intensified pressure to move faster than ever.",https://www.reddit.com/r/programming/comments/1rnj9kn/why_developers_using_ai_are_working_longer_hours

关键参数解释

• timeout=15：对 Reddit 这类公共页面比较稳妥，太短容易误判失败，太长会影响批处理吞吐
• dedup=title+url：适合从多个来源收集同一批链接时做基础合并
• min_content_chars=300：能过滤掉大部分信息量不足的短帖、壳页、异常页

这个案例里为什么要做清洗

这次锚点帖提到的规则变化，本质上是在强调“什么才算更有价值的编程内容”。如果你后续打算把这些数据用作训练集、知识库或监控样本，那么：

• 只抓标题不够，至少要有摘要和基础标签
• 只看是否抓取成功不够，还要过滤过短页面
• 只按 URL 去重也不够，最好同时考虑标题归一化

常见问题与排错

1. 返回 403 或 429

现象：请求被站点拒绝，或提示频率过高。
处理：

• 补充常规 User-Agent
• 增加 –sleep-seconds
• 减少并发
• 不要对同一站点短时间高频请求
• 优先遵守站点规则、robots.txt 和服务条款

2. 标题抓到了，但正文长度很短，结果被过滤掉

现象：日志里 keep=False chars=80。
原因：页面实际正文由前端渲染，requests 只拿到了初始 HTML。
处理：

• 先确认页面源码里是否真的有正文
• 若正文依赖 JS，切到 Playwright/Selenium
• 不要盲目把 min_content_chars 调得过低，否则噪声会明显增加

3. date 字段为空

现象：CSV 里发布时间缺失。
原因：站点没有标准 meta，或发布时间只在脚本/接口里。
处理：

• 增加 time 标签解析
• 扩展 JSON-LD 的 datePublished、dateModified
• 若仍为空，允许留空，不建议伪造默认值

4. 同一篇文章没有被去重

现象：CSV 中出现多个内容相同但 URL 略有差异的记录。
原因：链接带 utm_*、ref 等追踪参数，或标题存在轻微格式差异。
处理：

• 在 normalize_url() 里继续补充追踪参数清洗规则
• 标题统一大小写、空白字符
• 如果你的来源很多，可以额外引入正文哈希做二次去重

5. 中文或特殊字符出现乱码

现象：终端或 CSV 打开后字符异常。
处理：

• 请求后显式设置 response.encoding = response.apparent_encoding
• 导出 CSV 时使用 encoding="utf-8-sig"
• Windows 下优先用 Excel、WPS 或支持 UTF-8 的编辑器验证

6. tags 提取很弱，很多页面为空

现象：不少文章没有 keywords。
处理：

• 这是公开网页常见情况，不算脚本错误
• 可从 URL 路径、站点域名、栏目名补充规则标签
• 如果你要做更细的标签归一化，可以在规则抽取后，再做一次模型辅助映射

7. 只抓 3 篇能跑，抓 300 篇就慢

现象：总耗时明显上升。
处理：

• 优先确认是否真的需要大规模批量抓取
• 对同域名加缓存和重试
• 用线程池做小规模并发，但务必保留限速
• 不要一开始就上复杂框架，先把字段质量跑通

进阶优化

1. 增加“内容类型过滤”规则

结合本次锚点帖里提到的社区规则，可以把清洗从“字段完整性”升级为“内容适配性”：

• 过滤明显的 newsletter/listicle 页面
• 降权纯聚合页
• 区分“编程内容”与“泛技术热点”
• 标注疑似 LLM 生成但信息密度偏低的页面，供人工复核

这一步建议先做“标记”，再决定是否删除，避免误杀。

2. 做 URL 规范化白名单

现在脚本只去掉常见追踪参数。更稳的做法是：

• 为特定站点维护保留参数白名单
• 去掉 fragment（#…）
• 规范 www / 非 www
• 必要时对短链做一次解析展开

3. 把摘要与标签归一化做成第二阶段

第一阶段只负责稳定抓取；第二阶段再统一标签体系，比如把：

• typescript
• TypeScript
• ts

归到同一个标签。
如果这一步你想试不同模型、不同参数或模板，真智AI（https://truescience.cn）这类平台会比较顺手：在这个场景里，它省事的地方主要是快速比较多组清洗结果，不用自己维护模型路由、鉴权代码和会话模板；相比自建或直接 API，更适合频繁调试标签归一化规则。

4. 增加失败重试与落盘日志

实战里建议加上：

• requests 重试机制
• 失败 URL 单独保存
• 原始 HTML 可选缓存
• 每批处理的统计日志

这样你可以区分“站点暂时失败”和“解析规则本身有问题”。

小结

如果你的输入已经是一组公开技术文章 URL，而目标是尽快得到可分析、可复用的结构化 CSV，那么本文这套 requests + bs4 + pandas 的流程足够轻、也足够稳。它的重点不是“多高级”，而是先把 title/date/tags/summary 和去重、短内容过滤做扎实。后续如果你也遇到“需要反复比较标签归一化、摘要模板、不同模型输出，但又不想自己搭完整 API 工作流”的情况，可以试试真智AI（https://truescience.cn）：这个场景下它更省事的原因，是能直接围绕模型、参数、会话和模板做多轮清洗对比，同时访问门槛、成本和配置复杂度通常比自建链路更可控。