企业官网建设流程全解析

简介：以Java编写的百度新闻与今日头条爬虫程序包，可根据关键字批量抓取新闻并存入数据库，定位服务于Java爬虫初学者、资讯聚合开发者和需要定期采集新闻数据的个人或团队。资源共20个文件，16个Java源文件实现了URL收集、HTTP请求、HTML解析与数据库存储等核心逻辑，另有1个XML配置与1个properties文件负责工程配置和运行参数，说明文档介绍编译与使用方式；项目采用Maven工程结构，源码统一置于src/main下，压缩包仅19KB，代码体量小，适合直接阅读和二次开发。该程序完整走通了一个爬虫项目的常规流程：从关键字构造URL队列，到解析新闻标题与正文，再到入库持久化；同时兼顾robots协议、User-Agent等基础反爬措施，可帮助初学者建立工程化爬虫的基本框架。目前已有446人浏览学习，对想快速上手新闻类定向爬虫的开发者而言，不失为一份轻量且可运行的参考实现。

1. 关键字新闻爬虫的本质：先把“所有新闻”拆成关键词和时间范围

第一次拿到类似“百度新闻，今日头条爬虫，根据关键字爬取所有新闻并存如数据库.zip”这样的压缩包时，我第一反应不是先看代码，而是把需求翻译成可以度量的爬虫任务。“所有新闻”是个伪需求：新闻是无限增长的流，真正能落地的是“在给定关键词列表下，从某个时间点开始增量抓取”。想清楚这一点，数据库表结构、去重逻辑、定时策略才定得下来。这个项目的核心，其实是用关键词去搜索两个新闻源，再把搜索结果结构化，存进 MySQL。

适合读这篇内容的人，是手里已经有一个不完整的爬虫包、不知道怎么跑起来，或者打算自己写一个但已经在反爬和入库上栽过跟头的人。接下来按“怎么把数据掏出来、怎么建表、怎么运行、怎么避坑、怎么定时增量”这条主线展开，每个步骤尽量给可复现代码和参数解释。

2. 抓取百度新闻和今日头条之前：先搞清楚两种页面的数据载体

百度新闻和今日头条的搜索结果形态完全不一样。百度新闻页面偏服务端渲染，返回的是带新闻列表结构的 HTML，适合直接解析；今日头条的搜索结果页是前端异步渲染，直接去抓静态 HTML 大概率只能拿到一个框架，数据要么在内联的 JSON 初始化变量里，要么在单独的 XHR 接口里。用同一个解析模板去套两个站点，是我见过新手最容易踩的第一个坑。

在做任何解析之前，建议先用一个调试脚本把原始响应保存到本地，肉眼确认数据位置。这一步看着笨，实际是最高效的办法，能省掉后面反复猜字段名的时间。

2.1 百度新闻搜索结果：HTML 节点里直接解析

百度新闻网页版搜索 URL 可以用https://www.baidu.com/s基础路径，配合tn=news参数变成新闻搜索。常见的参数还有word表示关键词，pn表示页数偏移量，第一页是 0。抓取时最重要的不是 URL 多精确，而是请求头要像真实浏览器，否则很容易返回安全验证页。

先看调试请求的写法，我把响应按字节方式落盘，避免编码踩坑：

import requests session = requests.Session() session.headers.update({ "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" ), "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9", }) url = "https://www.baidu.com/s" params = { "tn": "news", "word": "人工智能", "pn": 0, } resp = session.get(url, params=params, timeout=10) print(resp.status_code, resp.url, resp.encoding) with open("baidu_news.html", "wb") as f: f.write(resp.content)

这段代码的逻辑是：通过一个requests.Session复用连接和 Cookie，headers里显式设置浏览器 User-Agent 和 Accept-Language，再把关键词传给params。保存时用resp.content而不是resp.text，是为了保留源站返回的原始字节，后面无论用什么工具打开，看到的都是网站原本的编码，不会因为本地终端问题产生二次乱码。

校验请求成功后，用 BeautifulSoup 解析标题和链接。下面是一个最小解析示例：

from bs4 import BeautifulSoup with open("baidu_news.html", "rb") as f: html = f.read() soup = BeautifulSoup(html, "lxml") items = soup.select("div.result") for item in items: a = item.select_one("h3 a") if not a: continue title = a.get_text(" ", strip=True) link = a.get("href") time_node = item.select_one(".c-color-gray") summary_node = item.select_one(".c-summary") print(title) print(link) print(time_node.get_text(strip=True) if time_node else "") print(summary_node.get_text(strip=True) if summary_node else "") print("---")

这段代码从本地文件中读取之前保存的 HTML，再用 BeautifulSoup 查找div.result外层容器。里面标题在h3 a标签内，摘要时间区域常见类是.c-color-gray和.c-summary。要说明的是，这类 class 名会随百度改版变化，最稳妥的判断方法还是用浏览器开发者工具选中一条新闻记录，对比外层节点结构。lxml解析速度比html.parser快很多，抓新闻场景下推荐直接用 lxml。

这里有一个老手也容易犯的错：百度新闻列表里的href经常不是最终新闻页地址，而是baidu.com/link?url=...这类跳转链接。如果使用最终的去重需求，必须额外发一次 GET 请求跟随跳转，取resp.url作为最终 URL，这个点在后面避坑章节再展开。

2.2 今日头条搜索结果：在前面找数据初始化 JSON

今日头条网页版的搜索地址是https://so.toutiao.com/search，常见参数是keyword和pd=information。直接用 requests 拿到响应后，先不要着急解析 HTML，把数据落盘看内容：

url = "https://so.toutiao.com/search" params = { "keyword": "人工智能", "pd": "information", "source": "search_tab", } resp = session.get(url, params=params, timeout=10) print(resp.status_code, resp.headers.get("Content-Type"), len(resp.content)) with open("toutiao_search.html", "wb") as f: f.write(resp.content)

这里source=search_tab是为了尽量模拟用户在新闻 tab 下搜索的行为。第一次请求后发现Content-Type是text/html，但正文里可能没有新闻列表，反而是一个被 JS 加载的空白骨架。此时可以继续在本地文件里搜索标题关键词，如果搜不到，就说明走的是 XHR 动态接口。

我的习惯是先在本地对关键词做一次前后文定位，确认能不能直接从 HTML 中提取：

grep -o "人工智能.\{0,200\}" toutiao_search.html | head -c 2000

如果你能搜到新闻标题，就说明数据确实存在当前 HTML 里，只是包在某个初始化变量中。常见的载体是window._SSR_DATA这类变量。可以用下面这段代码把整块 JSON 提取出来：

import re import json m = re.search(r"window\._SSR_DATA\s*=\s*(\{.*?\})\s*</script>", resp.text, re.S) if m: data_str = m.group(1) try: data = json.loads(data_str) print(json.dumps(data, ensure_ascii=False)[:5000]) except json.JSONDecodeError as e: print("JSON 解析失败，截取片段查看") print(data_str[:2000]) else: print("未找到 _SSR_DATA 变量")

这段代码的逻辑分两步：先用非贪婪正则把window._SSR_DATA = { ... }中的 JSON 字符串抓出来，再用json.loads转成 Python 对象。这里的关键点是“先找到数据嵌入位置，再写解析代码”，而不是对着网上可能过时的选择器硬套。

需要注意，今日头条页面结构变过多次，不同版本的字段名也不同。有的版本直接把列表放在data下，字段叫title、abstract、share_url；有的版本则多套一层search_result。如果这段 JSON 解析失败，不要反复调正则，直接打开本地 HTML 文件，用编辑器搜索title，看它实际被包在哪一层，这是最实际的排查方法。

2.3 到底用 requests 还是 Selenium：别让工具选错变成最大的坑

面对新闻爬虫，我会先默认选择 requests，而不是 Selenium。原因很直接：requests 加 Session 的写法轻量、便于控制并发和频率，也方便结合 SQLAlchemy 做批量入库；Selenium 启动浏览器实例后内存和 CPU 占用高，抓 1000 条新闻前的等待时间会明显拉长，处理不好反而容易被风控识别浏览器特征。

什么时候才需要 Selenium？我的判断标准有三条：一是 requests 拿到的 HTML 全文里连标题关键词都搜不到；二是接口请求需要较复杂的前端签名；三是页面必须滚动或点击交互才能触发加载。遇到这三种情况，才值得用 Selenium 或 Playwright 渲染驱动。就这个项目而言，两个站点的数据都能用 requests 取到，区别只是解析位置不同，所以优先把 requests 方案跑通。

还有一个容易被忽略的点：无论用哪种方式，都要注意请求频率。新闻源对搜索接口的限流通常比普通页面严格，短时间高频请求很容易触发验证。我一般会把单次请求间隔保持在 1 到 2 秒，并且用time.sleep(random.uniform(0.5, 1.5))打散节奏，避免固定间隔被识别。

3. 数据库表怎么建：字段设计、唯一键和 SQLAlchemy 批量入库

很多爬虫项目死在最后一步：数据抓到了，却因为表结构设计不合理，入库后又出现大量重复，或者查不到想要的维度。新闻数据入库前先想清楚一个问题：你想用这张表回答什么？是“某个关键词今天抓到多少条”，还是“某个来源在最近一周发了多少篇”，抑或“同一篇新闻最早什么时候出现”。不同问题决定了字段粒度。

这个项目我建议把单条记录定位成“某次搜索得到的一条新闻”，而不是“一篇唯一的新闻原文”。这样的设计保留了关键词维度的信息，既可以看到每个关键词的抓取结果，也能通过 URL 指纹做整体去重。

3.1 字段设计：先确定哪些必须存、哪些可空

新闻抓取入库的最小字段集合如下：主键、关键词、新闻标题、新闻链接、链接指纹、来源名、发布时间、抓取时间、摘要。其中链接指纹是去重的关键，必须建立唯一索引；关键词是重要的分组维度，也应该建索引。发布时间可能从页面解析不出来，所以允许为空；摘要同理。

这里最容易被忽略的是唯一键设计。URL 字段虽然理论上是唯一的，但 MySQL 对大的 VARCHAR 字段建索引有长度限制，而且不同页面可能给同一篇文章生成不同带参数的 URL，直接对url建唯一索引既低效又不可靠。更稳的方式是单独存一列url_hash，用 MD5 把 URL 变成 32 位定长字符串，再对这个字段建唯一索引。

3.2 用 SQLAlchemy 定义模型：连接参数和建表

现在用 SQLAlchemy 2.x 的声明式模型建表。代码里会用到declarative_base、字段类型和静态方法。先写模型文件：

from datetime import datetime import hashlib from sqlalchemy import ( Column, BigInteger, String, DateTime, Text, ) from sqlalchemy.orm import declarative_base Base = declarative_base() class NewsArticle(Base): __tablename__ = "news_article" id = Column(BigInteger, primary_key=True, autoincrement=True) keyword = Column(String(50), nullable=False, index=True) title = Column(String(500), nullable=False) url = Column(String(1000), nullable=False) url_hash = Column(String(32), nullable=False, unique=True) source_name = Column(String(100)) publish_time = Column(DateTime) crawl_time = Column(DateTime, default=datetime.now) summary = Column(Text) @staticmethod def make_url_hash(url: str) -> str: return hashlib.md5(url.encode("utf-8")).hexdigest()

这个模型对应第 3.1 节的表结构。keyword字段单独建了索引，是为了后续按关键词统计；url_hash设置unique=True后，数据库层面会生成唯一约束，重复写入会触发异常，这比在应用层做一次全表查询再判断是否已存在要可靠得多。

建表前还需要确定数据库连接。一般用create_engine创建引擎，再配合sessionmaker管理会话。连接字符串里平台层的五个核心参数都可以写成显式配置：

from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker DB_URI = ( "mysql+pymysql://root:your_password@127.0.0.1:3306/" "news_spider?charset=utf8mb4" ) engine = create_engine( DB_URI, pool_size=5, max_overflow=10, pool_pre_ping=True, echo=False, ) SessionLocal = sessionmaker( bind=engine, autoflush=False, expire_on_commit=False, )

pool_size=5表示连接池保持 5 个连接，max_overflow=10表示最多额外创建 10 个连接，应对短时峰值请求。pool_pre_ping=True会在每次从连接池取连接前执行一次轻量探测，防止 MySQL 的 wait_timeout 把连接断开后，程序拿到一个已经失效的连接。charset=utf8mb4必须写在连接字符串里，否则即使表建成了 utf8mb4，应用层写入时也可能出现中文乱码。

建表时有两种方式：一是直接在 MySQL 手动执行 CREATE TABLE 语句，二是让 SQLAlchemy 自动创建。开发阶段可以用后者快速迭代结构：

Base.metadata.create_all(engine)

这条命令会检查news_article表是否存在，不存在则按模型定义创建，已经存在则跳过。它的优点是不会改动已有表结构，缺点是你改了模型字段后它不会自动同步。所以我在项目初期依赖create_all，一旦表结构和数据稳定下来，后续改动都用显式 SQL 迁移。

3.3 批量入库：用 ON DUPLICATE KEY UPDATE 把更新和插入合并

逐条session.add()在数据量小的时候没问题，但新闻爬虫一次性可能抓几百条，逐条插入会产生大量数据库往返，速度明显变慢。更高效的做法是用 SQLAlchemy 的 MySQL 方言构造INSERT ... ON DUPLICATE KEY UPDATE语句，一次性批量提交。

写法如下：

from sqlalchemy.dialects.mysql import insert def batch_upsert_news(session, rows): if not rows: return for i in range(0, len(rows), 200): chunk = rows[i:i + 200] stmt = insert(NewsArticle).values(chunk) update_fields = { "title": stmt.inserted["title"], "url": stmt.inserted["url"], "source_name": stmt.inserted["source_name"], "publish_time": stmt.inserted["publish_time"], "summary": stmt.inserted["summary"], "crawl_time": stmt.inserted["crawl_time"], } stmt = stmt.on_duplicate_key_update(**update_fields) session.execute(stmt) session.commit()

调用时只需要把字典列表传进来：

rows = [ { "keyword": "人工智能", "title": "示例新闻标题", "url": "https://example.com/news/1", "url_hash": NewsArticle.make_url_hash("https://example.com/news/1"), "source_name": "示例来源", "publish_time": None, "summary": "摘要内容", }, ] with SessionLocal() as session: batch_upsert_news(session, rows)

这段代码的优势在于，如果插入的数据与已有数据在url_hash唯一索引上发生冲突，MySQL 会直接执行更新操作，把新的标题、摘要、抓取时间刷新进去。这样一来，同一个任务重复执行不会产生重复数据，第二次抓取相当于一次数据刷新。

chunk按每 200 条一组遍历，是为了控制单次 SQL 报文大小。MySQL 对单条 INSERT 能携带的字段数有上限，虽然 200 条远不到极限，但批量大小设在 200 到 500 之间时，性能和内存占用比较平衡。如果抓取源返回的摘要很长，TEXT 字段会占用较多报文空间，此时建议把批量大小调小到 100，避免报“packet too large”之类的错误。

4. 把压缩包里的代码跑起来：环境检查、依赖安装和最小命令行

这类压缩包里的代码通常不是一个可以直接双击运行的成品，它更接近一个半成品骨架。拿到包之后，我建议先不要急着解压运行，按下面步骤做一轮环境检查和文件清单检查，能避免一半以上的启动报错。

4.1 拿到压缩包后先看文件清单，再决定补哪些配置

在终端或者文件管理器里先看压缩包内文件结构。常见形式是：一个入口 python 文件、一个数据模型文件、一个依赖清单文件和一个 README 或配置示例。如果只看到单一 py 文件，也不要慌，说明它大概率把请求、解析和入库逻辑都写在了一起，运行前只需要补好数据库连接变量。

建议创建一个独立目录来解压，避免脚本生成的 HTML 调试文件和数据库配置混在桌面上：

mkdir -p news_spider && cd news_spider unzip ../news_spider.zip ls -la

ls -la会列出所有文件，包括隐藏配置项。如果看到requirements.txt，后续依赖安装就用它；如果没有，就自行安装一套最小依赖，覆盖 requests、BeautifulSoup、lxml、SQLAlchemy 和 pymysql 即可。

接下来确认 Python 和 MySQL 环境版本：

python --version pip --version mysql --version

这个项目需要 Python 3.8 以上，因为 SQLAlchemy 2.x 和 typing 写法都依赖较新的 Python 特性；MySQL 5.7 或 8.0 都能用，但建库时务必指定 utf8mb4 字符集。

4.2 创建数据库并安装依赖

先创建数据库。这里的字符集设置是重点，utf8mb4才能完整存储中文和 emoji 内容，只设utf8在遇到生僻字或特殊符号时会有风险：

mysql -uroot -p \ -e "CREATE DATABASE IF NOT EXISTS news_spider DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"

执行后会提示输入 root 密码。该语句的作用是创建news_spider数据库，默认字符集使用utf8mb4，排序规则选择utf8mb4_unicode_ci，后者在处理中文拼音排序时比utf8mb4_general_ci更规范。

接着创建虚拟环境并安装依赖，避免把包装进系统 Python 环境，污染其他项目：

python -m venv venv source venv/bin/activate # Windows 下用 venv\Scripts\activate pip install --upgrade pip pip install requests beautifulsoup4 lxml sqlalchemy pymysql apscheduler

如果你的压缩包内已经有requirements.txt，直接执行pip install -r requirements.txt更省事。一个最小的requirements.txt内容通常包含这些库：

requests beautifulsoup4 lxml SQLAlchemy PyMySQL APScheduler

没有写版本号的好处是安装时会自动拉取当前环境兼容的版本，缺点是未来某些库大版本更新可能带来接口变化。若追求可重复，应锁定到具体版本；若只是跑通项目，不指定版本更省心。

4.3 最小运行命令与多关键词参数设计

项目入口脚本通常需要支持命令行参数。下面是我常用的一种入口设计，兼顾单关键词和多关键词批量抓取：

import argparse import time import random def parse_args(): parser = argparse.ArgumentParser( description="百度新闻、今日头条关键词新闻爬虫" ) parser.add_argument( "--keyword", required=True, help="搜索关键词，多个词用逗号分隔", ) parser.add_argument( "--pages", type=int, default=3, help="每个关键词抓取多少页，默认 3 页", ) parser.add_argument( "--interval", type=float, default=1.0, help="两次请求之间的间隔秒数，默认 1.0 秒", ) return parser.parse_args() def run_keyword(keyword: str, pages: int, interval: float): # 实际请求和解析逻辑在第二、三章已经给出 # 这里主要是循环页数并调用入库函数 for page in range(pages): print(f"正在抓取关键词: {keyword}, 第 {page + 1} 页") time.sleep(interval + random.random()) if __name__ == "__main__": args = parse_args() keywords = [k.strip() for k in args.keyword.split(",") if k.strip()] for kw in keywords: run_keyword(keyword=kw, pages=args.pages, interval=args.interval)

这段入口脚本有三个关键参数：--keyword接收一个或多个关键词，多个关键词用逗号分隔；--pages控制每个关键词的抓取页数；--interval控制请求间隔。time.sleep(interval + random.random())的作用是在设定间隔基础上额外增加 0 到 1 秒随机延迟，让请求节奏更接近真人操作。

跑通全流程的最小命令是：

python crawl.py --keyword "人工智能" --pages 1 --interval 1.5

如果这条命令执行后，MySQL 的news_article表出现对应记录，说明从抓取到入库的链路已经完整。此时再放开多关键词和页数：

python crawl.py --keyword "人工智能,大数据,云计算" --pages 3 --interval 1.5

多个关键词会按顺序逐个抓取，每个关键词内部再按页码顺序执行。这个设计是刻意避免并发请求的，因为新闻搜索接口对并发很敏感，并发越高越容易被风控，顺序执行的耗时虽然长一些，但胜在稳定。

5. 新闻爬虫避坑：403、空数据、乱码、重复入库的四个高频事故

这一章写的是新闻爬虫开发中常见的真问题。每个问题都按“现象、原因、解决”三步拆开，排查时可以直接对照。

5.1 请求被拦截：响应里出现验证页

现象：requests返回状态码是 200，但解析不到任何新闻列表，打印出来发现页面标题是“百度安全验证”或类似内容。

原因：请求头缺少浏览器特征，或者请求频率过高被识别为脚本。这类验证不一定每次都触发，有时连续刷好几页才触发一次，看起来就像“玄学”。

解决：先检查请求头是否齐全，尤其是 User-Agent 和 Accept-Language。再用随机延迟把请求间隔拉开到 1 到 3 秒。遇到验证页时不要硬重试，我一般会临时跳过当前页，等待 10 秒后重新请求，连续失败 3 次就退出该关键词，避免被加重风控。

import time import random import requests def safe_get(session, url, params, max_retry=3): for attempt in range(max_retry): resp = session.get(url, params=params, timeout=10) if "安全验证" not in resp.text: return resp wait_time = 10 + attempt * 5 print(f"触发验证，等待 {wait_time} 秒后重试") time.sleep(wait_time) return None

max_retry=3是重试次数的上限，wait_time用 10 秒起步并逐次增加到 15 秒、20 秒，给风控一个冷却期。

5.2 今日头条返回空列表：大概率是风控而不是解析问题

现象：JSON 解析成功，也能打印出结构，但列表是空的，或者只有少数几条固定内容。

原因：今日头条搜索接口对频率更敏感，请求太快会导致服务端返回空数据。另一种常见情况是直接请求搜索接口时没有携带足够的 Cookie，服务器认为这不是一个完整会话。

解决：在发搜索请求前，先请求一次今日头条首页获取基础 Cookie，再复用同一个 Session 发起搜索。如果搜索频率控制不住，可以先把间隔拉到 3 秒以上，并且一次任务只抓少量关键词。还需要确认pd=information是否正确，因为不同 tab 对应的内容类型不同。

session.get("https://www.toutiao.com/", timeout=10) # 拿到基础 Cookie 后再执行搜索请求 resp = session.get(url, params=params, timeout=10)

这段代码里的前置请求不是多此一举，它让 Session 里带上初始 Cookie，后续搜索请求看起来更像从某个真实页面跳转过去的。

5.3 中文乱码：页面编码和数据库字符集要同时处理

现象：控制台和数据库里看到的是乱码或???，标题和摘要完全不可读。

原因：网络响应头没声明字符集，requests 默认按 ISO-8859-1 解码；或者数据库连接字符串没带charset=utf8mb4，应用层以 latin1 写入，数据库以 utf8mb4 读取，中文就会变成问号。

解决：双管齐下。请求层面用resp.encoding = resp.apparent_encoding让 requests 自动判断编码：

resp.encoding = resp.apparent_encoding print(resp.text[:500])

apparent_encoding基于响应字节的内容分析出的编码，对中文页面基本可靠。数据库层面要保证DB_URI里有charset=utf8mb4，建表语句也用utf8mb4。这两处都对了，中文乱码问题基本不会出现。

5.4 去重失效：链接跳转和同样标题的不同文章

现象：库里出现同一条新闻的多条记录，用url_hash去重却去不掉。

原因：第一，从搜索结果里直接拿到的链接是跳转链接，同一个新闻正文可能对应多个不同的跳转参数；第二，不同新闻站会转发同一篇统稿，标题一样但 URL 完全不同；第三，发布时间相同但来源不同，不是简单靠 URL 就能判重。

解决：对最终 URL 做 MD5 作为第一指纹，同时生成标题规范化指纹作为辅助判断。标题统一去掉首尾空格、压缩连续空格、转小写，然后取 MD5。入库时把两个 hash 同时写入记录，并在库里建立联合唯一索引。

import re import hashlib def normalize_title(title: str) -> str: title = re.sub(r"\s+", " ", title).strip() return title.lower() def title_hash(title: str) -> str: return hashlib.md5( normalize_title(title).encode("utf-8") ).hexdigest()

需要说明的是，标题 hash 去重存在误伤风险。同一个标题可能是不同媒体报道的不同事件，也可能是同名事件。所以实际使用时，我把最终 URL hash 作为主去重依据，标题 hash 只作为“疑似重复”信号，不参与唯一索引。如果团队内部对数据量要求高，可以把标题 hash 和发布时间联合起来建索引，这样更精准。

6. 从“跑通一次”到“每天跑”：定时增量与数据质量验证

到这里，整个爬虫已经能手工运行。接下来真正考验价值的是能否长期稳定运行。新闻数据只有持续抓取才有分析意义，所以最后两步很关键：定时增量任务，以及验证入库数据质量。

定时任务直接用 APScheduler 可以省去系统 crontab 的配置成本，尤其适合 Windows 和 Linux 跨平台场景。下面这段阻塞式调度器会每小时执行一次所有关键词的抓取任务：

from apscheduler.schedulers.blocking import BlockingScheduler from datetime import datetime def run_all_keywords(): # 从数据库配置表读取关键词列表 keywords = ["人工智能", "大数据"] for kw in keywords: run_keyword(keyword=kw, pages=2, interval=1.5) scheduler = BlockingScheduler() scheduler.add_job( run_all_keywords, "interval", hours=1, next_run_time=datetime.now(), ) scheduler.start()

hours=1表示每 1 小时执行一次，next_run_time=datetime.now()让任务启动后先立即执行一次，不需要等满一个周期。长期跑在服务器上时更建议用cron方式按固定时间执行，比如每天早上 8 点抓一次昨日新闻，这个参数改成小时或分钟级别的示例更直观。

抓完之后，用 SQL 做质量验证比打开表一行行看高效得多。下面这条查询能说明每个关键词在当天的入库量和去重后数量：

SELECT keyword, COUNT(*) AS total_cnt, COUNT(DISTINCT url_hash) AS uniq_cnt FROM news_article WHERE crawl_time >= CURDATE() GROUP BY keyword;

如果total_cnt和uniq_cnt差距过大，说明批量更新逻辑出了问题，或者 URL 指纹生成的时机不对。再按来源统计每日新闻分布，可以发现某个来源突然消失或暴增的异常情况：

SELECT source_name, COUNT(*) AS cnt FROM news_article WHERE crawl_time >= NOW() - INTERVAL 1 DAY GROUP BY source_name ORDER BY cnt DESC;

这个查询能直观看到今天入库的新闻来源分布。如果某一天全部记录都是同一个来源，说明解析器可能只匹配了某一种页面模板，需要及时检查。

我自己的习惯是每天跑完任务后，一定看一遍这两个 SQL 的结果，而不是只在爬虫报错时才关心数据。抓取脚本不报错不等于数据是完整的，只有验证过数量的爬虫才值得长期依赖。愿这一套从抓取到入库再到验证的流程，能帮你把这个压缩包变成真正能用的数据工具，希望帮到你。

本文还有配套的精品资源，点击获取