当前位置：首页 > news >正文

Python实战：5分钟搞定京东云短信API的发送、状态查询与回复处理

news 2026/7/15 6:28:04

Python实战：5分钟搞定京东云短信API的发送、状态查询与回复处理

短信功能在现代应用中扮演着至关重要的角色，从用户注册验证码到订单状态通知，再到营销活动推广，短信服务已经成为连接企业与用户的重要桥梁。对于Python开发者而言，快速集成短信功能不仅能提升开发效率，更能为业务带来即时、可靠的通信保障。本文将带你深入京东云短信API的实战应用，从发送短信到处理回执和用户回复，构建一个完整的端到端解决方案。

1. 环境准备与京东云配置

在开始编码之前，我们需要完成一些必要的准备工作。首先确保你的开发环境已经安装了Python 3.6或更高版本，这是京东云SDK的最低要求。

安装京东云Python SDK非常简单，只需运行以下命令：

pip install jdcloud_sdk

接下来，我们需要在京东云控制台进行一些配置：

登录京东云控制台，进入"短信服务"页面
创建应用并获取Access Key和Secret Key
申请短信签名（需要企业认证）
创建短信模板并等待审核通过

重要提示：短信签名和模板需要审核，建议提前准备，避免影响开发进度。审核通常需要1-2个工作日。

2. 短信发送功能实现

短信发送是短信服务最核心的功能。我们将从基础发送开始，逐步构建一个健壮的发送模块。

2.1 基础发送功能

首先，我们创建一个sms_sender.py文件，实现最基本的短信发送功能：

from jdcloud_sdk.core.credential import Credential from jdcloud_sdk.services.sms.client.SmsClient import SmsClient from jdcloud_sdk.services.sms.apis.BatchSendRequest import BatchSendParameters, BatchSendRequest class SMSSender: def __init__(self, access_key, secret_key, region_id='cn-north-1'): self.credential = Credential(access_key, secret_key) self.client = SmsClient(self.credential) self.region_id = region_id def send_verification_code(self, phone_number, code, template_id, sign_id): try: parameters = BatchSendParameters( regionId=self.region_id, templateId=template_id, signId=sign_id, phoneList=[phone_number], params=[str(code)] ) request = BatchSendRequest(parameters) resp = self.client.send(request) if resp.error is not None: print(f"发送失败: {resp.error.code} - {resp.error.message}") return False print(f"发送成功，序列号: {resp.result['sequenceNumber']}") return resp.result['sequenceNumber'] except Exception as e: print(f"发送异常: {str(e)}") return False

这个基础版本已经可以实现短信发送，但在实际项目中，我们需要考虑更多因素。

2.2 增强发送功能

让我们增强这个发送器，添加重试机制和更完善的错误处理：

import time from typing import List, Optional class EnhancedSMSSender(SMSSender): def __init__(self, access_key, secret_key, max_retries=3, retry_delay=1, region_id='cn-north-1'): super().__init__(access_key, secret_key, region_id) self.max_retries = max_retries self.retry_delay = retry_delay def send_batch(self, phone_numbers: List[str], template_id: str, sign_id: str, params: Optional[List[str]] = None) -> dict: attempts = 0 last_error = None while attempts < self.max_retries: try: parameters = BatchSendParameters( regionId=self.region_id, templateId=template_id, signId=sign_id, phoneList=phone_numbers, params=params or [] ) request = BatchSendRequest(parameters) resp = self.client.send(request) if resp.error is not None: last_error = f"{resp.error.code} - {resp.error.message}" attempts += 1 if attempts < self.max_retries: time.sleep(self.retry_delay) continue return { 'success': True, 'sequence_number': resp.result['sequenceNumber'], 'message_ids': resp.result['messageIds'] } except Exception as e: last_error = str(e) attempts += 1 if attempts < self.max_retries: time.sleep(self.retry_delay) return { 'success': False, 'error': last_error, 'attempts': attempts }

这个增强版本提供了以下改进：

支持批量发送
内置重试机制
更详细的返回信息
类型提示
更好的错误处理

3. 状态报告查询与处理

发送短信只是第一步，了解短信的送达状态同样重要。京东云提供了状态报告查询接口，让我们可以获取短信的最终状态。

3.1 基础状态查询

创建一个sms_status.py文件，实现状态查询功能：

from jdcloud_sdk.services.sms.apis.StatusReportRequest import StatusReportParameters, StatusReportRequest class SMSStatusChecker(SMSSender): def get_status_report(self, sequence_number: str, phone_numbers: List[str] = None) -> dict: try: parameters = StatusReportParameters( regionId=self.region_id, sequenceNumber=sequence_number ) if phone_numbers: parameters.setPhoneList(phoneList=phone_numbers) request = StatusReportRequest(parameters) resp = self.client.send(request) if resp.error is not None: return { 'success': False, 'error': f"{resp.error.code} - {resp.error.message}" } return { 'success': True, 'reports': resp.result['statusReports'] } except Exception as e: return { 'success': False, 'error': str(e) }

3.2 状态报告处理策略

在实际应用中，我们需要定期轮询状态报告，并处理各种可能的状态。以下是常见的状态码及其含义：

状态码	含义	建议处理方式
DELIVRD	已送达	无需处理
EXPIRED	已过期	记录日志，可能需要重新发送
DELETED	已删除	检查手机号是否有效
UNDELIV	未送达	检查手机号或联系运营商
UNKNOWN	未知状态	稍后重试查询
REJECTD	被拒绝	检查短信内容是否符合规范

我们可以实现一个状态处理器：

import logging from datetime import datetime, timedelta class SMSStatusProcessor(SMSStatusChecker): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.logger = logging.getLogger('sms_status') def process_status_reports(self, sequence_number: str, max_attempts=5, interval=60): attempts = 0 processed_numbers = set() while attempts < max_attempts: result = self.get_status_report(sequence_number) if not result['success']: self.logger.error(f"获取状态报告失败: {result['error']}") attempts += 1 time.sleep(interval) continue new_reports = [ report for report in result['reports'] if report['phone'] not in processed_numbers ] if not new_reports: self.logger.info("没有新的状态报告") break for report in new_reports: self._handle_single_report(report) processed_numbers.add(report['phone']) attempts += 1 if attempts < max_attempts: time.sleep(interval) def _handle_single_report(self, report): status = report['status'] phone = report['phone'] receive_time = report['receiveTime'] if status == 'DELIVRD': self.logger.info(f"短信已送达: {phone} at {receive_time}") elif status in ('EXPIRED', 'UNDELIV', 'REJECTD'): self.logger.warning(f"短信未送达({status}): {phone}") else: self.logger.info(f"短信状态未知({status}): {phone}")

4. 用户回复处理

对于双向短信服务，处理用户回复是必不可少的环节。京东云提供了回复查询接口，让我们可以获取用户对短信的回复。

4.1 基础回复查询

创建一个sms_reply.py文件，实现回复查询功能：

from jdcloud_sdk.services.sms.apis.ReplyRequest import ReplyParameters, ReplyRequest from typing import Dict, List class SMSReplyHandler(SMSSender): def get_replies(self, app_id: str, date: str, phone_numbers: List[str] = None) -> Dict: try: parameters = ReplyParameters( regionId=self.region_id, appId=app_id, dataDate=date ) if phone_numbers: parameters.setPhoneList(phoneList=phone_numbers) request = ReplyRequest(parameters) resp = self.client.send(request) if resp.error is not None: return { 'success': False, 'error': f"{resp.error.code} - {resp.error.message}" } return { 'success': True, 'replies': resp.result['replyContents'] } except Exception as e: return { 'success': False, 'error': str(e) }

4.2 回复处理与业务集成

用户回复通常需要与业务逻辑集成。例如，验证码验证、问卷调查回复、订单确认等。下面是一个处理回复的示例：

import json from collections import defaultdict class BusinessReplyHandler(SMSReplyHandler): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.handlers = defaultdict(list) def register_handler(self, keyword: str, handler): self.handlers[keyword.lower()].append(handler) def process_daily_replies(self, app_id: str, date: str = None): if date is None: date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d') result = self.get_replies(app_id, date) if not result['success']: self.logger.error(f"获取回复失败: {result['error']}") return False for reply in result['replies']: self._process_single_reply(reply) return True def _process_single_reply(self, reply): phone = reply['phone'] content = reply['content'].strip().lower() reply_time = reply['replyTime'] self.logger.info(f"收到回复: {phone} - {content}") # 查找匹配的处理程序 for keyword, handlers in self.handlers.items(): if keyword in content: for handler in handlers: try: handler(phone, content, reply_time) except Exception as e: self.logger.error(f"处理回复时出错: {str(e)}")

5. 完整项目集成示例

现在，我们将前面实现的各个模块整合到一个完整的项目中，模拟一个用户注册验证码的场景。

5.1 项目结构

sms_project/ ├── config/ │ └── settings.py # 配置文件 ├── core/ │ ├── __init__.py │ ├── sms_sender.py # 发送模块 │ ├── sms_status.py # 状态模块 │ └── sms_reply.py # 回复模块 ├── services/ │ └── user_service.py # 业务服务 └── main.py # 主程序

5.2 配置文件

config/settings.py:

# 京东云配置 JDCLOUD_ACCESS_KEY = 'your_access_key' JDCLOUD_SECRET_KEY = 'your_secret_key' JDCLOUD_REGION_ID = 'cn-north-1' # 短信模板配置 SMS_TEMPLATE_VERIFICATION = 'your_template_id' SMS_SIGN_ID = 'your_sign_id' # 应用ID SMS_APP_ID = 'your_app_id'

5.3 业务服务

services/user_service.py:

import random import string from datetime import datetime, timedelta from typing import Optional, Tuple from core.sms_sender import EnhancedSMSSender from core.sms_status import SMSStatusProcessor from core.sms_reply import BusinessReplyHandler from config import settings class UserService: def __init__(self): self.sms_sender = EnhancedSMSSender( settings.JDCLOUD_ACCESS_KEY, settings.JDCLOUD_SECRET_KEY ) self.status_processor = SMSStatusProcessor( settings.JDCLOUD_ACCESS_KEY, settings.JDCLOUD_SECRET_KEY ) self.reply_handler = BusinessReplyHandler( settings.JDCLOUD_ACCESS_KEY, settings.JDCLOUD_SECRET_KEY ) # 注册回复处理程序 self.reply_handler.register_handler('验证码', self.handle_verification_reply) # 内存中的验证码存储（生产环境应使用数据库） self.verification_codes = {} def generate_verification_code(self, length=6) -> str: return ''.join(random.choices(string.digits, k=length)) def send_verification_code(self, phone_number: str) -> Tuple[bool, Optional[str]]: code = self.generate_verification_code() sequence_number = self.sms_sender.send_verification_code( phone_number, code, settings.SMS_TEMPLATE_VERIFICATION, settings.SMS_SIGN_ID ) if sequence_number: self.verification_codes[phone_number] = { 'code': code, 'sent_at': datetime.now(), 'sequence_number': sequence_number, 'verified': False } return True, sequence_number return False, None def verify_code(self, phone_number: str, code: str) -> bool: record = self.verification_codes.get(phone_number) if not record: return False if record['verified']: return False # 验证码有效期5分钟 if datetime.now() - record['sent_at'] > timedelta(minutes=5): return False if record['code'] == code: record['verified'] = True return True return False def handle_verification_reply(self, phone: str, content: str, reply_time: str): # 提取验证码，假设回复格式为"验证码123456" code = content.replace('验证码', '').strip() if self.verify_code(phone, code): print(f"用户 {phone} 验证成功") else: print(f"用户 {phone} 验证失败") def check_delivery_status(self, phone_number: str): record = self.verification_codes.get(phone_number) if not record or 'sequence_number' not in record: return False self.status_processor.process_status_reports(record['sequence_number']) def process_replies(self): self.reply_handler.process_daily_replies(settings.SMS_APP_ID)

5.4 主程序

main.py:

import logging from services.user_service import UserService # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) def main(): user_service = UserService() # 模拟用户注册流程 phone_number = '13800138000' # 替换为实际手机号 # 发送验证码 success, seq_num = user_service.send_verification_code(phone_number) if success: print(f"验证码已发送，序列号: {seq_num}") # 检查发送状态 user_service.check_delivery_status(phone_number) # 处理用户回复（模拟） # 实际应用中，这部分通常由定时任务或事件驱动 user_service.process_replies() else: print("验证码发送失败") if __name__ == '__main__': main()

6. 高级技巧与最佳实践

在实际项目中使用短信服务时，以下技巧和最佳实践可以帮助你构建更健壮的系统。

6.1 性能优化

批量发送：尽可能使用批量接口发送短信，减少API调用次数
异步处理：将短信发送和状态检查放入后台任务，避免阻塞主流程
本地缓存：缓存常用模板和配置，减少不必要的API调用

6.2 错误处理与重试

网络错误：实现指数退避重试机制
限流处理：识别429状态码并适当降低请求频率
业务错误：区分可重试错误和不可重试错误

6.3 安全考虑

敏感信息：不要在日志中记录完整的手机号
密钥管理：使用环境变量或密钥管理服务存储Access Key和Secret Key
权限控制：遵循最小权限原则，为短信服务分配适当的权限

6.4 监控与告警

发送成功率：监控短信发送成功率，设置告警阈值
延迟监控：跟踪从发送到送达的时间
回复率：对于营销短信，监控用户回复率

7. 测试策略

完善的测试是保证短信服务可靠性的关键。以下是一些测试建议：

7.1 单元测试

import unittest from unittest.mock import MagicMock, patch from core.sms_sender import SMSSender class TestSMSSender(unittest.TestCase): @patch('jdcloud_sdk.services.sms.client.SmsClient') def test_send_verification_code_success(self, mock_client): # 配置mock mock_response = MagicMock() mock_response.error = None mock_response.result = {'sequenceNumber': 'test123'} mock_client.return_value.send.return_value = mock_response # 测试发送 sender = SMSSender('test_key', 'test_secret') result = sender.send_verification_code('13800138000', '123456', 'tpl123', 'sign123') # 验证结果 self.assertEqual(result, 'test123') @patch('jdcloud_sdk.services.sms.client.SmsClient') def test_send_verification_code_failure(self, mock_client): # 配置mock错误 mock_response = MagicMock() mock_response.error = MagicMock(code='500', message='Internal Error') mock_client.return_value.send.return_value = mock_response # 测试发送 sender = SMSSender('test_key', 'test_secret') result = sender.send_verification_code('13800138000', '123456', 'tpl123', 'sign123') # 验证结果 self.assertFalse(result) if __name__ == '__main__': unittest.main()