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当解析后端提速。理由如下表。方案上手成本解析稳定性适合场景我的评价requests 正则低极低结构极其规整的接口返回只适合临时救急不要长期维护requests BeautifulSoup(lxml)中高中小型站点、几千到几万条本项目的首选requests lxml XPath中高需要精确取属性、层级复杂熟练后比 BS4 快一截Scrapy高高十万条以上、需要并发与调度本项目属于杀鸡用牛刀无头浏览器Playwright/Selenium中高中数据靠 JS 渲染、必须执行脚本能不用就不用资源开销大无头浏览器这一项我特意列出来是因为很多人一看到页面数据抓不到就条件反射上浏览器。但帖子列表和帖子正文这类内容绝大多数情况下是服务端直出的——你右键查看网页源代码如果能在里面搜到帖子标题那就说明 HTML 里已经有了requests完全够用上浏览器纯属自己给自己增加负担。1.3 合规边界这是技术问题也是长期可维护性问题写抓取脚本绕不开这件事我把它放在设计阶段讲而不是最后加一句免责声明因为它直接决定了你的脚本能活多久。我的三条底线是这样的。第一先看 robots 协议。访问站点/robots.txt看User-agent: *下面的Disallow里有没有你要抓的路径。有的话就换目标别硬来。第二控制请求频率。单线程、每次请求之间随机间隔 1.53 秒一晚上跑个几千条对任何站点都不构成压力。反过来50 个线程怼上去结果一定是先被限速、再被拉黑你的任务反而完不成。第三只取公开内容不碰个人隐私。抓标题、正文、昵称这类公开展示的信息没问题手机号、身份证、私信、需要登录才能看到的内容一律不碰。抓下来的数据自己分析可以原样搬运发布就是另一回事了。提示如果目标页面明确要求登录才能查看或者频繁弹出行为验证正确的做法是停下来——降低频率、错峰运行或者干脆放弃这个数据源。任何必须想办法绕过的念头都应该在这一步掐掉。2. 抓取前的准备与页面结构分析2.1 URL 规律与分页参数是整个脚本的地基论坛类站点的 URL 设计几乎都遵循同一套思路用查询参数控制页码和排序。这个站点列表页的形态大致是/?kw关键词ieutf-8pn0其中pn就是偏移量而不是页码。这是个巨坑——很多人以为pn1是第二页结果pn0和pn1返回的内容几乎一样抓了一堆重复数据还找不到原因。正确的理解是pn表示跳过多少条每页固定 50 条所以第 1 页是pn0第 2 页是pn50第 n 页是pn(n-1)*50。写代码的时候我建议直接维护一个偏移量列表而不是用for page in range(1, 21)然后pnpage*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;q0.9, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.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(stripTrue) 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(stripTrue), 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, stripTrue), }) return postscontent.get_text(\n, stripTrue)这行里的\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, encodingutf-8) as f: return json.load(f) return {done_offsets: [], seen_tids: []} def save_state(state): tmp STATE_FILE .tmp with open(tmp, w, encodingutf-8) as f: json.dump(state, f, ensure_asciiFalse) 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}ieutf-8pn{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, encodingutf-8) as f: for r in records: f.write(json.dumps(r, ensure_asciiFalse) \n) # 全部抓完后转换 import pandas as pd df pd.read_json(posts.jsonl, linesTrue) df.to_csv(posts.csv, indexFalse, encodingutf-8-sig)encodingutf-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, tries3): for i in range(tries): try: r session.get(url, headersHEADERS, timeout10) 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_asciiFalse) \n) return []repr(e)而不是str(e)也是个小技巧前者会带上异常类型事后看日志时能一眼分清是AttributeError选择器不对还是TimeoutError网络问题排查效率完全不同。5. 常见问题与排查技巧实录跑了七八轮之后我整理了一份自己的速查表。这张表比任何教程都实用因为每一条都是真实踩过的。现象最可能的原因排查动作处理方式解析结果全是空列表选择器失效或页面改版CtrlU 看源码里搜标题文字重新定位节点改用带>import pandas as pd df pd.read_json(posts.jsonl, linesTrue) print(df.notna().mean().round(3)) # 各列非空率 print(df[tid].duplicated().sum()) # 重复条数应该是 0 print(df[content].str.len().describe()) # 正文长度分布第二件是抽样人工核对。随机抽 10 条把url复制到浏览器里打开肉眼比对标题和正文是否一致。这一步听起来笨但它能发现机器检查发现不了的问题——比如选择器抓到了楼主标签而不是正文或者楼层号取到了回复两个字。我每次跑新站点都会做这一步五分钟的事。第三件是简单统计。不用做多复杂的分析先看看发帖时间的小时分布、高频作者 Top 20、正文高频词这几个指标能帮你判断数据是否有偏。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))如果发现某个作者的帖子占比超过 20%那这批数据的代表性就有问题做文本分析时结论会严重偏向这个人的表达习惯。这时候要么扩大抓取量要么在分析阶段做降权。这类判断只能靠人来看工具帮不上忙。还有个实用的小扩展把tid和url单独导一份出来方便以后做增量更新。帖子是会不断有新回复的你第一次抓的时候回复数是 12一个月后可能变成 300。有了这份索引后续只需要对回复数变化超过阈值的帖子重新抓一次详情页成本不到全量重抓的十分之一。我自己在这个项目上最大的体会是抓取脚本的难度从来不在于能不能抓下来而在于能不能稳定、干净、可持续地抓下来。写了这么多年真正花时间的部分八成都在限速、重试、断点、校验这些看起来不酷的地方。但恰恰是这些部分决定了你的脚本是一次性的玩具还是可以每周定点跑一次的可靠工具。