尧图网站建设 
先说一个特别实际的场景你负责维护的系统里接了短信验证码功能一到联调测试就头大——测试手机在同事手里验证码有效期只有几分钟日志里查验证码又得去翻数据库。时间一长你就想能不能自己搭一个验证码接收平台把所有短信统一收进来、自动解析出验证码、在网页面板上直接看再留一个API给自动化测试脚本随时取用。这篇文章就是讲怎么从零搭出来而且会附上核心源代码照着敲一遍就能跑。这个平台的核心思路不是去“模拟收短信”而是走短信服务商的正规回调能力。如果你的验证码短信是某云服务商、某短信网关发出的服务商一般都提供“短信状态回调”或“上行回复回调”会把用户收到的短信内容实时POST到你指定的接口。平台要做的就是把各个服务商五花八门的回调格式统一收下来、解析出验证码、存进数据库再通过WebSocket实时推到前端页面展示。整个过程既可以用在公司内部联调环境也可以用于自动化测试、告警通知测试等场景。先说清楚边界这套东西只建议用在你自己有权的号码、正常的开发测试和企业应用场景里。把它用去接收不属于自己的短信、薅羊毛、绕过实名验证那是另一回事不在本文讨论范围内出了问题后果自负。1. 这个平台到底解决什么问题1.1 一个典型的联调痛点我见过太多团队在验证码上浪费时间。最常见的流程是这样的后端工程师说“短信发出去了”测试人员得拿出手机等短信等半天没收到一查发现是短信模板里变量名填错了有时候验证码到了但是被手机系统自动归类到通知栏没注意就过期了还有自动化测试脚本跑起来卡在“请输入验证码”这一步没人手动读短信就没法继续。这些问题表面上是“人肉处理验证码”太慢本质上是缺少一个统一接收、统一解析、统一查询的中间层。你需要的不是一台手机而是一个可以编程访问的“短信收件箱”。1.2 平台的目标与适用人群我的目标很明确做一个部署简单、能接收多个短信服务商回调、能自动提取验证码、能网页展示、能通过API查询的小型Web服务。它不需要很复杂但必须满足以下几个关键点短信到达后能实时看到最好是页面自动刷新不用手动点刷新按钮。验证码自动提取不需要复制整条短信再肉眼找数字。支持按手机号、平台、状态筛选方便大量测试号码时定位。提供简单HTTP接口能让Python/Shell等自动化脚本直接取验证码。适合谁我觉得主要是三类人一是后端开发联调时需要一个稳定的验证码查看工具二是测试工程师特别是写自动化用例的需要程序化获取验证码三是运维或者在短信平台做接入开发的需要一个调试工具来验证回调链路。对于完全没接触过Web开发的新手这篇文章也尽量把每一步写清楚按着做就行。2. 方案选型与整体架构2.1 为什么采用回调接收而不是自己造短信网关你可能想问能不能直接买一台短信猫/GSM模块插在服务器上自己接收运营商短信当然可以但是成本高、维护麻烦而且你需要实体SIM卡收一条就要扣一条短信费。对绝大多数开发者来说让短信服务商把短信内容回调给你的接口是最省事、成本最低的方案。现在主流的云短信服务商基本都支持几种回调短信状态报告回调推送下发状态、上行短信回调用户回复内容推送、还有部分服务商支持自定义推送。我们这次用“回调”来接收验证码前提是验证码由你自己通过该服务商发出然后服务商把完整的短信内容推送回来。如果你只是想在本地电脑上接一个真实手机号收验证码那属于另一个硬件方案这篇文章不涉及。2.2 技术栈选型为什么是FastAPI SQLite WebSocket技术选型上我故意选了非常朴素的一套Python FastAPI做后端SQLite做存储WebSocket做实时推送前端用原生HTML JavaScript不引入任何重框架。为什么这么做第一FastAPI写接口极其高效几行代码就能出一个带自动文档的API而且天然支持WebSocket和异步请求非常适合这个场景。第二SQLite是零配置文件型数据库单文件存储对于验证码这种轻量级数据几百上千条都没压力不需要单独装数据库服务。第三原生JS前端虽然简陋但部署时不用build、不用node_modules整个项目复制到任何机器都能跑对新人极度友好。如果以后数据量真的大了可以把SQLite换成PostgreSQL把单节点改成多实例但这是后话。第一步先把链路跑通。2.3 核心模块与工作流程整个平台的工作流程可以用下面这个思路理解短信服务商收到你发出的验证码短信后根据你在服务商后台配置的回调URL向平台的/api/callback/{platform}接口发起POST请求。FastAPI接收请求解析出手机号、短信内容并按平台适配字段名。解析模块从短信内容中用正则提取验证码。消息写入SQLite数据库同时通过WebSocket把新消息推送到所有在线前端页面。前端页面收到推送实时展示测试脚本也可以通过/api/messages接口拉取验证码。这个架构非常直白每个模块负责一件事出了问题也容易排查。3. 数据库与后端核心代码实现3.1 项目结构规划代码我做了简化但完整跑起来没问题。我的项目目录是这样sms-receiver/ ├── main.py # FastAPI入口路由、WebSocket ├── models.py # SQLAlchemy数据模型 ├── config.py # 配置项 ├── parser.py # 验证码解析模块 ├── requirements.txt # 依赖列表 └── static/ └── index.html # 前端页面先把依赖写进requirements.txtfastapi0.104.0 uvicorn0.24.0 sqlalchemy2.0.0安装就一行命令pip install -r requirements.txt3.2 数据模型设计数据模型是整个平台的地基。我设计了SmsMessage一张表字段不多但完全够用# models.py from datetime import datetime from sqlalchemy import create_engine, Column, Integer, String, DateTime from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker DATABASE_URL sqlite:///sms.db engine create_engine(DATABASE_URL, connect_args{check_same_thread: False}) SessionLocal sessionmaker(bindengine) Base declarative_base() class SmsMessage(Base): __tablename__ sms_messages id Column(Integer, primary_keyTrue, autoincrementTrue) platform Column(String(50), nullableFalse, indexTrue) # 来源短信服务商 phone Column(String(20), nullableFalse, indexTrue) # 接收号码 content Column(String(500), nullableFalse) # 短信原文 code Column(String(20), nullableTrue) # 解析出来的验证码 status Column(String(20), defaultunread) # unread / read created_at Column(DateTime, defaultdatetime.now) # 接收时间 Base.metadata.create_all(bindengine)这里有个细节phone字段我允许为空字符串因为某些服务商回调里可能不带上行号码只有目标号码这种时候别让程序报错存个空字符串保证短信内容不丢。code字段是可空的因为不是所有短信都有验证码比如通知类短信。status字段标记是否已读方便测试脚本“取一次就标记已读”避免重复使用同一个验证码。3.3 回调接收接口兼容多个服务商回调接口是整个平台的门面也是最容易出现兼容问题的地方。各家短信服务商的回调JSON格式都不一样有的字段叫mobile有的叫phone有的把短信内容放在content有的放在text。所以我的做法是在接口层做一层“字段映射”把能想到的可能字段都兼容一遍。# main.py 片段 from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect from fastapi.staticfiles import StaticFiles from models import SessionLocal, SmsMessage from parser import extract_code app FastAPI(title短信验证码接收平台) app.mount(/static, StaticFiles(directorystatic), namestatic) app.post(/api/callback/{platform}) async def sms_callback(platform: str, request: Request): data await request.json() # 兼容不同服务商的字段命名 phone ( data.get(phone) or data.get(mobile) or data.get(dest_id) or data.get(dest_code) or data.get(mobiles) or ) content ( data.get(content) or data.get(text) or data.get(msg) or data.get(message) or data.get(smsContent) or ) # 解析验证码 code extract_code(content) # 入库 db SessionLocal() try: msg SmsMessage( platformplatform, phonestr(phone), contentstr(content), codecode, ) db.add(msg) db.commit() db.refresh(msg) finally: db.close() # 推送给WebSocket在线页面 await manager.broadcast({ id: msg.id, platform: msg.platform, phone: msg.phone, content: msg.content, code: msg.code, status: msg.status, created_at: msg.created_at.strftime(%Y-%m-%d %H:%M:%S), }) return {code: 0, message: ok}有几点值得展开说一下or链式取值的写法看起来很随意但在兼容场景里非常实用。它会把第一个不为空的字段值取出来如果服务商A回调用mobile服务商B用phone那这行代码都能兼容。不过要注意如果某个字段的值是空字符串or会继续往后找所以空字符串不会被错误当作有效值。入库之后立刻通过WebSocket广播这样前端几乎是实时更新。有人可能问先广播再提交还是先提交再广播我选择先提交再广播因为广播时我用了msg.id和msg.created_at这两个字段只有提交后数据库才会生成顺序反了会出现id为None的情况。3.4 验证码解析逻辑验证码解析看起来简单实际上很考验细节。直接打印\d{6}可能会把短信里的其他数字也匹配出来比如“您本次订单金额100元验证码123456”这种情况就很尴尬。我的解析策略分优先级# parser.py import re def extract_code(content: str) - str | None: if not content: return None # 优先级1验证码提示语附近的数字 patterns [ r(?:验证码|校验码|动态码|短信码|安全码)[^\d]{0,10}?(\d{4,8}), r(?:为|是)[^\d]{0,5}?(\d{4,8}), ] for p in patterns: m re.search(p, content) if m: return m.group(1) # 优先级2短信正文里孤立出现的4-8位数字 m re.search(r(?!\d)(\d{4,8})(?!\d), content) if m: return m.group(1) return None第一个正则(?:验证码|校验码|动态码|短信码|安全码)[^\d]{0,10}?(\d{4,8})含义是先出现“验证码”这类关键词然后中间最多有10个非数字字符再捕获4到8位数字。这样做能匹配“验证码是123456”“您的验证码为1234565分钟内有效”这类常见模板。第二个正则(?:为|是)[^\d]{0,5}?(\d{4,8})处理的是没有“验证码”关键词的情况比如“本次验证码为123456”。最后的兜底才是匹配独立数字。这里用了(?!\d)和(?!\d)属于负向断言意思是数字前后不能紧挨着数字。这样“订单号1234567890”这种连续长数字就不会被误截成短验证码。为什么只匹配4到8位因为绝大多数平台的验证码是4到6位少数有8位动态密码超过8位的数字更可能是订单号、金额或其他业务编号就不硬猜了。实际使用中你会慢慢发现自己平台短信模板的特点到时候在patterns列表里加自己的规则就行。3.5 WebSocket实时推送后端推送我用了FastAPI自带的WebSocket支持关键代码是连接管理和广播# main.py 片段 from typing import List from fastapi import WebSocket class ConnectionManager: def __init__(self): self.active_connections: List[WebSocket] [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): if websocket in self.active_connections: self.active_connections.remove(websocket) async def broadcast(self, message: dict): for connection in self.active_connections: try: await connection.send_json(message) except Exception: pass manager ConnectionManager() app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: # 保持连接客户端可以定期发送ping await websocket.receive_text() except WebSocketDisconnect: manager.disconnect(websocket)有一个容易被新手忽略的点while True里的receive_text()是必须的。WebSocket连接建立后如果不做任何接收操作某些代理服务器或浏览器会自动判定连接空闲导致连接被断开。所以我在循环里等待客户端发来的任何文本消息前端可以每秒发一个ping一旦客户端主动断开就会抛出WebSocketDisconnect异常这时再清理连接即可。广播时的try...except Exception也是经验之谈。某个浏览器标签页如果已经关掉但连接还没有从列表里移除send_json可能会抛异常。我选择在广播时把异常吞掉避免因为一个失效连接导致整个推送流程崩掉。3.6 消息查询API除了实时推送自动化脚本还需要一个HTTP接口来拉取验证码。我的接口设计很简单# main.py 片段 from fastapi import Query app.get(/api/messages) async def get_messages( limit: int Query(20, ge1, le100), phone: str Query(, description按手机号筛选), code: str Query(, description按验证码筛选), status: str Query(, description按状态筛选 unread/read), ): db SessionLocal() try: query db.query(SmsMessage).order_by(SmsMessage.id.desc()) if phone: query query.filter(SmsMessage.phone phone) if code: query query.filter(SmsMessage.code code) if status: query query.filter(SmsMessage.status status) items query.limit(limit).all() return [ { id: item.id, platform: item.platform, phone: item.phone, content: item.content, code: item.code, status: item.status, created_at: item.created_at.strftime(%Y-%m-%d %H:%M:%S), } for item in items ] finally: db.close()这个接口对自动化脚本特别友好。比如测试脚本注册一个新手机号注册后循环调用这个接口传phone13800138000等上几秒就能拿到最新验证码然后动态填进登录表单里整个注册流程就全自动了。4. 前端展示页面实现4.1 页面交互怎么设计前端页面我用了一个static/index.html文件没有用任何构建工具。页面分成三个区域顶部是统计和清空按钮中间是筛选条件下面是消息列表。交互上最有价值的就是WebSocket实时推送——有新短信进来页面直接在最上面插入一条新记录同时把最新验证码用大号字体标出来。前端消息格式和后端广播的数据结构保持一致{ id: 12, platform: aliyun, phone: 13800138000, content: 您的验证码为1234565分钟内有效。, code: 123456, status: unread, created_at: 2025-01-15 10:30:00 }4.2 前端核心代码我摘取最关键的WebSocket部分和渲染逻辑!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title验证码接收平台/title style body { font-family: -apple-system, BlinkMacSystemFont, Segoe UI, sans-serif; max-width: 900px; margin: 20px auto; padding: 0 16px; background: #f6f8fa; } .item { background: #fff; border: 1px solid #e1e4e8; border-radius: 8px; padding: 12px 16px; margin-bottom: 10px; } .item .code { font-size: 28px; font-weight: bold; color: #0366d6; margin-right: 16px; } .item .meta { color: #6a737d; font-size: 13px; } .item .content { margin-top: 6px; word-break: break-all; } filter-section { margin-bottom: 16px; } input { padding: 6px 10px; margin-right: 8px; border: 1px solid #d0d7de; border-radius: 6px; } button { padding: 6px 16px; border: none; border-radius: 6px; background: #0366d6; color: #fff; cursor: pointer; margin-right: 8px; } /style /head body h2验证码接收平台/h2 div classfilter-section input idphoneFilter placeholder按手机号筛选 button onclickloadMessages()查询/button button onclickclearStorage()清空记录/button /div div idlist/div script const listEl document.getElementById(list); function renderMessage(item) { const div document.createElement(div); div.className item; div.innerHTML span styledisplay:flex;align-items:center; span classcode${item.code || 无}/span span classmeta${item.platform} | ${item.phone} | ${item.created_at}/span /span div classcontent${item.content}/div ; listEl.prepend(div); } function renderList(items) { listEl.innerHTML ; items.forEach(item renderMessage(item)); } // 加载历史消息 async function loadMessages() { const phone document.getElementById(phoneFilter).value.trim(); const url /api/messages?limit50 (phone ? phone${encodeURIComponent(phone)} : ); const resp await fetch(url); const data await resp.json(); renderList(data); } // WebSocket 实时推送 function connectWS() { const proto location.protocol https: ? wss:// : ws://; const ws new WebSocket(proto location.host /ws); ws.onmessage (event) { const item JSON.parse(event.data); renderMessage(item); }; ws.onclose () { setTimeout(connectWS, 3000); }; } // 清空记录这里只清空前端显示后端删除可按需加接口 async function clearStorage() { listEl.innerHTML ; } loadMessages(); connectWS(); /script /body /html主要功能就这么点。前端负责两件事打开页面时加载最近50条历史消息以及建立WebSocket连接等待新消息推送。onclose里做了3秒自动重连这样后端重启之后前端能自动恢复不用手动刷新页面。这里说明一下如果要在后端真正删除记录需要在后端加一个DELETE /api/messages/{id}或DELETE /api/messages接口前端再调用。我为了控制篇幅只演示了前端清空显示实际项目里按需补上删除接口就行实现逻辑和查询接口几乎一样。5. 从0到1跑通全流程5.1 环境准备和依赖安装实际操作时我建议先在本地电脑上跑通再考虑部署到服务器。环境要求Python 3.9我用的是3.11pip任意能访问本机的浏览器在项目目录下执行python -m venv venv source venv/bin/activate # Windows上是 venv\Scripts\activate pip install -r requirements.txt5.2 启动服务启动命令很简单uvicorn main:app --host 0.0.0.0 --port 8000启动之后浏览器访问http://localhost:8000/static/index.html就能看到页面。同时FastAPI自带的接口文档在http://localhost:8000/docs你可以在网页里直接测试/api/callback/{platform}接口。5.3 配置短信服务商回调这一步是联调的关键。到你的短信服务商控制台找到“回调配置”或“消息接收配置”填上你的回调地址。问题来了如果平台跑在本地电脑服务商的服务器没办法直接访问到你电脑的localhost这时候需要一个内网穿透工具。我一般用ngrok这类工具把本机8000端口映射成一个公网HTTPS地址然后把这个公网地址填到服务商后台。比如https://your-tunnel-domain.ngrok.io/api/callback/aliyun这里aliyun是路径参数代表平台来源。每个服务商填不同的路径就能在一个平台上区分消息来源。注意内网穿透工具只用于调试阶段上线部署到正式服务器后就不需要了。公网回调地址一定要用HTTPS否则部分服务商会拒绝调用。5.4 联调测试配置好之后从你的业务系统发起一条验证码短信。正常情况下几秒钟内页面就会弹出新消息顶部显示解析出的验证码。如果没显示按下面排查顺序走一遍服务商后台有没有“回调失败记录”看返回给服务商的HTTP状态码。后台会不会要求回调接口返回特定的成功标识比如{code:0}有些服务商还要求返回{code:0,msg:成功}字段对不上会一直重试。内网穿透工具的请求日志里有没有收到POST请求。排查完这几个点基本都能通。6. 常见问题与排查技巧6.1 回调请求进不来怎么办这是最常遇到的问题。回调进不来先别急着翻代码按这个优先级排查。先看内网穿透日志确认服务商的请求是否真的打到了你的机器。如果ngrok日志里一条请求都没有说明回调地址配置错了或者服务商后台还没审核通过回调配置。再看FastAPI的启动终端有没有类似POST /api/callback/aliyun 200的访问日志。如果有说明请求进来了如果没有说明请求被防火墙、安全组拦截了。最后才是看代码逻辑在回调函数第一行加个print(callback hit)确认函数有没有被执行。6.2 验证码解析不准确解析不准多半是短信模板和你写的正则对不上。比如有的短信写的是“您的安全码【123456】”有的写“验证码123456请勿泄露”。这时候我的建议是先打开数据库表sms_messages看content字段里存的短信原文对照原文调试正则表达式。也可以在线上直接测试正则Python里跑几行代码很快python -c from parser import extract_code; print(extract_code(您的验证码是1234565分钟内有效))要是不同模板差异太大就只能在patterns列表里多添加几条规则。注意正则的顺序很重要越具体的规则要放在越前面。6.3 WebSocket反复断开重连如果你看到页面前端日志里不断出现onclose和重新连接通常是网络中间设备影响了长连接。一种是代理或负载均衡层对空闲连接设置了超时比如60秒没有消息就断开解决办法是前端每秒发送一个ping文本让连接保持活跃。另一种是后端进程崩溃导致连接断开那就得去看后端日志确认是不是代码里有未捕获的异常。6.4 数据量变大之后怎么办验证码消息存到几万条以后SQLite依然能扛住但查询会逐渐变慢。最简单的优化办法是定期清理比如保留最近30天的数据。我在实际项目中写过一个小脚本用cron每天凌晨删除30天前的记录# clean.py from datetime import datetime, timedelta from models import SessionLocal, SmsMessage db SessionLocal() try: deadline datetime.now() - timedelta(days30) result db.query(SmsMessage).filter(SmsMessage.created_at deadline).delete() db.commit() print(fdeleted {result} old messages) finally: db.close()另外给created_at也建个索引删除和按时间查询会更快。不过对于演示项目这步可以先不做。7. 生产化加固与扩展思路7.1 回调签名验证真到了生产环境回调接口不能裸奔否则任何人都可以往你的平台伪造短信记录。服务商基本都支持回调签名或Token校验平台端也要做对应验证。以某服务商为例回调请求头里会带一个X-Signature生成规则是把timestamp secret做MD5或HMAC。你在后端接收到请求后先拿同样的密钥算一遍比对签名一致再处理。这一步不能省否则测试环境可能被人塞垃圾数据影响自动化脚本判断。7.2 存储层升级SQLite毕竟不支持高并发写入如果多个回调同时到达可能会出现database is locked的报错。生产环境建议把DATABASE_URL直接切换成PostgreSQLDATABASE_URL postgresql://user:passlocalhost:5432/sms_db代码里models.py的写法基本不用变SQLAlchemy SQLite和PostgreSQL之间可以平滑切换。唯一要注意的是PostgreSQL驱动需要额外安装pip install psycopg2-binary7.3 容器化部署我更喜欢用Docker来部署这个平台。写一个简单的DockerfileFROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]构建镜像并启动docker build -t sms-receiver . docker run -d -p 8000:8000 -v $(pwd):/app sms-receiver注意把sms.db数据库文件通过数据卷挂载出来不然容器一删除所有历史消息都没了。7.4 还能扩展什么功能这个平台的扩展空间其实很大。你可以给每个业务系统单独生成一个platform标识按项目维度统计短信量可以接入企业微信/钉钉机器人验证码到了之后直接推送到群聊可以增加一个“自动已读”接口测试脚本取完验证码后调用并标记已读下次查询默认跳过还可以对验证码做失败重试监控比如某个号码一分钟内重复收到验证码自动告警。我个人在实际操作中的体会是工具不在复杂在于顺手。这套源码我前后维护过几版从一开始只是给自己看验证码到后来被测试团队拿去接到自动化用例里整个过程没有加任何花哨功能核心就是“实时、准确、可查询”这六个字。你按着这篇文章搭完跑通第一单之后大概率也会有自己的想法那就顺着业务需求去加平台会越用越顺手。最后再分享一个小技巧调试回调接口时不要直接依赖真实短信服务商发短信费钱又慢。用服务商的“测试发送”功能或者用curl构造一个模拟回调请求又快又方便curl -X POST http://localhost:8000/api/callback/test \ -H Content-Type: application/json \ -d {phone: 13800138000, content: 您的验证码是888888请勿泄露。}能返回{code:0,message:ok}页面也收到新消息就说明整个链路没问题剩下的交给真实短信流入测试即可。 从零搭建短信验证码接收平台：回调解析与WebSocket实时推送
先说一个特别实际的场景你负责维护的系统里接了短信验证码功能一到联调测试就头大——测试手机在同事手里验证码有效期只有几分钟日志里查验证码又得去翻数据库。时间一长你就想能不能自己搭一个验证码接收平台把所有短信统一收进来、自动解析出验证码、在网页面板上直接看再留一个API给自动化测试脚本随时取用。这篇文章就是讲怎么从零搭出来而且会附上核心源代码照着敲一遍就能跑。这个平台的核心思路不是去“模拟收短信”而是走短信服务商的正规回调能力。如果你的验证码短信是某云服务商、某短信网关发出的服务商一般都提供“短信状态回调”或“上行回复回调”会把用户收到的短信内容实时POST到你指定的接口。平台要做的就是把各个服务商五花八门的回调格式统一收下来、解析出验证码、存进数据库再通过WebSocket实时推到前端页面展示。整个过程既可以用在公司内部联调环境也可以用于自动化测试、告警通知测试等场景。先说清楚边界这套东西只建议用在你自己有权的号码、正常的开发测试和企业应用场景里。把它用去接收不属于自己的短信、薅羊毛、绕过实名验证那是另一回事不在本文讨论范围内出了问题后果自负。1. 这个平台到底解决什么问题1.1 一个典型的联调痛点我见过太多团队在验证码上浪费时间。最常见的流程是这样的后端工程师说“短信发出去了”测试人员得拿出手机等短信等半天没收到一查发现是短信模板里变量名填错了有时候验证码到了但是被手机系统自动归类到通知栏没注意就过期了还有自动化测试脚本跑起来卡在“请输入验证码”这一步没人手动读短信就没法继续。这些问题表面上是“人肉处理验证码”太慢本质上是缺少一个统一接收、统一解析、统一查询的中间层。你需要的不是一台手机而是一个可以编程访问的“短信收件箱”。1.2 平台的目标与适用人群我的目标很明确做一个部署简单、能接收多个短信服务商回调、能自动提取验证码、能网页展示、能通过API查询的小型Web服务。它不需要很复杂但必须满足以下几个关键点短信到达后能实时看到最好是页面自动刷新不用手动点刷新按钮。验证码自动提取不需要复制整条短信再肉眼找数字。支持按手机号、平台、状态筛选方便大量测试号码时定位。提供简单HTTP接口能让Python/Shell等自动化脚本直接取验证码。适合谁我觉得主要是三类人一是后端开发联调时需要一个稳定的验证码查看工具二是测试工程师特别是写自动化用例的需要程序化获取验证码三是运维或者在短信平台做接入开发的需要一个调试工具来验证回调链路。对于完全没接触过Web开发的新手这篇文章也尽量把每一步写清楚按着做就行。2. 方案选型与整体架构2.1 为什么采用回调接收而不是自己造短信网关你可能想问能不能直接买一台短信猫/GSM模块插在服务器上自己接收运营商短信当然可以但是成本高、维护麻烦而且你需要实体SIM卡收一条就要扣一条短信费。对绝大多数开发者来说让短信服务商把短信内容回调给你的接口是最省事、成本最低的方案。现在主流的云短信服务商基本都支持几种回调短信状态报告回调推送下发状态、上行短信回调用户回复内容推送、还有部分服务商支持自定义推送。我们这次用“回调”来接收验证码前提是验证码由你自己通过该服务商发出然后服务商把完整的短信内容推送回来。如果你只是想在本地电脑上接一个真实手机号收验证码那属于另一个硬件方案这篇文章不涉及。2.2 技术栈选型为什么是FastAPI SQLite WebSocket技术选型上我故意选了非常朴素的一套Python FastAPI做后端SQLite做存储WebSocket做实时推送前端用原生HTML JavaScript不引入任何重框架。为什么这么做第一FastAPI写接口极其高效几行代码就能出一个带自动文档的API而且天然支持WebSocket和异步请求非常适合这个场景。第二SQLite是零配置文件型数据库单文件存储对于验证码这种轻量级数据几百上千条都没压力不需要单独装数据库服务。第三原生JS前端虽然简陋但部署时不用build、不用node_modules整个项目复制到任何机器都能跑对新人极度友好。如果以后数据量真的大了可以把SQLite换成PostgreSQL把单节点改成多实例但这是后话。第一步先把链路跑通。2.3 核心模块与工作流程整个平台的工作流程可以用下面这个思路理解短信服务商收到你发出的验证码短信后根据你在服务商后台配置的回调URL向平台的/api/callback/{platform}接口发起POST请求。FastAPI接收请求解析出手机号、短信内容并按平台适配字段名。解析模块从短信内容中用正则提取验证码。消息写入SQLite数据库同时通过WebSocket把新消息推送到所有在线前端页面。前端页面收到推送实时展示测试脚本也可以通过/api/messages接口拉取验证码。这个架构非常直白每个模块负责一件事出了问题也容易排查。3. 数据库与后端核心代码实现3.1 项目结构规划代码我做了简化但完整跑起来没问题。我的项目目录是这样sms-receiver/ ├── main.py # FastAPI入口路由、WebSocket ├── models.py # SQLAlchemy数据模型 ├── config.py # 配置项 ├── parser.py # 验证码解析模块 ├── requirements.txt # 依赖列表 └── static/ └── index.html # 前端页面先把依赖写进requirements.txtfastapi0.104.0 uvicorn0.24.0 sqlalchemy2.0.0安装就一行命令pip install -r requirements.txt3.2 数据模型设计数据模型是整个平台的地基。我设计了SmsMessage一张表字段不多但完全够用# models.py from datetime import datetime from sqlalchemy import create_engine, Column, Integer, String, DateTime from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker DATABASE_URL sqlite:///sms.db engine create_engine(DATABASE_URL, connect_args{check_same_thread: False}) SessionLocal sessionmaker(bindengine) Base declarative_base() class SmsMessage(Base): __tablename__ sms_messages id Column(Integer, primary_keyTrue, autoincrementTrue) platform Column(String(50), nullableFalse, indexTrue) # 来源短信服务商 phone Column(String(20), nullableFalse, indexTrue) # 接收号码 content Column(String(500), nullableFalse) # 短信原文 code Column(String(20), nullableTrue) # 解析出来的验证码 status Column(String(20), defaultunread) # unread / read created_at Column(DateTime, defaultdatetime.now) # 接收时间 Base.metadata.create_all(bindengine)这里有个细节phone字段我允许为空字符串因为某些服务商回调里可能不带上行号码只有目标号码这种时候别让程序报错存个空字符串保证短信内容不丢。code字段是可空的因为不是所有短信都有验证码比如通知类短信。status字段标记是否已读方便测试脚本“取一次就标记已读”避免重复使用同一个验证码。3.3 回调接收接口兼容多个服务商回调接口是整个平台的门面也是最容易出现兼容问题的地方。各家短信服务商的回调JSON格式都不一样有的字段叫mobile有的叫phone有的把短信内容放在content有的放在text。所以我的做法是在接口层做一层“字段映射”把能想到的可能字段都兼容一遍。# main.py 片段 from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect from fastapi.staticfiles import StaticFiles from models import SessionLocal, SmsMessage from parser import extract_code app FastAPI(title短信验证码接收平台) app.mount(/static, StaticFiles(directorystatic), namestatic) app.post(/api/callback/{platform}) async def sms_callback(platform: str, request: Request): data await request.json() # 兼容不同服务商的字段命名 phone ( data.get(phone) or data.get(mobile) or data.get(dest_id) or data.get(dest_code) or data.get(mobiles) or ) content ( data.get(content) or data.get(text) or data.get(msg) or data.get(message) or data.get(smsContent) or ) # 解析验证码 code extract_code(content) # 入库 db SessionLocal() try: msg SmsMessage( platformplatform, phonestr(phone), contentstr(content), codecode, ) db.add(msg) db.commit() db.refresh(msg) finally: db.close() # 推送给WebSocket在线页面 await manager.broadcast({ id: msg.id, platform: msg.platform, phone: msg.phone, content: msg.content, code: msg.code, status: msg.status, created_at: msg.created_at.strftime(%Y-%m-%d %H:%M:%S), }) return {code: 0, message: ok}有几点值得展开说一下or链式取值的写法看起来很随意但在兼容场景里非常实用。它会把第一个不为空的字段值取出来如果服务商A回调用mobile服务商B用phone那这行代码都能兼容。不过要注意如果某个字段的值是空字符串or会继续往后找所以空字符串不会被错误当作有效值。入库之后立刻通过WebSocket广播这样前端几乎是实时更新。有人可能问先广播再提交还是先提交再广播我选择先提交再广播因为广播时我用了msg.id和msg.created_at这两个字段只有提交后数据库才会生成顺序反了会出现id为None的情况。3.4 验证码解析逻辑验证码解析看起来简单实际上很考验细节。直接打印\d{6}可能会把短信里的其他数字也匹配出来比如“您本次订单金额100元验证码123456”这种情况就很尴尬。我的解析策略分优先级# parser.py import re def extract_code(content: str) - str | None: if not content: return None # 优先级1验证码提示语附近的数字 patterns [ r(?:验证码|校验码|动态码|短信码|安全码)[^\d]{0,10}?(\d{4,8}), r(?:为|是)[^\d]{0,5}?(\d{4,8}), ] for p in patterns: m re.search(p, content) if m: return m.group(1) # 优先级2短信正文里孤立出现的4-8位数字 m re.search(r(?!\d)(\d{4,8})(?!\d), content) if m: return m.group(1) return None第一个正则(?:验证码|校验码|动态码|短信码|安全码)[^\d]{0,10}?(\d{4,8})含义是先出现“验证码”这类关键词然后中间最多有10个非数字字符再捕获4到8位数字。这样做能匹配“验证码是123456”“您的验证码为1234565分钟内有效”这类常见模板。第二个正则(?:为|是)[^\d]{0,5}?(\d{4,8})处理的是没有“验证码”关键词的情况比如“本次验证码为123456”。最后的兜底才是匹配独立数字。这里用了(?!\d)和(?!\d)属于负向断言意思是数字前后不能紧挨着数字。这样“订单号1234567890”这种连续长数字就不会被误截成短验证码。为什么只匹配4到8位因为绝大多数平台的验证码是4到6位少数有8位动态密码超过8位的数字更可能是订单号、金额或其他业务编号就不硬猜了。实际使用中你会慢慢发现自己平台短信模板的特点到时候在patterns列表里加自己的规则就行。3.5 WebSocket实时推送后端推送我用了FastAPI自带的WebSocket支持关键代码是连接管理和广播# main.py 片段 from typing import List from fastapi import WebSocket class ConnectionManager: def __init__(self): self.active_connections: List[WebSocket] [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): if websocket in self.active_connections: self.active_connections.remove(websocket) async def broadcast(self, message: dict): for connection in self.active_connections: try: await connection.send_json(message) except Exception: pass manager ConnectionManager() app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: # 保持连接客户端可以定期发送ping await websocket.receive_text() except WebSocketDisconnect: manager.disconnect(websocket)有一个容易被新手忽略的点while True里的receive_text()是必须的。WebSocket连接建立后如果不做任何接收操作某些代理服务器或浏览器会自动判定连接空闲导致连接被断开。所以我在循环里等待客户端发来的任何文本消息前端可以每秒发一个ping一旦客户端主动断开就会抛出WebSocketDisconnect异常这时再清理连接即可。广播时的try...except Exception也是经验之谈。某个浏览器标签页如果已经关掉但连接还没有从列表里移除send_json可能会抛异常。我选择在广播时把异常吞掉避免因为一个失效连接导致整个推送流程崩掉。3.6 消息查询API除了实时推送自动化脚本还需要一个HTTP接口来拉取验证码。我的接口设计很简单# main.py 片段 from fastapi import Query app.get(/api/messages) async def get_messages( limit: int Query(20, ge1, le100), phone: str Query(, description按手机号筛选), code: str Query(, description按验证码筛选), status: str Query(, description按状态筛选 unread/read), ): db SessionLocal() try: query db.query(SmsMessage).order_by(SmsMessage.id.desc()) if phone: query query.filter(SmsMessage.phone phone) if code: query query.filter(SmsMessage.code code) if status: query query.filter(SmsMessage.status status) items query.limit(limit).all() return [ { id: item.id, platform: item.platform, phone: item.phone, content: item.content, code: item.code, status: item.status, created_at: item.created_at.strftime(%Y-%m-%d %H:%M:%S), } for item in items ] finally: db.close()这个接口对自动化脚本特别友好。比如测试脚本注册一个新手机号注册后循环调用这个接口传phone13800138000等上几秒就能拿到最新验证码然后动态填进登录表单里整个注册流程就全自动了。4. 前端展示页面实现4.1 页面交互怎么设计前端页面我用了一个static/index.html文件没有用任何构建工具。页面分成三个区域顶部是统计和清空按钮中间是筛选条件下面是消息列表。交互上最有价值的就是WebSocket实时推送——有新短信进来页面直接在最上面插入一条新记录同时把最新验证码用大号字体标出来。前端消息格式和后端广播的数据结构保持一致{ id: 12, platform: aliyun, phone: 13800138000, content: 您的验证码为1234565分钟内有效。, code: 123456, status: unread, created_at: 2025-01-15 10:30:00 }4.2 前端核心代码我摘取最关键的WebSocket部分和渲染逻辑!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title验证码接收平台/title style body { font-family: -apple-system, BlinkMacSystemFont, Segoe UI, sans-serif; max-width: 900px; margin: 20px auto; padding: 0 16px; background: #f6f8fa; } .item { background: #fff; border: 1px solid #e1e4e8; border-radius: 8px; padding: 12px 16px; margin-bottom: 10px; } .item .code { font-size: 28px; font-weight: bold; color: #0366d6; margin-right: 16px; } .item .meta { color: #6a737d; font-size: 13px; } .item .content { margin-top: 6px; word-break: break-all; } filter-section { margin-bottom: 16px; } input { padding: 6px 10px; margin-right: 8px; border: 1px solid #d0d7de; border-radius: 6px; } button { padding: 6px 16px; border: none; border-radius: 6px; background: #0366d6; color: #fff; cursor: pointer; margin-right: 8px; } /style /head body h2验证码接收平台/h2 div classfilter-section input idphoneFilter placeholder按手机号筛选 button onclickloadMessages()查询/button button onclickclearStorage()清空记录/button /div div idlist/div script const listEl document.getElementById(list); function renderMessage(item) { const div document.createElement(div); div.className item; div.innerHTML span styledisplay:flex;align-items:center; span classcode${item.code || 无}/span span classmeta${item.platform} | ${item.phone} | ${item.created_at}/span /span div classcontent${item.content}/div ; listEl.prepend(div); } function renderList(items) { listEl.innerHTML ; items.forEach(item renderMessage(item)); } // 加载历史消息 async function loadMessages() { const phone document.getElementById(phoneFilter).value.trim(); const url /api/messages?limit50 (phone ? phone${encodeURIComponent(phone)} : ); const resp await fetch(url); const data await resp.json(); renderList(data); } // WebSocket 实时推送 function connectWS() { const proto location.protocol https: ? wss:// : ws://; const ws new WebSocket(proto location.host /ws); ws.onmessage (event) { const item JSON.parse(event.data); renderMessage(item); }; ws.onclose () { setTimeout(connectWS, 3000); }; } // 清空记录这里只清空前端显示后端删除可按需加接口 async function clearStorage() { listEl.innerHTML ; } loadMessages(); connectWS(); /script /body /html主要功能就这么点。前端负责两件事打开页面时加载最近50条历史消息以及建立WebSocket连接等待新消息推送。onclose里做了3秒自动重连这样后端重启之后前端能自动恢复不用手动刷新页面。这里说明一下如果要在后端真正删除记录需要在后端加一个DELETE /api/messages/{id}或DELETE /api/messages接口前端再调用。我为了控制篇幅只演示了前端清空显示实际项目里按需补上删除接口就行实现逻辑和查询接口几乎一样。5. 从0到1跑通全流程5.1 环境准备和依赖安装实际操作时我建议先在本地电脑上跑通再考虑部署到服务器。环境要求Python 3.9我用的是3.11pip任意能访问本机的浏览器在项目目录下执行python -m venv venv source venv/bin/activate # Windows上是 venv\Scripts\activate pip install -r requirements.txt5.2 启动服务启动命令很简单uvicorn main:app --host 0.0.0.0 --port 8000启动之后浏览器访问http://localhost:8000/static/index.html就能看到页面。同时FastAPI自带的接口文档在http://localhost:8000/docs你可以在网页里直接测试/api/callback/{platform}接口。5.3 配置短信服务商回调这一步是联调的关键。到你的短信服务商控制台找到“回调配置”或“消息接收配置”填上你的回调地址。问题来了如果平台跑在本地电脑服务商的服务器没办法直接访问到你电脑的localhost这时候需要一个内网穿透工具。我一般用ngrok这类工具把本机8000端口映射成一个公网HTTPS地址然后把这个公网地址填到服务商后台。比如https://your-tunnel-domain.ngrok.io/api/callback/aliyun这里aliyun是路径参数代表平台来源。每个服务商填不同的路径就能在一个平台上区分消息来源。注意内网穿透工具只用于调试阶段上线部署到正式服务器后就不需要了。公网回调地址一定要用HTTPS否则部分服务商会拒绝调用。5.4 联调测试配置好之后从你的业务系统发起一条验证码短信。正常情况下几秒钟内页面就会弹出新消息顶部显示解析出的验证码。如果没显示按下面排查顺序走一遍服务商后台有没有“回调失败记录”看返回给服务商的HTTP状态码。后台会不会要求回调接口返回特定的成功标识比如{code:0}有些服务商还要求返回{code:0,msg:成功}字段对不上会一直重试。内网穿透工具的请求日志里有没有收到POST请求。排查完这几个点基本都能通。6. 常见问题与排查技巧6.1 回调请求进不来怎么办这是最常遇到的问题。回调进不来先别急着翻代码按这个优先级排查。先看内网穿透日志确认服务商的请求是否真的打到了你的机器。如果ngrok日志里一条请求都没有说明回调地址配置错了或者服务商后台还没审核通过回调配置。再看FastAPI的启动终端有没有类似POST /api/callback/aliyun 200的访问日志。如果有说明请求进来了如果没有说明请求被防火墙、安全组拦截了。最后才是看代码逻辑在回调函数第一行加个print(callback hit)确认函数有没有被执行。6.2 验证码解析不准确解析不准多半是短信模板和你写的正则对不上。比如有的短信写的是“您的安全码【123456】”有的写“验证码123456请勿泄露”。这时候我的建议是先打开数据库表sms_messages看content字段里存的短信原文对照原文调试正则表达式。也可以在线上直接测试正则Python里跑几行代码很快python -c from parser import extract_code; print(extract_code(您的验证码是1234565分钟内有效))要是不同模板差异太大就只能在patterns列表里多添加几条规则。注意正则的顺序很重要越具体的规则要放在越前面。6.3 WebSocket反复断开重连如果你看到页面前端日志里不断出现onclose和重新连接通常是网络中间设备影响了长连接。一种是代理或负载均衡层对空闲连接设置了超时比如60秒没有消息就断开解决办法是前端每秒发送一个ping文本让连接保持活跃。另一种是后端进程崩溃导致连接断开那就得去看后端日志确认是不是代码里有未捕获的异常。6.4 数据量变大之后怎么办验证码消息存到几万条以后SQLite依然能扛住但查询会逐渐变慢。最简单的优化办法是定期清理比如保留最近30天的数据。我在实际项目中写过一个小脚本用cron每天凌晨删除30天前的记录# clean.py from datetime import datetime, timedelta from models import SessionLocal, SmsMessage db SessionLocal() try: deadline datetime.now() - timedelta(days30) result db.query(SmsMessage).filter(SmsMessage.created_at deadline).delete() db.commit() print(fdeleted {result} old messages) finally: db.close()另外给created_at也建个索引删除和按时间查询会更快。不过对于演示项目这步可以先不做。7. 生产化加固与扩展思路7.1 回调签名验证真到了生产环境回调接口不能裸奔否则任何人都可以往你的平台伪造短信记录。服务商基本都支持回调签名或Token校验平台端也要做对应验证。以某服务商为例回调请求头里会带一个X-Signature生成规则是把timestamp secret做MD5或HMAC。你在后端接收到请求后先拿同样的密钥算一遍比对签名一致再处理。这一步不能省否则测试环境可能被人塞垃圾数据影响自动化脚本判断。7.2 存储层升级SQLite毕竟不支持高并发写入如果多个回调同时到达可能会出现database is locked的报错。生产环境建议把DATABASE_URL直接切换成PostgreSQLDATABASE_URL postgresql://user:passlocalhost:5432/sms_db代码里models.py的写法基本不用变SQLAlchemy SQLite和PostgreSQL之间可以平滑切换。唯一要注意的是PostgreSQL驱动需要额外安装pip install psycopg2-binary7.3 容器化部署我更喜欢用Docker来部署这个平台。写一个简单的DockerfileFROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]构建镜像并启动docker build -t sms-receiver . docker run -d -p 8000:8000 -v $(pwd):/app sms-receiver注意把sms.db数据库文件通过数据卷挂载出来不然容器一删除所有历史消息都没了。7.4 还能扩展什么功能这个平台的扩展空间其实很大。你可以给每个业务系统单独生成一个platform标识按项目维度统计短信量可以接入企业微信/钉钉机器人验证码到了之后直接推送到群聊可以增加一个“自动已读”接口测试脚本取完验证码后调用并标记已读下次查询默认跳过还可以对验证码做失败重试监控比如某个号码一分钟内重复收到验证码自动告警。我个人在实际操作中的体会是工具不在复杂在于顺手。这套源码我前后维护过几版从一开始只是给自己看验证码到后来被测试团队拿去接到自动化用例里整个过程没有加任何花哨功能核心就是“实时、准确、可查询”这六个字。你按着这篇文章搭完跑通第一单之后大概率也会有自己的想法那就顺着业务需求去加平台会越用越顺手。最后再分享一个小技巧调试回调接口时不要直接依赖真实短信服务商发短信费钱又慢。用服务商的“测试发送”功能或者用curl构造一个模拟回调请求又快又方便curl -X POST http://localhost:8000/api/callback/test \ -H Content-Type: application/json \ -d {phone: 13800138000, content: 您的验证码是888888请勿泄露。}能返回{code:0,message:ok}页面也收到新消息就说明整个链路没问题剩下的交给真实短信流入测试即可。