企业官网建设流程全解析

上一篇聊完基础篇，很多人留言说：照着示例能把一个静态页面抓下来，但一旦页面变成动态数据、数据量超过几百条、或者连续跑几分钟就被封，整个脚本就像玩具一样不顶用。这其实是很多初级爬虫项目从“能跑”到“稳跑”之间最真实的坎。

这篇下篇只做一件事：把单页抓取的脚本，升级成一个能持续产出数据、出错能自愈、并发不失控、断点能续跑的“小工程”。我会从动态页面抓取、数据落库、反爬应对、并发提速、任务调度这些角度逐个拆解，代码都是可以直接抄着改的，案例也用虚拟的场景来说明。适合已经会 requests 和 BeautifulSoup 基础用法、想进阶或准备用爬虫做数据积累的人，如果还没看完上篇，建议先补一下基础再回来。

1. 下篇要解决的问题：爬虫工程化入门

1.1 从“一页能抓”到“持续能抓”到底差在哪

很多人以为爬虫最难的是解析页面，其实解析只是一部分。一个项目真正耗时的是：如何应对动态渲染的页面、如何把数据稳定落入数据库、如何在不触发对方风控的前提下把任务跑完。这几件事其实对应着生产环境里的四个关键词：稳定性、可恢复性、可控速度、数据质量。

我见过不少初级开发者，写了一个能抓首页的脚本就以为大功告成，结果第二天数据变了、接口加了签名，脚本立刻哑火。甚至有人一次性把几千个请求用 for 循环怼出去，几分钟后整个IP段都被限制访问。所以下篇的第一个内容，就是把这些问题拆开，逐个给方案。

从另一个角度看，爬虫本质上是一个“请求—解析—存储”的重复闭环。上篇帮你把闭环跑通，下篇帮你把闭环做厚——每一层都加上容错、日志和对异常情况的兜底。这样当目标网站改版、某个字段缺失、或者网络抖动时，你不会手足无措。

1.2 本篇技术栈全景与选型逻辑

先给一篇本篇涉及的技术清单，方便心里有数：

这套组合不是唯一解，但一定是最容易上手、调试成本最低的路线。比如真实的大规模采集系统会用 Scrapy 这种框架，但在你还没把“采集”这件事本身琢磨明白之前，直接上框架反而容易一头雾水。下面的内容都基于原生 Python 和常用第三方库展开，理解了原理后再去看 Scrapy，会轻松很多。

2. 动态页面抓取：别再看静态源码，学会抓接口

2.1 用开发者工具判断数据来自哪一层

遇到一个页面，先别急着写代码。用浏览器打开目标页面，按下 F12 打开开发者工具，找到“网络/Network”面板，刷新页面，观察所有请求。如果数据的 HTML 结构能在“查看源代码”里直接找到，这是纯服务端渲染，requests 直接请求 URL 就能拿到。如果源代码里没有内容，或者内容是在页面加载后几秒才出现，那基本是 JavaScript 动态产生或通过接口加载的。

最简单的判断方式：在页面右键“查看网页源代码”，搜索一个页面里肉眼可见的标题文字。搜不到，说明数据来自接口或 JS 渲染。这时候切换到 Network 面板，在筛选框里选 XHR 或 Fetch，刷新页面，看到一堆 JSON 请求。点开看 Preview/响应，如果里面正好有你要的标题和列表数据，那就直接对接这个接口，比渲染页面再解析要快得多，也更稳定。

接口请求 URL 通常长得像https://api.example.com/v1/list?page=1&size=20。浏览器里能看到所有请求参数和请求头。把请求头复制下来，在 requests 里模拟一遍，就能拿到 JSON 数据。这个思路可以覆盖大多数资讯类、列表类网站的数据采集。

2.2 直接对接 Json 接口的写法

拿到接口后，代码其实很简单。以某新闻资讯站点的列表接口为例，结构和大部分平台的 JSON 接口类似：

import requests import json api_url = "https://api.example.com/v1/article/list" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0 Safari/537.36", "Referer": "https://www.example.com/news/list", "X-Requested-With": "XMLHttpRequest", } params = { "page": 1, "size": 20, } resp = requests.get(api_url, headers=headers, params=params, timeout=10) data = resp.json() items = data["data"]["list"] for item in items: print(item["title"], item["url"], item["publish_time"])

这种写法有几个点需要注意：第一，resp.json() 之前最好先判断 resp.status_code，避免拿到错误页面时报奇怪的错；第二，接口返回的字段名要确认清楚，嵌套层级也要先打出来看看；第三，接口地址通常带一些时间戳或签名参数，如果目标网站做了签名校验，直接模拟就会失败，这种情况再走下面的渲染方案。

从稳定性角度看，抓接口比抓渲染后的页面更可靠，因为很多动态页面只是套了一层 JS 模板，真正有价值的数据都在接口里。把接口摘出来，等于拿到一个稳定数据源。

2.3 需要渲染时的兜底方案

如果接口加了复杂的签名校验，或者数据是跳转几次以后才出现在页面里，直接抓接口就不太可行。另一个思路是模拟浏览器完整执行 JavaScript，拿到最终渲染结果。Playwright 是目前我比较推荐的工具，比 Selenium 更现代，自动等待机制也更好用。

from playwright.sync_api import sync_playwright with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page() page.goto("https://www.example.com/news/article/12345", timeout=30000) page.wait_for_selector("div.article-content") html = page.content() browser.close()

用 page.content() 拿到完整 HTML 以后，就可以丢给 BeautifulSoup 解析了。这里有个细节：wait_for_selector 是在等某一个关键节点出现，比固定 sleep 几秒可靠得多，页面加载快就能少等，加载慢也不容易空白返回。Playwright 还支持拦截接口请求，如果页面里调了好几个接口，你可以从浏览器的 network 事件里直接把响应提出来，这比等渲染更高效。

Selenium 的原理基本一样，这里不展开两套并讲。如果你只需要偶尔处理一两个动态页，Playwright 足够；如果你的整个站点都是动态渲染，而且页面很多，那就建议别一页一页去渲染，而是用渲染入口模块只处理那些必要的页面，其他页面能走接口就走接口——渲染方案的资源消耗约为普通请求的几十倍。

2.4 解析实操中容易忽略的细节

不管静态页面还是动态渲染页面，解析的时候有几个细节极其影响体验。第一个是空值判断：接口返回的某一条可能没有作者、没有发布时间，直接按字段取值就会 KeyError。写代码的时候尽量用item.get("author")或者item.get("author") or ""，让缺失字段变成空字符串而不是直接崩溃。

第二个是分页参数。很多网站的分页不是简单的 page+1，可能是 cursor 翻页、offset 翻页、或者带了签名的时间戳。建议先手动改参数观察返回结果变化，确定翻页的规律再写循环。第三，列表页和详情页最好分开解析。列表页只拿标题、链接、摘要，详情页再单独请求并解析正文。这样即使详情页失败，也不会影响列表页数据的采集进度重建。

3. 数据落库：结果要能反复用，才算有价值

3.1 建模先行：先设计字段字典再写解析

很多新手爬到数据之后 print 一下就算完事了，稍微好一点的写进 CSV。但当数据量大了以后，你会发现没有结构的输出就是垃圾。我建议拿到一个采集目标时，先坐下来把需要保存的字段列出来，形成一个字典结构，再照着这个结构写解析代码。

以前面提到的新闻文章为例，一个典型的字段字典是：

article_item = { "title": "", "url": "", "source": "", "author": "", "publish_time": "", "summary": "", "content": "", "category": "", }

写解析代码的时候，每一步都往这个结构里填值。这样做的最大好处是后续无论写 CSV、MySQL 还是 MongoDB，代码几乎不需要改动，只要把 items 列表传给存储函数就行。另一个好处是排查问题非常快：哪条数据哪个字段明显不正常，直接定位到对应解析函数。

如果抓取目标非常大，我还会在字典里加一个_raw字段，存原始响应片段，方便以后字段解析有问题时回头核对。这个字段后续入库时可以丢弃，但在调试阶段很有用。

3.2 字段清洗的通用套路

清洗是为了让数据能被直接使用，而不是拿到手还要人工再处理一遍。常见场景有三个：去 HTML 标签、去掉多余的空白字符、把时间格式统一。

去掉 HTML 标签可以先用 BeautifulSoup 提取 text，再把多个空白归一。比如详情页的正文内容，常常夹杂一堆 \n、\t 和连续空格：

from bs4 import BeautifulSoup import re def clean_html(raw_html): soup = BeautifulSoup(raw_html, "html.parser") text = soup.get_text(separator=" ") text = re.sub(r"\s+", " ", text).strip() return text

时间格式统一更关键，新闻文章常见的格式可能是 “2025-03-12 10:30:00”、“2025年3月12日”、“刚刚”、“昨天” 这种相对时间。这都属于时间清洗的范畴，建议把能解析的转成标准 ISO 格式或YYYY-MM-DD HH:MM:SS。解析不了的就留空，不要随便猜。

还有一类清洗是针对列表型字段，比如文章的标签 tags 是个数组，直接存 CSV 会变得很难看。建议用json.dumps(tags, ensure_ascii=False)序列化成字符串来存，或者在有 MongoDB 的环境里直接用数组格式存储。

3.3 CSV 与 Excel 的应急方案

数据量比较少、又没有数据库环境时，CSV 是很好的轻量方案。注意两个细节：第一，CSV 文件一定要指定encoding="utf-8-sig"，这样 Excel 打开才不会出现乱码；第二，写文件时使用newline=""，避免出现空白行。

import csv fields = ["title", "url", "source", "author", "publish_time", "content"] with open("articles.csv", "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=fields) writer.writeheader() writer.writerows(item_list)

如果你更习惯用 Excel 表格处理，也可以把同样的数据写到pandas.DataFrame，再用to_excel()导出。不过 Excel 格式对单 cell 长度有限制，正文内容特別长的文章不建议存 Excel，CSV 反而更通用。

3.4 真正该用的 MySQL 与 MongoDB 写入方案

当数据量达到几万条甚至更多，并且需要按时间、分类去查询时，关系型数据库是更好的选择。以 MySQL 为例，先建一张文章表，核心是要给 url 字段加唯一索引，这是去重的第一道防线。

CREATE TABLE article ( id BIGINT PRIMARY KEY AUTO_INCREMENT, title VARCHAR(500) NOT NULL, url VARCHAR(1000) NOT NULL, source VARCHAR(100), author VARCHAR(100), publish_time DATETIME, content MEDIUMTEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_url (url(255)) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

写入时用 INSERT ... ON DUPLICATE KEY UPDATE，可以做到“存在就更新，不存在就插入”，天然实现增量更新：

import pymysql conn = pymysql.connect( host="localhost", user="root", password="your_password", database="spider", charset="utf8mb4", ) cursor = conn.cursor() insert_sql = """ INSERT INTO article (title, url, source, author, publish_time, content) VALUES (%s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE title = VALUES(title), content = VALUES(content) """ data = [ (item["title"], item["url"], item["source"], item["author"], item["publish_time"], item["content"]) for item in item_list ] cursor.executemany(insert_sql, data) conn.commit() cursor.close() conn.close()

如果页面结构经常变、字段不固定，比如追踪商品价格变动、采集招聘信息，字段一会儿多一会儿少，我更推荐 MongoDB。它的update_one配合$set和 upsert 参数，可以很优雅地做去重和更新：

from pymongo import MongoClient client = MongoClient("mongodb://localhost:27017") col = client["spider"]["article"] for item in item_list: col.update_one( {"url": item["url"]}, # 查询条件 {"$set": item}, # 更新内容 upsert=True, # 不存在则插入 )

给 url 建唯一索引是必须的，否则高并发场景下会出现重复数据。MongoDB 的灵活性和写入速度都比较适合爬虫数据这种“先收了再说”的场景，缺点是不太适合复杂联表统计。我个人习惯是结构化明显的数据用 MySQL，字段变化频繁的用 MongoDB。

4. 反爬应对：稳定采集的前提是“像正常用户”

4.1 被识别的主要原因

服务端识别爬虫并不神秘，无非是看你请求的“特征”和人类使用习惯差多远。最常见的识别维度包括：User-Agent 是不是一个未知的库标识、请求头是否缺少主流浏览器该有的字段、请求频率是不是固定且密集、Cookie 有没有经过真实登录流程、同一个 IP 在短时间内的请求数是否异常。

如果你用默认的python-requestsUA 去请求一个做了基础防护的网站，对方在日志里一眼就能认出来。更严格的服务端还会看浏览器指纹，比如是否有 webdriver 特征、Canvas 指纹是否异常等。这都属于高级反爬，先打好基础再研究不迟。

4.2 请求头部完整态与 User-Agent 池

第一件事，把默认请求头补成完整。真实浏览器至少会带上 Accept、Accept-Language、User-Agent、Referer 等字段。requests 的默认 header 太精简，很容易被识别。我一般这样构造：

import random def build_headers(): ua_list = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_0) AppleWebKit/537.36 Chrome/121.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0", ] return { "User-Agent": random.choice(ua_list), "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Referer": "https://www.example.com/", }

每次请求随机挑一个 UA，看起来更像不同用户。但要注意，UA 池不是万能的，很多网站还会校验其他请求头与 UA 是否匹配。比如你用了 Chrome 的 UA，却用了一个 Python 特色的 header 组合，反而容易暴露。尽量直接用 Browser 默认头部，减少自行删改。

Cookie 也是关键。如果是访客可见的数据，一般带上基础的 session cookie 就行。如果数据需要登录后才能看，就需要先登录获取 cookie，再在每次请求中带上。

4.3 Session 与 Cookie 与登录态管理

requests 的 Session 对象会自动保存服务端返回的 Cookie，并在后续请求中带上，这对保持会话很有用：

session = requests.Session() session.headers.update(build_headers()) # 第一次请求会种下 cookie resp = session.get("https://www.example.com/news") # 后续请求自动携带 cookie resp2 = session.get("https://www.example.com/news/detail/1001")

遇到真正需要登录的场景，我的做法是先用浏览器手动登录一次，把登录成功后的 cookie 导出成一个字典，然后交给 requests 使用。这样在开发阶段比较省事，也不用把账号密码写在代码里。Cookie 一般有有效期，过期后 requests 拿到的就是登录页面，这时候需要在代码里判断响应内容中是否包含“登录”或跳转关键词，发现异常时触发重新登录逻辑。

用 Playwright 或 Selenium 自动化登录获取 Cookie 也是一种常见方案：浏览器里执行登录脚本，登录成功后用context.cookies()导出 cookie，再注入到 requests 中。注意这种方式要求在抓取模块和渲染模块之间传 Cookie，尽量做成一个公共函数统一管理，别分散在代码各处。

4.4 请求频率与随机延迟

比 header 更重要的，是请求频率。很多网站对 Request 频率的敏感度远超对 UA 的敏感度。一个真实用户就算一直点列表页，也很难做到每秒 5 个请求。所以代码里必须加入延迟控制，而且最好带随机性。

import time import random def polite_delay(): time.sleep(random.uniform(1.0, 2.5))

随机延迟不能太规律，否则“每次都睡 1.5 秒”照样会被识别。你可以基于一个初始间隔加抖动，比如 base_delay = 2.0，每次在 1.2 到 2.8 秒之间波动。这是最简单也最有效的低成本反识别手段。

如果目标是大型网站，还需要进一步做频率总控。可以用一个外部计数器，每秒钟最多发出 N 个请求，超出就等待。这样能把“全站请求速率”控制在一个平稳的范围，而不是一会儿猛冲一会儿空窗。

4.5 验证码、滑块与其他硬骨头

反爬做到一定程度就会出现验证码，滑块、点选、图形验证码都是常见形式。我的态度是：不要为了绕过这些机制投入太多精力，更不要把绕过验证码作为项目核心。能做的很有限，而且成本很高。

现实的做法是降低触发概率。比如降低请求频率、减少并发数、增加随机延迟、控制单 IP 单机请求量。真正被 IP 维度限制时，单机方案几乎无解，更可行的是把抓取周期拉长，或者改用分布式多机调度，让每个出口的请求量都减小。但这属于架构层面，不适合在零基础入门里硬讲，这里只把思路提出来。

如果验证码已经挡住了你的流程，就说明目标站点的数据价值太高，或者你的触发频率太激进。这时候应该停下来重新评估：是不是只需要其中一部分数据？能不能换一个公开的数据源？

5. 并发提速：多线程跑任务，也要有纪律

5.1 慢在哪：网络 IO 才是瓶颈

单线程爬虫最大的问题是：每次请求发出后，要等网络往返结束才能继续下一步。这段等待时间占用极少 CPU，纯粹是 IO 等待。多线程的意义就是把多个网络请求的等待时间重叠起来，同一时间里有多个请求在飞，总耗时自然大幅下降。

以抓取 1000 个详情页为例，单线程平均每页 2 秒，总共约 33 分钟。改成 5 个线程同时抓，理论上能压缩到 7 分钟左右。当然实际没法完全线性，因为受网络带宽、目标服务器连接数限制，但几十倍提速是可以见到的。

多线程不是万能的，如果目标服务器根本吃不消，那 “提速” 就变成 “加速被封”。所以并发必须和限速配合使用。下面这个写法是我在多数项目里采用的公式：并发数 × 每请求平均耗时大致等于每秒请求数，把这个每秒请求数控制在目标网站可接受的范围。

5.2 线程池的推荐写法

Python 里最推荐的并发写法是concurrent.futures.ThreadPoolExecutor，语言原生、写法简洁、结果收集方便，不需要手动去维护一堆线程对象。

from concurrent.futures import ThreadPoolExecutor, as_completed urls = [f"https://www.example.com/news/detail/{i}" for i in range(1, 101)] def fetch_detail(url): resp = session.get(url, timeout=10) return parse_detail(resp.text) with ThreadPoolExecutor(max_workers=4) as pool: futures = {pool.submit(fetch_detail, url): url for url in urls} for future in as_completed(futures): url = futures[future] try: item = future.result() save_item(item) except Exception as exc: print(f"抓取失败: {url}, 错误: {exc}")

这段代码的关键是as_completed——谁先完成就先处理谁，不会因为某一条卡住而阻塞整个线程池。异常处理单独放在取结果的地方，这样单条失败只会记日志，不会让整个任务中断。

关于线程数的选择，建议从 3 到 5 起步，逐步增加。小网站可能 2 个线程就够了，大网站反而需要克制。千万别一上来就 50 个线程满速跑，那是给自己挖坑。

5.3 并发数与请求速率的平衡

并发和限速很容易互相矛盾：写多线程是为了快，限速让请求变慢。怎么平衡？我的做法是设置一个全局的“时钟”来控制每秒最大请求数。

可以用一个简单的令牌桶思路：每秒放行 N 个请求，超过就等待。代码大致如下：

import threading import time class RateLimiter: def __init__(self, qps): self.interval = 1.0 / qps self.lock = threading.Lock() self.next_time = time.time() def wait(self): with self.lock: now = time.time() if now < self.next_time: time.sleep(self.next_time - now) self.next_time = time.time() + self.interval

每个线程在发请求前调用limiter.wait()，就把全站 QPS 锁在了设定值附近。比如RateLimiter(qps=2)表示每秒最多 2 个请求。这种全局控制在多线程环境里极其重要，否则并发数一多，瞬时 QPS 就会飙到失控。

我在实际项目里的组合一般是：ThreadPoolExecutor(max_workers=6) 配合 qps=3~5。具体数字根据目标网站的响应速度和数据量去调。如果发现大量请求超时或者被拒绝，第一反应不是加线程，而是降 QPS。

5.4 落库阶段也是 IO

多线程时代，数据写入往往是新的瓶颈。如果你每抓一条就连接一次 MySQL、插入一次，开销会非常大。正确做法是：抓取线程只把结果放入一个内存队列，由专门的写入线程统一批量落库。下面是一个简单的队列结构：

import queue result_queue = queue.Queue() def worker_save(): while True: item = result_queue.get() if item is None: break save_to_mysql(item) result_queue.task_done()

批量写入可以用executemany，把几十条数据一次性提交，减少网络往返和 commit 次数。MongoDB 也一样，尽量用insert_many或批量 update，不要每条都单独等待确认。

提到队列，它还承担了一个重要作用：解耦抓取速度和存储速度。即使某一次数据库连接慢了一点，也不会拖累正在进行中的请求。

5.5 异步是后续方向

如果你想更进一步，可以用 asyncio + aiohttp 实现单线程异步并发。相比多线程，异步的线程切换开销更小，能支撑更大的并发数，但写法和排错难度也上升一个台阶。零基础阶段不建议一上来就学异步，先掌握多线程的工程化写法，等能稳定跑完一个完整项目后，再考虑异步优化。

6. 从脚本到任务：调度、失败重试与日志

6.1 一次运行 vs 每日定时

爬虫最终要解决的是持续更新问题。很多数据源每天都会产生新内容，如果只手动跑一次脚本，那数据永远是旧的。所以要把爬虫写成一个可重复执行的任务，每次执行只处理新增或变化的页面，这就是增量抓取。

增量抓取的实现思路，记录上次抓取的截止时间或最后一条 ID。下一次运行从“游标”位置继续向后抓，抓到已存在数据库中的数据就停止。配合数据库唯一索引，即使重复抓了，写入也不会产生垃圾数据。

把脚本放进定时任务，就完成了从“手动执行”到“自动运行”的升级。

6.2 用 Queue 维护失败任务

一个健壮的爬虫必须处理失败重试。我的做法是准备一个失败任务队列，请求失败或解析失败的任务先放进去，一轮主流程结束后再统一重试，最多重试 3 次。超过重试次数的任务写入日志，等后续人工处理。

task_queue = queue.Queue() fail_queue = queue.Queue() def run_task(task): try: item = fetch_and_parse(task) save_item(item) except Exception as exc: fail_queue.put((task, str(exc))) def retry_failed(): while not fail_queue.empty(): task, exc = fail_queue.get() for attempt in range(1, 4): try: run_task(task) break except Exception: print(f"重试第 {attempt} 次失败: {task}")

把正常抓取和重试分开写“稍显啰嗦”，但工程化的背后就是这种“宁慢勿漏”的思路。批量跑任务时，单条失败往往不是致命问题，反而是日志里追踪不到的隐性失败最麻烦。

6.3 日志：爬虫也需要监控

日志是爬虫项目最容易忽略但最值得投入的部分。不要用print打完就结束，应该用logging模块记录请求状态、耗时、失败原因和重试次数。当任务跑完却发现有一部分数据缺失时，日志能直接告诉你发生了什么。

import logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[ logging.FileHandler("spider.log", encoding="utf-8"), logging.StreamHandler(), ], ) logger = logging.getLogger(__name__) logger.info("开始抓取列表页，页码: %s", page) logger.warning("请求超时: %s", url) logger.error("解析失败: %s", url)

日志里除了错误信息，还应该记录进度：当前抓到第几条、总计多少条、耗时多少。这样即使脚本在服务器后台跑，你也能随时拉日志掌握进度。

一个我常用的技巧是：每完成一个批次，就把当前页码和抓取时间写进一个run_status表或文件。下次启动时先读取这个状态，从上次断点继续。这比从零重抓要省非常多的时间和流量。

6.4 断点续爬的实现思路

断点续爬的本质是“把进度持久化”。对已经抓过的 URL，入库前先做去重判断，入库后自然就腾出了去重标记。而对分页任务，可以做更轻量的状态记录：

import json, os STATE_FILE = "spider_state.json" def save_state(page): with open(STATE_FILE, "w") as f: json.dump({"last_page": page, "last_time": time.time()}, f) def load_state(): if os.path.exists(STATE_FILE): with open(STATE_FILE) as f: return json.load(f) return {"last_page": 0, "last_time": 0}

每次翻页成功后就保存一次状态。下次运行时读取 last_page，从该页码继续。如果中途断开，也不需要重头开始，损失最多一页。

6.5 定时任务与部署

定时触发最简单的方式是用操作系统的 crontab。比如每天凌晨 2 点执行一次采集：

0 2 * * * cd /path/to/project && /usr/bin/python3 main.py >> run.log 2>&1

如果你跑在 Windows 上，可以用计划任务程序。更友好一点的方案是 APScheduler，直接把调度逻辑写进 Python 脚本里，不依赖系统配置。下面是一个最小示例：

from apscheduler.schedulers.blocking import BlockingScheduler def job(): logger.info("定时任务启动") run_spider() scheduler = BlockingScheduler(timezone="Asia/Shanghai") scheduler.add_job(job, "cron", hour=2, minute=0) scheduler.start()

部署上可以先用一台电脑跑，后续数据量大再考虑迁移到云服务器。脚本最好加上 Python 虚拟环境和常用库依赖清单文件 requirements.txt，这样换机器时能快速复现运行环境。

7. 一个小而完整的实战：新闻站增量采集

7.1 需求描述

假设我要采集某新闻门户的新闻列表和详情页，每天更新两次。字段包括标题、URL、来源、作者、发布时间、正文内容，存储到 MySQL。列表页是静态服务端渲染，详情页返回完整 HTML，目前没有任何 JS 渲染需求。这是很典型的基础采集场景，适合把所有工程化技巧串起来。

7.2 整体流程与核心代码

整体流程可以拆成四个函数：fetch_list_page 获取列表页，parse_list_page 解析列表得到详情页链接，fetch_detail 获取详情页，parse_detail 解析详情并返回结构化字典。最后在 main 里用线程池并发处理详情页，并统一落库。

def fetch_list_page(page): resp = session.get(f"https://www.example.com/news/list/{page}", timeout=10) resp.raise_for_status() return resp.text def parse_list_page(html): soup = BeautifulSoup(html, "html.parser") links = [] for a in soup.select("div.news-list a.item-title"): links.append(a.get("href")) return links def fetch_detail(detail_url): resp = session.get(detail_url, timeout=10) resp.raise_for_status() return parse_detail(resp.text, detail_url) def parse_detail(html, url): soup = BeautifulSoup(html, "html.parser") item = { "title": soup.select_one("h1.article-title").get_text(strip=True), "url": url, "source": (soup.select_one("span.source") or {}).get_text(strip=True), "author": (soup.select_one("span.author") or {}).get_text(strip=True), "publish_time": parse_publish_time(soup), "content": clean_html(str(soup.select_one("div.article-content"))), } return item def main(): detail_urls = [] for page in range(1, 6): html = fetch_list_page(page) detail_urls.extend(parse_list_page(html)) polite_delay() with ThreadPoolExecutor(max_workers=4) as pool: futures = {pool.submit(fetch_detail, url): url for url in detail_urls} for future in as_completed(futures): try: item = future.result() save_to_mysql(item) except Exception as exc: logger.error("处理失败: %s, 错误: %s", futures[future], exc)

main 函数里，列表页循环和详情页线程池之间要加礼貌延迟。列表页一般是轻量请求，频率可以相对高一点，但详情页是大页面，最好设置更宽松的等待。

7.3 实战中容易出现的几个问题

第一个坑：列表页里的详情链接可能是相对地址，直接 requests 会报错。组装时要用urllib.parse.urljoin(base_url, link)。第二个坑：详情页如果有分页（一篇文章分好几页），解析时要把多页内容拼接，否则只抓了第一段。第三个坑：发布时间的格式如果特别复杂，解析到一半可能会异常，我会先转为字符串再做正则匹配，解析失败就把时间字段留空，不要影响主流程。

import re from datetime import datetime def parse_publish_time(soup): raw = soup.select_one("time.article-time") if not raw: return None text = raw.get_text(strip=True) m = re.search(r"(\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2})", text) if m: return datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S") return None

7.4 这个流程能扩展成什么

这套流程稍微改改就能用到很多方向：电商平台的价格监控改成定时跑商品页、招聘网站的职位聚合改成解析列表页和职位详情、舆情监控改成按关键词搜索再抓新闻正文。核心不变的是：列表页拿索引、详情页拿完整数据、入库前清洗去重、定时调度跑增量。

下次你遇到一个新的采集目标，不用急着写代码，先按这个流程图想一遍：数据从哪个接口或页面来，字段有哪些，存到哪里，多久跑一次，失败怎么重试。想清楚这些，代码写起来就有章法多了。

8. 踩坑记录与问题排查速查表

8.1 常见问题表

8.2 我的几点经验

写爬虫这几年，最大的体会是：能稳定跑完一年的爬虫，靠的不是多花哨的绕过技巧，而是把每个环节的异常都考虑进去。我现在的习惯是，任何新爬虫写完第一版之后，先故意模拟几个错误场景——断网、字段缺失、页面改版——看它会不会崩。不会崩，说明工程化及格了。

另外一个很实用的小技巧：把采集目标的数据量先掂量清楚。如果只需要几千条数据，完全没必要上多线程和复杂的入库逻辑，脚本足够；如果需要每天更新几十万条，那再考虑线程池、队列和分布式架构。为不存在的规模提前设计，反而是负担。

如果你正在从“会写爬虫”走向“能写好爬虫”，建议把这篇里的代码当成起点，去改造一个自己手里真正需要的案例。只有在真实项目里踩一次数据错乱、反爬升级、定时任务失灵的坑，才会真正理解这些工程化设计为什么存在。跑通之后你会回来告诉我：原来下篇讲的东西，没有一句是废话。