Python爬虫抓取论坛帖子：requests+BeautifulSoup实践

做数据这块的人，大概率都遇到过这样的需求：某个老牌的兴趣分区论坛里，有一批关于产品吐槽、使用体验、选购纠结的帖子，市场或者运营的同事想拿去做一轮文本分析，或者你自己想扒点语料练手。这时候第一个念头通常就是——写个 Python 爬虫，把论坛帖子内容拉下来。标题里说的"Python 爬虫获取论坛帖子内容"，说的就是这么一件事：用 requests 把列表页和详情页的 HTML 拿回来，用解析库把标题、作者、发布时间、正文、回复数拆成结构化字段，再落成 CSV 或者 JSONL 供后续处理。它不算什么高技术难度的活儿，但坑全在细节上：分页参数怎么算、页面上哪些是服务端渲染的、抓快了会怎么被"劝退"、解析器选哪一套不容易翻车。

这篇东西适合两类人看。一是刚学完 requests 和 BeautifulSoup、想找一个真实站点练手的朋友，我会把每一步的意图和判断依据讲清楚；二是已经在跑抓取任务、但脚本老是中途挂掉、数据缺一块的老手，后半部分的限速、退避、断点续爬和排查表更对你有用。下面提到的站点统一用tieba.example.com作占位，你把域名换成自己实际要访问的那个就行，代码逻辑完全通用。

1. 项目整体设计与思路拆解

1.1 先把"要抓什么"这件事钉死

很多人一上来就写requests.get，写了三十行发现字段对不上，回头改结构，白干。我的习惯是先花二十分钟做需求清单，把"输出长什么样"写在纸上。这个项目我最后定的输出是两张表：一张帖子维表，一张回帖事实表。

帖子维表按帖子一行，字段是tid（帖子唯一标识）、title、author、reply_count、view_count、last_reply_time、url。回帖事实表按楼层一行，字段是tid、floor、author、post_time、content、is_楼中楼。分成两张表而不是揉成一张，理由很实在：一个热门帖可能有上千楼，揉成一张表会让帖子级别的字段重复上千次，后面做去重和统计都难受。

还有一个容易被忽略的点是抓取粒度。论坛动辄几百万帖，全站抓既没意义也不现实。我一般会先做一轮"列表页扫描"，只抓前 N 页（比如前 20 页，约 1000 条帖子），拿到tid和reply_count之后做一次筛选：回复数小于 3 的跳过，内容明显是广告或者纯表情的跳过。这一步能把详情页的请求量砍掉六七成，是整个方案里性价比最高的优化。

注意：筛选逻辑要在列表页阶段完成，不要等抓到详情页再丢弃。一次多余的详情页请求，成本是列表页的十几倍。

1.2 技术选型的取舍：为什么不一开始就上重型框架

工具选型上我踩过两种极端的坑。一种是新手最常见的：听说 Scrapy 强大，直接上框架，结果被settings.py、middlewares、pipelines这一套配置绕晕，一个下午过去还没跑出第一条数据。另一种是我自己的坑：图省事全用requests + 正则，正则写了两百行，页面一改版全线崩溃。

这个项目的体量（几千到几万条记录、单机、不需要分布式）决定了正确的答案是中间路线：requests负责网络层，BeautifulSoup负责解析层，中间用lxml当解析后端提速。理由如下表。

无头浏览器这一项我特意列出来，是因为很多人一看到页面数据"抓不到"就条件反射上浏览器。但帖子列表和帖子正文这类内容，绝大多数情况下是服务端直出的——你右键"查看网页源代码"，如果能在里面搜到帖子标题，那就说明 HTML 里已经有了，requests完全够用，上浏览器纯属自己给自己增加负担。

1.3 合规边界：这是技术问题，也是长期可维护性问题

写抓取脚本绕不开这件事，我把它放在设计阶段讲，而不是最后加一句免责声明，因为它直接决定了你的脚本能活多久。我的三条底线是这样的。

第一，先看 robots 协议。访问站点/robots.txt，看User-agent: *下面的Disallow里有没有你要抓的路径。有的话就换目标，别硬来。

第二，控制请求频率。单线程、每次请求之间随机间隔 1.5～3 秒，一晚上跑个几千条，对任何站点都不构成压力。反过来，50 个线程怼上去，结果一定是先被限速、再被拉黑，你的任务反而完不成。

第三，只取公开内容，不碰个人隐私。抓标题、正文、昵称这类公开展示的信息没问题；手机号、身份证、私信、需要登录才能看到的内容，一律不碰。抓下来的数据自己分析可以，原样搬运发布就是另一回事了。

提示：如果目标页面明确要求登录才能查看，或者频繁弹出行为验证，正确的做法是停下来——降低频率、错峰运行，或者干脆放弃这个数据源。任何"必须想办法绕过"的念头，都应该在这一步掐掉。

2. 抓取前的准备与页面结构分析

2.1 URL 规律与分页参数，是整个脚本的地基

论坛类站点的 URL 设计几乎都遵循同一套思路：用查询参数控制页码和排序。这个站点列表页的形态大致是/?kw=关键词&ie=utf-8&pn=0，其中pn就是偏移量，而不是页码。这是个巨坑——很多人以为pn=1是第二页，结果pn=0和pn=1返回的内容几乎一样，抓了一堆重复数据还找不到原因。

正确的理解是：pn表示"跳过多少条"，每页固定 50 条，所以第 1 页是pn=0，第 2 页是pn=50，第 n 页是pn=(n-1)*50。写代码的时候我建议直接维护一个偏移量列表，而不是用for page in range(1, 21)然后pn=page*50，后者一旦搞错了起始值，整批数据的页码就全错位了。

PAGE_SIZE = 50 PAGE_COUNT = 20 offsets = [i * PAGE_SIZE for i in range(PAGE_COUNT)] # 0, 50, 100, ...

详情页那边还有一层嵌套分页：帖子正文在第一页，回帖超过一定数量后需要?pn=继续翻。这层的偏移量通常是每页 30 层左右，具体数值以实际页面为准，不要凭记忆写死，翻两页数一下就知道了。我习惯在配置区把这几个数字都提出来，注释写清楚"从页面实测得来"，后面改起来一目了然。

2.2 用开发者工具定位节点，比猜选择器靠谱十倍

打开目标列表页，F12 打开开发者工具，用元素选择器（那个小箭头图标）点中一条帖子标题，看它在 DOM 里长什么样。你会看到类似这样的结构：外面一个li，带j_thread_list这个 class，还有>HEADERS = { "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-Language": "zh-CN,zh;q=0.9", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", }

Cookie 这块我的态度比较明确：能不带就不带。很多人习惯把浏览器里的完整 Cookie 复制进脚本，图的是"看起来更像真人"。但这带来两个后果：一是 Cookie 过期后脚本会静默地抓到登录页或者空内容，你排查半天以为是选择器失效；二是把账号凭证硬编码进代码，一旦代码外泄就是安全隐患。除非目标页面明确要求登录后可见（这种情况应当重新评估合规性），否则匿名访问加合理频率就够了。

另外建议用requests.Session()而不是裸requests.get()。Session 会自动复用 TCP 连接，一轮几百个请求下来能省掉大量握手开销，顺带把 Cookie 容器也管住了，代码还更整洁。

3. 核心代码实现：从列表页到详情页

3.1 列表页解析：把能拿的字段一次性拿全

列表页的价值不只是"拿到详情页链接"。列表页上其实已经有很多字段了——标题、作者、回复数、最后回复时间——这些如果不在这里拿走，后面还得再进详情页抓一遍，纯浪费。所以我的列表解析函数会尽量把可见字段全收。

from urllib.parse import urljoin from bs4 import BeautifulSoup BASE = "https://tieba.example.com" def text_of(node): """节点存在就返回去空白文本，不存在返回空串。""" return node.get_text(strip=True) if node else "" def parse_list(html, keyword): soup = BeautifulSoup(html, "lxml") rows = [] for li in soup.select("li.j_thread_list"): a = li.select_one("a.j_th_tit") if not a: continue rows.append({ "tid": li.get("data-tid", ""), "keyword": keyword, "title": a.get_text(strip=True), "url": urljoin(BASE, a.get("href", "")), "author": text_of(li.select_one("a.frs-author-name")), "reply_count": text_of(li.select_one(".threadlist_rep_num")), }) return rows

这里有个小设计值得说：text_of这个辅助函数。新手写解析最常犯的错是直接li.select_one(".xxx").get_text()，一旦某个节点不存在（比如匿名发帖没有作者名），立刻AttributeError，整个循环崩掉，前面抓的全白费。用一个"安全取值"函数包一层，把异常变成空字符串，脚本的健壮性会上一个台阶。这个习惯我从写第一个爬虫一直保留到现在。

还有一个细节：reply_count拿到的是字符串，可能是 "12" 也可能是 "1.2万"。落库前必须做一次规范化，否则后面排序、筛选全乱。

import re def parse_count(s): s = s.strip() if not s: return 0 m = re.match(r"^([\d.]+)\s*(万)?$", s) if not m: return 0 n = float(m.group(1)) return int(n * 10000) if m.group(2) else int(n)

3.2 详情页解析：楼层、作者、正文三件套

详情页的结构比列表页复杂，因为要区分主楼和回复楼，还要处理楼中楼。我的做法是先用一个粗粒度的选择器把所有"楼层块"选中，再在块内部提取字段，这样即使页面外面套了几层无关的 div，也不会影响结果。

def parse_detail(html, tid): soup = BeautifulSoup(html, "lxml") posts = [] for div in soup.select("div.l_post"): content = div.select_one("div.d_post_content") if not content: continue # 广告位或空楼层，跳过 posts.append({ "tid": tid, "floor": text_of(div.select_one("span.tail-info")), "author": text_of(div.select_one("a.p_author_name")), "content": content.get_text("\n", strip=True), }) return posts

content.get_text("\n", strip=True)这行里的"\n"参数经常被忽略，但它很有用。正文里往往有多个<br>或者多个<div>分段，默认的get_text()会把它们粘成一整行，"今天下雨了明天可能晴"这种语义就丢了。加上换行分隔符，段落关系就保留下来了，后面做分词、做情感分析时质量明显不同。

楼中楼（回复里的回复）在 DOM 里通常是嵌套在楼层的子节点中，选择器要更深一层，比如div.lzl_list。这块要不要抓，取决于你的分析目标。如果只是做帖子级别的主题聚类，楼中楼可以不抓；如果要研究对话结构，就必须抓，而且要注意楼中楼没有独立的楼层号，得自己用tid + 父楼层 + 序号拼一个复合主键，否则入库时会撞唯一索引。

注意：正文里常混着表情图片、@提及、超链接。get_text会把@某某保留下来，但图片会被丢掉，只留下一个空的 img 标签。如果你的分析需要知道"这条回复带了图"，得在get_text之前单独判断一下content.select("img")的数量。

3.3 翻页、去重与断点续爬，决定脚本能不能跑通宵

一次跑几千条请求，中途遇到网络抖动、临时限速、机器休眠是必然事件。没有断点机制的脚本，一旦在第 800 条挂掉，你就得从头再来一遍——既浪费时间，又给对方站点增加无谓的压力，这在礼貌上也不合适。

我的做法是把"进度"落成一个很小的 JSON 文件，记录已完成的偏移量和已抓到的tid集合。

import json, os STATE_FILE = "state.json" def load_state(): if os.path.exists(STATE_FILE): with open(STATE_FILE, "r", encoding="utf-8") as f: return json.load(f) return {"done_offsets": [], "seen_tids": []} def save_state(state): tmp = STATE_FILE + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False) os.replace(tmp, STATE_FILE) # 原子替换，防止写一半断电损坏

os.replace这个写法值得单独提一句。直接open(..., "w")覆盖原文件，如果正好在写入过程中进程被杀，你会得到一个被截断的 JSON，下次启动直接解析失败。先写临时文件再原子替换，是处理状态文件的通用套路，成本极低，收益很大。

去重则用seen_tids这个集合。列表页翻页时偶尔会因为你懂的或不懂的原因返回重复帖子（比如置顶帖在每页都出现），进详情页之前先查一下集合，能省下不少重复请求。

state = load_state() seen = set(state["seen_tids"]) for off in offsets: if off in state["done_offsets"]: continue html = fetch(f"{BASE}/?kw={kw}&ie=utf-8&pn={off}", session) if not html: continue for row in parse_list(html, kw): if not row["tid"] or row["tid"] in seen: continue if parse_count(row["reply_count"]) < 3: continue seen.add(row["tid"]) # 进详情页... state["done_offsets"].append(off) state["seen_tids"] = list(seen) save_state(state)

3.4 数据落地：为什么我默认选 JSONL 而不是 CSV

落盘格式这件事，很多人随手就写 CSV，然后在中途遇到两个问题：正文里的换行和逗号把列冲散；抓了五千条之后程序挂了，CSV 文件尾部是半行残缺数据，整个文件都没法用。

JSONL（每行一个 JSON 对象）恰好解决这两个问题。它天然支持嵌套和换行，追加写的时候一行一条，即使中途崩溃，最坏情况也只是最后一行不完整，前面的数据完好无损。等全部跑完，再用 pandas 一次性转成 CSV 交付。

import json def append_jsonl(path, records): with open(path, "a", encoding="utf-8") as f: for r in records: f.write(json.dumps(r, ensure_ascii=False) + "\n") # 全部抓完后转换 import pandas as pd df = pd.read_json("posts.jsonl", lines=True) df.to_csv("posts.csv", index=False, encoding="utf-8-sig")

encoding="utf-8-sig"这个细节是给 Excel 用户准备的。不带 BOM 的 UTF-8 文件，用 Excel 直接双击打开中文会变乱码，加上 BOM 就正常了。这不是技术问题，是协作问题——你的数据最终大概率要给不写代码的同事看，这一步能省掉好几轮沟通。

4. 稳定性优化：让脚本能安安稳稳跑完

4.1 限速策略：随机间隔比固定间隔好用

固定sleep(2)是最容易写的，但它有个隐患：请求节奏过于规律。我的习惯是用random.uniform(1.5, 3.0)，让间隔在区间内浮动。这里不是为了对抗什么，纯粹是因为匀速请求在网络层面更容易形成尖峰，而随机间隔的分布更平滑，对站点更友好，也更不容易触发临时的频率限制。

import random, time def polite_sleep(): time.sleep(random.uniform(1.5, 3.0))

再进一步，我会给整个任务设一个"每小时上限"。比如每小时最多 600 个请求，跑满就sleep到下一个整点。这个约束写在配置文件里，比散落在代码各处的sleep好维护得多，也方便你随时根据自己的时间安排调整——比如设成"晚上八点后自动停"，机器就不用整夜开着。

4.2 指数退避与异常分层，这是重试逻辑的关键

遇到403或者429，说明对方已经在明确表达不欢迎了。这时候最糟糕的应对是"立刻重试"，那等于火上浇油。正确的做法是停下来，等更久，再试一次；如果还不行，就放弃这一条、继续下一条，而不是卡死在这里。

import requests, time def fetch(url, session, tries=3): for i in range(tries): try: r = session.get(url, headers=HEADERS, timeout=10) except requests.RequestException as e: print(f"[网络异常] {url} -> {e}") time.sleep(5 * (i + 1)) continue if r.status_code == 200: if not r.encoding or r.encoding.lower() == "iso-8859-1": r.encoding = r.apparent_encoding return r.text if r.status_code in (403, 429): wait = 30 * (2 ** i) # 30s, 60s, 120s print(f"[限速] {r.status_code}，等待 {wait}s：{url}") time.sleep(wait) continue if 500 <= r.status_code < 600: time.sleep(10 * (i + 1)) continue print(f"[跳过] {r.status_code}：{url}") return None return None

这段代码里有个容易被忽略的处理：r.encoding的判断。有些站点的响应头里 charset 写得不规范，甚至干脆不写，requests会默认按 ISO-8859-1 解码，你拿到的中文就是一堆问号和乱码。先判断再回退到apparent_encoding，能救回大部分乱码场景。

异常分层的意义在于"不同错误不同待遇"。网络超时是可以重试的，429是要退避的，404是没必要重试的（页面确实没了），500一般是服务端抽风、稍等就好。如果所有异常一视同仁地sleep(3)然后重试，你的脚本会在一堆无意义的请求上耗掉一整晚。

4.3 解析层的容错：别让一条脏数据毁掉整批任务

抓取任务最常见的崩溃点其实不在网络层，而在解析层。某个楼层结构异常、某个帖子的正文是纯图片没有文本节点、某个标题里混了特殊字符导致编码失败——任何一个都会抛出异常，把整个循环打断。

我的标准做法是把"单条处理"整体包进try，并且把失败记录单独落盘，方便事后补抓。

def safe_handle(row, session, bad_fp): try: html = fetch(row["url"], session) if not html: raise ValueError("empty response") return parse_detail(html, row["tid"]) except Exception as e: bad_fp.write(json.dumps( {"tid": row["tid"], "url": row["url"], "err": repr(e)}, ensure_ascii=False) + "\n") return []

repr(e)而不是str(e)也是个小技巧，前者会带上异常类型，事后看日志时能一眼分清是AttributeError（选择器不对）还是TimeoutError（网络问题），排查效率完全不同。

5. 常见问题与排查技巧实录

跑了七八轮之后，我整理了一份自己的速查表。这张表比任何教程都实用，因为每一条都是真实踩过的。

from collections import Counter import jieba # 高频作者 print(Counter(df["author"]).most_common(20)) # 正文高频词（去掉停用词后） words = [] for text in df["content"].dropna(): words.extend([w for w in jieba.cut(text) if len(w) > 1]) print(Counter(words).most_common(50))