Skip to content

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 False

2. 短信模板管理

在实际应用中,我们需要管理多个短信模板:

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实现短信自动化,包括基础短信发送、批量发送个性化短信、定时发送短信等功能。通过阿里云、腾讯云等第三方短信服务,我们可以轻松实现短信的自动化发送,用于客户通知、验证码验证、提醒服务等场景。

在实际应用中,我们需要注意以下几点:

  1. 安全性:保护API密钥,防止短信轰炸,遵守相关法规
  2. 可靠性:实现重试机制,监控发送状态,处理错误
  3. 灵活性:支持多种短信模板,支持个性化参数,支持定时发送
  4. 成本控制:合理设置发送频率,避免不必要的发送

通过本文的学习,读者应该能够掌握使用Python进行短信自动化的基本技能,为构建更复杂的自动化营销系统打下基础。