尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

Python+Twilio短信通知系统实战:自动化告警与消息推送

Python+Twilio短信通知系统实战:自动化告警与消息推送 写东西之前我先问自己一个问题做一个在线服务或者自动化脚本的时候你最怕什么我最怕的是出了问题没人知道。数据库满了没人报凌晨跑批挂了没人报服务器被爬虫打爆了也没人报——等到白天用户开始骂了才意识到昨晚就炸了。我搭这个用Python和Twilio构建的短信通知系统就是为了把“事后被骂”变成“事前被通知”。说白了就是给程序安一个传话筒关键事件发生时让它主动发消息到你的手机上。这篇文章不是那种高深架构课而是一套可以直接抄作业的落地实践。不管你是做运维告警、订单状态提醒、预约通知还是自己的量化策略、定时脚本要出结果这套系统的思路都能直接套用。我会把方案选型、环境准备、代码实现、踩坑记录全部写出来新手照着步骤能跑通老手也能拿来当模块接进现有服务。1. 整体设计思路为什么选Twilio而不是自己搭短信网关1.1 先拆需求短信通知系统到底要解决什么问题很多人在动手之前容易犯一个错误一上来就研究短信 API 怎么调结果做完发现根本不知道自己在为什么而做。我当初规划这个系统时先花了一个晚上把需求拆成了四个维度可靠性通知本身就是容错手段如果通知这个环节自己都不稳定那整个系统等于白搭。短信必须能发出去发完还要能确认状态。及时性从事件发生到手机收到短信中间不应该有明显延迟。对告警类场景来说1分钟和10分钟是两种完全不同的体验。成本可控短信不是免费的每条都有成本。不能因为写了个死循环就一夜之间把预算烧光。结果可追踪发出的短信是“已送达”还是“失败”不能靠猜。需要有办法拿到发送状态失败时能重试。有了这四个维度后面所有技术选型都围绕它们展开。你会发现很多看似复杂的设计本质上都是在为这四个目标服务。1.2 方案选型Twilio、云厂商短信、自建网关怎么选先说结论我选了Twilio因为它把“发送短信”这件事抽象得极其干净API 设计几乎是所有通信服务里的教科书级别。做个对比你就明白了方案优点缺点适合场景自建短信网关买硬件或对接运营商完全可控大批量时单价低前期成本高对接运营商周期长维护复杂短信公司、日均百万级吞吐平台国内云厂商短信如阿里云、腾讯云国内发送速度快签名审核合规需要备案、签名审核国际号码支持较弱国内业务、用户营销触达TwilioAPI 干净、文档全、状态回调完善、支持全球号码价格折合人民币不便宜非中文文档开发者个人项目、海外业务、告警通知代比较下来“自建网关”对个人项目来说就是灾难你花在硬件和运营商对接上的时间足够把整个系统写完十遍国内云厂商适合国产业务但它的签名审核和模板审核流程有时候会让人崩溃Twilio 则恰好击中“开发者友好”这个点——注册就能拿到一个测试号跑通最小闭环只需要十几分钟。当然选择Twilio也意味着要接受它的缺点单价偏高、控制台是全英文。但对一个“把短信当告警工具”而不是“营销工具”用的场景来说这些缺点完全可接受。1.3 系统由哪些模块组成短信通知系统听着像一个单独的服务实际上拆开来看是几个独立模块的组合发送模块封装 Twilio SDK对外提供 send 接口。模板模块维护一批可参数化的消息模板避免在代码里写死消息内容。事件源模块可以是定时任务、监控脚本、业务回调等触发通知产生。状态回调模块接收 Twilio 的送达状态上报更新记录。日志与重试模块记录每次发送的请求ID、状态失败时自动补发。这五个模块各司其职。我在第一版里没有写状态回调模块结果短信发出去 10 分钟没送到都不知道后来吃了亏才补上的。后面我会逐一讲清楚每个模块怎么实现。2. 环境准备从 Python 环境到 Twilio 账号开通2.1 Python 环境准备与虚拟环境隔离Twilio 官方提供一个 Python SDK需要 Python 3.8 以上版本。这里有个小坑很多人手里的 Python 是系统自带的 2.7或者版本老旧直接用 pip 装依赖会报一堆看不懂的错。我的建议是先确认版本再动手python3 --version如果输出的是 3.8 及以上恭喜你直接进入下一步。如果你的机器还没有装 Python或者版本太低去 Python 官网下载对应系统的新版本装上就行。装完记得确认pip3可用。然后创建一个虚拟环境。这一步非常重要不是可选项。虚拟环境就是给项目一个独立的小房间不会跟你全局环境里的包互相打架。我踩过最惨的坑某次直接在全局环境里装了一堆依赖后来把系统自带的包给覆盖了莫名其妙的Bug持续了一个多星期才排查出来。mkdir sms-notifier cd sms-notifier python3 -m venv venv source venv/bin/activate # Windows 系统用 venv\Scripts\activate激活之后命令行前面会出现(venv)前缀这时候你再装任何包都是装进这个小房间里随时可以推倒重建干净又卫生。2.2 Twilio 账号注册与凭证获取打开 Twilio 官网注册一个账号。注册过程会要求验证邮箱和手机号这些按流程走即可。注册完成后你进入控制台第一件事不是急着看代码而是找两个关键信息Account SID相当于你的账号ID公开也无妨。Auth Token相当于你的账号密码绝对不能泄露。这两个值在控制台首页就能看到沿着 Dashboard 页面往下找就行。注意 Auth Token 旁边有个“眼睛”图标点击可以显示/隐藏你要复制完整的一串。接下来还需要一个可发送短信的号码。Twilio 在一次免费试用期里会给你一个测试号码这个号码可以往你注册时验证过的手机号发短信做开发测试完全够用。生产的实时使用通常需要购买一个正式号码Twilio 控制台里号码资源也购买选一个便宜的美国或英国号码就行。拿到号码之后建议立刻把所有敏感信息用环境变量管理起来绝对不要把 Auth Token 写在代码里。我自己见过有人在公共Git仓库里提交了 token几分钟内就被爬虫扫到账号直接被拿去刷短信账单瞬间爆掉。在项目目录创建.env文件TWILIO_ACCOUNT_SID你的AccountSID TWILIO_AUTH_TOKEN你的AuthToken TWILIO_PHONE_NUMBER你的Twilio号码 ALERT_PHONE_NUMBER你的接收手机号然后安装 python-dotenv让 Python 在启动时自动加载这个文件pip install python-dotenv2.3 安装 Twilio SDK 并验证账号连通性环境变量准备好之后安装 SDKpip install twilio然后写一个最简测试验证账号通不通import os from dotenv import load_dotenv from twilio.rest import Client load_dotenv() client Client( os.environ[TWILIO_ACCOUNT_SID], os.environ[TWILIO_AUTH_TOKEN] ) message client.messages.create( body你好这是第一条测试短信。, from_os.environ[TWILIO_PHONE_NUMBER], toos.environ[ALERT_PHONE_NUMBER], ) print(message.sid)运行它。如果手机上收到短信说明账号、凭证、号码都正常可以继续往下写了。如果你用的还是试用号短信内容后面会附带一行“Sent from your Twilio trial account”之类的推广文字这是正常现象换成正式号码后就会消失。3. 核心代码实现从第一条短信到完整的通知模块3.1 先跑通最小闭环理解发送逻辑上面那段测试代码看起来没什么技术含量但它其实是整个系统的最小闭环。这里面有一个关键概念值得先说清楚Twilio 的messages.create不是把短信直接丢给手机运营商就完了它会返回一个MessageSid——这个 SID 是这条短信在当前账号下的唯一标识。任何一次短信发送不管成功失败系统都会生成一个 SID。这个 SID 至关重要因为后面查状态、查账单、排查故障全都要靠它。第一版代码我犯过一个大意光发了短信就完事没把 SID 存下来。后来有用户说没收到短信我连是哪条、有没有发成功都查不到只能干瞪眼。所以从第一行代码开始就把message.sid打印出来、记到日志里这是一个会让你后面省很多事的习惯。另外一个容易忽略的参数是to的格式。Twilio 要求传完整国际格式比如中国的手机号要写成8613800138000这种以国家区号开头的格式本地格式直接丢进去会报错。3.2 把代码封装成可复用的通知服务类最小闭环跑通之后最自然的思考就是把这段发短信的逻辑封装成一个服务类。为什么要封装因为你后面会有很多地方调用定时任务要发日报、监控脚本要发告警、业务代码要发验证码。如果每个地方都去写一遍client.messages.create(...)将来要改发送渠道或者加日志的时候你得一个文件一个文件地改追悔莫及。有意义的一次封装大概是这样的import os import logging from dotenv import load_dotenv from twilio.rest import Client logger logging.getLogger(__name__) class SmsNotifier: def __init__(self): load_dotenv() self.client Client( os.environ[TWILIO_ACCOUNT_SID], os.environ[TWILIO_AUTH_TOKEN] ) self.from_number os.environ[TWILIO_PHONE_NUMBER] def send(self, to_number: str, body: str) - str: try: message self.client.messages.create( bodybody, from_self.from_number, toto_number, ) logger.info(短信已发送 to%s sid%s, to_number, message.sid) return message.sid except Exception as e: logger.error(短信发送失败 to%s error%s, to_number, e) raise这个类的关键设计有两点。第一初始化时把所有配置读取集中在构造函数里以后要换环境只改.env不用动业务代码第二对外只暴露一个send(to_number, body)方法内部把 SDK 的创建、发送、异常处理全部藏起来调用方根本感受不到 Twilio 的复杂性。我还习惯性地加了日志。别小看这一行logger.info它能让整个系统在运行过程中留下痕迹排查问题的时候你会发现它是最大的救星。3.3 管理多套通知模板参数化让消息更有用等到系统要发的消息种类多起来你会发现直接在业务代码里写死消息文本是个灾难。比如订单通知的文案、告警的文案、日报的文案全是硬编码改一个字都要翻遍代码库。这时候模板管理就该登场了。最简单有效的方式是用字符串模板加上参数填充Python 的str.format或者 f-string 就能搞定class MessageTemplates: ORDER_NOTIFY 您的订单 {order_id} 状态已更新为{status}预计送达时间{eta}。 ALERT_CPU [告警] 服务器 {host} CPU 使用率已达 {usage}%请及时检查。 DAILY_REPORT 【日报】今日订单数{order_count}销售额{revenue} 元异常订单{abnormal_count} 条。发送的时候只需要做填充notifier SmsNotifier() body MessageTemplates.ALERT_CPU.format(hostweb-01, usage95%) notifier.send(os.environ[ALERT_PHONE_NUMBER], body)这样做的好处是显而易见的文案集中管理、统一维护每个模板的参数列表一目了然发送时只关心数据组装不用关心文案格式。如果你的项目里文案特别多还可以进一步把模板搬到配置文件里用 YAML 或 JSON 维护业务代码和文案彻底解耦。对个人项目来说用类常量已经足够不用过度设计。3.4 发送状态追踪加一个回调接口掌握每一个消息的最终去向只发短信不追踪状态就像寄了快递不查物流全靠猜。Twilio 提供了非常完整的消息状态回调机制它会往你指定的 URL 发一个 HTTP 请求告诉你这条短信的最新状态比如sent已发送到运营商、delivered手机已收到、failed失败、undelivered无法送达。回调接口的实现不复杂用 Flask 写一个小服务就行from flask import Flask, request app Flask(__name__) app.route(/sms/status, methods[POST]) def sms_status(): data request.form message_sid data.get(MessageSid) status data.get(MessageStatus) error_code data.get(ErrorCode) # 这里可以把状态写入数据库或日志系统 print(f消息 {message_sid} 状态更新为 {status}错误码 {error_code}) return OK, 200在 Twilio 控制台里配置这个回调地址路径选择消息状态回调填上你部署好的公网地址即可。如果你暂时没有公网服务器可以先用知名开发工具在本地临时暴露一个公网地址来测试Twilio 官方文档里也有类似推荐。等系统正式上线再把这个接口部署到一台有公网 IP 的机器上。有了状态回调你能做到的事情就很实用了短信失败时自动写日志连续几次失败触发更高优先级的告警后台统计送达率。这类数据积累起来之后你的通知系统才算真正有了“监控自己的眼睛”。4. 进阶实操定时任务、事件触发与异常自愈4.1 定时发送用 APScheduler 实现日报推送短信通知系统最常见的进阶需求就是定时推送。比如每天上午九点把昨天的运行报告发到手机上或者每个整点检查一次服务器状态。定时任务方案我建议用 APScheduler它比time.sleep靠谱得多支持 cron 表达式还能长期驻留运行。安装pip install apscheduler然后写一个最简单的每日定时任务from apscheduler.schedulers.blocking import BlockingScheduler from datetime import datetime from notifier import SmsNotifier from templates import MessageTemplates def send_daily_report(): order_count 128 revenue 34567.8 abnormal_count 2 body MessageTemplates.DAILY_REPORT.format( order_countorder_count, revenuerevenue, abnormal_countabnormal_count ) SmsNotifier().send(os.environ[ALERT_PHONE_NUMBER], body) if __name__ __main__: scheduler BlockingScheduler() scheduler.add_job( send_daily_report, triggercron, hour9, minute0 ) print(定时任务已启动等待触发...) scheduler.start()这里要提醒一个新手容易忽略的点定时任务执行的函数如果抛出异常APScheduler 默认会吞掉并继续等下一次触发不会让你感知到。这就是为什么任务函数里必须要有 try/except 和日志。另一个务实的小技巧是测试定时任务时别干等时间点把trigger临时改成interval、间隔设成几分钟先验证函数本身跑得通再切回 cron 表达式能节省不少时间。如果你本身在做量化交易策略或者其他的数据采集任务思路完全一样策略计算完成之后把关键指标拼成短信发出来。我见过有人把 Twilio 接在聚宽、米筐这类量化平台的定时运行任务后面每天收盘自动把持仓收益和风险指标发到手机上体验远比打开电脑看报告来得直接。4.2 事件驱动写一个轻量监控脚本变化发生时立即通知定时任务适合“按固定时间点通知”但有些场景需要“变化发生时立刻通知”。比如商品价格跌破心理价位、网站服务响应变慢、服务器磁盘空间低于阈值这些事件你不知道具体什么时候会发生只能靠轮询去“看着它”。事件驱动型的经典实现是一个 while 循环加轮询下面这个例子展示如何监控一个商品价格并在跌破阈值时发短信import time import os import requests from notifier import SmsNotifier PRICE_URL https://api.example.com/price THRESHOLD 200.0 CHECK_INTERVAL 300 # 每5分钟查一次 def fetch_current_price(): resp requests.get(PRICE_URL, timeout10) resp.raise_for_status() return float(resp.json()[price]) def main(): notifier SmsNotifier() last_notified False # 防抖标志位价格一直在低位时只通知一次 while True: try: price fetch_current_price() if price THRESHOLD and not last_notified: notifier.send( os.environ[ALERT_PHONE_NUMBER], f[提醒] 当前价格为 {price} 元已跌破目标价 {THRESHOLD} 元。 ) last_notified True elif price THRESHOLD: last_notified False except Exception as e: print(f轮询失败: {e}) time.sleep(CHECK_INTERVAL) if __name__ __main__: main()这段代码里有三个细节值得展开。第一是请求超时时间timeout10必不可少否则网络异常时脚本会卡死在请求上第二是last_notified防抖标志很多人在这个位置选择“每次低于阈值都发”结果价格在临界点附近震荡时手机被短信轰炸到怀疑人生第三是整个循环里包了一层 try/except轮询脚本长期运行网络波动和接口故障都是常态不能让一个异常干掉整个循环。从这里也就自然引出事件源的话题事件源不只限于轮询。你的业务系统里有很多现成事件——用户下单、支付回调、定时任务失败、Webhook 接收等——在事件处理函数里调用notifier.send()就是最朴素的事件驱动通知。我自己在运行爬虫任务时最常用的做法是把抓取异常直接接进这个通知系统爬虫一挂短信立刻就到再也不用定时查看日志。4.3 异常处理与自愈短信发送失败时自动重试短信发送不是百分之百成功的号码格式错误、运营商网关抖动、账号余额不足都可能让发送失败。如果你的通知系统没有重试机制一次失败可能就意味着一条重要告警被彻底吞掉。我推荐用 tenacity 这个 Python 库来管理重试逻辑它可以把重试策略写得很优雅pip install tenacity改造后的发送方法import os import logging from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type logger logging.getLogger(__name__) class SmsNotifier: # ... 初始化部分省略 ... retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max30), retryretry_if_exception_type(Exception), reraiseTrue ) def send(self, to_number: str, body: str) - str: message self.client.messages.create( bodybody, from_self.from_number, toto_number, ) logger.info(短信发送成功 sid%s, message.sid) return message.sid这里有三个重试参数的含义要说明一下stop_after_attempt(3)最多尝试 3 次防止无限重试把你账号的余额耗尽wait_exponential是指数退避策略第一次失败后等 2 秒第二次失败后等 4 秒给 Twilio 和运营商一点恢复时间reraiseTrue表示 3 次尝试全部失败之后把最后一次异常继续抛出让上层调用方知道发送彻底失败了。关于重试我特别想强调一点千万不要对“失败原因永久性”的异常做重试。比如号码格式错误、手机号不存在这类错误你再重试一百次结果都一样只会白白浪费短信配额和 API 配额。我通常会在代码里先做一个号码格式的基础校验明显不合法的号码直接拒绝发送只有网络异常、超时这类临时性错误才允许重试。把“可重试”和“不可重试”区分清楚是通知系统健壮性的分水岭。如果重试还是失败最后的兜底方案是落盘。你至少要把失败的短信内容写进一个本地队列文件等故障恢复后再补发。当然补发时要注意按时间先后排序别把过时的告警和新告警混在一起。我的经验是告警类消息超过 30 分钟就不再补发了过时的告警反而会干扰判断直接把消息存到日志里留作事后分析就好。5. 常见问题与排查技巧实录5.1 常见错误码和排查思路用 Twilio 的过程中你迟早会遇到各种报错。我把常见的错误情况整理成了一张速查表做成了表格方便遇到问题快速对照错误现象常见错误/返回结果原因解决办法发送时抛出 21211 错误“Invalid To Phone Number”接收号码格式不对改写成 86 开头的国际格式发送时抛出 21610 错误“Unable to create record”错误余额不足或账号受限控制台查看账单充值或检查套餐收到短信但内容带了推广后缀消息末尾多一段文字正在使用试用号购买正式号码后自动消失发送成功但手机收不到状态回调显示 failed运营商通道问题或号码被屏蔽联系 Twilio 支持确认号码发送能力回调接口偶尔报 404无法找到接口地址路由路径配错检查控制台配置的回调 URL 与实际路由一致本地通知多次重复到达防抖失效判断标志位逻辑没写对确认状态去重和防抖代码在整个循环中生效这里最建议做的防御性工作是发送前普通格式校验import re def validate_phone_number(number: str) - bool: # 简单校验国际格式 开头后跟国家代码和号码 return bool(re.match(r^\\d{8,15}$, number))校验不过就早一点返回错误信息既节省 Twilio API 调用也避免把垃圾数据发给外部系统。5.2 调试技巧避免测试时把手机刷爆整个搭建过程中我犯过的最傻的一个错在一个循环测试里忘记给发送加延迟结果一分钟内给手机号发了二十多条“测试”短信手机疯狂震动最后还被运营商风控给临时禁了短信接收。之后我给自己定了一条铁律所有调试场景一律用 Mock 替代真实发送。写一个假的 Notifier 非常简单class FakeNotifier: def send(self, to_number: str, body: str) - str: print(f[FAKE] to{to_number} body{body}) return FAKE_SID_ body[:10]业务逻辑依赖的是send方法传入FakeNotifier和真实 Notifier 能跑同样的流程。这样你在开发阶段可以放心地写循环、测定时任务、测异常重试一条短信都不会真发出去。等所有逻辑都验证过了再把 Fake 替换成真实实例做一次端到端冒烟测试——只发一条。另外我习惯把日志级别调成 DEBUG每次发送都输出to、body前缀、sid、耗时这些信息。肉眼扫日志就能看出发送链路是否正常比断点调试更直观尤其适合调试那种“跑一会才出问题”的长驻循环任务。5.3 合规与频控别让通知变成骚扰短信通知系统本身是工具但使用不当就会成为骚扰工具。我见过有人给同一个用户重复推送十几条同质化广告也见过告警系统在半夜因为同一个错误连续轰炸值班员手机。这个领域虽然不像营销短信那样有严格的监管要求但作为开发者还是要守住两条底线第一凡是面向用户的短信通知必须给用户退订能力。至少要支持回复指定关键字退订或者提供后续不再接收通知的选择。这不是为了过审而是基本的尊重。第二合理设置频率上限。Twilio 的 Messaging Service 本身就支持设置默认速率限制你可以控制在每秒或每分钟最多发送多少条。即使用户场景确实是高频告警也应该做分组和降噪——同一个告警 5 分钟内重复出现时只发一次并合并累计次数。上一节提到的last_notified防抖逻辑也是这个思路。在接收端也要做好准备。如果你是给自己做告警就更是如此重要通知和生产消息分开用不同通知渠道。通常短信是即时性最强的方式但它的缺点是只能承载极短文本。如果一条告警短信没有写清楚是哪个服务、什么时间、影响范围接收人反而要花更长时间去查上下文。所以模板设计时要保证核心信息必须在前 30 个字内明确表达出来。回看我搭建这套系统的过程真正难的部分其实不是 Twilio SDK 的调通而是想清楚“什么情况下该发这条短信”“怎么确认它真正送达到位”“发不出去时自己怎么办”。这几点想透了代码量其实很少。我最后还有一个使用习惯想分享正式投入生产之后我会每隔一段时间主动给自己发一条测试短信确认号码没有被运营商或 Twilio 风控限制。这种最不起眼的定期自检反而避免过最尴尬的一次事故——系统一直报正常其实短信通道早已静默失效。希望这套系统能帮你省下几个被折腾的凌晨。
返回列表