做一面实时 X 推文墙：没人提醒你的那几件事
做一面实时 X 推文墙：没人提醒你的那几件事
这个搜索词底下是三个不同的项目：
- 活动大屏 —— 把关于你活动的推文投在舞台背后的屏幕上
- 实时监控 —— 有人说了什么就告警，不涉及显示
- 开直播 —— 在 X 上直播视频，那是写操作，不是我们做的事
这篇讲前两个。 ★取推文大概只占整个工作的 20%。★ 剩下的是当真实观众发现你那面墙之后会发生什么。
先问：你到底需要多实时？
这个答案决定了架构，而人们几乎总是把它定得过高。
| 延迟 | 做法 | 什么时候对 |
|---|---|---|
| 秒级 | 流式连接 | 直播活动、交易、事故响应 |
| 1-5 分钟 | 轮询 | ★几乎其它一切★ |
| 15 分钟以上 | 定时批处理 | 报表、分析 |
⚠️ ★活动大屏不需要秒级延迟。★ 一条推文在写出来 90 秒后出现在屏幕上，对任何看屏幕的人来说和"即时"没有区别。为一场两小时的会议搭一套流式管道，是你不需要的工程。
秒级真正要紧的地方： 突发新闻告警、盯一次发布、任何有人要立刻据此行动的场景。轮询与流式的完整权衡。
如果你确实需要秒级，那是一条持久连接而不是反复请求 —— 我们正是为此提供了流。
★三个真正会搞垮活动大屏的问题★
这部分在每一篇"做推文墙"的教程里都是缺失的，而它们每一个都让某人在观众面前难堪过。
1. 审核 —— 会毁掉职业生涯的那个
你把活动标签投在会场屏幕上。观众里有人注意到了，于是带着你的标签发了一条不堪入目的内容。 它出现在你 CEO 身后的屏幕上。
★这不是假设。对任何规模足够大的活动，这是无审核推文墙的默认结局。★
唯一安全的架构是一个队列：
取数 → 暂存 → 人工批准 → 显示
⚠️ 自动过滤本身不够。 敏感词表拦不住有创意的拼写，而文本过滤器根本不检查图片内容。 ★一个盯着队列、手上有"批准"按钮的人，才是真正有效的控制。★
务实的折中： 预先批准的账号(讲者、赞助商、员工)自动通过，其余全部进队列。
2. 去重
轮询返回的结果会重叠。没有去重，同一条推文会反复循环上墙，显示看起来就像坏了。
★按推文 ID 去重，不要按文本★ —— 转发和几乎相同的推文之间总有细微差别，按文本匹配要么漏掉、要么错误地把不同推文合并。
3. 显示节流
在主题演讲期间，推文到达的速度会超过任何人的阅读速度。 一面把每条推文立刻渲染出来的墙，会变成看不清的一片模糊。
★节流到可读的节奏 —— 每 4-8 秒一条★，多余的排队。冷场时循环播放最近的高互动推文，而不是显示一块空屏。
一份能跑的实现
import requests, time, json, pathlib
from collections import deque
BASE = "https://api.socialapi.tech"
KEY = "your_api_key"
HDRS = {"X-API-Key": KEY}
APPROVED = pathlib.Path("approved.jsonl") # 墙上渲染的内容
TRUSTED = {"yourbrand", "yourceo", "sponsor1"} # 这些自动通过
class Wall:
def __init__(self, query, min_faves=0):
self.query = query
self.min_faves = min_faves
self.seen = set()
self.pending = deque()
def poll(self):
r = requests.get(f"{BASE}/v1/search/advanced",
params={"query": self.query, "product": "Latest",
"limit": 50},
headers=HDRS, timeout=60)
r.raise_for_status()
new = 0
for post in r.json()["data"]:
if post["id"] in self.seen:
continue # ★按 ID 去重★
self.seen.add(post["id"])
if (post.get("like_count") or 0) < self.min_faves:
continue
if post["author"]["username"].lower() in TRUSTED:
self.approve(post) # 预批准账号
else:
self.pending.append(post) # ★其余全部等待★
new += 1
return new
def approve(self, post):
with APPROVED.open("a", encoding="utf-8") as f:
f.write(json.dumps(post, ensure_ascii=False) + "\n")
wall = Wall('#yourevent2026 -filter:retweets', min_faves=1)
while True:
found = wall.poll()
print(f"新增 {found} 条 · {len(wall.pending)} 条待审")
time.sleep(60) # ★大屏用 60 秒完全够★
每道防线在干什么：
- ★
TRUSTED自动批准★ —— 让墙在冷场时也有内容，而不必对所有人敞开 - ★
pending队列★ —— 任何来自未知账号的内容都要等人工 - ★
min_faves★ —— 在噪音到达任何人之前就过滤掉零互动内容 - ★
-filter:retweets★ —— 防止一条热门推文用副本占满屏幕
⚠️ 注意这里刻意缺少了什么：未知账号的自动通过路径。(作者资料随每条结果返回，所以可信判断是一次字段查找而不是第二次调用。) 那不是疏漏 —— 那正是整个安全性所在。
不带显示的实时监控
这个搜索词的另一半。没有墙、没有队列 —— 你只想知道有人说了什么。
架构更简单，因为没有观众可以让你难堪：轮询、去重、超过阈值就告警。★失败模式从"屏幕上出现脏话"变成了"告警多到没人看"★ —— 这在怎么设计不吵的告警里讲了。
大家会问的问题
怎么做一个实时推特信息流？ 定时轮询搜索、按推文 ID 去重、渲染。在任何内容上屏之前加一道审核队列。
怎么给活动做推文墙？ 同样的机制，再加三道防线：人工审核、去重、显示节流。
怎么把推文投到屏幕上？ 取数、存下已批准的、从你自己的存储里渲染，而不是每次刷新都调接口。
需要审核队列吗？ ★任何面向公众的场合都需要。★ 一面无审核的标签墙，在真实活动上迟早会显示出你不想出现在舞台背后的东西。
能自动审核吗？ 部分能。 敏感词表拦不住有创意的拼写，而且不检查图片。一个手上有批准按钮的人才是有效的控制。
推文墙需要多实时？ 一分钟就够。 看屏幕的人分辨不出 5 秒和 60 秒的差别。
轮询和流式有什么区别？ 轮询是反复问；流式是握着一条连接接收推送。完整对比。
实时信息流该多久轮询一次？ 大屏 30-60 秒。更频繁更贵，而且视觉上没有任何差别。
同一条推文为什么反复出现？ 缺去重。 按推文 ID 去重 —— 按文本匹配不可靠。
怎么防止一个人刷屏？ 限制每个作者在时间窗内的条数，并排除转发。
冷场时怎么办？ 循环播放最近的高互动已批准推文，而不是显示空屏。
推文在屏幕上该多久换一条？ 每 4-8 秒一条。 更快就读不了了。
能按互动量过滤吗？
能 —— 一个 min_faves 阈值能在噪音进入队列之前就滤掉。
怎么把实时流嵌到网站上？ 简单时间线用官方嵌入组件；只有需要过滤或多账号时才自己做 —— 对比在这里。
能同时显示多个话题标签吗？
能 —— 用 OR 合并成一条查询，或者跑几条再合并去重。
怎么实时追踪一个标签？ 把标签当搜索查询轮询。如果是要衡量而不是显示，见标签分析。
最好的推文墙软件是哪个？ 有几个托管产品。只有当你需要现成产品做不到的定制过滤或品牌化时才自己做。
能用于电视直播吗？ 机制相同，但审核要更严 —— 广播对失误的容忍度比会议屏幕低得多。
怎么只显示带图的推文？
在查询里加 filter:images —— 操作符集。
需要做缓存吗？ 需要。 从你自己的存储渲染，不要每次屏幕刷新都调接口。
如果接口变慢了会怎样？ 继续从你的存储渲染。 墙不该依赖一次实时请求才能显示东西。
怎么在推特上开直播？ 那是从 App 里做视频广播 —— 一个写操作，和显示推文无关。★任何只读接口都做不到。★
能实时流式接收推文吗？ 用持久连接可以。做墙不需要；给突发新闻告警可能需要。
流式的延迟大概多少？ 从发布到送达几秒。轮询要在此之上再加你的间隔。
一面墙能处理多少推文？ 约束是显示速率，不是取数。 每 5 秒一条的话，一小时显示 720 条，和到达多少无关。
能用手机审核吗？ 把队列做成一个简单网页就到处都能用。 大多数团队就是这么做的。
该显示回复吗？
墙上通常不要 —— 回复脱离上下文难以理解。用 -filter:replies 排除。
怎么避免显示已删除的推文？ 如果批准和渲染之间有时间差，渲染前再核一次。 一条推文可能在你取到之后被删。
能在墙上显示粉丝数吗？ 作者资料随每条推文返回，所以能 —— 不过对观众来说通常没什么用。
如果有人在批准之后改成了冒犯内容怎么办？ 他能编辑或删除，但你存下的副本照样会渲染。★要留一个能立刻把某条从轮播里移除的开关。★
活动前怎么测试这面墙？ 拿一个繁忙的公开标签跑一小时。 那能在问题真正要紧之前暴露出节流和去重的毛病。
做实时流需要开发者账号吗？ 读公开数据不需要。 只有写操作(比如直播)才需要 —— 开发者账号涉及什么。
跑一面墙要花多少钱？ 每小时轮询次数 × 小时数。60 秒间隔就是每小时 60 次 —— 调用量的成本。
能离线跑一个备份吗？ 预先批准一批推文，连接断了就轮播它们。 ★做直播活动，这个一定要有。★
如果你要做一个
取数就是一个定时的搜索调用。★决定它能不能用的，是审核、去重和节流 —— 而第一个是"一次好的演示"和"一次事故"之间的差别。★
我们的 API 覆盖取数这一段 —— 支持完整操作符语法的搜索、用于判阈值的互动数据，以及每条结果都带作者资料，所以"可信账号自动通过"是一次字段判断而不是第二次调用。真正需要秒级时，用流代替轮询。
它做不到的： 直播视频，或者替你审核。前者是写操作；后者是只有人该做的决定。