
3步搞定网络发短信:手写实现解决API版本变动痛点
版本升级后 API 全变了?别慌,今天带你手写实现网络发短信核心逻辑,彻底摆脱对第三方SDK的依赖。
项目目标与痛点分析
做后端开发的朋友肯定遇到过这种情况:昨天还能正常发送短信,今天一更新依赖包,报错信息直接告诉你“Method not found”。这就是典型的API版本迭代带来的阵痛。很多开发者习惯直接调用阿里云、腾讯云或Twilio的SDK,这些库虽然封装得不错,但底层通信协议一旦变更,或者厂商调整了接口规范,你的代码就得跟着重写。
更麻烦的是,很多老旧项目还在用几年前的SDK版本,新版本的鉴权机制、参数结构完全变了。这时候,如果你懂底层的HTTP通信和签名算法,自己手写实现发送逻辑,就能牢牢掌握主动权。本文将以Python为例,从零搭建一个不依赖任何短信SDK的网络发短信模块。我们的目标很明确:通过标准HTTP请求,手动构造签名,实现短信发送。这不仅解决了版本兼容问题,还让你彻底理解短信网关的工作原理。
目录结构与依赖规划
为了保证代码的可复现性,我们规划如下目录结构:
sms_sender/
├── main.py # 主程序入口
├── core/
│ ├── __init__.py
│ ├── signer.py # 签名算法实现
│ ├── api_client.py# HTTP请求封装
├── config.py # 配置管理
└── requirements.txt # 依赖库
我们只需要最基础的依赖,避免引入重型框架。requests 库用于处理HTTP请求,hashlib 用于MD5/SHA256签名,hmac 用于HMAC-SHA256计算。这些标准库在Python环境中默认可用或极易安装,确保了代码的轻量化和高可移植性。
在 requirements.txt 中,我们仅列出:
requests=2.28.0
注意,我们特意排除了所有厂商提供的SDK。这就是“手写实现”的核心价值:去中间件化,直接对接网关API。
核心代码实现与逐行讲解
1. 配置管理:隔离敏感信息
在 config.py 中,我们定义网关地址和密钥。这里以模拟网关为例,实际使用时替换为你申请的AccessKey和SecretKey。
# config.py
import os
# 从环境变量读取,避免硬编码泄露
class Config:
# 网关端点,不同厂商地址不同
API_ENDPOINT = os.getenv(SMS_API_URL, https://api.example-sms.com/v1/send)
ACCESS_KEY = os.getenv(SMS_ACCESS_KEY, your_access_key)
SECRET_KEY = os.getenv(SMS_SECRET_KEY, your_secret_key)
# 签名算法版本,部分网关区分v1和v2
SIGN_VERSION = v2
2. 签名算法:手写核心逻辑
这是最容易被SDK掩盖的部分。大多数短信网关采用 HMAC-SHA256 进行签名。我们需要手动构造待签名字符串,通常包含:HTTP方法、请求路径、查询参数排序后的字符串、时间戳。
在 core/signer.py 中实现:
# core/signer.py
import hashlib
import hmac
import time
import urllib.parse
from config import Config
class SMSSigner:
短信网关签名器
手动实现HMAC-SHA256签名逻辑
@staticmethod
def _generate_timestamp():
生成当前时间戳(秒级)
return int(time.time())
@staticmethod
def _sort_params(params: dict) - str:
对参数进行字典序排序并拼接
注意:排除None值,key和value均进行URL编码
if not params:
return
# 过滤空值
filtered = {k: v for k, v in params.items() if v is not None}
# 按key字典序排序
sorted_items = sorted(filtered.items())
# 拼接成 key=valuekey=value 格式
# 注意:这里使用的是原始值,部分网关要求编码后的值,需根据文档调整
return .join([f{k}={v} for k, v in sorted_items])
@classmethod
def sign_request(cls, method: str, path: str, params: dict) - dict:
生成请求头中的签名信息
返回包含签名、时间戳、密钥ID的字典
timestamp = cls._generate_timestamp()
# 构造待签名串: Method\nPath\nQueryString\nTimestamp
# 具体格式取决于网关文档,此处以常见规范为例
query_string = cls._sort_params(params)
# 构建签名原始串
# 换行符使用 \n
raw_string_to_sign = f{method}\n{path}\n{query_string}\n{timestamp}
# 计算HMAC-SHA256
key = Config.SECRET_KEY.encode('utf-8')
msg = raw_string_to_sign.encode('utf-8')
# hmac.new(key, msg, digestmod)
signature = hmac.new(key, msg, hashlib.sha256).hexdigest()
return {
X-Signature: signature,
X-Timestamp: str(timestamp),
X-Access-Key: Config.ACCESS_KEY,
X-Sign-Version: Config.SIGN_VERSION
}
逐行解析关键点:
_sort_params 方法确保了参数顺序的一致性,这是签名验证通过的前提。很多新手在这里翻车,是因为忽略了参数排序或URL编码的规则。
raw_string_to_sign 的构造格式是核心。不同的短信服务商(如阿里云、AWS SNS)格式略有差异,必须严格参照官方文档。
hmac.new 的第三个参数 digestmod 指定了哈希算法,必须与网关要求一致。
3. API客户端:HTTP请求封装
在 core/api_client.py 中,我们封装发送逻辑:
# core/api_client.py
import requests
import json
from config import Config
from core.signer import SMSSigner
class SMSClient:
短信发送客户端
基于requests库的手写实现
def __init__(self):
self.endpoint = Config.API_ENDPOINT
self.signer = SMSSigner()
def send_sms(self, phone: str, message: str, template_code: str = None) - dict:
发送短信
:param phone: 接收号码
:param message: 短信内容
:param template_code: 模板代码,部分网关需要
:return: 响应结果字典
# 1. 构造请求参数
params = {
phoneNumber: phone,
content: message,
timestamp: self.signer._generate_timestamp() # 确保时间戳参与签名
}
if template_code:
params[templateCode] = template_code
# 2. 提取路径和查询参数用于签名
# 假设 endpoint 为 https://api.example.com/v1/send
# 我们需要分离出 path (/v1/send) 和 query params
# 这里简化处理,假设所有参数都在body或query中
# 实际项目中,需根据网关要求决定参数放URL还是Body
# 假设参数放在URL Query中
method = POST
path = /v1/send # 需从endpoint中解析,此处简化
# 3. 生成签名头
headers = self.signer.sign_request(method, path, params)
# 4. 发送请求
# 注意:如果参数在Body中,params应放入data,而非headers签名源
# 此处演示参数在Query String的情况
try:
response = requests.post(
self.endpoint,
headers=headers,
params=params, # 自动序列化到URL Query
timeout=5
)
response.raise_for_status() # 抛出HTTP错误
return response.json()
except requests.exceptions.RequestException as e:
return {
success: False,
error: str(e)
}
避坑指南:
超时设置:务必设置 timeout,防止网关无响应导致线程挂起。
参数位置:签名时使用的参数列表,必须与实际发送的参数完全一致。如果参数放在JSON Body中,签名串的构造方式可能不同,需查阅具体文档。
异常处理:raise_for_status() 能将非200状态码转化为异常,便于统一捕获。
运行与测试
创建 main.py 进行测试:
# main.py
import sys
from core.api_client import SMSClient
def main():
# 初始化客户端
client = SMSClient()
# 测试发送
# 注意:此处需配置真实的环境变量或修改config.py中的默认值
result = client.send_sms(
phone=13800138000,
message=这是一条测试短信,
template_code=SMS_123456
)
# 打印结果
print(json.dumps(result, indent=4, ensure_ascii=False))
if result.get(success):
print(短信发送成功!)
sys.exit(0)
else:
print(f发送失败: {result.get('error')})
sys.exit(1)
if __name__ == __main__:
main()
运行前,确保设置了环境变量:
export SMS_API_URL=https://api.example-sms.com/v1/send
export SMS_ACCESS_KEY=test_key
export SMS_SECRET_KEY=test_secret
python main.py
如果返回 SignatureDoesNotMatch,请检查:
时间戳是否过期(部分网关允许5分钟误差)。
参数排序是否正确。
换行符是否使用了 \n 而非 \r\n。
优化扩展与进阶技巧
手写实现的最大优势在于可定制性。以下是几个优化方向:
重试机制:网络不稳定时,增加指数退避重试。
import time
import random
def send_with_retry(self, phone, message, retries=3):
for i in range(retries):
result = self.send_sms(phone, message)
if result.get(success):
return result
# 指数退避
wait_time = (2 ** i) + random.uniform(0, 1)
time.sleep(wait_time)
return {success: False, error: Max retries exceeded}
连接池复用:requests 默认每次请求都新建连接,使用 Session 对象可复用TCP连接,提升性能。
class SMSClient:
def __init__(self):
self.session = requests.Session()
# ...
日志记录:在 sign_request 中增加调试日志,输出原始签名串,便于排查问题。
import logging
logger = logging.getLogger(__name__)
# 在sign_request中
logger.debug(fRaw String: {raw_string_to_sign})
多厂商适配:通过策略模式,将签名算法抽象为接口,不同厂商实现不同的Signer类,实现代码解耦。
小结
通过手写实现网络发短信功能,我们不仅解决了API版本升级带来的兼容性问题,更深入理解了短信网关的通信机制。这种底层掌控力,在面对任何第三方服务变动时,都能让你快速响应,而非被动等待SDK更新。
在实际项目中,建议将签名逻辑单独封装为工具类,便于维护和测试。同时,务必重视密钥管理,避免硬编码。
你更常用哪种写法?是直接依赖厂商SDK,还是像本文这样手写HTTP请求与签名?评论区交流你的经验。