Python案例怎么发送短信通知？

本文目录导读：

1. 方法一：使用第三方短信服务商API
2. 方法二：使用第三方HTTP API接口
3. 方法三：使用邮件转短信网关（适用于特定运营商）
4. 方法四：完整的短信发送工具类
5. 注意事项

我来介绍几种Python发送短信通知的方法：

使用第三方短信服务商API

使用Twilio（国际服务，国内慎用）

from twilio.rest import Client
# Twilio账户信息
account_sid = 'your_account_sid'
auth_token = 'your_auth_token'
client = Client(account_sid, auth_token)
def send_sms_via_twilio(phone_number, message):
    try:
        message = client.messages.create(
            body=message,
            from_='+1234567890',  # Twilio提供的号码
            to=phone_number
        )
        print(f"短信发送成功，SID: {message.sid}")
        return True
    except Exception as e:
        print(f"发送失败: {e}")
        return False
# 使用示例
send_sms_via_twilio('+8613800138000', '您好，这是一条测试短信！')

使用阿里云短信服务（推荐国内使用）

import json
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
class AliyunSMS:
    def __init__(self, access_key_id, access_key_secret):
        self.client = AcsClient(
            access_key_id,
            access_key_secret,
            'cn-hangzhou'
        )
    def send_sms(self, phone_numbers, template_code, template_param):
        request = CommonRequest()
        request.set_accept_format('json')
        request.set_domain('dysmsapi.aliyuncs.com')
        request.set_method('POST')
        request.set_protocol_type('https')
        request.set_version('2017-05-25')
        request.set_action_name('SendSms')
        request.add_query_param('PhoneNumbers', phone_numbers)
        request.add_query_param('SignName', '你的短信签名')
        request.add_query_param('TemplateCode', template_code)
        request.add_query_param('TemplateParam', json.dumps(template_param))
        try:
            response = self.client.do_action(request)
            result = json.loads(response)
            if result['Code'] == 'OK':
                print(f"短信发送成功！")
                return True
            else:
                print(f"发送失败: {result['Message']}")
                return False
        except Exception as e:
            print(f"发送异常: {e}")
            return False
# 使用示例
sms = AliyunSMS('your_access_key_id', 'your_access_key_secret')
sms.send_sms(
    '13800138000',
    'SMS_123456789',  # 短信模板ID
    {'code': '123456'}  # 模板参数
)

使用聚合数据短信API

import requests
import json
class JuheSMS:
    def __init__(self, api_key):
        self.api_key = api_key
        self.url = 'http://v.juhe.cn/sms/send'
    def send_sms(self, phone, tpl_id, tpl_value):
        params = {
            'mobile': phone,
            'tpl_id': tpl_id,
            'tpl_value': json.dumps(tpl_value),
            'key': self.api_key
        }
        try:
            response = requests.post(self.url, data=params)
            result = response.json()
            if result['error_code'] == 0:
                print(f"短信发送成功！")
                return True
            else:
                print(f"发送失败: {result['reason']}")
                return False
        except Exception as e:
            print(f"发送异常: {e}")
            return False
# 使用示例
sms = JuheSMS('your_api_key')
sms.send_sms(
    '13800138000',
    'tpl_id_123',  # 模板ID
    {'code': '123456', 'minute': '5'}  # 模板变量
)

使用第三方HTTP API接口

import requests
import hashlib
import time
class CustomSMSAPI:
    def __init__(self, api_url, app_id, app_secret):
        self.api_url = api_url
        self.app_id = app_id
        self.app_secret = app_secret
    def generate_sign(self, params):
        """生成签名"""
        # 按key排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 拼接字符串
        sign_str = '&'.join([f'{k}={v}' for k, v in sorted_params])
        # 添加secret
        sign_str += f'&key={self.app_secret}'
        # MD5加密
        return hashlib.md5(sign_str.encode()).hexdigest().upper()
    def send_sms(self, phone, content):
        params = {
            'app_id': self.app_id,
            'phone': phone,
            'content': content,
            'timestamp': str(int(time.time()))
        }
        # 生成签名
        params['sign'] = self.generate_sign(params)
        try:
            response = requests.post(self.api_url, data=params)
            result = response.json()
            if result.get('code') == 0:
                print(f"短信发送成功！")
                return True
            else:
                print(f"发送失败: {result.get('msg')}")
                return False
        except Exception as e:
            print(f"发送异常: {e}")
            return False
# 使用示例
api = CustomSMSAPI(
    'http://api.example.com/send',
    'your_app_id',
    'your_app_secret'
)
api.send_sms('13800138000', '您好，这是一条测试短信！')

使用邮件转短信网关（适用于特定运营商）

import smtplib
from email.mime.text import MIMEText
class EmailToSMS:
    def __init__(self, smtp_server, smtp_port, email_user, email_pass):
        self.smtp_server = smtp_server
        self.smtp_port = smtp_port
        self.email_user = email_user
        self.email_pass = email_pass
    def send_sms(self, phone_number, carrier, message):
        """通过邮件转短信发送"""
        # 运营商网关地址
        carrier_gateways = {
            '移动': f'{phone_number}@139.com',
            '联通': f'{phone_number}@wo.cn',
            '电信': f'{phone_number}@189.cn'
        }
        if carrier not in carrier_gateways:
            print(f"不支持的运营商: {carrier}")
            return False
        to_email = carrier_gateways[carrier]
        # 创建邮件
        msg = MIMEText(message)
        msg['Subject'] = '短信通知'
        msg['From'] = self.email_user
        msg['To'] = to_email
        try:
            # 发送邮件
            server = smtplib.SMTP_SSL(self.smtp_server, self.smtp_port)
            server.login(self.email_user, self.email_pass)
            server.sendmail(self.email_user, [to_email], msg.as_string())
            server.quit()
            print(f"短信发送成功！")
            return True
        except Exception as e:
            print(f"发送失败: {e}")
            return False
# 使用示例
sms = EmailToSMS(
    'smtp.qq.com',
    465,
    'your_email@qq.com',
    'your_email_password'
)
sms.send_sms('13800138000', '移动', '您好，这是一条测试短信！')

完整的短信发送工具类

import requests
import json
import logging
from typing import Optional
class SMSClient:
    """短信发送客户端"""
    def __init__(self, provider: str, config: dict):
        """
        初始化短信客户端
        :param provider: 服务商名称 (aliyun, twilio, juhe等)
        :param config: 配置信息
        """
        self.provider = provider
        self.config = config
        self.logger = logging.getLogger(__name__)
        # 初始化对应的客户端
        if provider == 'aliyun':
            self._init_aliyun()
        elif provider == 'twilio':
            self._init_twilio()
        elif provider == 'juhe':
            self._init_juhe()
    def _init_aliyun(self):
        from aliyunsdkcore.client import AcsClient
        from aliyunsdkcore.request import CommonRequest
        self.client = AcsClient(
            self.config['access_key_id'],
            self.config['access_key_secret'],
            self.config.get('region', 'cn-hangzhou')
        )
        self.sign_name = self.config['sign_name']
    def _init_twilio(self):
        from twilio.rest import Client
        self.client = Client(
            self.config['account_sid'],
            self.config['auth_token']
        )
        self.from_number = self.config['from_number']
    def _init_juhe(self):
        self.base_url = 'http://v.juhe.cn/sms/send'
    def send(self, phone: str, template_id: str, params: dict) -> bool:
        """
        发送短信
        :param phone: 手机号码
        :param template_id: 模板ID
        :param params: 模板参数
        :return: 是否发送成功
        """
        try:
            if self.provider == 'aliyun':
                return self._send_aliyun(phone, template_id, params)
            elif self.provider == 'twilio':
                return self._send_twilio(phone, params.get('message', ''))
            elif self.provider == 'juhe':
                return self._send_juhe(phone, template_id, params)
            else:
                raise ValueError(f"不支持的服务商: {self.provider}")
        except Exception as e:
            self.logger.error(f"发送短信失败: {e}")
            return False
    def _send_aliyun(self, phone, template_id, params):
        from aliyunsdkcore.request import CommonRequest
        request = CommonRequest()
        request.set_accept_format('json')
        request.set_domain('dysmsapi.aliyuncs.com')
        request.set_method('POST')
        request.set_protocol_type('https')
        request.set_version('2017-05-25')
        request.set_action_name('SendSms')
        request.add_query_param('PhoneNumbers', phone)
        request.add_query_param('SignName', self.sign_name)
        request.add_query_param('TemplateCode', template_id)
        request.add_query_param('TemplateParam', json.dumps(params))
        response = self.client.do_action(request)
        result = json.loads(response)
        return result['Code'] == 'OK'
    def _send_twilio(self, phone, message):
        response = self.client.messages.create(
            body=message,
            from_=self.from_number,
            to=phone
        )
        return response.status == 'sent' or response.status == 'queued'
    def _send_juhe(self, phone, template_id, params):
        payload = {
            'mobile': phone,
            'tpl_id': template_id,
            'tpl_value': json.dumps(params),
            'key': self.config['api_key']
        }
        response = requests.post(self.base_url, data=payload)
        result = response.json()
        return result['error_code'] == 0
# 使用示例
if __name__ == '__main__':
    # 配置日志
    logging.basicConfig(level=logging.INFO)
    # 创建短信客户端
    sms_client = SMSClient('aliyun', {
        'access_key_id': 'your_access_key_id',
        'access_key_secret': 'your_access_key_secret',
        'sign_name': '你的签名',
        'region': 'cn-hangzhou'
    })
    # 发送短信
    success = sms_client.send(
        '13800138000',
        'SMS_123456789',
        {'code': '123456'}
    )
    if success:
        print("短信发送成功！")
    else:
        print("短信发送失败！")

注意事项

1. API密钥保护：敏感信息不要硬编码在代码中，使用环境变量或配置文件
2. 频率限制：遵守服务商的发送频率限制，避免被封
3. 异常处理：添加完善的异常处理机制
4. 日志记录：记录发送日志便于排查问题
5. 短信模板：国内服务商需要使用审核通过的短信模板
6. 费用控制：注意控制发送数量和费用

选择服务商时，建议根据你的目标用户群体（国内/国际）、预算和功能需求来决定。