Appearance
4.2.短信自动化
在现代营销和客户服务中，短信通知是一种快速、直接的沟通方式。通过Python，我们可以实现短信的自动化发送，用于客户通知、验证码验证、提醒服务等场景。
本文将介绍如何使用Python实现短信自动化，包括基础短信发送、批量发送个性化短信以及定时发送短信等功能。
短信发送服务简介
在Python中，我们主要使用第三方云服务提供的API来发送短信。常用的短信服务提供商包括：
1.阿里云短信服务：提供稳定可靠的短信发送服务，支持多种短信类型
2.腾讯云短信服务：提供全球短信覆盖，支持多语言短信发送
3.Twilio：国际知名的通信服务提供商，支持短信、语音通话等
4.云片网：提供短信、语音、流量服务，适合国内企业使用
安装所需依赖：
bash
pip install aliyunsdkcore aliyunsdksms qcloudsms_py requests基础短信发送功能
使用阿里云发送短信
阿里云短信服务提供了完整的短信发送API，以下是使用阿里云SDK发送短信的基础示例：
python
from aliyunsdkcore.client import AcsClient
from aliyunsdksms.request.v20170525 import SendSmsRequest
import json
def send_aliyun_sms(access_key_id, access_secret, phone_numbers, sign_name, template_code, template_param=None):
"""
使用阿里云发送短信
Args:
access_key_id: 阿里云AccessKey ID
access_secret: 阿里云AccessKey Secret
phone_numbers: 接收号码（字符串或列表）
sign_name: 短信签名名称
template_code: 短信模板ID
template_param: 模板参数（字典格式）
Returns:
发送结果
"""
try:
# 创建AcsClient实例
client = AcsClient(access_key_id, access_secret, "cn-hangzhou")
# 创建SendSmsRequest实例
request = SendSmsRequest.SendSmsRequest()
# 设置接收号码
if isinstance(phone_numbers, list):
phone_numbers = ','.join(phone_numbers)
request.set_PhoneNumbers(phone_numbers)
# 设置短信签名
request.set_SignName(sign_name)
# 设置短信模板ID
request.set_TemplateCode(template_code)
# 设置模板参数
if template_param:
request.set_TemplateParam(json.dumps(template_param))
# 发送请求
response = client.do_action_with_exception(request)
print("短信发送成功")
return response.decode('utf-8')
except Exception as e:
print(f"发送短信时出错: {e}")
return None
# 使用示例
# 注意：实际使用时应将敏感信息存储在环境变量或配置文件中
# result = send_aliyun_sms(
# access_key_id="your_access_key_id",
# access_secret="your_access_secret",
# phone_numbers=["13800138000"],
# sign_name="您的短信签名",
# template_code="SMS_123456789",
# template_param={"code": "123456"}
# )
# print(result)使用腾讯云发送短信
腾讯云也提供了完善的短信服务，以下是使用腾讯云发送短信的基础示例：
python
from qcloudsms_py import SmsSingleSender
from qcloudsms_py.httpclient import HTTPError
import json
def send_tencent_sms(appid, appkey, phone_number, template_id, template_params):
"""
使用腾讯云发送短信
Args:
appid: 腾讯云短信应用ID
appkey: 腾讯云短信应用密钥
phone_number: 接收号码
template_id: 短信模板ID
template_params: 模板参数列表（如["123456", "10"]）
Returns:
发送结果
"""
try:
# 创建短信发送对象
ssender = SmsSingleSender(appid, appkey)
# 发送短信
result = ssender.send_with_param(
86,
phone_number,
template_id,
template_params,
"",
"",
""
)
print("短信发送成功")
return result
except HTTPError as e:
print(f"HTTP请求失败: {e.code}, {e.response}")
return None
except Exception as e:
print(f"发送短信时出错: {e}")
return None
# 使用示例
# 注意：实际使用时应将敏感信息存储在环境变量或配置文件中
# result = send_tencent_sms(
# appid=123456,
# appkey="your_appkey",
# phone_number="13800138000",
# template_id=123456,
# template_params=["123456", "10"]
# )
# print(result)批量发送个性化短信
批量发送短信
以下是一个使用阿里云短信服务批量发送短信的示例：
python
import time
import random
from datetime import datetime
def batch_send_aliyun_sms(config_file, csv_file):
"""
批量发送短信（阿里云版）
Args:
config_file: 配置文件路径
csv_file: 包含客户信息的CSV文件
"""
try:
# 加载配置文件（这里简化为硬编码）
config = {
'access_key_id': 'your_access_key_id',
'access_secret': 'your_access_secret',
'sign_name': '您的短信签名',
'template_code': 'SMS_123456789'
}
# 读取客户数据（这里简化为硬编码）
customers = [
{'name': '张三', 'phone': '13800138000', 'code': '123456'},
{'name': '李四', 'phone': '13900139000', 'code': '654321'}
]
# 初始化短信客户端
from aliyunsdkcore.client import AcsClient
from aliyunsdksms.request.v20170525 import SendSmsRequest
client = AcsClient(config['access_key_id'], config['access_secret'], "cn-hangzhou")
# 批量发送短信
sent_count = 0
for customer in customers:
try:
# 构建请求
request = SendSmsRequest.SendSmsRequest()
request.set_PhoneNumbers(customer['phone'])
request.set_SignName(config['sign_name'])
request.set_TemplateCode(config['template_code'])
# 构建模板参数
template_param = {
"name": customer['name'],
"code": customer['code'],
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
request.set_TemplateParam(json.dumps(template_param))
# 发送短信
response = client.do_action_with_exception(request)
print(f"已发送短信给: {customer['name']} - {customer['phone']}")
sent_count += 1
# 模拟随机延迟（避免触发频率限制）
time.sleep(random.uniform(0.5, 1.5))
except Exception as e:
print(f"发送短信给 {customer['name']} 失败: {e}")
print(f"批量发送完成，共发送 {sent_count} 条短信")
return True
except Exception as e:
print(f"批量发送短信时出错: {e}")
return False
# 使用示例
# batch_send_aliyun_sms("config.json", "customers.csv")使用Pandas处理客户信息
在实际应用中，我们通常会从CSV文件读取客户信息。以下是使用Pandas读取客户信息并发送短信的示例：
python
import pandas as pd
def read_customer_info(csv_file):
"""
从CSV文件读取客户信息
Args:
csv_file: CSV文件路径
Returns:
客户信息DataFrame
"""
try:
# 读取CSV文件
df = pd.read_csv(csv_file)
# 检查必需字段
required_columns = ['name', 'phone', 'code']
for col in required_columns:
if col not in df.columns:
raise ValueError(f"缺少必需的列: {col}")
print(f"成功读取 {len(df)} 条客户信息")
return df
except Exception as e:
print(f"读取客户信息时出错: {e}")
return None
def send_personalized_sms(df, config):
"""
根据客户信息发送个性化短信
Args:
df: 客户信息DataFrame
config: 短信服务配置
Returns:
成功发送的数量
"""
sent_count = 0
try:
# 初始化短信客户端
from aliyunsdkcore.client import AcsClient
from aliyunsdksms.request.v20170525 import SendSmsRequest
client = AcsClient(config['access_key_id'], config['access_secret'], "cn-hangzhou")
# 批量发送短信
for _, row in df.iterrows():
try:
# 构建请求
request = SendSmsRequest.SendSmsRequest()
request.set_PhoneNumbers(row['phone'])
request.set_SignName(config['sign_name'])
request.set_TemplateCode(config['template_code'])
# 构建模板参数
template_param = {
"name": row['name'],
"code": row['code'],
"product": row.get('product', '未知产品'),
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
request.set_TemplateParam(json.dumps(template_param))
# 发送短信
response = client.do_action_with_exception(request)
print(f"已发送短信给: {row['name']} - {row['phone']}")
sent_count += 1
# 模拟随机延迟（避免触发频率限制）
time.sleep(random.uniform(0.5, 1.5))
except Exception as e:
print(f"发送短信给 {row['name']} 失败: {e}")
return sent_count
except Exception as e:
print(f"发送短信时出错: {e}")
return sent_count
# 示例主函数
if __name__ == "__main__":
# 简化配置（实际应从配置文件加载）
config = {
'access_key_id': 'your_access_key_id',
'access_secret': 'your_access_secret',
'sign_name': '您的短信签名',
'template_code': 'SMS_123456789'
}
# 读取客户信息
df = read_customer_info("customers.csv")
if df is not None:
# 发送个性化短信
sent_count = send_personalized_sms(df, config)
print(f"共发送 {sent_count} 条短信")定时发送短信功能
使用schedule模块定时发送短信
schedule 是一个轻量级的 Python 定时任务调度库，非常适合用来实现定时发送短信的功能。
python
import schedule
import time
import random
import pandas as pd
from datetime import datetime
def scheduled_sms_task():
"""
定时短信发送任务
"""
print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 开始执行定时短信发送任务...")
try:
# 加载客户信息
df = pd.read_csv("scheduled_customers.csv")
# 简化配置（实际应从配置文件加载）
config = {
'access_key_id': 'your_access_key_id',
'access_secret': 'your_access_secret',
'sign_name': '您的短信签名',
'template_code': 'SMS_123456789'
}
# 发送个性化短信
sent_count = send_personalized_sms(df, config)
print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 共发送 {sent_count} 条短信")
except Exception as e:
print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')] 短信任务执行失败: {e}")
def run_scheduler():
"""
运行定时任务调度器
"""
print("启动短信定时任务系统...")
# 安排每天特定时间执行任务
schedule.every().day.at("09:00").do(scheduled_sms_task)
# 无限循环检查任务
while True:
schedule.run_pending()
time.sleep(1)
# 使用示例
if __name__ == "__main__":
# 启动定时任务调度器
run_scheduler()运行定时任务调度器run_scheduler()函数，该函数会安排每天上午9点执行scheduled_sms_task()函数。schedule.run_pending()会检查是否有任务需要执行，如果有，则执行。time.sleep(1)会暂停1秒，以便检查下一个任务。
使用APScheduler实现更复杂的定时任务
APScheduler是一个更强大的Python定时任务调度库，支持更复杂的定时任务调度，如Cron表达式、间隔执行、一次性执行等。
python
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
import logging
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
def advanced_scheduler():
"""使用APScheduler实现更复杂的定时任务调度"""
# 创建调度器
scheduler = BlockingScheduler()
# 添加每天上午9点执行的任务
scheduler.add_job(
scheduled_sms_task,
CronTrigger(hour=9, minute=0),
id='daily_morning_task',
name='每日上午短信任务'
)
# 添加每周一下午2点执行的任务
scheduler.add_job(
lambda: scheduled_sms_task('weekly_customers.csv', 'SMS_987654321'),
CronTrigger(day_of_week='mon', hour=14, minute=0),
id='weekly_task',
name='每周促销短信任务'
)
# 添加每月1号执行的任务
scheduler.add_job(
lambda: scheduled_sms_task('monthly_customers.csv', 'SMS_567891234'),
CronTrigger(day=1, hour=10, minute=0),
id='monthly_task',
name='每月会员短信任务'
)
# 启动调度器
try:
logging.info('高级定时任务调度器已启动')
scheduler.start()
except (KeyboardInterrupt, SystemExit):
logging.info('调度器已停止')
scheduler.shutdown()短信发送状态监控与错误处理
在实际应用中，我们需要监控短信发送状态并处理可能出现的错误。以下是一个简单的短信发送状态监控与错误处理示例：
python
import json
import logging
import time
from datetime import datetime
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('sms_log.txt'),
logging.StreamHandler()
]
)
class SmsMonitor:
"""短信发送状态监控类"""
def __init__(self):
self.success_count = 0
self.fail_count = 0
self.retry_count = 0
self.start_time = None
self.end_time = None
self.failed_records = []
def start_monitoring(self):
"""开始监控"""
self.start_time = datetime.now()
logging.info(f"开始短信发送任务: {self.start_time}")
def end_monitoring(self):
"""结束监控"""
self.end_time = datetime.now()
duration = (self.end_time - self.start_time).total_seconds()
logging.info(f"短信发送任务结束: {self.end_time}")
logging.info(f"总耗时: {duration:.2f} 秒")
logging.info(f"成功发送: {self.success_count} 条")
logging.info(f"发送失败: {self.fail_count} 条")
logging.info(f"重试次数: {self.retry_count} 次")
# 保存失败记录
if self.failed_records:
with open('failed_sms.json', 'w', encoding='utf-8') as f:
json.dump(self.failed_records, f, ensure_ascii=False, indent=2)
logging.info(f"失败记录已保存至: failed_sms.json")
def record_success(self, phone, template_code):
"""记录发送成功"""
self.success_count += 1
logging.info(f"短信发送成功: {phone}, 模板: {template_code}")
def record_failure(self, phone, template_code, error, retry=False):
"""记录发送失败"""
if retry:
self.retry_count += 1
logging.warning(f"短信重试失败: {phone}, 模板: {template_code}, 错误: {error}")
else:
self.fail_count += 1
logging.error(f"短信发送失败: {phone}, 模板: {template_code}, 错误: {error}")
# 记录失败信息
self.failed_records.append({
'phone': phone,
'template_code': template_code,
'error': str(error),
'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'retry': retry
})
def send_with_retry(phone, template_code, template_param, config, max_retries=3, retry_interval=5):
"""带重试机制的短信发送"""
monitor = SmsMonitor()
monitor.start_monitoring()
# 初始化短信客户端
from aliyunsdkcore.client import AcsClient
from aliyunsdksms.request.v20170525 import SendSmsRequest
client = AcsClient(config['access_key_id'], config['access_secret'], "cn-hangzhou")
retries = 0
while retries <= max_retries:
try:
# 构建请求
request = SendSmsRequest.SendSmsRequest()
request.set_PhoneNumbers(phone)
request.set_SignName(config['sign_name'])
request.set_TemplateCode(template_code)
if template_param:
request.set_TemplateParam(json.dumps(template_param))
# 发送短信
response = client.do_action_with_exception(request)
response_dict = json.loads(response.decode('utf-8'))
# 检查发送结果
if response_dict.get('Code') == 'OK':
monitor.record_success(phone, template_code)
break
else:
error = f"错误码: {response_dict.get('Code')}, 消息: {response_dict.get('Message')}"
if retries < max_retries:
logging.warning(f"短信发送失败，准备重试 ({retries+1}/{max_retries}): {error}")
retries += 1
time.sleep(retry_interval)
else:
monitor.record_failure(phone, template_code, error, retry=True)
break
except Exception as e:
if retries < max_retries:
logging.warning(f"短信发送异常，准备重试 ({retries+1}/{max_retries}): {e}")
retries += 1
time.sleep(retry_interval)
else:
monitor.record_failure(phone, template_code, str(e), retry=True)
break
monitor.end_monitoring()
return monitor.success_count > 0实际应用场景
1. 验证码发送
在用户注册、登录、找回密码等场景中，我们需要发送验证码短信：
python
import random
import string
def generate_verification_code(length=6):
"""生成随机验证码"""
return ''.join(random.choices(string.digits, k=length))
def send_verification_code(phone, config):
"""发送验证码短信"""
# 生成验证码
code = generate_verification_code()
# 设置验证码有效期（分钟）
expire_minutes = 5
# 发送短信
template_param = {
"code": code,
"expire": str(expire_minutes)
}
result = send_aliyun_sms(
access_key_id=config['access_key_id'],
access_secret=config['access_secret'],
phone_numbers=phone,
sign_name=config['sign_name'],
template_code=config['template_code_verification'],
template_param=template_param
)
# 返回验证码和发送结果
return {
'code': code,
'expire_minutes': expire_minutes,
'success': result is not None
}2. 营销活动通知
在促销活动、新品发布等场景中，我们需要向客户发送营销短信：
python
def send_marketing_message(customer_file, config):
"""发送营销活动通知"""
# 读取客户信息
df = pd.read_csv(customer_file)
# 按客户等级分组
vip_customers = df[df['level'] == 'VIP']
regular_customers = df[df['level'] == 'Regular']
# 为VIP客户发送专属优惠
for _, customer in vip_customers.iterrows():
template_param = {
"name": customer['name'],
"discount": "8折",
"code": f"VIP{customer['id']}"
}
send_aliyun_sms(
access_key_id=config['access_key_id'],
access_secret=config['access_secret'],
phone_numbers=customer['phone'],
sign_name=config['sign_name'],
template_code=config['template_code_vip'],
template_param=template_param
)
# 避免频率限制
time.sleep(random.uniform(0.5, 1.5))
# 为普通客户发送一般优惠
for _, customer in regular_customers.iterrows():
template_param = {
"name": customer['name'],
"discount": "9折",
"code": f"REG{customer['id']}"
}
send_aliyun_sms(
access_key_id=config['access_key_id'],
access_secret=config['access_secret'],
phone_numbers=customer['phone'],
sign_name=config['sign_name'],
template_code=config['template_code_regular'],
template_param=template_param
)
# 避免频率限制
time.sleep(random.uniform(0.5, 1.5))3. 订单状态通知
在电商、物流等场景中，我们需要向客户发送订单状态变更通知：
python
def send_order_status_notification(order_id, customer_phone, status, config):
"""发送订单状态通知"""
# 根据状态选择不同的模板
template_code = {
'payment_received': 'SMS_PAYMENT_RECEIVED',
'shipped': 'SMS_SHIPPED',
'delivered': 'SMS_DELIVERED',
'cancelled': 'SMS_CANCELLED'
}.get(status)
if not template_code:
print(f"未知的订单状态: {status}")
return False
# 构建模板参数
template_param = {
"order_id": order_id,
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
# 如果是发货状态，添加物流信息
if status == 'shipped':
template_param["logistics"] = "顺丰快递"
template_param["tracking_number"] = f"SF{random.randint(1000000, 9999999)}"
# 发送短信
result = send_aliyun_sms(
access_key_id=config['access_key_id'],
access_secret=config['access_secret'],
phone_numbers=customer_phone,
sign_name=config['sign_name'],
template_code=template_code,
template_param=template_param
)
return result is not None短信发送最佳实践
1. 安全性考虑
在实际应用中，我们需要注意短信发送的安全性：
python
import os
from dotenv import load_dotenv
import hashlib
import time
# 加载环境变量
load_dotenv()
def get_secure_config():
"""从环境变量获取配置信息"""
return {
'access_key_id': os.getenv('SMS_ACCESS_KEY_ID'),
'access_secret': os.getenv('SMS_ACCESS_SECRET'),
'sign_name': os.getenv('SMS_SIGN_NAME'),
'template_code_verification': os.getenv('SMS_TEMPLATE_VERIFICATION'),
'template_code_marketing': os.getenv('SMS_TEMPLATE_MARKETING'),
'template_code_notification': os.getenv('SMS_TEMPLATE_NOTIFICATION')
}
def prevent_sms_bombing(phone, redis_client, max_sms=5, period_seconds=300):
"""防止短信轰炸"""
# 生成Redis键
key = f"sms:limit:{phone}"
# 获取当前发送次数
count = redis_client.get(key)
if count is None:
# 首次发送，设置计数器为1，有效期为period_seconds
redis_client.setex(key, period_seconds, 1)
return True
elif int(count) < max_sms:
# 未超过限制，计数器加1
redis_client.incr(key)
return True
else:
# 超过限制
return False2. 短信模板管理
在实际应用中，我们需要管理多个短信模板：
python
class SmsTemplateManager:
"""短信模板管理类"""
def __init__(self, config_file):
"""初始化短信模板管理器"""
with open(config_file, 'r', encoding='utf-8') as f:
self.templates = json.load(f)
def get_template(self, template_name):
"""获取模板信息"""
if template_name in self.templates:
return self.templates[template_name]
else:
raise ValueError(f"未找到模板: {template_name}")
def render_template(self, template_name, params):
"""渲染模板参数"""
template = self.get_template(template_name)
# 检查必需参数
for required_param in template.get('required_params', []):
if required_param not in params:
raise ValueError(f"缺少必需参数: {required_param}")
# 返回模板信息和参数
return {
'template_code': template['template_code'],
'sign_name': template.get('sign_name'),
'params': params
}小结
本文介绍了如何使用Python实现短信自动化，包括基础短信发送、批量发送个性化短信、定时发送短信等功能。通过阿里云、腾讯云等第三方短信服务，我们可以轻松实现短信的自动化发送，用于客户通知、验证码验证、提醒服务等场景。
在实际应用中，我们需要注意以下几点：
- 安全性：保护API密钥，防止短信轰炸，遵守相关法规
- 可靠性：实现重试机制，监控发送状态，处理错误
- 灵活性：支持多种短信模板，支持个性化参数，支持定时发送
- 成本控制：合理设置发送频率，避免不必要的发送
通过本文的学习，读者应该能够掌握使用Python进行短信自动化的基本技能，为构建更复杂的自动化营销系统打下基础。