☰
TrendRadar实战:多平台热搜监控与自动推送,打破信息差
2026/9/29 15:46:22 网站建设 项目流程

今年上半年我做内容运营,有一次因为晚了一步,彻底错过了当天刷屏的一个热点。那个话题早上八点多就挂在微博热搜前三,我十点半才从同事转发里看到,等素材整理完、稿子发出去,话题已经掉出了前十,打开率惨得没法看。后来我认真复盘这件事,发现问题的根源不是"我刷得不够勤",而是所有信息源都在被动等着我去打开。看清楚这一点之后,我决定自己写一套工具,让它盯着各平台的热搜榜单,把值得看的内容主动推到我手机上。这套工具就是今天要分享的TrendRadar。

TrendRadar 做的事情不复杂:每几分钟抓一次微博、知乎、百度、B站等平台的热搜榜,把重复话题合并去重,按我配置的关键词和排名变化打分,再把结果实时推到钉钉群和微信里。整套东西代码量不算大,跑在一台最低配云主机上就能 24 小时工作,属于典型的"小工具解决大问题"。这篇文章会把完整搭建过程、核心代码逻辑、部署方式和这半年我踩过的坑都写出来,想自己做一套信息监控工具的运营、产品、开发和自媒体朋友都可以直接参考。

1. 为什么是"热搜":信息差的真实场景与 TrendRadar 的定位

1.1 一次"慢半拍"的代价

做内容这行有个很现实的情况:热点本身不是秘密,谁都知道它存在,关键差别是"你什么时候知道"。同样是追一个热点,早两个小时看到,你有时间准备角度、截图、找素材、做标题;晚两个小时看到,你只能跟在别人内容后面发一篇差不多的东西,平台流量早就分完了。

那次失误给我的教训很明显:人脑记不住六个平台的实时榜单,更不可能每十分钟手动刷一遍。你想保持信息的实时性,靠"勤快"是做不到的,必须靠自动化。另外还有一个更隐蔽的问题:就算把榜单刷出来了,上面几十条内容里真正和你领域相关的可能只有两三条,靠肉眼筛选既累又容易漏。所以我需要的不是"更多信息",而是一台能自动筛选并快速通知我的雷达。

1.2 什么才算"打破信息差":三个可量化指标

工具开始写之前,我先给自己定了一套衡量标准,否则做着做着就不知道到底有没有用。

  • 时效延迟:从一条热点进入平台榜单,到推送到达我的手机,目标是控制在十分钟以内。超过这个时间,热点可能已经开始在朋友圈发酵,我的先发优势就没了。
  • 覆盖率:至少要覆盖六个以上主流平台。原因很简单,不同平台的热搜生态差别很大,只盯微博容易漏掉 B 站、知乎这类垂直圈层的热点。
  • 噪音率:推送里我不关心的内容比例,目标低于百分之三十。如果推送十条里有七条都不想看,那工具就变成了新的噪音源,我会很快把它关掉。

这三个指标后来成为我所有设计决策的出发点:凡是能降低延迟、扩大覆盖面、减少噪音的方案,优先做;凡是会增加维护负担或者让推送变"吵"的方案,砍掉。TrendRadar 这个名字也是这时候定的,它本质上就是一台信息雷达,目标不是给你越多越好,而是给你"刚刚好"。

2. 链路拆解:一条热搜从平台数据到手机通知的完整旅程

2.1 模块划分与数据流向

动手写第一行代码前,我先在脑子里把整个系统拆成了六个模块,这个划分后来证明非常关键:

  1. 采集器:定时请求各平台热搜榜单,拿到原始数据。
  2. 清洗与规范化:把各平台不同的数据格式统一成同一种结构,包括标题、热度值、链接、排名。
  3. 去重与存储:把新数据写入 SQLite 数据库,同时判断它是全新事件还是已有事件的更新。
  4. 规则引擎:根据关键词命中、排名变化、平台权重算出每一条的推送分值。
  5. 推送器:把达到阈值的条目通过钉钉或企业微信机器人发出去。
  6. 调度器:统一控制采集频率、推送频率,以及失败重试。

数据流向是一条直线:采集器 -> 清洗器 -> 数据库 -> 规则引擎 -> 推送器。不需要消息队列,不需要分布式任务,因为单机处理的量级实在太小了:一次采集最多几十条记录,一分钟跑一轮也才几百条。这个数据规模用消息队列纯属自己给自己找麻烦。

2.2 技术选型:为什么是 Python + SQLite 而不是重框架

选型的时候我认真对比过几套方案。第一反应是上 Scrapy 加 Redis,后来又想过要不要引入 Celery 做定时任务,最后全部否掉了。这套工具的实时性压力远低于爬虫系统,数据量也不大,最合适的方案反而是最简单的那套:

组件我的选择备选方案选择理由
语言Python 3.10Go解析 JSON 方便,requests 生态成熟,个人开发效率高
抓取requests + 轻量正则Scrapy榜单页面结构简单,不需要框架的调度和中间件
存储SQLiteMySQL / Redis单文件、零配置、备份方便,几十万条数据无压力
定时APSchedulercrontab / Celery跨平台一致,进程内管理任务,逻辑清晰
推送钉钉/企微 Webhook邮件、Server酱免费、延迟低、支持 Markdown,手机端即时到达

最容易被忽略的一点是运维成本。我一个业余时间维护的项目,如果引入 Redis 和 Celery,等于给自己增加一个需要长期照看的服务。SQLite 文件放服务器上,坏了就直接拷贝走,完全没有管理负担。这也是我后来一直坚持的原则:个人工具没有 KPI 压力,唯一的 KPI 是你愿不愿意长期用下去,而"省心"是愿意用下去的前提。

3. 数据源接入实操:多平台热搜的抓取、去重与入库

3.1 数据源选型清单

不同的平台适合不同的信息需求。我做内容运营,关注的更多是泛娱乐和社会话题,所以选了这几个:

平台获取方式更新频率适合哪类人
微博热搜公开 JSON 接口分钟级,随时变动综合热点、突发事件、娱乐话题
知乎热榜公开 JSON 接口半小时到一小时一级科技、职场、生活方式深度话题
百度热搜网页榜单解析小时级下沉市场、大众搜索兴趣
B站热门网页榜单解析小时级视频创作、年轻人圈层热点
抖音热点网页热点榜半小时级短视频、娱乐向内容

这里要特别说明:我只接入平台公开的榜单页或官方提供的 JSON 接口,不做任何破解签名、绕过防线之类的操作,请求频率也压在五分钟以上一次,对平台几乎没有压力。做这套工具的底线是只采集公开数据、只服务自己,不涉及任何个人隐私,也不把抓取结果拿去做商业用途。这样我用着放心,也不需要担心平台来找麻烦。

3.2 一个最小可用的采集器实现

核心采集代码很简单,以微博热搜和知乎热榜为例。微博有个公开接口可以直接拿榜单数据,知乎热榜也有对应的公开 JSON 接口,返回的数据都是标准 JSON,解析起来非常省事。

import requests import time HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://weibo.com/" } def fetch_weibo_hot(): url = "https://weibo.com/ajax/side/hotSearch" resp = requests.get(url, headers=HEADERS, timeout=10) resp.raise_for_status() items = resp.json()["data"]["realtime"] result = [] for idx, item in enumerate(items[:30], start=1): title = item.get("word", "").strip() if not title: continue heat = item.get("num", 0) # 微博热搜链接可以直接用话题词拼接 link = f"https://s.weibo.com/weibo?q=%23{title}%23" result.append({ "platform": "weibo", "rank": idx, "title": title, "heat": heat, "url": link, }) return result def fetch_zhihu_hot(): url = "https://www.zhihu.com/api/v3/feed/topstory/hot-lists/total?limit=50" resp = requests.get(url, headers={"User-Agent": HEADERS["User-Agent"]}, timeout=10) resp.raise_for_status() items = resp.json()["data"] result = [] for idx, item in enumerate(items[:30], start=1): target = item["target"] title = target.get("title", "").strip() if not title: continue heat = target.get("heat", 0) link = target.get("url", "") result.append({ "platform": "zhihu", "rank": idx, "title": title, "heat": heat, "url": link, }) return result

这段代码里有两个容易踩的细节。第一是必须在请求头里带 User-Agent 和 Referer,如果只带 User-Agent,微博接口偶尔会返回空数据,我当时排查了快半小时才定位到是 Referer 缺失。第二是首页榜单虽然显示五十条,但我只取前三十条入库,因为排名太靠后的内容很快就会掉出榜单,推送价值很低,还会增加数据库里的噪音。

3.3 去重逻辑:同一个话题在不同平台的不同写法

数据源接了三个平台之后,我很快遇到一个真正需要动脑的问题:同一个事件在微博叫"某品牌联名推出新品",在知乎可能叫"如何看待某品牌联名",在百度可能叫"某品牌联名款"。如果只按标题精确匹配,这些就会被当成三条不同的事件,推送三遍,体验极差。

我采用的方案分两步:

第一步是标题归一化。先把标题做一轮清洗:统一全角半角、去掉所有标点空格、去掉括号里的补充说明、把"如何看待""如何评价"这类知乎常见前缀剥掉。清洗之后得到一个"核心标题"。

第二步是计算指纹。对核心标题取 SHA-1 哈希,作为这条事件在数据库里的指纹。新的采集数据进来时,先看指纹是否已经存在;如果存在,就认为这是旧事件的更新,只更新排名和热度,不新增记录。

import hashlib import re def normalize_title(title: str) -> str: # 统一全角/半角 trans = str.maketrans(",。!?、()", ",.!?()") title = title.translate(trans) # 去掉常见前缀 title = re.sub(r"^(如何看待|如何评价|怎么看待)\s*", "", title) # 去掉所有空白和标点 title = re.sub(r"[\s,,。!!??、()()::;;\"\"]", "", title) return title.lower() def fingerprint(title: str) -> str: return hashlib.sha1(normalize_title(title).encode("utf-8")).hexdigest()

这里要提醒一下:归一化不能做太狠。我一开始把"苹果发布会"和"苹果 发布会"统一成同样的指纹,结果把两个不同的事件合并了,反而漏掉了一条重要新闻。我的经验是宁可少合一角度的题目,也不要激进归一化;跨平台合并准确率不够时,用"关键词命中同一个行业词"来辅助,而不是强行合并标题。

3.4 SQLite 表结构与为什么不用 Redis

数据库我用 SQLite,就两张表。

CREATE TABLE IF NOT EXISTS hot_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, fingerprint TEXT NOT NULL, platform TEXT NOT NULL, title TEXT NOT NULL, url TEXT, heat INTEGER DEFAULT 0, rank INTEGER DEFAULT 0, status TEXT DEFAULT 'normal', -- new / up / down / normal / gone score INTEGER DEFAULT 0, first_seen_at DATETIME, last_seen_at DATETIME, push_count INTEGER DEFAULT 0 ); CREATE UNIQUE INDEX idx_fingerprint ON hot_events(platform, fingerprint); CREATE TABLE IF NOT EXISTS push_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, event_id INTEGER, pushed_at DATETIME, content TEXT );

用 Redis 做这件事当然也完全可以,但对我来说是过度设计。热点事件一天最多几千条记录,SQLite 轻松扛住,而且我可以直接把整个数据库文件下载到本地分析,这在 Redis 里反而麻烦。个人项目的核心原则是"能少一个依赖就少一个依赖"。一张表记录事件本身,一张表记录推送历史,后面排查"为什么这条没推"或者"这条推了几次"都非常方便,直接查 SQL 就行。

4. 消息推送:渠道选型、模板设计与避免打扰

4.1 推送渠道选型对比

数据抓回来之后,剩下的关键问题是怎么把它送到人眼前。我认真对比了四个可行的渠道:

渠道到达速度费用消息格式适合场景主要限制
钉钉自定义机器人秒级免费支持 Markdown个人/小团队信息聚合每分钟 20 条限制
企业微信群机器人秒级免费支持 Markdown已经在用企微的团队同钉钉,限制类似
Server酱秒级免费额度纯文本/Markdown推送到微信服务号第三方服务依赖
邮件 SMTP秒级到分钟级免费支持 HTML日报/周报汇总容易被手机通知栏折叠

我最后选了钉钉机器人,核心原因是 Webhook 足够简单、免费、秒级到达,而且 Markdown 格式可以把多条消息放在一条里推送,正好规避它的频率限制。实际操作中我建了两个群:一个"重点群"只接收关键词命中的高优先级消息,一个"普通群"接收全部去重后的榜单消息。早晨我只打开重点群就行,普通群用来晚上复盘。

4.2 钉钉机器人接入:加签、发送与限流应对

新版钉钉自定义机器人要求加签,这也是很多教程没讲清楚的地方。加签流程是:把当前毫秒级时间戳加上机器人密钥,做 HMAC-SHA256 签名,再把签名拼在 Webhook 地址后面。

import time import hmac import hashlib import base64 import urllib.parse import requests SECRET = "PASTE_YOUR_SECRET" ACCESS_TOKEN = "PASTE_YOUR_TOKEN" def sign_url(): timestamp = str(round(time.time() * 1000)) string_to_sign = f"{timestamp}\n{SECRET}" hmac_code = hmac.new(SECRET.encode("utf-8"), string_to_sign.encode("utf-8"), digestmod=hashlib.sha256).digest() sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) return ( f"https://oapi.dingtalk.com/robot/send?access_token={ACCESS_TOKEN}" f"&timestamp={timestamp}&sign={sign}" ) def send_dingtalk_markdown(title: str, text: str): data = { "msgtype": "markdown", "markdown": { "title": title, "text": text, }, } resp = requests.post(sign_url(), json=data, timeout=10) resp.raise_for_status() result = resp.json() if result.get("errcode", -1) != 0: raise RuntimeError(f"钉钉推送失败: {result}")

这里我最开始踩过一个很大的坑:把每一条热搜都单独发一条消息,一分钟内连续推三十条,结果钉钉直接返回错误码拒绝服务,原话是"触发限流"。正确的做法是合并推送:每十五分钟把这一轮分数最高的十到十五条攒成一条 Markdown 消息发出去,遇到排名突然飙升的紧急事件再单独插一条。这样既不会触发限流,用户也更容易形成阅读习惯。

4.3 消息模板:让一条通知在三秒内讲清楚三件事

推送消息不能只甩标题列表,至少要让接收者一眼看清三件事:这条热搜来自哪个平台、当前排在第几位、排名相比上次是涨了还是跌了。我最终使用的模板长这样:

#### 热搜雷达 14:02 · 重点关注 6 条 **[微博热搜 #2 ↑5] 某品牌联名新品发布 热度 980万** https://s.weibo.com/weibo?q=%23某品牌联名新品发布%23 **[知乎热榜 #7 NEW] 如何看待这次联名背后的定价策略** https://www.zhihu.com/question/123456 **[百度热搜 #12 ↓3] 联名款首日销售数据出炉** https://www.baidu.com/s?wd=联名款首日销售

每条都遵循"平台 + 排名 + 排名变化 + 标题 + 热度 + 链接"的结构。排名变化我用↑5、↓3、NEW这种符号,比写"较上次上升5位"省空间得多,扫一眼就能看出来是该关注的还是已经过气的。

有一点我想特别强调:模板里一定要带链接。钉钉的消息可以点链接直达原平台,这意味着运营拿到消息后可以直接在手机上点开看详情、截素材,整套流程从"通知-跳转-使用"是闭环的。如果只发标题不带链接,看到消息还要自己去搜,效率会大打折扣。

4.4 防打扰策略:推送频率、冷却时间与夜间静默

工具上线第一周,我收到的推送每天接近八十条,手机一直在震,烦到差点把整个项目删掉。后来我给系统加了三条硬规则,推送数量立刻降下来,体验也质变了:

  • 合并批次:同一批次收集十五分钟,统一推送一次,正常情况下每天推送不超过十次。
  • 冷却时间:同一个事件二十四小时内最多推送两次。第一次是刚上榜或排名飙升时,第二次是排名再次大幅提升时,其余情况只更新数据库不推送。
  • 夜间静默:晚上十一点到早上七点之间只记录不推送。热点晚上确实经常爆发,但凌晨两点看一条推送并不会让你做出更好的决策,只会影响睡眠。

这三条规则才是整套系统里让我觉得"终于能用了"的关键。做信息推送工具,最难的不是让你看到更多,而是帮你把打扰控制在可接受范围内。

5. 部署上线:定时调度、异常恢复与成本控制

5.1 运行方案对比:本机、云主机与云函数

代码写完之后,我面临在哪运行的问题。对比过三种方案:

方案成本优点缺点
本机电脑跑免费开发调试方便关机就停,笔记本合盖就断
最低配云主机每月几十元真正 7x24 小时运行,完全可控需要简单了解 Linux
云函数定时触发有免费额度按调用计费,几乎零成本阈值限制较多,Webhook 网络出口需自行确认

我最终选择了一台最低配的国内云主机,原因是它不需要我操心平台限制,cron 想怎么配就怎么配,出了问题还能直接 SSH 上去看日志。如果你完全是新手,建议先在本地电脑把采集和推送跑通,再迁移到云主机,不要一上来就买服务器。

5.2 用 systemd 和 cron 守住进程

部署到云主机后,我用系统自带的 cron 定时触发主脚本,路径统一放在/opt/trendradar下。一个最简单的 crontab 配置:

*/5 * * * * cd /opt/trendradar && /opt/trendradar/venv/bin/python main.py >> logs/run.log 2>&1

这里有几个容易被新手忽略的细节。第一是必须使用虚拟环境里的 Python 绝对路径,直接写python大概率会调用系统自带的版本,缺依赖。第二是日志重定向不能省,2>&1直接把标准错误也写进日志,排查问题全靠它。第三是如果某些任务的耗时可能超过五分钟,要在程序里加一个简单的文件锁,防止上一次还没跑完下一次又启动了。

如果你希望进程崩溃后自动重启,还可以用一个 systemd 服务配合Restart=always来守护。不过我实测下来,一个普通 Python 脚本配合 cron 加日志已经足够了,脚本崩了下次 cron 触发时会重新启动,不会造成实质影响。

5.3 失败重试、日志与异常快照

线上运行最怕的不是脚本报错,而是静默失败:推送接口超时了,程序吞掉异常继续跑,结果你收不到任何通知。为了解决这个,我做了三件事:

  1. 关键请求重试:推送和采集的 requests 调用统一外包一层,失败后按 1 秒、3 秒、10 秒的间隔重试三次,三次仍失败才记录并放弃。
  2. 日志分级:正常采集写 INFO,榜单数据异常写 WARNING,推送失败写 ERROR。每天早晨我会花一分钟看昨晚的错误日志。
  3. 异常快照:每次崩溃都在logs/last_error.txt里保存完整的异常堆栈和当时的上下文数据。这对定位"偶发空列表"这类问题特别管用。

我第一次在网上部署服务时特别喜欢写except Exception: pass,觉得不报错就是好了。运行半个月后才发现,静默吞掉的异常让整个系统形同虚设。后来我把所有异常都记录到日志,然后再决定哪些是真的无所谓、哪些必须重试。个人项目同样需要这个纪律。

5.4 成本与安全合规的底线

整套系统每月的成本就是一台云主机的几十元,其余全是免费的公开接口和免费推送渠道。为了把成本压到最低,我没有给域名备案做复杂的事情,直接用 IP 访问或者干脆不需要访问入口,反正代码都在服务器上本地跑。

合规这块我再强调一遍:只采集公开榜单数据、请求频率不低于五分钟一次、不存储任何用户隐私、推送只发给自己或自己所在的群。这套工具的价值不在于"能拿到什么别人拿不到的数据",而在于"把公开数据用更合理的方式组织和触达"。我不会把它扩展到任何与隐私相关的方向上,这也是我能一直放心让它跑着的底气。

6. 上线半年后的实测复盘:哪些设计有用,哪些还需要改进

6.1 真实运行数据一览

TrendRadar 目前已经跑了半年多,覆盖六个平台,每天采集超过两百轮。我统计了近三十天的数据:

  • 日均抓取事件约 120 条,去重后实际新增事件约 40 条。
  • 关键词命中率约 23%,也就是每天有 9-10 条和我行业直接相关。
  • 推送延迟中位数在 4 分钟左右,也就是说一条热点上榜后,最慢十分钟内我能在手机上看到。
  • 重点群推送的点击率约 30%,普通群约 10%,说明关键词过滤确实有效。

最有价值的变化不是"我看得快了",而是"我判断要不要跟这个热点的时间大幅缩短了"。以前我需要刷榜、比对多个平台、判断话题真假和发展趋势,现在推送里连热度、排名变化和链接都带好了,我只需要看三秒就能决定要不要跟进。这个效率提升是实打实的。

6.2 踩坑记录:四个我差点放弃的问题

第一个坑是钉钉限流。一开始我把每条热搜单独推送,一分钟推三十条,直接被钉钉拒绝。后面把所有内容合并成一条消息,用批次推送代替逐条推送,这个问题才算根治。顺便说一句,钉钉自定义机器人的限流是每分钟二十条,不是文档里写的那么多,别心存侥幸。

第二个坑是标题归一化过度。我最早把清洗做得太激进,"苹果发布会"和"苹果 发布会"被当成同一个指纹,有两次真正不同的活动被我合并掉,漏掉了重要消息。后来我把清洗逻辑改保守,只处理标点和常见前缀,把跨平台合并的权重交给"关键词命中"来解决。宁可漏合一个,也不要错合一个。

第三个坑是新上榜事件的比较异常。一个事件第一次进入数据库时没有"上一次排名",我当时在比较排名变化时直接拿来算差值,程序立刻报错。处理很简单:如果是status = NEW的事件,不计算排名涨跌,直接给一个初始加分;其余状态才比较差值。

第四个坑是夜间接口偶发返回空数据。有些平台的榜单接口在凌晨会偶尔返回空列表,如果程序把空结果当成"当前没有热点",会把数据库里的老事件全部标记成出榜,导致第二天早上推送逻辑全乱了。我的处理方式是:如果一次采集得到的数据为空,跳过本轮更新,保留上次数据。宁可重复数据,也不能错误地"清空"事件状态。

6.3 下一步迭代:从"推送"进化到"判断"

目前的版本已经满足日常使用,但我已经在计划下一版。一个方向是给热搜打行业标签,比如科技、娱乐、财经、社会,这样可以直接按标签分流到不同的群,而不是只靠关键词匹配。另一个方向是做热度趋势曲线,把一条热搜从上榜到出榜的完整生命周期记录下来,用来判断"这个话题还有没有长尾流量空间",这对内容选题尤其有用。

从实际使用体验来看,最值得迭代的不是采集速度,也不是推送速度,而是理解能力。现在系统能告诉我"什么在热",但还不能告诉我"为什么热"以及"还能热多久"。后两个问题才是内容决策里真正值钱的部分。不过那是一段更长的路,先把雷达做好,再慢慢给它加"思考"的能力。

最后分享一点个人体会。工具跑起来之后,我最深刻的感受其实是:推送类工具最大的敌人不是技术复杂度,而是"用户的注意疲劳"。如果手机每天早上醒来都有二十条推送等你处理,那它就从雷达变成了负担。所以我自己反而在刻意做减法:把不相关平台的推送关掉、把关键词设得更严格、只保留一个群的实时权限。信息差从来不是越多越好,而是该知道的时候知道、不该被打扰的时候安静。这也算是 TrendRadar 给我上的最值钱的一课。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询