实战:资讯站列表与详情采集

资讯、博客、公告类站点是采集练习里最典型的目标:列表页有分页、详情页有正文、内容持续更新,需要增量与去重。本章按「先找接口、再解析 HTML」的顺序做一遍完整实现。

目标拆解

环节产出验收标准
列表页每页的文章链接与标题能连续翻到第 N 页且无重复
详情页标题、作者、发布时间、正文、原文链接正文长度达标,字段非空率 > 95%
去重同一文章只入库一次重跑一遍条数不增加
增量只采新增文章第二次运行请求数远小于第一次

第一步:优先找接口,找不到再解析 HTML

打开目标页,按 F12 → Network → 筛选 Fetch/XHR,翻一次页,观察是否有返回 JSON 的请求——接口返回结构化数据,不用面对改版,也不用写复杂选择器。找到后用 curl 复现,确认不依赖登录、签名与特殊 Header:

curl -s 'https://example.com/api/news?page=1&size=20' -H 'Accept: application/json' | head -c 500
路线优点缺点什么时候选
公开 JSON 接口结构稳定、字段齐全需要有接口且未做校验首选,先用第 07 章的方法确认
解析 HTML一定有、不依赖接口改版即失效、要维护选择器没有可用接口时的兜底
站点 RSS / 开放数据最稳定、最合规字段有限、可能延迟有就优先用
官方 API(需申请)合法、稳定、有配额需要申请与授权商业用途必须走这条路

第二步:完整实现

# news_spider.py —— Python 3.10+,依赖 requests、beautifulsoup4
import json, time
from pathlib import Path
import requests
from bs4 import BeautifulSoup

BASE = "https://example.com"
STATE = Path("state/news_state.json")
OUT = Path("output/news/news.jsonl")
MIN_LEN, INTERVAL, MAX_PAGES = 200, 2.0, 5      # 正文阈值 / 请求间隔 / 单次最多翻页
session = requests.Session()
session.headers.update({"User-Agent": "MyResearchBot/1.0 (+contact@example.com)"})   # 声明身份

def fetch(url: str, retry: int = 3) -> str | None:
    """带限速与退避重试的 GET;连续失败返回 None 并跳过"""
    for attempt in range(1, retry + 1):
        try:
            resp = session.get(url, timeout=10)
            resp.encoding = resp.apparent_encoding or "utf-8"
            if resp.status_code == 200 and "text/html" in resp.headers.get("Content-Type", ""):
                return resp.text
            print(f"[{resp.status_code}] {url}")
        except requests.RequestException as exc:
            print(f"第 {attempt} 次请求失败:{exc}")
        time.sleep(INTERVAL * attempt)          # 退避等待,越失败越慢
    return None

def parse_list(html: str) -> list[str]:
    """列表页只负责产出详情链接(选择器按目标站替换)"""
    soup = BeautifulSoup(html, "html.parser")
    hrefs = [a.get("href", "") for a in soup.select("ul.news-list li a")]
    return [BASE + h if h.startswith("/") else h for h in hrefs if h.startswith(("/", BASE))]

def parse_detail(html: str, url: str) -> dict | None:
    """详情页:正文提取 + 长度阈值校验,不达标视为解析失败"""
    soup = BeautifulSoup(html, "html.parser")
    node = soup.select_one("div.article-content") or soup.select_one("article")
    if node is None:
        return None
    for bad in node.select("script, style, .recommend, .ad"):
        bad.decompose()                          # 剔除广告与推荐位,正文才干净
    content = "\n".join(p.get_text(strip=True) for p in node.find_all("p") if p.get_text(strip=True))
    if len(content) < MIN_LEN:
        return None
    return {"url": url, "title": (soup.select_one("h1") or soup).get_text(strip=True),
            "author": (soup.select_one("span.author") or soup).get_text(strip=True),
            "pub_time": (soup.select_one("span.time") or soup).get_text(strip=True),
            "content": content, "content_len": len(content),
            "collect_time": time.strftime("%Y-%m-%d %H:%M:%S")}

def main() -> None:
    state = json.loads(STATE.read_text(encoding="utf-8")) if STATE.exists() else {"seen_urls": []}
    seen, saved, failed = set(state.get("seen_urls", [])), 0, 0
    OUT.parent.mkdir(parents=True, exist_ok=True)
    for page in range(1, MAX_PAGES + 1):
        html = fetch(f"{BASE}/news/list?page={page}")
        if not html:
            break
        links = [u for u in parse_list(html) if u not in seen]
        if not links:                            # 本页没有新链接,说明已到增量边界
            print(f"第 {page} 页无新链接,停止翻页")
            break
        for url in links:
            item = parse_detail(fetch(url) or "", url)
            time.sleep(INTERVAL)                 # 详情页额外限速,宁可慢也别打扰站点
            if item is None:
                failed += 1
                continue
            with OUT.open("a", encoding="utf-8") as f:
                f.write(json.dumps(item, ensure_ascii=False) + "\n")
            seen.add(url)
            saved += 1
    state["seen_urls"] = list(seen)[-5000:]      # 有界化,避免状态文件无限膨胀
    STATE.parent.mkdir(parents=True, exist_ok=True)
    STATE.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
    print(f"新增 {saved} 篇,解析失败 {failed} 篇")

if __name__ == "__main__":
    main()

字段设计与增量策略

字段类型说明
urlstring原文链接,唯一键,必须保留
title / authorstring标题与非空;作者取不到时留空而非编造
pub_timedatetime发布时间,统一格式后用于增量判断
content / content_lentext / int正文与长度,长度用于质量校验
collect_timedatetime采集时间,便于回溯与补跑

推荐组合:URL 指纹去重 + 时间戳断点。指纹保证不重复、重跑安全;时间戳断点把请求量压到最低,但要留缓冲窗口——站点时间乱序时会漏采。全量重跑加唯一键去重虽然不漏采,但请求量大,不礼貌也不经济。

常见坑

现象原因处理
正文只抓到一句摘要只取了首个 p 或命中摘要块find_all("p") 合并并加长度阈值
翻页到第 3 页就重复站点忽略 page 参数恒返回第一页对比页面 HTML 指纹,相同则停止
中文乱码未按响应头或实际编码解码apparent_encoding 或显式指定
第二次运行又全采一遍状态文件未落盘或被清先写 .tmp 再原子改名

合规提示

这类站点最容易出现「整站转载」的越界行为,要守住四条线:只采集公开可访问的文章页面,遵守 robots.txt 与站点服务条款,保留原文链接与来源标识;控制频率与并发(单线程、每次请求间隔 2 秒是保守基线),不追求「全站镜像」;不绕过登录、验证码与付费墙,付费墙后的内容一律不采;转载或再分发正文前必须获得授权——《著作权法》保护新闻作品与原创内容,聚合展示与全文转载是不同性质的行为。涉及个人信息处理的需符合《个人信息保护法》与《数据安全法》;有官方 API 或开放数据就用它们,比解析 HTML 更稳也更合规。

小结:资讯站采集的顺序是「先找接口、再解析 HTML」,列表页只负责产出链接,详情页用选择器加正文长度阈值做质量判断;去重靠 URL 指纹、增量靠时间戳断点;把合规当设计约束而非事后补充——只采公开文章、控制频率、保留来源链接、不整站转载。

笔记加载中…