SC 连发、电话提醒、动态发布这几件事最近在一个游戏主播社区里被聊得比较多。起因是一个叫“鬼叔”的主播连发 SC 提醒另一个叫“小豪”的玩家别碰某类内容小豪后来发动态解释自己当时没有看到电话等打开记录才发现夜里 12 点收到过提醒。这事在社区里讨论度很高但抛开围观情绪它其实暴露了一个很实际的技术问题直播平台的 SC、私信、动态、电话提醒分散在不同渠道一个人很难实时盯住所有消息。这次我们就把这个场景做成一个正经项目来聊。目标不是吃瓜而是围绕“SC/弹幕监听、关键词过滤、电话语音提醒、动态自动发布、批量任务队列”搭建一套直播互动全渠道通知服务。它适合主播、社群运营、直播数据分析、以及想给自己的小工具加实时通知能力的人。文章会从架构设计、环境准备、启动方式、功能测试一直讲到接口调用和常见排错所有代码都给通用示例具体的平台 API 参数需要你按实际接入的服务商文档替换。1. 核心能力速览先说结论这套通知服务能做什么一会直接看功能测试部分。能力项说明项目定位直播互动消息接收、过滤、多渠道通知服务消息来源SC / 弹幕 / 私信 / Webhook需按直播平台官方 API 接入消息过滤支持关键词、用户 ID、金额区间、时间窗口过滤通知渠道站内回调、Webhook、邮件、电话语音提醒动态发布自动生成动态草稿支持 dry_run 人工审核后再发布批量任务基于队列的批量通知与失败重试运行环境Python 3.9Redis 可选跨平台启动方式命令行启动 配置文件也支持 API 服务模式是否需要 GUI不需要适合场景主播消息同频、社群运营、直播数据监控、自动化提醒这里的“支持”指的是通用架构支持不是某个具体平台已经封装好的现成插件。实际接入时你需要确认目标平台是否提供官方消息推送接口以及接口的调用频率、授权方式和数据字段。不要试图绕过平台限制去抓取用户隐私数据这是底线。2. 适用场景与使用边界2.1 这套服务适合谁主播和直播间运营需要第一时间知道重要 SC 或私信但又不可能一直盯着后台。社群管理员消息量大需要根据关键词把“重要消息”捞出来同步到多个群或负责人。做直播数据分析的开发者需要一个稳定的消息采集和归档入口SC 金额、用户、时间都是结构化数据。个人开发者想给自己的小工具加通知能力比如收到指定类型消息后触发电话语音提醒。2.2 不适合的场景不适合做批量骚扰或自动回复真人。所有自动触达用户的行为都要有明确授权和服务条款依据。不适合绕过直播平台 API 做爬虫式抓取。平台接口规则一直在变硬抓既不稳定也有封号风险。不适合对用户隐私数据做长期留存和分析。SC 内容、用户 ID 属于敏感信息保存和删除策略必须合规。2.3 合规边界直接看这几点使用直播平台官方提供的 API不要逆向私有不公开接口。电话语音提醒、短信通知必须确保接收者已同意接收并保留退订机制。动态自动发布建议先生成草稿人工确认后再发布。如果涉及第三方素材或用户内容发布前要确认授权。这套系统本身是工具用途是否合规取决于使用方式。下面的部署过程也按“先本地测试、再小范围验证、最后接入正式环境”的顺序来。3. 系统总体设计先不着急写代码我们把整个通知链路拆开。一套直播互动通知服务典型的数据流是直播平台消息推送 | v 消息接收模块 (Webhook / 长连接 / 轮询) | v 消息过滤模块 (关键词、金额、用户、时间) | v 通知路由模块 (邮件 / Webhook / 电话语音) | v 动态发布模块 (生成草稿 / 人工确认 / 发布) | v 任务队列与日志 (Redis Queue 日志文件)从设计上分成四个核心模块消息接收层负责对接平台的消息推送统一转换成内部 JSON 结构。过滤路由层根据配置判断消息是否值得通知以及走哪个通知渠道。通知执行层调用邮件、Webhook、电话语音服务的发送逻辑。任务管理层把耗时操作放进队列失败自动重试。这样的好处是每个模块都能单独测试。消息接收挂了不影响通知执行通知服务商限流了也不会丢掉消息本体。4. 环境准备与前置条件4.1 基础环境操作系统Windows、Linux、macOS 都可以建议开发机和服务器都用 Linux。Python3.9 或更高版本建议 3.10。pip升级到最新版。Redis可选如果消息量大建议装一个用于任务队列和去重。网络环境服务器能访问直播平台 API 和通知服务商接口即可。4.2 账号与依赖你需要准备一个直播平台的开发者账号/应用能申请到消息推送或事件订阅的权限。一个通知服务商账号比如支持电话语音的云通信服务或者自备 SIP 语音网关。如果想发邮件通知准备 SMTP 配置。检查 Python 环境python --version pip --version创建虚拟环境python -m venv venv source venv/bin/activate # Linux/macOS venv\Scripts\activate # Windows安装依赖pip install requests pyyaml redis再准备一个requirements.txt方便后续部署requests2.31.0 pyyaml6.0 redis5.0.0 apscheduler3.10.05. 安装部署与启动方式5.1 项目目录结构建议按模块拆分别把代码全部塞在一个文件里live-notify/ ├── app.py # 入口启动接收和调度 ├── config.yaml # 配置文件 ├── requirements.txt ├── logger.py # 日志模块 ├── modules/ │ ├── receiver.py # 消息接收Webhook/轮询 │ ├── filter.py # 消息过滤 │ ├── notifier.py # 通知执行 │ ├── dynamo_pub.py # 动态发布模块 │ └── queue.py # 任务队列封装 └── tests/ ├── test_filter.py └── test_notifier.py这个结构不复杂但每个模块的职责是清楚的。5.2 配置文件示例config.yaml是核心所有过滤规则、通知渠道、队列参数都放在这里app: host: 127.0.0.1 port: 8000 debug: true receiver: type: webhook # 平台消息接收方式 path: /api/webhook secret: your-webhook-secret filter: keywords: [SC, 提醒, 别碰] min_amount: 1.0 # 最低金额单位按平台实际 users: [] # 空数组表示不按用户过滤 time_window: 60 # 同一个用户去重时间窗口单位秒 notify: channels: webhook: enabled: true url: https://example.com/callback email: enabled: false smtp_host: smtp.example.com smtp_port: 465 from_addr: notifyexample.com to_addr: adminexample.com phone: enabled: true provider: your-phone-provider api_key: replace-with-your-key call_number: 10086 # 接收提醒的号码 publisher: enabled: true dry_run: true # 先生产草稿不真正发布 queue: redis_url: redis://127.0.0.1:6379/0 max_retry: 3 retry_delay: 5注意provider、api_key、call_number都是占位需要替换成你实际使用的电话服务商参数。5.3 启动消息接收服务入口文件app.py负责启动 HTTP 服务和后台任务import yaml from modules.receiver import start_receiver from modules.queue import start_worker def main(): with open(config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) start_receiver(config) start_worker(config) if __name__ __main__: main()启动命令python app.py启动成功后日志会输出监听地址和端口。首次测试建议把debug设为true这样每条消息都会打印完整中间结果方便排查过滤逻辑。6. 功能测试与效果验证6.1 消息接收测试先把服务跑起来用 curl 模拟一条平台推送的 SC 消息curl -X POST http://127.0.0.1:8000/api/webhook \ -H Content-Type: application/json \ -H X-Secret: your-webhook-secret \ -d { type: superchat, user: user_1001, amount: 66.0, content: 鬼叔连发SC提醒测试一下, ts: 1730000000 }判断标准日志里能看到消息接收成功数据结构完整。2025-01-01 12:00:01,001 INFO received message: {type: superchat, ...}如果收不到先确认端口和路径是否正确再看secret校验逻辑。6.2 关键词过滤测试过滤模块的逻辑是SC 消息金额大于等于min_amount且内容命中keywords中的任意一个才进入通知流程。代码示意def match_filter(msg, config): keywords config[filter][keywords] min_amount config[filter][min_amount] if msg.get(type) ! superchat: return False if msg.get(amount, 0) min_amount: return False content msg.get(content, ) return any(kw in content for kw in keywords)测试用例测试消息金额是否包含关键词预期结果普通弹幕0是不过滤不通知SC“提醒一下”6是通过过滤SC“随便聊聊”66否不通过6.3 电话语音提醒测试这是很多人关心的功能。接到重要 SC 后系统自动触发电话语音提醒相当于“夜里 12 点提醒你那个电话”的自动化版本。调用电话服务商接口的封装import requests def send_phone_call(config, message): phone_conf config[notify][channels][phone] url https://api.your-phone-provider.com/call payload { api_key: phone_conf[api_key], call_number: phone_conf[call_number], text: f检测到重要直播消息{message}, retry: 2 } response requests.post(url, jsonpayload, timeout30) response.raise_for_status() return response.json()测试时建议先填写真实的测试号码并在电话服务商后台把语音内容改成“测试通知”确认能正常接听后再接入正式直播间消息。如果电话没响排查顺序是电话服务商余额和套餐是否正常。接收号码是否经过授权。语音模板是否需要提前审核。日志里是否出现send_phone_call的异常。6.4 动态发布测试动态发布模块默认开启dry_run意思是只生成动态内容草稿不真实发送。这一步非常关键能让自动发布“保留一道人工闸门”。def build_draft(msg): draft { content: f收到重要消息 {msg[user]}{msg[content]}, status: draft } return draft操作步骤在config.yaml中开启publisher.enabled。保持dry_run: true。触发一条 SC看日志里是否生成草稿。确认内容无误后再把dry_run改成false接入真实发布账号。6.5 批量任务测试批量任务的场景是这样的凌晨积压了一批 SC 和私信系统按队列逐个执行通知而不是同时并发把所有消息都发出去避免接口限流。启动 worker 之后可以用一个简单脚本往队列里塞测试任务from modules.queue import enqueue for i in range(10): task { task_id: ftest_{i}, type: notify, payload: {content: f消息{i}} } enqueue(task)然后在日志里观察 worker 是否按顺序消费了 10 条任务失败重试是否生效。7. 接口 API 与批量任务7.1 对外 API 设计除了接收平台 Webhook这套服务也可以对外提供 API方便你接到其他系统里。常用接口建议方法路径作用POST/api/webhook接收平台消息推送POST/api/notify手动触发一次通知POST/api/task创建一个批量任务GET/api/task/{id}查询任务状态GET/health健康检查7.2 手动触发通知手动触发通知的接口示例适合拿来测试通知渠道curl -X POST http://127.0.0.1:8000/api/notify \ -H Content-Type: application/json \ -d { channel: phone, message: 这是一条测试语音提醒 }7.3 创建批量任务批量任务接口适合在直播结束后统一补发提醒curl -X POST http://127.0.0.1:8000/api/task \ -H Content-Type: application/json \ -d { type: notify_batch, items: [ {channel: email, message: SC1}, {channel: phone, message: SC2} ] }返回结果示例{ task_id: task_123456, status: queued, created_at: 1730000000 }7.4 Python 调用模板如果你想把通知能力集成到现有 Python 工具里可以直接用 requests 调用上述 APIimport requests BASE_URL http://127.0.0.1:8000 def create_notify_task(messages): payload { type: notify_batch, items: messages } resp requests.post(f{BASE_URL}/api/task, jsonpayload, timeout10) resp.raise_for_status() return resp.json() if __name__ __main__: messages [ {channel: phone, message: 重要SC请查看}, {channel: webhook, message: 新私信提醒} ] print(create_notify_task(messages))7.5 批量任务队列设计批量任务不建议用线程池无脑并发更稳妥的做法是任务进入 Redis 队列。Worker 以固定速率消费比如每秒最多 5 个任务。失败任务进入重试队列重试最多 3 次。重试仍失败的任务写入失败日志人工查看。这样可以保护通知服务商接口避免短时间大量请求触发限流。8. 资源占用与性能观察这套系统本身不涉及 GPU 推理资源占用主要看消息量和通知频率。8.1 观察哪些指标CPU消息过滤和 JSON 解析非常轻量单机普通配置足够。内存如果 Redis 队列堆积大量任务内存会缓慢上升需要监控队列长度。网络带宽电话语音通知不会消耗本机带宽但 Webhook 回调和平台消息推送会占少量带宽。8.2 影响性能的关键因素消息吞吐量如果直播间每分钟有几千条弹幕建议过滤逻辑尽量前置只保留 SC 和指定关键词消息进入后续流程。通知服务商限流电话语音接口通常有并发限制批量通知必须走队列。日志写入日志级别建议使用 INFO避免 DEBUG 日志在高吞吐下占满磁盘。8.3 如何观察启动服务后用命令观察进程资源top -p $(pgrep -f app.py)Redis 队列长度检查redis-cli llen task_queue如果队列长度持续上涨说明消费速度跟不上需要加快消费速率或延长重试间隔。9. 常见问题与排查方法问题现象可能原因排查方式解决方案Webhook 收不到消息路径写错、secret 不对、平台未配置订阅查看服务日志curl 手动模拟推送修正路径和配置参数过滤后没有通知关键词不匹配、金额低于阈值开启 debug 日志打印过滤前后数据调整关键词或调低金额电话语音提醒未触发服务商 key 错误、号码未授权、模板未审核检查日志中的异常和 HTTP 状态码更换测试号码确认语音模板动态发布失败账号授权过期、内容触发平台审核看发布接口返回的错误码重新授权或改为 dry_run 人工发布队列任务大量失败通知服务商限流查看 worker 日志中的 retry 记录降低消费速率增加重试间隔Redis 连接不上Redis 未启动或地址配置错误redis-cli ping启动 Redis 服务检查配置端口被占用8000 端口已有服务lsof -i:8000修改 config.yaml 中的端口10. 最佳实践与使用建议10.1 消息格式统一不要直接把平台返回的原始 JSON 向下游传递。建议在消息接收层就转换为统一格式比如统一成{type, user, amount, content, ts}后面所有模块只认这一种结构这样换平台时只改接收层的解析逻辑。10.2 先 dry_run 再上线动态发布和电话提醒这两类高影响操作必须先跑 dry_run。动态生成草稿给人工看电话只打给测试号码确认无误后再放开正式通知。10.3 日志和任务可追溯每条消息都要有唯一 ID日志里记录完整链路消息收到 - 过滤通过 - 生成通知任务 - 通知成功出问题时能快速定位是哪个环节断了。10.4 限流与熔断电话通知、邮件通知这类外部服务必须做限流。建议单个用户 5 分钟内最多通知 1 次。全局通知速率不超过服务商限制的 80%。连续失败超过 10 次自动暂停该渠道并告警。10.5 隐私与授权SC 内容、用户 ID、手机号都是敏感信息。建议日志中打码用户 ID。不保存无需留存的原始文本。定期清理历史数据。电话提醒功能必须获得接收者明确同意。10.6 合规发布动态自动发布动态建议满足三个条件内容不包含未经授权转载的素材。不冒充他人。不发布误导信息。即使内容由系统生成最终发布责任仍然在操作者。11. 总结与下一步这套直播互动通知服务最值得试的地方是把“SC/私信/电话/动态”这些分散渠道统一到一个配置里。部署后你先验证三件事Webhook 能不能收到消息、关键词过滤是否准确、电话语音提醒能不能打通。最容易踩的坑是平台 API 变动和通知服务商限流所以一开始就做好队列和日志后面会省很多事。接下来可以继续扩展的方向把 SC 数据做成交互式报表统计高价值用户接入更细粒度的消息路由比如不同金额走不同通知渠道或者把动态发布改成“AI 辅助草稿 人工审批”的工作流。建议收藏备用等需要给自己的直播运营加提醒能力时按这篇文章的流程重新走一遍就行。