简介面向微博数据采集场景的一份Python爬虫脚本用户只需输入关键词脚本便会自动构造搜索请求并翻页抓取相关微博同时收集结果中的关键字段如rid可用于舆情监测、热点追踪、学术研究等小规模社交媒体数据整理。压缩包仅含1个py文件大小约2KB无复杂依赖轻量易用方便快速部署或二次修改。目前已有1593人浏览学习适合具备基础Python语法、希望接触网页爬虫或需要特定关键词微博样本的开发者参考。脚本代码结构清晰涵盖关键词参数构造、翻页循环、响应解析与结果输出等核心模块运行后可直接得到指定字段数据既能为真实项目提供脚本基础也能作为理解微博搜索接口调用逻辑的入门范例帮助读者掌握常见请求头设置与反爬应对思路。1. 微博关键词爬虫难在让请求看起来像真人微博关键词爬虫听起来是典型的入门需求一个关键词、一个搜索框、一页结果。真跑上一个月难点全浮出来登录态、分页上限、时间字段还有搜索接口隔段时间就换一张脸。做过舆情采集的工程师都清楚这条链路的关键从来不是“能爬”而是“稳定地增量”今天能取到第 10 页明天也许就收到 418关键词换一个返回结构又变一个样。这篇文章按一线采集工程师会做的事情展开用 requests 请求搜索页用 mid 做去重主键数据落 SQLite定时任务交给 APScheduler最后给被验证过的参数与排错顺序。适合三类人做舆情与竞品分析、需要自己抓数据验证判断的工程师想从手工复制粘贴里解放出来的运营以及要给团队搭最小可运行采集原型的后端开发。2. 微博关键词爬虫的请求构造从搜索入口到返回体处理微博搜索页有两个常用入口s.weibo.com网页版和m.weibo.cn移动端接口。网页版返回的是 HTML结构稳定但解析稍啰嗦移动端接口直接返回 JSON解析方便却对登录态和请求频率更敏感。做关键词爬虫的常见做法是让网页版当主力、JSON 接口当备选避免某一天接口调整导致任务全线停摆。开始写代码之前先确认几件事搜索词必须做 URL 编码页面必须携带带登录态的 CookieReferer不能省略。这三条缺一条返回结果要么登录跳转、要么验证码、要么只有少量公开内容。下面把最小请求先跑通。2.1 用 requests 构造一条可用的微博关键词搜索请求依赖只需要requests解析 HTML 时再引入beautifulsoup4。下面这段代码把 Cookie、Header 和关键词请求封装在一个 Session 里避免每次请求重复设置。import requests import urllib.parse s requests.Session() s.headers.update({ 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, Referer: https://s.weibo.com/, }) s.cookies.set(SUB, 这里填浏览器里的SUB值) s.cookies.set(SUBP, 这里填浏览器里的SUBP值) def fetch_search_page(keyword: str, page: int 1) - str: q urllib.parse.quote(keyword) url fhttps://s.weibo.com/weibo?q{q}page{page} resp s.get(url, timeout15) resp.raise_for_status() resp.encoding utf-8 return resp.text html fetch_search_page(瑞幸咖啡, 1) print(len(html))这段代码的核心逻辑是urllib.parse.quote把中文关键词转成 URL 安全编码Session复用 TCP 连接与 Cookieencoding显式指定 utf-8 避免中文乱码。SUB和SUBP来自浏览器登录 weibo.com 后按 F12在 Application 面板里找到s.weibo.com的 Cookie把这两个字段的值复制出来即可。需要注意User-Agent不要用 requests 默认值。默认 UA 在微博的访问日志里特征非常明显很容易被一票拒绝。Referer填https://s.weibo.com/是为了让请求看起来像从搜索页内部发起。2.2 搜索请求关键参数q、page、timescope 的搭配微博搜索页 URL 上的参数不多但每个都有坑。把常用参数整理成下面这张表调参时直接对照。参数作用常用取值与常见坑q搜索关键词必须 URL 编码多个词用空格分隔想精确匹配就在词外加上双引号page页码从 1 开始超过 50 页后结果大量重复不再适合继续翻页timescope时间窗口格式为custom:2024-06-01-0:2024-06-02-23结束时间的“小时”不要写 24xsort排序方式切换“实时”Tab 时会出现实际值是xsorthot对登录态要求更高pagebar加载更多老版本翻页会用到pagebar0/1新版本以浏览器地址栏实际 URL 为准timescope是抓历史数据最重要的参数。比如要采集 2024 年 6 月 1 日到 6 月 2 日的数据URL 里的写法是https://s.weibo.com/weibo?q%E7%91%9E%E5%B9%B8timescopecustom:2024-06-01-0:2024-06-02-23page1注意日期和时间之间是短横线日期与日期之间是冒号小时范围从 0 到 23。写错任何一个符号搜索页都会忽略时间条件直接返回默认结果。2.3 从 HTML 里摘出 mid、正文和时间字段拿到 HTML 后用 BeautifulSoup 按搜索页的卡片结构解析。搜索结果的每一条微博通常包在div.card-wrap里action-type属性是feed_list_item微博的数字 ID 在mid属性上。from bs4 import BeautifulSoup def parse_search_html(html: str) - list[dict]: soup BeautifulSoup(html, html.parser) rows [] for card in soup.select(div.card-wrap[action-typefeed_list_item]): mid card.get(mid) if not mid: continue txt_node card.select_one(p.txt) text txt_node.get_text(stripTrue) if txt_node else from_node card.select_one(div.from) time_raw from_node.get_text( , stripTrue) if from_node else rows.append({ mid: mid, text: text, time_raw: time_raw, url: fhttps://weibo.com/{mid}, }) return rowsmid是整条数据的主键后面入库去重全靠它。p.txt是微博正文div.from里混着作者名、发布时间和来源设备先用get_text整个拿出来后续再按需清洗。页面结构偶尔会微调如果 CSS 选择器失效用 F12 查看当前卡片的实际 class把select_one里的参数改成对应值就行。3. 微博关键词爬虫的翻页、排序与增量去重翻页看起来只是page加一实际跑到一定深度就会发现问题页码越深重复内容越多甚至可能出现某一页返回完全相同的卡片列表。这个现象不是代码问题而是搜索索引的排序策略决定的。关键词爬虫要长期稳定产出必须在翻页策略和去重逻辑上做设计而不是无脑拉取。3.1 翻页深挖的上限与回溯历史数据的窗口切法综合排序下前 50 页基本能覆盖一个关键词的高热度内容。超过 50 页后返回结果会出现大量重复有些页码直接空白。想追溯更早的数据常见做法是用timescope把时间切成小窗口再逐段翻页。举个例子想回溯 2024 年 5 月的全部数据不要一次请求整月。把 5 月拆成 10 个三天的小窗口每个窗口内翻 20 页再合并去重。这样能有效绕开“搜索太宽导致深度结果稀疏”的问题也方便按时间段断点续采。切窗口时关键词越热门窗口可以切得越小冷门关键词直接按月请求也不会超页数。3.2 用 mid 做唯一键为什么不能依赖正文或 URL微博的搜索结果里同一条微博可能同时出现在“综合”和“实时”两个 Tab跨页也会重复。去重时最可靠的是mid这是微博数字 ID在 HTML 的mid属性和移动端 JSON 的mblog.id里都能拿到。要注意 URL 里的那串字符串是bid不是mid两者都能点开同一条微博但不能混用。每抓一页就把这一页里的mid和已入库的mid集合做比对。只保留新出现的mid旧数据跳过。这样即使某次任务崩溃后重新跑同一个时间窗口也不会产生重复入库。3.3 增量断点用时间窗口和已见 mid 集合控制采集范围把上面的逻辑串成一段可复用的增量采集函数。start_time和end_time是字符串形式的时间窗口函数内部逐页请求、去重、产出新数据遇到空页或已全部见过的mid就停止。import random import time def fetch_incremental(keyword: str, start_time: str, end_time: str): # start_time / end_time 形如 2024-06-01-0 / 2024-06-02-23 seen set() for page in range(1, 51): q urllib.parse.quote(keyword) url ( https://s.weibo.com/weibo f?q{q}timescopecustom:{start_time}:{end_time}page{page} ) html s.get(url, timeout15).text rows parse_search_html(html) if not rows: break new_rows [r for r in rows if r[mid] not in seen] if not new_rows: break seen.update(r[mid] for r in new_rows) yield from new_rows time.sleep(random.uniform(1, 2))这段代码利用了生成器的特性调用方每消费一条数据函数才继续请求下一页。seen集合只在一个时间窗口内有效窗口结束就释放内存。random.uniform(1, 2)控制相邻两页请求间隔 1 到 2 秒这个节奏比固定 2 秒更接近真实浏览行为也不容易触发频率限制。3.4 用 m.weibo.cn 的 JSON 接口做兜底网页版解析偶尔会碰上线结构变动这时可以用移动端接口快速确认数据是否正常。m.weibo.cn的搜索接口返回 JSONcontainerid需要把100103type1q关键词整体编码后再拼进 URL。def fetch_mobile_json(keyword: str, page: int) - dict: containerid 100103type1q urllib.parse.quote(keyword) url ( https://m.weibo.cn/api/container/getIndex f?containerid{urllib.parse.quote(containerid)} fpage_typesearchallpage{page} ) resp s.get(url, timeout15) return resp.json()这里最容易写错的地方是containerid里的符号。如果直接拼进 URL井号后的参数会被当成外层参数解析导致搜索词丢失。所以要先quote整个 containerid 再拼接。接口返回的data.cards里card_type 9的卡片就是微博正文card.mblog里包含id、text、created_at和转评赞数据。接口返回码也需要提前认识code含义处理方式0正常读取 cards 即可-100登录态失效重新登录复制新的 SUB-1参数异常检查 containerid 编码是否正确100 或 400风控触发降低频率暂停一段时间再继续4. 微博关键词爬虫的数据入库与定时采集数据拿回来不落库重启一次就全丢。个人项目和百十来万条的数据量级SQLite 是性价比最高的选择单文件、零运维、Python 标准库直接支持。等数据量大到需要多人同时查询时再迁移到 MySQL 也不迟。4.1 用 SQLite 建表字段按搜索结果的原始结构设计建表时把正文、时间、转评赞、用户信息和原始 JSON 都存下来。mid作为主键天然保证同一条微博只会出现一次。CREATE TABLE IF NOT EXISTS weibo_post ( mid TEXT PRIMARY KEY, keyword TEXT NOT NULL, user_id TEXT, user_name TEXT, text TEXT, created_at TEXT, reposts_count INTEGER DEFAULT 0, comments_count INTEGER DEFAULT 0, attitudes_count INTEGER DEFAULT 0, raw_json TEXT, first_seen_at TEXT DEFAULT (datetime(now)) ); CREATE INDEX IF NOT EXISTS idx_keyword_created ON weibo_post(keyword, created_at);raw_json保存整条微博的原始返回方便后续字段扩展。比如今天只需要正文和时间明天想分析“被转发数大于 100 的微博”直接从raw_json里算就行不用重新抓。索引建在keyword created_at上是因为最常见的查询是“某关键词某天发了什么”。4.2 INSERT OR IGNORE 与 ON CONFLICT 两种幂等写入增量采集时同一批mid可能被多个关键词命中。写入要用幂等方式第一版用INSERT OR IGNOREimport sqlite3 conn sqlite3.connect(weibo.db) conn.execute( INSERT OR IGNORE INTO weibo_post (mid, keyword, user_id, user_name, text, created_at, reposts_count, comments_count, attitudes_count, raw_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , (mid, keyword, user_id, user_name, text, created_at, reposts_count, comments_count, attitudes_count, raw_json), ) conn.commit()INSERT OR IGNORE的语义是“主键冲突就跳过”适合首次入库。但如果后续重新抓到一个已经存在的mid转评赞数据正好更新了这种写法不会更新旧数据。需要更新时改用ON CONFLICT(mid) DO UPDATEINSERT INTO weibo_post (mid, keyword, user_id, user_name, text, created_at, reposts_count, comments_count, attitudes_count, raw_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(mid) DO UPDATE SET reposts_count excluded.reposts_count, comments_count excluded.comments_count, attitudes_count excluded.attitudes_count, text excluded.text;注意raw_json没有更新意思是首次抓到的原始结构始终保留后续只刷新统计数字。这个设计在追热点时很好用后来重跑同一关键词不会覆盖早期快照。4.3 定时采集APScheduler 与 crontab 二选一定时任务有两个主流方案。进程内用 APScheduler 灵活系统级用 crontab 稳定。单机单人维护我更推荐 APScheduler因为可以热更新关键词列表不用每次改配置都去编辑 crontab。from apscheduler.schedulers.blocking import BlockingScheduler def collect_job(): for kw in load_keywords(): collect_one_keyword(kw, window1h) scheduler BlockingScheduler(timezoneAsia/Shanghai) scheduler.add_job(collect_job, cron, minute*/10) scheduler.start()先把两种方式放在一起对比方式适用场景注意点APScheduler多关键词、需要动态调整任务随 Python 进程一起退出需配合 systemd 或 nohup 保活crontab简单固定任务、长期无人值守要写绝对路径Python 虚拟环境的解释器路径别写错手动执行调试单次采集加--once参数方便前台观察日志crontab 的写法也很简单半小时一次*/30 * * * * cd /data/weibo_spider /usr/bin/python3 run.py logs/run.log 21cd到项目目录是为了保证相对路径配置可用日志重定向到run.log方便事后排查。4.4 关键词列表和采集窗口的配置化不要把关键词写死在代码里。用一个 YAML 文件管理关键词和时间窗口运营也能自己改。keywords: - name: 瑞幸咖啡 timescope: - name: AIGC timescope: custom:2024-06-01-0:2024-06-30-23 interval_minutes: 10 max_pages: 20timescope留空表示使用默认时间范围填了就用定制窗口。读取配置用yaml.safe_load采集任务每轮循环读取一次这样修改配置后下一个调度周期自动生效不需要重启进程。这个模式在追踪竞品数据时特别实用把多个品牌名丢进keywords定时跑一遍Excel 报表就自动出来了。5. 微博关键词爬虫的并发设计、限流与最后一道自检关键词一多逐页串行采集的耗时就会变得很难看。但并发不是越大越好微博对请求频率的容忍度比普通网站低。常见的做法是小并发线程池加随机延时而不是无脑上协程。5.1 线程池加随机延时比协程更容易控制节奏requests 是阻塞式 IO协程收益有限线程池反而直观。max_workers建议 2 到 3配合每页之间 1.5 到 3.5 秒的随机延时比 10 线程无间隔稳定得多。from concurrent.futures import ThreadPoolExecutor def safe_fetch_keyword(kw: str, page: int): time.sleep(random.uniform(1.5, 3.5)) html fetch_search_page(kw, page) return parse_search_html(html) with ThreadPoolExecutor(max_workers2) as pool: tasks [ pool.submit(safe_fetch_keyword, kw, page) for kw in keywords for page in range(1, 4) ] for task in tasks: rows task.result() save_rows(rows)线程数不是越高越好。max_workers2时同一时间只有两个请求在飞微博的压力阈值可以接受到 10 时连续请求密度迅速升高很快会触发 418。只有关键词列表膨胀到几十个、单机排队时间明显超出采集周期时才考虑把任务拆分给多台机器那就是分布式爬虫的范畴了这个体量先不碰。5.2 被限流时先看响应头别急着上代理 IP请求失败时先看返回状态码和响应头。常见的 418 是风控拦截504 是网关超时502 多数是微博侧本身不稳定。如果响应头里有Retry-After按它给的时间退避比自定义重试更有效。排错顺序建议是先确认 Cookie 没过期再检查请求间隔是否小于 3 秒然后看关键词是否触发了实时排序。都正常但 418 依旧才考虑换 IP。使用代理 IP 时要特别注意同一个 Cookie 在多个 IP 间跳动反而更容易触发验证码。正确做法是让一个 Cookie 对应一个固定出口 IPSession 保持不变频率不变只换网络路径。5.3 相对时间、短链和 Emoji三个高频清洗点搜索页返回的时间通常是“x分钟前”“今天 12:30”这类相对时间入库前必须转成绝对时间戳否则排序和按时间筛选都会出错。import re from datetime import datetime, timedelta def normalize_time(raw: str) - str: raw raw.strip() now datetime.now() if 分钟前 in raw: minutes int(re.search(r\d, raw).group()) return (now - timedelta(minutesminutes)).strftime(%Y-%m-%d %H:%M:%S) if 小时前 in raw: hours int(re.search(r\d, raw).group()) return (now - timedelta(hourshours)).strftime(%Y-%m-%d %H:%M:%S) if 今天 in raw: return now.strftime(%Y-%m-%d ) raw.replace(今天, ).strip() if 昨天 in raw: return (now - timedelta(days1)).strftime(%Y-%m-%d ) raw.replace(昨天, ).strip() return raw print(normalize_time(今天 08:30))短链的处理也容易忽略正文里的“https//t.cn/xxxx”不是完整跳转地址如果要展开需要从卡片的a[href]里取真实地址或者请求短链后读响应头的Location。Emoji 在入库时报错通常是因为数据库字符集不支持四字节 UTF-8MySQL 要指定utf8mb4而不是utf8SQLite 没有这个问题。5.4 用自检脚本守住增量链路定时任务跑久了最大的风险不是爬虫代码出错而是“看似正常但没新数据”。写一个两小时维度的自检脚本挂在 crontab 里能第一时间发现链路静默。# check_incremental.py import sqlite3 from datetime import datetime, timedelta conn sqlite3.connect(weibo.db) cutoff (datetime.now() - timedelta(hours2)).strftime(%Y-%m-%d %H:%M:%S) row conn.execute( SELECT COUNT(*), MAX(first_seen_at) FROM weibo_post WHERE first_seen_at ?, (cutoff,), ).fetchone() print({count_in_2h: row[0], last_seen_at: row[1]}) if row[0] 0: print(warn: 2小时内没有新入库请检查Cookie是否过期) exit(1)脚本输出去重后入库数量和时间窗口的最新落库时间。连续两个周期计数为 0就去浏览器重新登录一次把 SUB 换掉再手动跑一次单关键词探针通常就能定位是 Cookie 失效还是关键词本身太冷。本文还有配套的精品资源点击获取
微博关键词爬虫稳定增量采集：从请求构造到SQLite入库的工程实践
简介面向微博数据采集场景的一份Python爬虫脚本用户只需输入关键词脚本便会自动构造搜索请求并翻页抓取相关微博同时收集结果中的关键字段如rid可用于舆情监测、热点追踪、学术研究等小规模社交媒体数据整理。压缩包仅含1个py文件大小约2KB无复杂依赖轻量易用方便快速部署或二次修改。目前已有1593人浏览学习适合具备基础Python语法、希望接触网页爬虫或需要特定关键词微博样本的开发者参考。脚本代码结构清晰涵盖关键词参数构造、翻页循环、响应解析与结果输出等核心模块运行后可直接得到指定字段数据既能为真实项目提供脚本基础也能作为理解微博搜索接口调用逻辑的入门范例帮助读者掌握常见请求头设置与反爬应对思路。1. 微博关键词爬虫难在让请求看起来像真人微博关键词爬虫听起来是典型的入门需求一个关键词、一个搜索框、一页结果。真跑上一个月难点全浮出来登录态、分页上限、时间字段还有搜索接口隔段时间就换一张脸。做过舆情采集的工程师都清楚这条链路的关键从来不是“能爬”而是“稳定地增量”今天能取到第 10 页明天也许就收到 418关键词换一个返回结构又变一个样。这篇文章按一线采集工程师会做的事情展开用 requests 请求搜索页用 mid 做去重主键数据落 SQLite定时任务交给 APScheduler最后给被验证过的参数与排错顺序。适合三类人做舆情与竞品分析、需要自己抓数据验证判断的工程师想从手工复制粘贴里解放出来的运营以及要给团队搭最小可运行采集原型的后端开发。2. 微博关键词爬虫的请求构造从搜索入口到返回体处理微博搜索页有两个常用入口s.weibo.com网页版和m.weibo.cn移动端接口。网页版返回的是 HTML结构稳定但解析稍啰嗦移动端接口直接返回 JSON解析方便却对登录态和请求频率更敏感。做关键词爬虫的常见做法是让网页版当主力、JSON 接口当备选避免某一天接口调整导致任务全线停摆。开始写代码之前先确认几件事搜索词必须做 URL 编码页面必须携带带登录态的 CookieReferer不能省略。这三条缺一条返回结果要么登录跳转、要么验证码、要么只有少量公开内容。下面把最小请求先跑通。2.1 用 requests 构造一条可用的微博关键词搜索请求依赖只需要requests解析 HTML 时再引入beautifulsoup4。下面这段代码把 Cookie、Header 和关键词请求封装在一个 Session 里避免每次请求重复设置。import requests import urllib.parse s requests.Session() s.headers.update({ 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, Referer: https://s.weibo.com/, }) s.cookies.set(SUB, 这里填浏览器里的SUB值) s.cookies.set(SUBP, 这里填浏览器里的SUBP值) def fetch_search_page(keyword: str, page: int 1) - str: q urllib.parse.quote(keyword) url fhttps://s.weibo.com/weibo?q{q}page{page} resp s.get(url, timeout15) resp.raise_for_status() resp.encoding utf-8 return resp.text html fetch_search_page(瑞幸咖啡, 1) print(len(html))这段代码的核心逻辑是urllib.parse.quote把中文关键词转成 URL 安全编码Session复用 TCP 连接与 Cookieencoding显式指定 utf-8 避免中文乱码。SUB和SUBP来自浏览器登录 weibo.com 后按 F12在 Application 面板里找到s.weibo.com的 Cookie把这两个字段的值复制出来即可。需要注意User-Agent不要用 requests 默认值。默认 UA 在微博的访问日志里特征非常明显很容易被一票拒绝。Referer填https://s.weibo.com/是为了让请求看起来像从搜索页内部发起。2.2 搜索请求关键参数q、page、timescope 的搭配微博搜索页 URL 上的参数不多但每个都有坑。把常用参数整理成下面这张表调参时直接对照。参数作用常用取值与常见坑q搜索关键词必须 URL 编码多个词用空格分隔想精确匹配就在词外加上双引号page页码从 1 开始超过 50 页后结果大量重复不再适合继续翻页timescope时间窗口格式为custom:2024-06-01-0:2024-06-02-23结束时间的“小时”不要写 24xsort排序方式切换“实时”Tab 时会出现实际值是xsorthot对登录态要求更高pagebar加载更多老版本翻页会用到pagebar0/1新版本以浏览器地址栏实际 URL 为准timescope是抓历史数据最重要的参数。比如要采集 2024 年 6 月 1 日到 6 月 2 日的数据URL 里的写法是https://s.weibo.com/weibo?q%E7%91%9E%E5%B9%B8timescopecustom:2024-06-01-0:2024-06-02-23page1注意日期和时间之间是短横线日期与日期之间是冒号小时范围从 0 到 23。写错任何一个符号搜索页都会忽略时间条件直接返回默认结果。2.3 从 HTML 里摘出 mid、正文和时间字段拿到 HTML 后用 BeautifulSoup 按搜索页的卡片结构解析。搜索结果的每一条微博通常包在div.card-wrap里action-type属性是feed_list_item微博的数字 ID 在mid属性上。from bs4 import BeautifulSoup def parse_search_html(html: str) - list[dict]: soup BeautifulSoup(html, html.parser) rows [] for card in soup.select(div.card-wrap[action-typefeed_list_item]): mid card.get(mid) if not mid: continue txt_node card.select_one(p.txt) text txt_node.get_text(stripTrue) if txt_node else from_node card.select_one(div.from) time_raw from_node.get_text( , stripTrue) if from_node else rows.append({ mid: mid, text: text, time_raw: time_raw, url: fhttps://weibo.com/{mid}, }) return rowsmid是整条数据的主键后面入库去重全靠它。p.txt是微博正文div.from里混着作者名、发布时间和来源设备先用get_text整个拿出来后续再按需清洗。页面结构偶尔会微调如果 CSS 选择器失效用 F12 查看当前卡片的实际 class把select_one里的参数改成对应值就行。3. 微博关键词爬虫的翻页、排序与增量去重翻页看起来只是page加一实际跑到一定深度就会发现问题页码越深重复内容越多甚至可能出现某一页返回完全相同的卡片列表。这个现象不是代码问题而是搜索索引的排序策略决定的。关键词爬虫要长期稳定产出必须在翻页策略和去重逻辑上做设计而不是无脑拉取。3.1 翻页深挖的上限与回溯历史数据的窗口切法综合排序下前 50 页基本能覆盖一个关键词的高热度内容。超过 50 页后返回结果会出现大量重复有些页码直接空白。想追溯更早的数据常见做法是用timescope把时间切成小窗口再逐段翻页。举个例子想回溯 2024 年 5 月的全部数据不要一次请求整月。把 5 月拆成 10 个三天的小窗口每个窗口内翻 20 页再合并去重。这样能有效绕开“搜索太宽导致深度结果稀疏”的问题也方便按时间段断点续采。切窗口时关键词越热门窗口可以切得越小冷门关键词直接按月请求也不会超页数。3.2 用 mid 做唯一键为什么不能依赖正文或 URL微博的搜索结果里同一条微博可能同时出现在“综合”和“实时”两个 Tab跨页也会重复。去重时最可靠的是mid这是微博数字 ID在 HTML 的mid属性和移动端 JSON 的mblog.id里都能拿到。要注意 URL 里的那串字符串是bid不是mid两者都能点开同一条微博但不能混用。每抓一页就把这一页里的mid和已入库的mid集合做比对。只保留新出现的mid旧数据跳过。这样即使某次任务崩溃后重新跑同一个时间窗口也不会产生重复入库。3.3 增量断点用时间窗口和已见 mid 集合控制采集范围把上面的逻辑串成一段可复用的增量采集函数。start_time和end_time是字符串形式的时间窗口函数内部逐页请求、去重、产出新数据遇到空页或已全部见过的mid就停止。import random import time def fetch_incremental(keyword: str, start_time: str, end_time: str): # start_time / end_time 形如 2024-06-01-0 / 2024-06-02-23 seen set() for page in range(1, 51): q urllib.parse.quote(keyword) url ( https://s.weibo.com/weibo f?q{q}timescopecustom:{start_time}:{end_time}page{page} ) html s.get(url, timeout15).text rows parse_search_html(html) if not rows: break new_rows [r for r in rows if r[mid] not in seen] if not new_rows: break seen.update(r[mid] for r in new_rows) yield from new_rows time.sleep(random.uniform(1, 2))这段代码利用了生成器的特性调用方每消费一条数据函数才继续请求下一页。seen集合只在一个时间窗口内有效窗口结束就释放内存。random.uniform(1, 2)控制相邻两页请求间隔 1 到 2 秒这个节奏比固定 2 秒更接近真实浏览行为也不容易触发频率限制。3.4 用 m.weibo.cn 的 JSON 接口做兜底网页版解析偶尔会碰上线结构变动这时可以用移动端接口快速确认数据是否正常。m.weibo.cn的搜索接口返回 JSONcontainerid需要把100103type1q关键词整体编码后再拼进 URL。def fetch_mobile_json(keyword: str, page: int) - dict: containerid 100103type1q urllib.parse.quote(keyword) url ( https://m.weibo.cn/api/container/getIndex f?containerid{urllib.parse.quote(containerid)} fpage_typesearchallpage{page} ) resp s.get(url, timeout15) return resp.json()这里最容易写错的地方是containerid里的符号。如果直接拼进 URL井号后的参数会被当成外层参数解析导致搜索词丢失。所以要先quote整个 containerid 再拼接。接口返回的data.cards里card_type 9的卡片就是微博正文card.mblog里包含id、text、created_at和转评赞数据。接口返回码也需要提前认识code含义处理方式0正常读取 cards 即可-100登录态失效重新登录复制新的 SUB-1参数异常检查 containerid 编码是否正确100 或 400风控触发降低频率暂停一段时间再继续4. 微博关键词爬虫的数据入库与定时采集数据拿回来不落库重启一次就全丢。个人项目和百十来万条的数据量级SQLite 是性价比最高的选择单文件、零运维、Python 标准库直接支持。等数据量大到需要多人同时查询时再迁移到 MySQL 也不迟。4.1 用 SQLite 建表字段按搜索结果的原始结构设计建表时把正文、时间、转评赞、用户信息和原始 JSON 都存下来。mid作为主键天然保证同一条微博只会出现一次。CREATE TABLE IF NOT EXISTS weibo_post ( mid TEXT PRIMARY KEY, keyword TEXT NOT NULL, user_id TEXT, user_name TEXT, text TEXT, created_at TEXT, reposts_count INTEGER DEFAULT 0, comments_count INTEGER DEFAULT 0, attitudes_count INTEGER DEFAULT 0, raw_json TEXT, first_seen_at TEXT DEFAULT (datetime(now)) ); CREATE INDEX IF NOT EXISTS idx_keyword_created ON weibo_post(keyword, created_at);raw_json保存整条微博的原始返回方便后续字段扩展。比如今天只需要正文和时间明天想分析“被转发数大于 100 的微博”直接从raw_json里算就行不用重新抓。索引建在keyword created_at上是因为最常见的查询是“某关键词某天发了什么”。4.2 INSERT OR IGNORE 与 ON CONFLICT 两种幂等写入增量采集时同一批mid可能被多个关键词命中。写入要用幂等方式第一版用INSERT OR IGNOREimport sqlite3 conn sqlite3.connect(weibo.db) conn.execute( INSERT OR IGNORE INTO weibo_post (mid, keyword, user_id, user_name, text, created_at, reposts_count, comments_count, attitudes_count, raw_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , (mid, keyword, user_id, user_name, text, created_at, reposts_count, comments_count, attitudes_count, raw_json), ) conn.commit()INSERT OR IGNORE的语义是“主键冲突就跳过”适合首次入库。但如果后续重新抓到一个已经存在的mid转评赞数据正好更新了这种写法不会更新旧数据。需要更新时改用ON CONFLICT(mid) DO UPDATEINSERT INTO weibo_post (mid, keyword, user_id, user_name, text, created_at, reposts_count, comments_count, attitudes_count, raw_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(mid) DO UPDATE SET reposts_count excluded.reposts_count, comments_count excluded.comments_count, attitudes_count excluded.attitudes_count, text excluded.text;注意raw_json没有更新意思是首次抓到的原始结构始终保留后续只刷新统计数字。这个设计在追热点时很好用后来重跑同一关键词不会覆盖早期快照。4.3 定时采集APScheduler 与 crontab 二选一定时任务有两个主流方案。进程内用 APScheduler 灵活系统级用 crontab 稳定。单机单人维护我更推荐 APScheduler因为可以热更新关键词列表不用每次改配置都去编辑 crontab。from apscheduler.schedulers.blocking import BlockingScheduler def collect_job(): for kw in load_keywords(): collect_one_keyword(kw, window1h) scheduler BlockingScheduler(timezoneAsia/Shanghai) scheduler.add_job(collect_job, cron, minute*/10) scheduler.start()先把两种方式放在一起对比方式适用场景注意点APScheduler多关键词、需要动态调整任务随 Python 进程一起退出需配合 systemd 或 nohup 保活crontab简单固定任务、长期无人值守要写绝对路径Python 虚拟环境的解释器路径别写错手动执行调试单次采集加--once参数方便前台观察日志crontab 的写法也很简单半小时一次*/30 * * * * cd /data/weibo_spider /usr/bin/python3 run.py logs/run.log 21cd到项目目录是为了保证相对路径配置可用日志重定向到run.log方便事后排查。4.4 关键词列表和采集窗口的配置化不要把关键词写死在代码里。用一个 YAML 文件管理关键词和时间窗口运营也能自己改。keywords: - name: 瑞幸咖啡 timescope: - name: AIGC timescope: custom:2024-06-01-0:2024-06-30-23 interval_minutes: 10 max_pages: 20timescope留空表示使用默认时间范围填了就用定制窗口。读取配置用yaml.safe_load采集任务每轮循环读取一次这样修改配置后下一个调度周期自动生效不需要重启进程。这个模式在追踪竞品数据时特别实用把多个品牌名丢进keywords定时跑一遍Excel 报表就自动出来了。5. 微博关键词爬虫的并发设计、限流与最后一道自检关键词一多逐页串行采集的耗时就会变得很难看。但并发不是越大越好微博对请求频率的容忍度比普通网站低。常见的做法是小并发线程池加随机延时而不是无脑上协程。5.1 线程池加随机延时比协程更容易控制节奏requests 是阻塞式 IO协程收益有限线程池反而直观。max_workers建议 2 到 3配合每页之间 1.5 到 3.5 秒的随机延时比 10 线程无间隔稳定得多。from concurrent.futures import ThreadPoolExecutor def safe_fetch_keyword(kw: str, page: int): time.sleep(random.uniform(1.5, 3.5)) html fetch_search_page(kw, page) return parse_search_html(html) with ThreadPoolExecutor(max_workers2) as pool: tasks [ pool.submit(safe_fetch_keyword, kw, page) for kw in keywords for page in range(1, 4) ] for task in tasks: rows task.result() save_rows(rows)线程数不是越高越好。max_workers2时同一时间只有两个请求在飞微博的压力阈值可以接受到 10 时连续请求密度迅速升高很快会触发 418。只有关键词列表膨胀到几十个、单机排队时间明显超出采集周期时才考虑把任务拆分给多台机器那就是分布式爬虫的范畴了这个体量先不碰。5.2 被限流时先看响应头别急着上代理 IP请求失败时先看返回状态码和响应头。常见的 418 是风控拦截504 是网关超时502 多数是微博侧本身不稳定。如果响应头里有Retry-After按它给的时间退避比自定义重试更有效。排错顺序建议是先确认 Cookie 没过期再检查请求间隔是否小于 3 秒然后看关键词是否触发了实时排序。都正常但 418 依旧才考虑换 IP。使用代理 IP 时要特别注意同一个 Cookie 在多个 IP 间跳动反而更容易触发验证码。正确做法是让一个 Cookie 对应一个固定出口 IPSession 保持不变频率不变只换网络路径。5.3 相对时间、短链和 Emoji三个高频清洗点搜索页返回的时间通常是“x分钟前”“今天 12:30”这类相对时间入库前必须转成绝对时间戳否则排序和按时间筛选都会出错。import re from datetime import datetime, timedelta def normalize_time(raw: str) - str: raw raw.strip() now datetime.now() if 分钟前 in raw: minutes int(re.search(r\d, raw).group()) return (now - timedelta(minutesminutes)).strftime(%Y-%m-%d %H:%M:%S) if 小时前 in raw: hours int(re.search(r\d, raw).group()) return (now - timedelta(hourshours)).strftime(%Y-%m-%d %H:%M:%S) if 今天 in raw: return now.strftime(%Y-%m-%d ) raw.replace(今天, ).strip() if 昨天 in raw: return (now - timedelta(days1)).strftime(%Y-%m-%d ) raw.replace(昨天, ).strip() return raw print(normalize_time(今天 08:30))短链的处理也容易忽略正文里的“https//t.cn/xxxx”不是完整跳转地址如果要展开需要从卡片的a[href]里取真实地址或者请求短链后读响应头的Location。Emoji 在入库时报错通常是因为数据库字符集不支持四字节 UTF-8MySQL 要指定utf8mb4而不是utf8SQLite 没有这个问题。5.4 用自检脚本守住增量链路定时任务跑久了最大的风险不是爬虫代码出错而是“看似正常但没新数据”。写一个两小时维度的自检脚本挂在 crontab 里能第一时间发现链路静默。# check_incremental.py import sqlite3 from datetime import datetime, timedelta conn sqlite3.connect(weibo.db) cutoff (datetime.now() - timedelta(hours2)).strftime(%Y-%m-%d %H:%M:%S) row conn.execute( SELECT COUNT(*), MAX(first_seen_at) FROM weibo_post WHERE first_seen_at ?, (cutoff,), ).fetchone() print({count_in_2h: row[0], last_seen_at: row[1]}) if row[0] 0: print(warn: 2小时内没有新入库请检查Cookie是否过期) exit(1)脚本输出去重后入库数量和时间窗口的最新落库时间。连续两个周期计数为 0就去浏览器重新登录一次把 SUB 换掉再手动跑一次单关键词探针通常就能定位是 Cookie 失效还是关键词本身太冷。本文还有配套的精品资源点击获取