Python留言采集爬虫实战：断点续爬与增量入库指南

简介Renminwang-Message-Crawler-3 压缩包提供了一套基于 Python 的留言板留言爬虫实现面向想掌握网络数据抓取、多进程并行处理与 Selenium 模拟浏览器操作的开发者。代码设计了模拟登录、动态加载等待、留言解析、多进程任务分配和数据存储等环节适合用于学习爬虫框架搭建与反爬策略应对。压缩包整体大小约 1.8MB为 RAR 格式便于直接下载解压。截止目前已有 589 人学习浏览可作为进阶爬虫项目的实践参考。通过阅读这份代码读者能理解 GIL 对多线程的限制以及 multiprocessing 模块的用法同时掌握 Selenium 在 JavaScript 动态页面中的实际应用并对 cookie 维持、元素定位和批量抓取形成完整认知。1. 先弄明白这套留言消息爬虫在爬什么对象、场景与价值边界某个夜里部署的留言采集程序突然不动了日志里反复出现Connection reset by peer消息数据停在凌晨 3 点之前的最后一条。这类问题几乎所有做过留言板采集的人都会遇到。这套名为 Message-Crawler 的留言消息采集包解决的就是把某新闻留言平台上公开的留言消息按栏目、按页稳定地抓下来并落库——不仅是抓一遍还要能断点续爬、去重更新、长期增量运行。它的应用场景集中在舆情监控、新闻线索汇总、留言回复状态跟踪这几类适合已经能写基础 Python 脚本、想直接拿一套完整流程去改的开发者。数据本身是公开页面展示的内容难度不在“能不能爬到”而在“怎么持续稳定地爬到不翻车”。2. 解压与工程结构拿到压缩包后先办这三件事2.1 解压后目录长什么样才算「能跑」常见的做法是先把压缩包解开到一个独立目录再确认里面几个关键文件是否齐全。很多留言采集工具会打包成 rar 格式所以第一步是安装 unrar 并解压。mkdir -p ~/message_crawler cd ~/message_crawler unrar x message-crawler-3.rar tree -L 2如果没有 unrarDebian/Ubuntu 上执行apt install unrarCentOS 上执行yum install unrar。解压完成后用tree -L 2看目录层级。一个能跑起来的留言采集包通常至少包含这几个文件crawler.py 是主入口config.ini 存站点配置和数据库连接requirements.txt 列依赖storage.sql 是建表语句crawl_state.json 是断点续爬的状态文件。如果包里的版本没带 requirements.txt自己补一个就行内容就是 requests、lxml、pymysql 三个库。这里有个细节压缩包名字带 “3”通常说明这是第三个迭代版本。第三版相比早期版本最常见的变化就是把散落在代码里的 URL、UA、数据库密码都收进了配置文件并且新增了断点状态文件。看到解压目录里同时存在 config.ini 和 crawl_state.json基本可以判断这个版本不是为了“跑一次”设计的而是为了“长期挂在服务器上每天跑”设计的。2.2 依赖与配置请求头、存储路径、日志级别该设在哪依赖安装建议用虚拟环境避免把系统 Python 环境搞乱。python3 -m venv venv source venv/bin/activate pip install requests lxml pymysql装完后打开 config.ini重点检查几个配置项。我一般会把站点的列表页模板、详情页模板、请求延时、超时时间、数据库连接都放在这里代码里不出现任何硬编码的 URL 和密码。import configparser cfg configparser.ConfigParser() cfg.read(config.ini, encodingutf-8) base_url cfg.get(site, base_url) list_url cfg.get(site, list_url) # 列表页模板{page} 是页码占位符 detail_url cfg.get(site, detail_url) # 详情页模板{msg_id} 是消息ID占位符 delay cfg.getfloat(crawler, delay) # 相邻两次请求的间隔单位秒 timeout cfg.getint(crawler, timeout) # 单次请求超时 db_cfg { host: cfg.get(mysql, host), port: cfg.getint(mysql, port), user: cfg.get(mysql, user), password: cfg.get(mysql, password), database: cfg.get(mysql, database), }这段代码做的事就是把配置和逻辑分离。为什么不要硬编码因为换一台机器、换一个栏目、换一个数据库只需要改 config.ini就能重新跑起来不用动代码。特别是数据库密码这类敏感信息放进代码里一旦传到 Git 仓库就是事故。几个核心配置项的参数说明配置项作用推荐值delay相邻请求间隔单位秒1.5 到 3.0timeout单次请求超时单位秒10list_url列表页 URL 模板https://example.com/message/index_{page}.htmluser_agent访客 UA真实浏览器的 UA 字符串retry_times单页失败重试次数3delay 是这套配置里最关键的参数。留言平台的压力远小于电商网站但防护策略往往更敏感1 秒以内的间隔跑半小时就可能触发限制。我一般起步设 2 秒跑稳定了再慢慢降到 1.5 秒低于 1 秒就不建议了。2.3 首次运行前验证先发一个探测请求在写完整爬虫之前先用一个短脚本验证网络通不通、页面结构对不对、UA 是否被接受。import requests s requests.Session() s.headers.update({User-Agent: cfg.get(site, user_agent)}) url list_url.format(page1) resp s.get(url, timeoutcfg.getint(crawler, timeout)) print(status:, resp.status_code) print(length:, len(resp.text)) print(encoding:, resp.encoding) print(sample:, resp.text[:200])这里用requests.Session()而不是直接requests.get()目的是让后续请求自动携带第一次响应下发的 Cookie。很多留言平台第一次访问会种一个会话 Cookie不带这个 Cookie 直接抓详情页很容易被判定为异常访客。打印出 status、length、encoding 三个值200、长度大于 2000、编码是 utf-8 或 gbk基本就能往下走。如果 status 是 302说明被重定向到了别的页面多半是 UA 被识别了。3. 把留言消息真正跑起来从列表页到详情页的完整流程3.1 列表页和详情页的 URL 规律v3 改了什么留言平台的消息列表通常有固定的分页规律常见格式是index_1.html、index_2.html这样递增页码详情页则用detail_12345.html这样带消息 ID。v3 相比早期版本最实用的改动就是不再模拟点击“下一页”而是直接把 URL 模板放进配置里按页码递增。这样断点续爬只需要记录最后一次成功的页码重新启动后从last_page 1继续而不是重新解析一遍页面里的“下一页”链接。另一个常见做法是列表页支持按栏目和日期过滤URL 里带上column_id和start_date参数。如果目标是长期增量采集建议在配置文件里预留这两个参数位虽然不一定每个平台都支持但支持的话能省掉很多过滤无效数据的功夫。3.2 解析与入库标题、正文、回复状态、发布时间怎么落库列表页解析用 lxml 写 XPath 是主流方案。下面是典型的解析函数选择器里的 class 名需要按实际页面结构调整。import re from lxml import html def parse_list_page(html_text, base_url): doc html.fromstring(html_text) items [] for li in doc.xpath(//ul[contains(class,msg-list)]/li): try: title_node li.xpath(.//a[contains(class,msg-title)])[0] title title_node.text_content().strip() href title_node.get(href) if not href.startswith(http): href base_url href msg_id re.search(r(\d)\.html, href).group(1) pub_date li.xpath(.//span[contains(class,date)]/text())[0].strip() status li.xpath(.//span[contains(class,status)]/text())[0].strip() items.append({ msg_id: msg_id, title: title, href: href, pub_date: pub_date, status: status }) except (IndexError, AttributeError): continue return itemsmsg_id 是从详情页链接里提取的数字它是后续去重的业务主键。status 字段一般是“已回复”“待回复”“已转交”这类文本入库前建议统一映射成数字。这个映射可以用字典做{待回复: 0, 已回复: 1, 已转交: 2}避免数据库里出现一堆中文重复值。入库时用 pymysql 执行 INSERT核心是ON DUPLICATE KEY UPDATEdef save_message(conn, msg, content, reply_text): sql INSERT INTO messages (msg_id, title, content, reply_status, reply_content, publish_time, crawl_time) VALUES (%s, %s, %s, %s, %s, %s, NOW()) ON DUPLICATE KEY UPDATE reply_status VALUES(reply_status), reply_content VALUES(reply_content), crawl_time NOW() with conn.cursor() as cur: cur.execute(sql, (msg[msg_id], msg[title], content, msg[status], reply_text, msg[pub_date])) conn.commit()这段 SQL 的逻辑是msg_id 没存在过就插入新纪录msg_id 已存在则只更新回复状态、回复内容和抓取时间。这样第二次运行同一批页面时不会产生重复行而是把平台上新增的回复补进来。留言类数据最有价值的部分就是回复状态变化比如“待回复”变成“已回复”这个更新逻辑能让你追踪到每条留言的办理进度。3.3 断点续爬与增量拉取程序挂了不用从头再来留言平台的数据量一般几万条起如果每次中断都从头抓既浪费时间又增加被限流的风险。断点续爬的做法是维护一个 JSON 状态文件每抓完一页立刻写入页码。import os, json state_file crawl_state.json if os.path.exists(state_file): with open(state_file, r, encodingutf-8) as f: state json.load(f) else: state {last_page: 0} start_page int(state.get(last_page, 0)) 1 total_pages get_total_pages() # 从第一页解析出的总页数 for page in range(start_page, total_pages 1): page_url list_url.format(pagepage) ok crawl_one_page(page_url, page) if ok: state[last_page] page with open(state_file, w, encodingutf-8) as f: json.dump(state, f, ensure_asciiFalse, indent2) time.sleep(delay)这里的关键是写入时机必须每页成功后就写不能等整个循环结束再写。血的教训是循环跑了几百页之后程序被系统杀掉状态还停在初始值重启只能从头开始。数据没丢就是后悔药但没必要靠运气。增量拉取则更省资源。列表页按时间倒序排列时新留言的 msg_id 比旧留言大。启动时先读取上一次抓到的最大 msg_id一旦当前页第一条记录的 msg_id 不大于它说明这一页开始往后的都是旧数据本轮直接停止。这里的状态文件就需要同时记录last_msg_id和last_page两个字段。3.4 请求节奏与重试把翻车概率压到最低留言采集的翻车大多不是网络问题而是请求频率问题。一个稳健的请求函数应该同时处理超时、状态码异常、响应体过短三种情况。import random import time def fetch_with_retry(session, url, retries3, delay1.5): for attempt in range(retries): try: resp session.get(url, timeout10) if resp.status_code 200 and len(resp.text) 2000: return resp print(f第{attempt 1}次返回异常: status{resp.status_code}, len{len(resp.text)}) except requests.exceptions.RequestException as e: print(f第{attempt 1}次请求异常: {e}) time.sleep(delay * (attempt 1)) return Nonelen(resp.text) 2000是一个很实用的防御条件。正常留言列表页至少几千字节而拦截页通常只有几百字节。状态码 200 不代表拿到了列表页可能是拦截页。重试间隔用指数退避第一次失败等 1.5 秒第二次等 3 秒第三次等 4.5 秒给服务端留出恢复时间。成功返回后主循环里再 sleep 一个基础延时两次请求之间至少隔 1.5 到 2 秒。4. 留言数据的落地与校验MySQL 表设计、并发取舍、对账方法4.1 建表 SQL主键、唯一索引和字段类型怎么选留言数据落库前先想清楚表结构。下面是练手项目里常用的建表语句字段覆盖了标题、正文、回复状态、回复内容和发布时间。CREATE TABLE IF NOT EXISTS messages ( id INT AUTO_INCREMENT PRIMARY KEY, msg_id VARCHAR(64) NOT NULL COMMENT 平台原始消息ID, title VARCHAR(512) NOT NULL DEFAULT , content TEXT, reply_status TINYINT DEFAULT 0 COMMENT 0待回复 1已回复 2已转交, reply_content TEXT, publish_time DATETIME DEFAULT NULL, crawl_time DATETIME DEFAULT NULL, UNIQUE KEY uk_msg_id (msg_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;主键用自增 id唯一索引建在 msg_id 上这是留言采集的标准做法。平台的消息 ID 才是业务上的唯一标识自增 id 只是物理定位用。如果主键直接设成 msg_id后续如果平台调整 ID 规则迁移数据时会更麻烦。utf8mb4 字符集是为了防止留言正文里的 emoji 写入报错留言内容里出现 emoji 的概率比想象中高。reply_status 用 TINYINT 而不是字符串是为了后续按状态聚合时更快也避免同一状态出现“已回复”和“已 回复”这种脏数据。4.2 并发取舍单线程为主别急着上多线程很多开发者拿到爬虫第一反应是上多线程加速但在留言采集这个场景里我建议先算一笔账。单线程加 2 秒延时一小时抓 1800 条一天 4 万条足够支撑一个中小型舆情监控项目。留言平台的数据量通常就是几万到几十万条单线程全量抓一遍也就一晚上。多线程带来的收益有限代价却很具体请求频率升高更容易触发验证拦截多个线程共用一个 Session 时 Cookie 串扰可能拿到别人的会话状态MySQL 连接冲突导致插入报错。如果确实要加速控制在 3 个线程以内并且每个线程用独立的 Session。from concurrent.futures import ThreadPoolExecutor sessions [requests.Session() for _ in range(3)] for i, s in enumerate(sessions): s.headers.update({User-Agent: user_agent_list[i % len(user_agent_list)]}) def fetch_one(url, session): return fetch_with_retry(session, url) with ThreadPoolExecutor(max_workers3) as pool: futures [pool.submit(fetch_one, url, sessions[i % 3]) for i, url in enumerate(urls)] for future in futures: result future.result() if result is not None: handle_page(result.text)这段代码的关键是sessions[i % 3]让每个线程固定使用自己的 Session避免 Cookie 互相覆盖。max_workers 设成 3 是经验值超过 5 之后限流概率会明显上升收益却很小。4.3 数据校验采集量与页面总数对账采集跑完不是终点要能证明数据是完整的。留言列表页底部通常会显示“共 X 条留言”或者“共 X 页”这是对账的参考值。SELECT COUNT(*) AS total, COUNT(DISTINCT msg_id) AS unique_total, MIN(publish_time) AS min_time, MAX(publish_time) AS max_time FROM messages; SELECT reply_status, COUNT(*) AS cnt FROM messages GROUP BY reply_status;第一条 SQL 看 total 和 unique_total 是否一致不一致说明有重复数据要去查主键设置。MIN 和 MAX 时间能确认采集的时间窗口是否覆盖目标范围。第二条 SQL 按回复状态分组用来验证状态映射是否正确比如正常情况下“已回复”的占比应该和平台上显示的比例大致相当。如果对账发现库里数量比页面显示的少先别急着怀疑爬虫。留言平台本身会删除违规留言历史数据下架也会导致页面总数减少差距在 2% 以内基本属于正常现象。5. 避坑与排查留言采集里最常踩的五个坑5.1 乱码控制台正常入库后全变问号现象程序在终端里打印标题一切正常但写入 MySQL 后查询出来全是乱码或问号。原因通常是两步requests 拿到响应后没有正确设置编码或者 MySQL 表字符集不是 utf8mb4。解决方法是请求返回后立即指定编码再检查建表语句。在解析函数之前加一行resp.encoding utf-8如果页面实际是 GBK就把resp.encoding设成gbk。拿不准的时候用resp.apparent_encoding让 requests 根据页面字节内容自动判断。另外确认连接数据库时设置了charsetutf8mb4pymysql 连接串里漏掉这个参数即使表结构没问题也会乱码。5.2 重复数据跑第二次后记录翻倍现象同一份采集脚本跑两遍数据库里的记录数量直接翻倍。原因基本可以锁定为主键设置有问题。多数留言平台的消息 ID 才是唯一键如果用自增 id 作为唯一依据那每次插入都是新纪录。解决方法是给 msg_id 建唯一索引插入语句改成ON DUPLICATE KEY UPDATE。改动之后第二次跑相同页面时已存在的记录只会更新回复状态不会新增行。5.3 请求拦截状态码 200但解析出来是空列表现象程序没有报错list_items 返回空数组日志里打印的响应长度只有几百字节。原因不是代码逻辑错了而是请求频率过高服务端返回了验证拦截页。拦截页的状态码也是 200但页面内容和留言列表完全不同。解决方法是先看响应长度len(resp.text) 2000基本可以判定被拦这时立即停止程序而不是继续抓。休息 20 到 30 分钟再恢复同时把延时调大。验证拦截页是服务端的明确信号不建议硬闯做留言数据采集讲究的是长期主义。5.4 日期格式不统一时间排序乱套现象publish_time 字段排序时同一天的记录有的排在早上有的排在晚上仔细一看数据有的是2025-01-01 10:00:00有的是2025-01-01 10:00还有的只有日期没有时间。原因很简单平台编辑录入时格式混用。解决方法是入库前统一用 datetime 解析。from datetime import datetime def parse_dt(s): s s.strip() for fmt in (%Y-%m-%d %H:%M:%S, %Y-%m-%d %H:%M, %Y-%m-%d): try: return datetime.strptime(s, fmt) except ValueError: continue return Noneparse_dt 返回 None 时可以打印原始值做人工排查而不是直接跳过因为日期是留言分析里最常用的维度。5.5 重启后从头抓状态文件写晚了现象程序跑了几百页后手动杀掉重启发现日志又显示从 page 1 开始。原因大概率是状态文件在循环结束后才写入或者写入时用了追加模式导致 JSON 解析失败。解决方法是每页成功后立即写状态文件并在启动时打印起始页码用于确认。另一个隐蔽问题是多进程同时跑同一任务两个进程同时写状态文件会互相覆盖。加文件锁可以解决。import fcntl with open(state_file, w, encodingutf-8) as f: fcntl.flock(f, fcntl.LOCK_EX) json.dump(state, f, ensure_asciiFalse, indent2)这个锁是进程级的确保同时只有一个进程能写入状态。如果只是单进程跑普通写入就够了。6. 进阶用法把 v3 升级成每天自动跑的增量采集服务6.1 定时调度与日志轮转留言采集最舒服的形态是每天凌晨自动跑增量早上起来看结果。用 crontab 就可以实现30 3 * * * cd /opt/message_crawler /usr/bin/python3 crawl.py --incremental logs/crawl.log 21凌晨 3 点半这个时间点网络空闲也不会和平台白天的编辑更新冲突。增量模式直接读状态文件里的 last_msg_id只抓新增部分通常几分钟就跑完。日志按天切割用 logging 的 TimedRotatingFileHandler避免单个日志文件膨胀到几个 G。import logging from logging.handlers import TimedRotatingFileHandler handler TimedRotatingFileHandler(logs/crawl.log, whenmidnight, backupCount7) logging.basicConfig(handlers[handler], levellogging.INFO, format%(asctime)s %(levelname)s %(message)s)backupCount 设成 7保留一周日志排查问题时足够用。日志里要记录每个阶段结束的页码和数据量格式越规整后面写巡检脚本越省事。6.2 自动对账每天跑完发一条统计结果定时任务跑完后加一个小脚本做数据巡检统计当天新增了多少条、总共有多少条、按回复状态分布如何。巡检不通过时输出告警信息。def daily_check(): total get_total_count() today_count get_today_count() print(f[CHECK] total{total}, today{today_count}) if today_count 0: print([WARN] 今日新增为0请检查列表页是否改版)这里的告警只输出到日志就行不用做太复杂的通知机制。我自己每次部署这类包都会在 cron 里多加一行记录结束时间的日志哪怕只是echo $(date %F\ %T) done。因为踩过太多次“看似在跑其实程序早已退出”的坑——进程还在但循环因为某个异常 break 了日志里看不出任何问题。留言数据采集拼的不是技巧是稳定和细心每一条能对上的数据都比花哨的并发方案更有价值。希望帮到你。本文还有配套的精品资源点击获取