前阵子做一个微博舆情分析的小项目,需要在几天内把某个话题相关的微博、热门评论全部拉下来。一开始手动复制粘贴,大概点了半小时就意识到这活儿不能这么干。后来我花了一个下午顺手写了个微博聚合采集工具,把搜索结果、指定主页的历史帖子、以及每条帖子下的评论统一跑通,一次跑完几万条数据稳稳落库。这篇文章把这个工具从思路到实现拆开,给有类似需求的读者一个可以直接上手的方案。
微博的开放接口对个人开发者基本不友好,很多数据权限申请下来要等很久,而临时做个采集工具往往不是业务诉求,只是支撑分析的数据获取手段。所以这类工具的形态通常是:登录后模拟正常行为,对微博网页端或客户端接口做请求,拿到JSON数据后解析、清洗、落库,最后导出成表格。这类工具适合新媒体运营、市场调研、舆情分析、学术研究甚至个人兴趣项目管理,只要是“需要批量拿微博公开内容”的场景,都可以复用同一套思路。
下面我从需求拆解、技术选型、核心实现、问题排查四个角度完整展开,中间会给出可复用的代码片段,并标注我在实际调试中踩过的坑。
1. 项目整体思路与需求拆解
1.1 需求场景与使用人群
先说为什么需要这样一个“聚合采集”工具,而不是直接用官方的导出功能或者花钱买商业版SaaS。
微博本身是有开放平台的,但个人开发者想申请高级接口权限,难度比较大。普通接口能拿到的数据又很有限,搜索、评论、用户时间线基本都受限。还有一个现实问题:即使有权限,通过API拿的数据和网页端看到的也不是完全一致的,某些字段在API里根本拿不到。这种情况下,自己写一个基于登录态的数据采集脚本,反而更可控。
需求来源其实很集中,我接触到的几个典型场景:
- 新媒体运营需要监控竞品或行业大号的每日发帖情况,记录传播数据;
- 舆情分析师要针对特定关键词做定向抓取,观察话题发酵路径和评论情绪;
- 学术研究者做社交网络分析,需要同一时间段内的大量帖子及其评论关系;
- 电商运营做品牌声量调研,要从微博评论里提取用户痛点。
这些场景的共同点是:数据量级在万级以上,对字段有定制化要求,而且往往是一次性任务。商业化的采集工具不是不能用,但按月付费、字段固定、不支持深度定制,遇到一次性的调研任务其实挺不划算。
1.2 核心需求拆解:搜索、主页、评论三件事
标题里写了三个主要功能:批量采集搜索帖子、主页帖子、评论。拆开看其实是三条独立的采集路径:
第一条,搜索帖子。给定一个关键词,返回所有包含该关键词的微博列表,并按时间排序或热度排序。搜索适合做话题追踪和热点监控。
第二条,主页帖子。给定一个用户ID或主页链接,采集该用户发布过的微博。这里需要注意两个细节:微博有两种发布形态,原创和转发;主页里的微博还可能包含长文、图片、视频等富文本内容,采集时需要区分。
第三条,评论采集。给定一条帖子或一批帖子,采集这些帖子下的全部评论。评论又有二级结构,楼中楼是常见形态,需要决定是否展开或者只取顶层。
这三条路径看起来各不相同,但实际上可以抽象成一个统一的“请求-解析-存储”框架。搜索和主页本质是帖子列表接口,评论本质是评论列表接口,区别只在于接口路径、返回结构和翻页方式。这个抽象很关键,它决定了后面代码的复用性。我的最终实现里,所有采集任务全部收敛成一个基类,每个任务只覆盖URL构造、参数生成和字段映射三个方法,新增一个采集方向半小时内就能搞定。
1.3 为什么强调“聚合”
标题里的“聚合”两个字,我理解有两层含义。
第一层是把多个来源的数据合并到一起。搜索帖子、主页帖子、评论来自不同的接口,字段也不完全一致,如果不做归一化,后续分析会很痛苦。例如搜索接口返回里有“微博正文”字段,主页接口里可能叫“text”,评论接口里叫“comment_text”。聚合的第一步是把它们映射到同一个字段名。
第二层是去重和关联。搜索结果的某条帖子可能同时出现在某个用户的微博主页里,两个接口都抓到同一帖子是很常见的情况。如果做舆情分析,重复计数会让数据直接失真;如果做传播分析,评论和帖子的关联关系必须稳定。所以聚合不仅是拼接,还需要一套统一的去重规则和主键策略。
这一点在我最初写的版本里被忽略了,结果就是两张表里同一篇帖子被抓了两遍,总数看着很唬人,实际可用度很低。后来重构时把主键统一为微博的MID值,所有表都挂上这个主键,问题才彻底解决。
2. 技术选型与核心架构设计
2.1 技术栈选择与实际理由
这个项目的技术栈我选的是:Python 3.10 + httpx + lxml + SQLite。辅助用了pandas做最后的表格导出。
每选一个组件都有具体原因:
- Python不用多说,做数据采集最顺手的语言,解析库、并发库、数据处理库齐全;
- httpx比requests强在两个地方:支持HTTP/2,虽然微博接口目前大多还是HTTP/1.1,但留了余地;支持异步,如果以后需要并发采集可以直接复用;
- lxml用于解析HTML页面中的部分数据,同时XPath在解析嵌套结构时比正则可靠;
- SQLite作为存储,零配置文件即开即用,几十万条数据完全撑得住,后续要迁移到MySQL也不费劲;
- pandas负责把查询结果导出成Excel或CSV,用户在最终交付时通常要这个格式。
不用Selenium或Playwright,不是不行,而是不值得。浏览器级渲染方案面对动态加载页面确实稳定,但每条数据都要开浏览器、加载页面资源,速度比纯接口请求慢两个数量级。以搜索100页数据为例,接口请求大概几分钟,浏览器方案可能要半小时以上,而且对服务器内存有要求。
2.2 数据接口的发现与参数字段分析
微博的网页端和移动端背后都是JSON接口。在浏览器登录微博后,打开开发者工具的网络面板,页面上的每一次加载、每一条微博的展开、每一页评论的翻页,都会对应一条真实的HTTP请求。我们要做的事情很简单:找到关键请求,看懂参数,用代码模拟它。
以我实际使用的情形为例,微博手机网页版的数据请求通常有几个固定参数:uid(用户ID)、pagebar(翻页标记或页码)、containerid(内容流的容器ID,不同板块对应不同ID)、since_id(下一页游标)。
这里有一个非常重要的经验:接口参数会升级和改名,你在网上搜到的旧参数可能已经失效,最可靠的方式始终是打开浏览器抓包看真实请求。我写得再详细也只能代表我调试当天的情况,实际复现时一定要自己动手定位接口。当然,接口的整体思路和字段含义是稳定的,理解了为什么有这些参数,即使参数名变了也能很快找到规律。
2.3 数据模型与存储设计
采集到的数据我设计了四张表:
帖子主表保存微博正文和传播数据,字段为MID、用户ID、用户名、正文清洗后的纯文本、原始正文JSON、发布时间、转发数、评论数、点赞数、来源设备、抓取时间。
评论表保存评论内容和统计信息,字段为评论ID、所属MID、评论用户ID、评论用户名、评论内容、评论时间、点赞数、是否含图、回复目标评论ID。
用户表保存采集过程中碰到的用户信息,字段为用户ID、昵称、简介、粉丝数、关注数、微博数、认证类型、主页链接。
抓取任务表记录每次任务的配置和状态,字段为任务类型、目标关键词或URL、本次新增数量、本次去重数量、开始时间、结束时间、状态。
用户表和任务表在初版设计里是没有的,后来发现没有用户表导致导出的Excel里只有用户昵称,没有粉丝数等关键维度,分析时还得二次补数据。任务表则是为了断点续采。微博采集经常遇到跑到一半登录失效的情况,如果没有任务记录,重启后不知道哪些页面已经跑过,全部重来非常浪费时间。
2.4 请求层与异常处理的统一封装
采集程序最容易翻车的地方不在业务逻辑,而在网络请求本身。接口超时、偶发报错、登录态失效、返回结构变化,任何一种情况都可能导致整个任务中断。所以我在正式代码里做了一层请求封装,所有HTTP请求统一走同一个函数。
class WeiboClient: def __init__(self, cookie: str): self.client = httpx.Client( headers={ "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15", "Cookie": cookie, "Referer": "https://m.weibo.cn/", }, timeout=10.0, follow_redirects=True, ) def get_json(self, url: str, params: dict): for retry in range(3): try: resp = self.client.get(url, params=params) if resp.status_code == 200: return resp.json() # 如果返回登录跳转,说明Cookie失效 if resp.status_code == 302 or "passport" in resp.text: raise CookieExpiredError("登录态失效") except (httpx.TimeoutException, httpx.NetworkError): time.sleep(2 + retry) return None这段代码里有一个容易被忽略的设计:重试只针对超时和网络错误,不针对Cookie失效。Cookie失效属于不可恢复错误,重试再多次也没有意义,正确的做法是抛异常,让任务表把它标记为“需要人工介入”的状态。我之前有过一个版本是把所有异常都放重试逻辑里,结果Cookie过期后还在无脑重试,白白浪费了半小时。
3. 核心功能实现与代码细节
3.1 登录态维护与Cookie持久化
采集微博数据的前提是有效的登录态。这里不是指一定要通过代码登录,而是说先用浏览器扫码登录微博,从开发者工具里把请求时使用的Cookie复制出来,写入配置。
我用的方案是把Cookie存在一个本地JSON文件里,每次请求前读取,如果发现响应返回未登录或者账号异常,就提示重新放Cookie。自动化的账号密码登录不是不能做,但微博的登录验证码机制一直在变,成本很高,而手动扫码效率其实很高。
import json COOKIE_FILE = "weibo_cookie.json" def load_cookie(): with open(COOKIE_FILE, "r") as f: return json.load(f)这里有几个细节需要注意。Cookie会过期,微博网页端的Cookie有效期通常几天到几周不等,因此最好在采集开始前先用一个轻量请求测试有效性。其次,不要把Cookie提交到公共仓库,它等于你账号的短期通行证,泄露之后对方可以直接操作你的账号。我一般在脚本里用环境变量或者单独的本地配置,而不是写死在代码里。
3.2 搜索帖子采集
搜索功能的核心是构造搜索请求并处理翻页。
接口层面,搜索的URL和重定向逻辑比较多,我实际使用的是移动端搜索接口,因为返回结构清晰且没有多余渲染。构造参数时主要有关键词、排序类型、时间范围、页码或游标。
关键点在于翻页。搜索结果的翻页有两个维度:按时间排序时通常采用时间线游标,按热度排序时可能直接给页码。我在实现时统一用一个page参数叠加since_id参数,每次请求后从返回内容中提取下一翻页标记,直到翻页标记为空或达到最大页数。
def fetch_search_posts(client, keyword, page=1, since_id=""): params = { "keyword": keyword, "page": page, "since_id": since_id, "sort_type": "time", "typeall": 1, } data = client.get_json(SEARCH_URL, params=params) if not data: return [], "" cards = extract_cards_from_json(data) posts = [normalize_search_post(card) for card in cards if card] next_since_id = data.get("next_cursor", "") return posts, next_since_id实际跑下来,搜索接口一天内请求太频繁会触发账号暂时限制。我在正式脚本里加了一个随机延时,每次请求之间sleep 2到5秒,并且把单次任务的最大请求数做了限制,宁可慢一点,也不要在短时间内打爆接口。
3.3 主页帖子采集
主页帖子采集的目标是某个用户发布的全部微博。接口通常以containerid和since_id作为分页游标。
这里的坑主要有两个:第一,微博主页的帖子列表会做合并,转发和原创在同一个列表里,需要根据字段判断;第二,长微博可能只显示摘要,完整内容需要单独请求详情接口。
我在代码里对每条帖子做了类型标记,is_original字段区分原创和转发,full_text先取列表中的内容,如果发现截断标记,再补发一次详情请求。
由于主页帖子采集面对的是单一用户,数据量取决于该用户的发帖频率和微博总数。几万条的大号跑完可能需要一段时间,我增加了进度日志功能,每采集100条打印一次进度,方便监控。
def fetch_user_posts(client, uid, containerid, max_pages=500): all_posts = [] since_id = "" for page in range(1, max_pages + 1): params = {"uid": uid, "containerid": containerid, "since_id": since_id} data = client.get_json(PROFILE_URL, params=params) cards = parse_profile_cards(data) posts, since_id = extract_posts_from_cards(cards) all_posts.extend(posts) if not since_id or len(posts) == 0: break time.sleep(random.uniform(2, 5)) return all_posts这里建议打日志时把“当前页数”“累计条数”“剩余游标”都打出来。因为用户主页的帖子总量和翻页深度是未知的,如果不打日志,程序看起来就像卡死了一样,实际上只是接口响应慢。
3.4 评论采集与楼中楼处理
评论采集的难度主要在分页和层级。评论接口一般也是游标分页,但评论量大的帖子可能需要翻几十页。
对于楼中楼,我默认只采顶层评论,如果确实需要分析对话结构,再针对热门评论单独展开子评论。这个取舍是为了控制请求数量,一条3万评论的帖子如果全部展开子评论,请求量爆炸式增长,很容易触发限制。
评论数据里最有价值的部分是评论文本和点赞数。文本清洗需要处理表情符号、短链接、@提及,这些在分析时往往需要单独处理。我的清洗规则是先去掉HTML标签和转义字符,再提取纯文本,同时对表情符做映射——把[笑cry]这类文本保留到独立字段emoji_text,方便后续情感分析时识别情绪信号。
def clean_comment_text(raw_text): text = re.sub(r"<[^>]+>", "", raw_text) text = html.unescape(text) emojis = re.findall(r"\[[^\]]+\]", text) text = re.sub(r"\[[^\]]+\]", "", text).strip() return text, emojis这个清洗函数的坑在于表情符号的正则必须放在去除普通标点之前。如果先去掉所有符号,表情就全部丢失了。另外微博文本里的链接是短链形式,直接去掉会丢失来源信息,我一般是保留原始链接字段,清洗后的纯文本里去掉链接,两个字段并存。
4. 聚合逻辑与数据落地
4.1 字段归一化与去重策略
无论数据来自搜索、主页还是评论,进入数据库之前都要统一字段。
我在代码里建了一个normalize_posts函数,负责把不同来源的字典映射为统一结构。核心字段包括MID、UID、用户名、时间、转发数、评论数、点赞数、正文、纯文本、原始JSON。
去重方面,我使用MID作为唯一键,SQLite插入时用INSERT OR IGNORE。这样即使同一个帖子被搜索和主页两个任务抓到,第二次插入直接跳过,数据的数量和唯一性都有保证。
INSERT OR IGNORE INTO posts (mid, uid, username, created_at, reposts, comments, likes, content, text, raw) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)这个策略简单但有效。唯一要注意的是,MID必须是从接口返回里直接获取的原始字符串,不要自己拼接或计算。我在初版犯过一个错误:用“用户ID+时间戳+随机数”拼了一个主键,导致同一帖子在不同任务里生成了不同的主键,去重彻底失效。
4.2 增量采集与断点续跑
很多场景下不需要每次全量采集,例如每周跑一次竞品监控,只关心上周新增的帖子。增量采集的实现方式是记录每次任务最大的时间游标或最新MID,下次拉取时只处理晚于该游标的数据。
断点续跑则依赖任务表。每次任务开始前查询任务表,如果发现相同关键词或相同用户存在未完成任务,就从上次记录的翻页游标继续。这个功能在一次抓5000条大号微博时帮了我大忙,因为登录状态中断过一次,恢复后直接续上,没有重新开始。
任务表的设计很简单,但有一个细节值得留意:状态字段不要只记“成功”和“失败”,要记录“未完成任务-已采集到第X页-当前游标”这样精确的信息。不然断点续跑只能从第一页重新开始,时间全花在重复请求上了。
4.3 数据导出与分析衔接
采集的数据最终要交付给别人用,或者进入分析流程。我支持两种导出方式:CSV和Excel。Excel格式用pandas直接写,列名保持中文字段,方便非技术人员打开。
df = pd.read_sql("SELECT * FROM posts", conn) df.to_excel("微博数据_导出.xlsx", index=False)如果后续要做词频、情感分析,建议不要直接用Excel,而是把清洗后的纯文本字段导入Jieba或SnowNLP。我自己的流程是采集完成后输出一份统计摘要,包括总帖子数、去重数、评论数、时间范围、最热帖子前三,这样交付时一眼就能看出数据质量。
5. 常见问题与排查技巧实录
5.1 高频请求引发账号限制
最常遇到的问题就是请求太频繁,响应里出现验证码页面或者直接返回错误提示。我的应对方案:第一,所有请求之间强制随机延时;第二,每个任务之间设置冷却时间;第三,把单次任务的请求上限写为可配置参数,默认2000次,防止程序失控。
如果你还是被限制了,不要继续硬跑,停下来休息一段时间再做新的尝试。这和访问频控是同一个逻辑,短时间内的连续失败只会加重限制时长。
5.2 Cookie失效与登录状态检测
Cookie失效的表现通常是某个请求返回未登录的JSON结构。我在请求层做了一个统一判断:如果响应关键字段缺失且包含登录跳转标记,就抛出一个自定义异常,任务表标记状态为“需要重新登录”,程序挂起等待操作员处理。
操作员只需要更新配置文件里的Cookie,然后重新跑任务,断点续跑逻辑会自动从上次中断的位置继续。
class CookieExpiredError(Exception): pass这个异常类很简单,但它让整个程序的错误处理逻辑清晰了很多。网络超时重试,Cookie失效挂起,数据解析失败跳过当前条,三种异常三种策略,互不混淆。
5.3 评论采集不全
评论不全的原因通常是翻页游标处理不对。微博评论接口的游标格式可能有Base36编码,直接使用还是需要转换取决于接口版本。如果你发现评论数明显少于页面显示,优先检查翻页参数是否在每页请求后正确更新。
另一个可能性是只采到了热门评论,没有采集全部评论。部分接口默认返回热度排序的热门前几页,需要显式设置排序方式为“按时间”,才能拿到全部评论。这个坑我在一个数据量很大的热门帖子下踩过,热门评论只有几十条,按时间排序后拉出三千多条完整评论。
5.4 数据乱码和编码问题
中文数据在写入CSV时如果处理不当,Excel打开容易乱码。解决办法是CSV写入时使用utf-8-sig编码,Excel读取就能正确识别中文。另外,SQLite的文本存储默认UTF-8,如果从接口返回的数据里有非法编码,要先做异常处理,用errors='ignore'或errors='replace'兜底,避免整条记录写入失败。
我还在解析层加了JSON解析异常的处理。微博接口有时会返回一段HTML而不是JSON,通常是触发了安全拦截。这种情况下json.loads会直接抛异常,需要捕获后跳过,而不是让整个程序崩溃。
5.5 翻页死循环与最大页数保护
还有一种隐蔽的问题:接口返回的since_id一直是同一个值,程序会陷入翻页死循环。原因可能是本地参数缓存没有正确更新,也可能是接口对某些条件下的数据不再返回新的游标。我的解决办法是加一层判断,如果连续5页的since_id完全相同,就直接终止任务并给出提示。
这个保护机制我建议所有采集任务都加上。很多看似卡死的程序不是真的卡死了,而是在某个循环里无脑请求同一页数据,白白消耗请求配额。
6. 扩展应用与合规提示
6.1 从采集到分析的可扩展方向
这套工具写完以后可以往多个方向扩展,我列几个实际可行且收益明显的:
多平台聚合。可以扩展到今日头条、小红书、知乎等平台,统一的数据模型和采集框架可以直接复用。每个平台不同的只是接口和解析层,核心的存储、去重、导出逻辑不用动。
定时任务化。把采集脚本部署到服务器,用cron或计划任务定期执行,就能形成自动化的监控系统。注意长时间运行后Cookie维护问题,可以每天定时检查一次登录状态。
语义分析接入。采集完成后接入大模型API做评论情感分类、摘要抽取。我之前试过把评论聚合后直接调大模型生成话题观点总结,效果比传统关键词统计好得多。只需要把清洗后的文本批量发给模型,再接收分类结果写回数据库。
6.2 合规与数据使用边界
最后聊一下合规问题。采集公开数据本身是常见的技术行为,但使用数据时需要遵守几个底线:不侵犯个人隐私,不用于商业变现或恶意竞争,不突破平台的反爬机制,不对目标平台造成资源压力。我的建议是控制采集频率,尊重平台的用户协议,采集到的数据仅用于内部研究和分析,不要公开传播原始数据。
就我自己跑过的任务来看,这个工具最大的价值不只是省时间,而是让“数据获取”从一件需要反复手工操作的事,变成了一条稳定可复用的流水线。每次接到新的采集需求,我只需要配置关键词或用户ID,改一下任务参数就能跑,交付数据的速度和质量都比以前手动整理强很多。如果你也在做类似的事,这套思路可以节省大量调试成本,值得自己动手搭一遍。