1. 动手之前先把“要抓什么”钉死做 Python 爬取这类短文本社交平台的内容最容易翻车的地方不是代码写不出来而是需求没定义清楚。我见过太多人上来就写requests.get跑通了一条链接就以为大功告成结果真正跑批的时候发现正文被折叠了、图片只拿到缩略图、点赞数是“1万”这种模糊值、时间全是“3小时前”这种相对表达。所以第一步不是写代码是拿一张纸把要落库的字段一个一个列出来想清楚每个字段从哪里来、长什么样、存成什么类型。这里说的“这类平台”指的是以短文本加图片为主、带点赞评论转发互动、有列表流和时间线的社交内容社区。它的数据形态有几个共性主贴有唯一 ID正文有长短之分互动数据是三个并列的计数图片是带多套尺寸的 CDN 链接时间既有绝对时间戳也有相对展示文本。把这几个共性摸透代码结构基本就定型了。1.1 一条动态的结构拆解我习惯先把单条动态在数据结构上“解剖”一遍。一条完整的帖子在数据层面大致分成四块第一块是身份标识包括帖子 ID、作者 ID、作者昵称。帖子 ID 是整个采集系统的锚点翻页、去重、增量更新、详情补全全部依赖它。这个 ID 通常是纯数字或者数字加字母的定长串一定要按字符串存别按整数存——我吃过这个亏某次用INT存 ID结果遇到超过 19 位的 ID 直接被截断导出一批数据全乱了回头补数据补了整整两天。第二块是内容主体包括标题有些平台的短内容没有独立标题取正文前 N 个字做标题、正文 HTML、纯文本正文、话题标签、 提及、外链。正文这块要特别注意接口返回的往往是带标签的 HTML 片段直接入库以后做文本分析会很难受所以我在采集阶段就同时存两份一份原始 HTML 留档一份清洗后的纯文本用于检索。第三块是互动指标点赞、评论、转发三个计数。这里有个坑不少平台在列表页返回的是“1.2万”这种展示值只有详情接口才给精确数字。如果你只抓列表得到的是一堆字符串想要精确值必须走详情。我的做法是列表页存展示值详情页覆盖成精确值并额外加一个字段标记这条数据是否已经补全。第四块是媒体资源图片、视频封面、长图。图片链接通常带尺寸路径段比如同一张图有缩略、中等、原图三个 URL只要替换路径里的关键片段就能拿到原图。这个规律需要在抓包阶段手动验证几次确认替换规则稳定后再写进代码。1.2 字段清单与存储类型设计把上面的拆解落成一张表采集目标就清晰了字段名类型来源备注post_idVARCHAR(32)列表接口主键增量判断依据author_idVARCHAR(32)列表接口用于按作者聚合author_nameVARCHAR(64)列表接口昵称可能含表情符号titleVARCHAR(255)正文截取无独立标题时取前 40 字content_htmlTEXT详情接口原始片段留档用content_textTEXT清洗后检索、分词用publish_timeDATETIME时间解析统一转成绝对时间like_countINT详情接口展示值需二次解析comment_countINT详情接口同上repost_countINT详情接口同上imagesTEXT媒体字段JSON 数组存 URL 列表is_detailTINYINT本地标记0 未补全 / 1 已补全crawl_timeDATETIME本地生成记录采集时刻这张表有几个设计取舍值得说一下。images我没有单独建一张图片表而是用 JSON 数组塞进 TEXT 字段原因是图片在分析场景里绝大多数时候是整体读取很少需要按图片做关联查询拆表反而增加 join 成本。当然如果你的下游要做图片去重或者以图搜图那就老老实实拆成post_image表用post_id seq做主键。is_detail这个标记字段是我后来才加的加了之后整个流程的健壮性提升非常明显。因为详情补全这一步是最容易失败的环节有了这个标记失败的数据下次跑批会自动被捞出来重试不用手动去翻日志找哪条漏了。publish_time一定要在建表阶段就定死为DATETIME别偷懒存字符串。我早期图省事存了原始文本后面做按天统计的时候写了一大堆CASE WHEN去转换“昨天”“3小时前”这些值那酸爽不想再体验第二次。2. 路线选型三条路走哪条确定了抓什么接下来是技术上怎么抓。这类平台的内容采集主流就三条路解析服务端渲染的 HTML、直接调用前端用的数据接口、用浏览器自动化工具模拟真人操作。这三条路没有绝对优劣关键看你的数据量、更新频率和稳定性要求。2.1 三条路线的真实取舍静态 HTML 解析这条路优点是实现简单、依赖少一个requests加一个解析库就能跑。缺点是现在这类平台的服务端渲染内容越来越少很多关键字段是 JS 异步填充的直接拿到的 HTML 里根本没有点赞数。而且页面结构改版频率高今天写的选择器明天可能就失效。直接调数据接口是我最推荐的方式前提是你能通过抓包找到接口。这类平台的前端在滚动加载时一定会请求一个返回 JSON 的接口字段齐全、结构稳定、体积小。缺点是接口通常带签名参数或者需要登录态前期分析成本高一点。但只要把签名逻辑摸清楚后续维护成本是最低的。浏览器自动化适合接口加密特别复杂、或者需要执行 JS 才能拿到数据的场景。缺点是慢一个页面渲染动辄两三秒抓一万条数据要跑大半天资源占用也高开几个实例内存就顶到天花板。我的原则是能用接口就不用浏览器浏览器方案只作为最后的兜底。三条路的对比如下维度静态 HTML数据接口浏览器自动化开发成本低中中高单条耗时快最快慢2-5 秒字段完整度差好好抗改版能力差中好资源占用低低高适合数据量小大小到中2.2 我的分层架构与理由实际项目里我采用的是接口为主、浏览器为辅的混合方案整体分成四层接入层封装会话对象统一管理请求头、Cookie、代理配置、超时设置。所有对外请求都走这一层方便统一加限速和日志。解析层把接口返回的 JSON 映射成内部统一的数据结构。这一层的作用是隔离外部变化——哪天接口字段改名了只改这一层的映射关系下游完全无感。清洗层处理 HTML 转义、表情符号、相对时间、模糊计数这些脏数据。存储层负责落库、去重、断点状态维护。分层的价值在后期维护时体现得特别明显。我有一次遇到平台把点赞字段从like_count改成了attitudes_count因为解析层做了字段映射我只改了一行字典就恢复了业务代码完全没动。如果当初把所有字段解析都散在各个函数里那次改版至少要多花两个小时。选择接口路线还有一个隐含好处接口返回的是结构化数据图片 URL、话题标签、 列表都是现成的数组不需要自己写正则从 HTML 里抠。正则抠标签这事儿看着简单实际写起来要考虑嵌套、要考虑属性顺序、要考虑自闭合标签一不小心就抠出一堆残缺数据。3. 实操从会话建立到数据落库这一节是全文最“能抄作业”的部分。我按真实跑批的顺序把关键代码和踩坑点都摆出来。所有域名和参数名都用占位符替代你需要替换成自己抓包得到的目标值。3.1 会话建立与请求头伪装第一步是把会话搭起来。这里我坚持用requests.Session()而不是每次都requests.get原因是 Session 会自动保持 Cookie连接复用也能省掉大量 TCP 握手开销实测在抓几千条的规模下整体耗时能降三成左右。import requests import random import time BASE https://example.com HEADERS { User-Agent: (Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36), Accept: application/json, text/plain, */*, Accept-Language: zh-CN,zh;q0.9, Referer: f{BASE}/, X-Requested-With: XMLHttpRequest, } session requests.Session() session.headers.update(HEADERS) session.cookies.update({登录态相关cookie名: 登录态值})请求头里有几个字段不能省。User-Agent是基本礼貌缺了大概率直接被挡。Referer很多接口会校验尤其是图片下载不带上会返回 403 或者一张防盗链占位图。X-Requested-With告诉服务端这是个异步请求某些接口不带这个头会返回完整 HTML 页面而不是 JSON。注意登录态信息属于敏感凭证不要硬编码在源码里提交到公开仓库。我的做法是放在环境变量或者本地单独的配置文件里用.gitignore排除掉。一旦泄露轻则账号异常重则影响账号安全。会话建立后建议先发一个“探活”请求确认登录态有效再开始跑批。我自己写了个小函数请求一个轻量接口检查返回里有没有特定的登录失效标记。这个前置检查能省掉大量“跑到一半发现全是空数据”的尴尬。3.2 列表页拉取与翻页策略列表接口的核心是翻页参数。这类平台常见两种翻页方式一种是页码式page1,2,3另一种是游标式max_idxxxx或者since_idxxxx。游标式更常见也更稳因为页码式在数据频繁更新的场景下容易漏数据和重复数据。游标式翻页的逻辑很简单第一次请求不带游标从返回结果里拿到next_cursor下一次请求带上它直到next_cursor为空或者返回列表为空为止。def fetch_feed_page(cursorNone, max_retry3): params {count: 20} if cursor: params[max_id] cursor for attempt in range(max_retry): try: r session.get(f{BASE}/api/feed/list, paramsparams, timeout10) r.raise_for_status() data r.json() items data.get(data, {}).get(list, []) next_cursor data.get(data, {}).get(next_cursor) return items, next_cursor except Exception as e: wait 2 ** attempt random.uniform(0, 1) print(f第 {attempt1} 次失败{wait:.1f}s 后重试{e}) time.sleep(wait) return [], None这段代码里有几个细节值得展开。count参数不要贪大我试过设成 100结果服务端直接截断返回 20 条白白浪费一次请求。20 到 30 是比较稳妥的区间。重试用的是指数退避加随机抖动2 ** attempt保证退避时间越来越长random.uniform(0, 1)加抖动是为了避免多个任务同时重试形成尖峰。这个策略在限流场景下特别管用比固定间隔重试的恢复率高不少。翻页过程中还有个隐性坑游标可能重复。在数据写入特别频繁的时段翻页游标有可能在相邻两次请求里返回同一个值导致死循环。我加了个保护——记录最近的游标值如果连续三次拿到相同游标就主动跳出并记录日志。3.3 详情补全与长文处理列表接口返回的字段是残缺的正文可能被截断成前 140 字点赞数可能是“1.2万”这种模糊值图片可能只有缩略图。所以要有一个详情补全的环节。补全的触发条件是is_detail 0逐条调用详情接口。这里有个性能优化点详情接口一般支持批量查询一次传多个 ID能显著降低请求数。我在能批量的场景下都用批量一次传 10 到 20 个 ID请求数直接降一个数量级。def fetch_detail_batch(id_list): ids ,.join(id_list) r session.get(f{BASE}/api/post/detail, params{ids: ids}, timeout10) r.raise_for_status() return r.json().get(data, {}).get(list, [])长文处理是另一个容易忽略的点。这类平台的正文有“全文”和“摘要”两种形态列表返回的往往是摘要末尾带个“展开全文”的标记。如果不去掉这个标记直接入库后续做文本分析时会出现大量重复的噪声尾巴。我的处理是先判断是否包含展开标记包含的话就走详情接口拿全文同时把标记文本从正文里剔除。正文里的 HTML 转义也需要注意。接口返回的正文经常带amp;、lt;、#39;这类实体直接入库以后做关键词匹配会匹配不到。我的清洗顺序是先做 HTML 反转义再剥标签再压缩连续空白。import html import re def clean_text(raw): if not raw: return text html.unescape(raw) text re.sub(rbr\s*/?, \n, text, flagsre.I) text re.sub(r[^], , text) text re.sub(r[ \t\u3000], , text) text re.sub(r\n{3,}, \n\n, text) return text.strip()顺序很重要。如果先剥标签再反转义那些转义过的尖括号会被误当成标签剥掉内容就丢了。我一开始就是顺序搞反丢了一批带代码片段的数据后来对比原始 JSON 才发现问题。3.4 时间解析与计数还原时间是采集里最琐碎的一块。这类平台在列表里展示的是相对时间“刚刚”“5分钟前”“3小时前”“昨天 14:20”“03-15”跨年的还会显示“2023-03-15”。要在采集阶段就统一成绝对时间不要留到分析阶段处理。from datetime import datetime, timedelta def parse_time(raw, nowNone): now now or datetime.now() raw (raw or ).strip() if not raw: return None if raw 刚刚: return now m re.match(r(\d)\s*秒前, raw) if m: return now - timedelta(secondsint(m.group(1))) m re.match(r(\d)\s*分钟前, raw) if m: return now - timedelta(minutesint(m.group(1))) m re.match(r(\d)\s*小时前, raw) if m: return now - timedelta(hoursint(m.group(1))) if raw.startswith(昨天): hm re.search(r(\d{1,2}):(\d{2}), raw) base now - timedelta(days1) if hm: return base.replace(hourint(hm.group(1)), minuteint(hm.group(2)), second0, microsecond0) return base.replace(hour0, minute0, second0, microsecond0) for fmt in (%m-%d, %Y-%m-%d, %m-%d %H:%M, %Y-%m-%d %H:%M): try: dt datetime.strptime(raw, fmt) if %Y not in fmt: dt dt.replace(yearnow.year) if dt now: dt dt.replace(yearnow.year - 1) return dt except ValueError: continue return None这段代码里有个容易漏的点%m-%d格式解析出来的年份是 1900必须补上当前年份。补完还要判断是否大于当前时间如果是就减一年——因为跨年的时候“12-31”出现在 1 月份逻辑上应该是去年的。计数还原相对简单但要处理“万”“亿”两种单位def parse_count(raw): if isinstance(raw, int): return raw raw str(raw or 0).strip().replace(,, ) if raw.endswith(万): return int(float(raw[:-1]) * 10000) if raw.endswith(亿): return int(float(raw[:-1]) * 100000000) try: return int(raw) except ValueError: return 03.5 图片下载与落库去重图片这块的坑主要集中在三点防盗链、原图替换、格式识别。防盗链靠Referer解决下载时必须带上平台域名。原图替换靠 URL 路径段替换抓包时对比几次就能找到规律比如把路径里的thumb或者small换成large、original。这个规则我一般写成常量字典放在配置里改版时只改配置。import hashlib import os IMG_DIR ./images os.makedirs(IMG_DIR, exist_okTrue) def download_image(url, post_id, seq): headers {Referer: f{BASE}/, User-Agent: HEADERS[User-Agent]} try: r session.get(url, headersheaders, timeout20, streamTrue) if r.status_code ! 200: return None ctype r.headers.get(Content-Type, ) if image not in ctype: return None buf bytearray() for chunk in r.iter_content(8192): buf.extend(chunk) if len(buf) 8 * 1024 * 1024: return None md5 hashlib.md5(buf).hexdigest() ext ctype.split(/)[-1].split(;)[0] or jpg path os.path.join(IMG_DIR, f{post_id}_{seq}_{md5[:8]}.{ext}) with open(path, wb) as f: f.write(buf) return path except Exception as e: print(f图片下载失败 {url}: {e}) return None几个细节streamTrue避免大图一次性加载进内存限制 8MB 上限防止异常文件撑爆磁盘用 MD5 做文件名既能去重又能避免同名覆盖校验Content-Type是因为有些防盗链返回的是 200 状态码但内容是一张错误图只看状态码会误判成功。落库我用的是 SQLite中小规模完全够用不用额外部署服务。建表时把post_id设成UNIQUE配合INSERT OR IGNORE实现天然去重import sqlite3 def init_db(path./data.db): conn sqlite3.connect(path) conn.execute( CREATE TABLE IF NOT EXISTS posts ( post_id TEXT PRIMARY KEY, author_name TEXT, title TEXT, content_text TEXT, content_html TEXT, publish_time TEXT, like_count INTEGER, comment_count INTEGER, repost_count INTEGER, images TEXT, is_detail INTEGER DEFAULT 0, crawl_time TEXT ) ) conn.execute(CREATE INDEX IF NOT EXISTS idx_time ON posts(publish_time)) conn.commit() return conn用TEXT PRIMARY KEY在 SQLite 里效果等同于UNIQUE索引INSERT OR IGNORE遇到重复主键静默跳过。这样即使同一页数据被重复拉取库里也不会出现脏数据。断点续爬的状态我用一个 JSON 文件维护记录最后一次成功的游标和已处理的 ID 集合import json 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 {last_cursor: None, done_ids: []} def save_state(state): with open(STATE_FILE, w, encodingutf-8) as f: json.dump(state, f, ensure_asciiFalse)ensure_asciiFalse别漏否则中文会被转成\uXXXX形式出问题的时候日志没法直接读。4. 稳定性限速、重试与封禁信号采集系统跑通只是开始能不能稳定跑完才是分水岭。我在这个环节踩的坑比写业务逻辑还多。4.1 请求节奏怎么定节奏控制的核心原则是像人不像机器。人浏览页面是有停顿的机械的固定间隔反而更像脚本。我一般用随机间隔基础值 1.5 秒上下浮动 1 秒def polite_sleep(base1.5, jitter1.0): time.sleep(base random.uniform(0, jitter))这个参数不是拍脑袋定的。我做过对比测试固定 0.3 秒间隔跑 200 条跑到 150 条左右开始出现大量 418 和 429改成 1.5±1 秒之后同样 200 条全程无异常耗时虽然从 1 分钟涨到 6 分钟但失败重试的时间成本远高于这个差值。宁可慢一点别跑一半全废。注意不要把间隔设得过短再指望重试来兜底重试次数一多请求总量反而超过从开始就慢慢跑的方案而且被封的风险显著上升。代理这块我不建议一上来就搞代理池。小规模采集几千条级别只要节奏控制得当单 IP 完全够用。真需要多 IP 的场景也要注意不同出口的 Cookie 隔离否则登录态会互相踢下线。4.2 重试策略与失败分类不是所有失败都值得重试。我把失败分成三类失败类型典型状态码处理方式网络抖动超时、连接重置退避后重试最多 3 次限流418、429长退避30s 起降低频次登录失效返回特定错误码不重试立刻停止并告警数据不存在404、空结果不重试标记跳过登录失效这一类特别重要如果混在普通重试里循环只会让账号异常状态持续更久。我的做法是在响应解析层统一检查一个错误标识字段命中就直接抛出自定义异常向上中断同时打印醒目日志。4.3 封禁信号的早期识别被封之前通常有征兆。我观察到的几个信号连续返回空列表但接口状态正常、图片下载全部返回占位图、响应体突然变成一段提示文案而不是 JSON、响应时间从几百毫秒突增到几秒。这些都是限流在升级的前兆。我的应对是在主循环里加一个健康度计数连续 5 次空结果或者连续 3 次异常响应就主动暂停 5 分钟并打印当前进度。暂停完之后从断点游标继续不从头开始。这个简单机制帮我避免了好几次“跑了一整晚结果什么都没抓到”的情况。5. 常见问题排查速查5.1 报错定位表把实际遇到过的典型问题整理成表出问题时可以直接对照现象可能原因排查与处理返回 403缺 Referer 或 UA补齐请求头重点检查 Referer返回 418 / 429触发限流增大间隔暂停 30 分钟再试返回 200 但 body 是 HTML登录态失效被重定向检查 Cookie 是否过期JSON 解析报错接口返回了非 JSON 内容打印r.text[:200]看原始内容点赞数全是 0用了列表接口的展示值补一次详情请求正文只有半截未处理折叠摘要走全文接口或清除截断标记图片全是同一张防盗链生效检查 Referer检查是否被限流时间解析返回 None出现了没覆盖的格式打印原始值补格式分支数据库插入报错字段超长或编码问题检查 emoji确认 utf8mb4翻页卡住死循环游标重复返回加游标重复检测连续相同则跳出5.2 数据质量的自查清单抓完之后别急着高兴先做三项自查第一项抽样人工比对。随机抽 5 到 10 条把库里的记录和页面上看到的内容逐字段对一遍。我靠这一步发现过“转发数是评论数”的字段错位问题接口字段名太像映射写反了不抽查根本发现不了。第二项统计分布检查。看publish_time的分布是不是集中在某个时间段看like_count有没有大量异常值比如全是 0 或者全是同一个数。分布异常基本都指向采集逻辑问题而不是真实数据形态。第三项覆盖率检查。算一下is_detail1的占比如果显著低于 100%说明详情补全有大批失败需要单独捞出来重跑。提示emoji 存储是最常见的隐性坑。昵称和正文里的表情符号超出基本多字节平面用utf8编码存会报错或者被截断MySQL 场景下必须用utf8mb4SQLite 默认就是 UTF-8 全量编码相对省心。6. 合规边界与个人经验这个环节我想放在最后聊因为它比任何技术细节都重要。采集公开内容做个人学习、数据分析、趋势研究和把别人的内容批量搬运出去二次分发是完全不同的两件事。前者在合理范围内后者会带来实实在在的问题。我在动手前会先看目标站点的 robots 协议确认哪些路径允许抓取会严格控制请求频率不给对方服务器造成压力会只采集公开可见的内容不碰需要额外权限才能看到的数据采集下来的数据只在自己本地做分析不导出传播尤其是涉及个人昵称、头像这类可以识别到个体的信息我在分析阶段都会做脱敏处理聚合到统计口径就停。另外一个我坚持的习惯是给每次采集任务设一个明确的数据量上限。跑之前就想好这次要多少条到了就停。这既是技术上的自我约束也避免了“反正是脚本跑的多抓点也无所谓”这种心态滑坡。脚本跑起来没有痛感但对面服务器是有真实负载的。从工程角度看这套东西最有价值的其实不是抓取本身而是那一套分层 状态记录 质量自查的结构。我后来把这套结构套用到好几个完全不同的采集需求上接入层换一下、解析层换一下清洗和存储层几乎原样复用。真正让我少加班的不是某个精妙的技巧而是把is_detail、断点状态、失败分类这些“笨办法”老老实实做扎实了。如果你要动手我的建议是先用 20 条数据把全链路跑通包括下载图片、落库、去重、断点恢复全部验证一遍再去放大规模。全链路上任何一环没验证规模一上去问题都会成倍放大。