310 lines
12 KiB
Python
310 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
阿里云 SMS 短信管理工具
|
||
功能: 签名管理、模板管理、发送记录查询、发送测试
|
||
依赖: pip install alibabacloud-dysmsapi20170525
|
||
"""
|
||
import argparse
|
||
import json
|
||
import sys
|
||
import os
|
||
from datetime import datetime, timedelta
|
||
|
||
def get_client():
|
||
from alibabacloud_dysmsapi20170525.client import Client
|
||
from alibabacloud_tea_openapi.models import Config
|
||
ak_id = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '')
|
||
ak_secret = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '')
|
||
if not ak_id or not ak_secret:
|
||
print('ERROR: Set ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET')
|
||
sys.exit(1)
|
||
config = Config(
|
||
access_key_id=ak_id,
|
||
access_key_secret=ak_secret,
|
||
endpoint='dysmsapi.aliyuncs.com',
|
||
)
|
||
return Client(config)
|
||
|
||
# ──────────────── 签名管理 ────────────────
|
||
|
||
def list_signs(args):
|
||
client = get_client()
|
||
from alibabacloud_dysmsapi20170525.models import QuerySmsSignListRequest
|
||
req = QuerySmsSignListRequest(page_index=args.page, page_size=args.size)
|
||
resp = client.query_sms_sign_list(req)
|
||
body = resp.body
|
||
if body.code != 'OK':
|
||
print(f'ERROR: {body.code} - {body.message}')
|
||
return
|
||
signs = body.sms_sign_list or []
|
||
print(f'\n📋 SMS 签名列表 (共 {len(signs)} 个)\n')
|
||
print(f'{"签名名称":<20} {"审核状态":<14} {"创建日期":<22} {"业务类型"}')
|
||
print('-' * 80)
|
||
status_map = {
|
||
'AUDIT_STATE_INIT': '审核中',
|
||
'AUDIT_STATE_PASS': '✅ 已通过',
|
||
'AUDIT_STATE_NOT_PASS': '❌ 被驳回',
|
||
'AUDIT_STATE_CANCEL': '已取消',
|
||
}
|
||
for s in signs:
|
||
st = status_map.get(s.audit_status, str(s.audit_status))
|
||
biz = s.business_type or ''
|
||
print(f'{s.sign_name:<20} {st:<14} {s.create_date:<22} {biz}')
|
||
|
||
def create_sign(args):
|
||
client = get_client()
|
||
from alibabacloud_dysmsapi20170525.models import AddSmsSignRequest
|
||
req = AddSmsSignRequest(
|
||
sign_name=args.name,
|
||
sign_source=args.source,
|
||
remark=args.remark,
|
||
)
|
||
resp = client.add_sms_sign(req)
|
||
body = resp.body
|
||
if body.code == 'OK':
|
||
print(f'✅ 签名 "{args.name}" 提交成功，等待审核')
|
||
else:
|
||
print(f'ERROR: {body.code} - {body.message}')
|
||
|
||
def delete_sign(args):
|
||
client = get_client()
|
||
from alibabacloud_dysmsapi20170525.models import DeleteSmsSignRequest
|
||
req = DeleteSmsSignRequest(sign_name=args.name)
|
||
resp = client.delete_sms_sign(req)
|
||
body = resp.body
|
||
if body.code == 'OK':
|
||
print(f'✅ 签名 "{args.name}" 已删除')
|
||
else:
|
||
print(f'ERROR: {body.code} - {body.message}')
|
||
|
||
# ──────────────── 模板管理 ────────────────
|
||
|
||
def list_templates(args):
|
||
client = get_client()
|
||
from alibabacloud_dysmsapi20170525.models import QuerySmsTemplateListRequest
|
||
req = QuerySmsTemplateListRequest(page_index=args.page, page_size=args.size)
|
||
resp = client.query_sms_template_list(req)
|
||
body = resp.body
|
||
if body.code != 'OK':
|
||
print(f'ERROR: {body.code} - {body.message}')
|
||
return
|
||
templates = body.sms_template_list or []
|
||
print(f'\n📋 SMS 模板列表 (共 {len(templates)} 个)\n')
|
||
print(f'{"模板Code":<20} {"模板名称":<20} {"类型":<8} {"审核状态":<14} {"内容"}')
|
||
print('-' * 100)
|
||
status_map = {
|
||
'AUDIT_STATE_INIT': '审核中',
|
||
'AUDIT_STATE_PASS': '✅ 通过',
|
||
'AUDIT_STATE_NOT_PASS': '❌ 驳回',
|
||
'AUDIT_STATE_CANCEL': '已取消',
|
||
}
|
||
type_map = {0: '验证码', 1: '通知', 2: '推广', 3: '国际', 6: '数字短信', 7: '其他'}
|
||
for t in templates:
|
||
st = status_map.get(t.audit_status, str(t.audit_status))
|
||
tp = type_map.get(t.template_type, str(t.template_type))
|
||
content = (t.template_content or '')[:40]
|
||
print(f'{t.template_code:<20} {t.template_name:<20} {tp:<8} {st:<14} {content}')
|
||
|
||
def create_template(args):
|
||
client = get_client()
|
||
# Use new CreateSmsTemplate API (AddSmsTemplate is deprecated)
|
||
from alibabacloud_dysmsapi20170525.models import CreateSmsTemplateRequest
|
||
sign_name = getattr(args, 'sign', None) or os.environ.get('ALIYUN_SMS_SIGN_NAME', '')
|
||
rule = getattr(args, 'rule', None) or ''
|
||
req = CreateSmsTemplateRequest(
|
||
template_type=args.type,
|
||
template_name=args.name,
|
||
template_content=args.content,
|
||
remark=args.remark,
|
||
related_sign_name=sign_name,
|
||
template_rule=rule if rule else None,
|
||
)
|
||
resp = client.create_sms_template(req)
|
||
body = resp.body
|
||
if body.code == 'OK':
|
||
code = getattr(body, 'template_code', None) or getattr(body, 'order_id', 'N/A')
|
||
print(f'✅ 模板 "{args.name}" 提交成功 (Code/Order: {code})，等待审核')
|
||
else:
|
||
print(f'ERROR: {body.code} - {body.message}')
|
||
|
||
def delete_template(args):
|
||
client = get_client()
|
||
from alibabacloud_dysmsapi20170525.models import DeleteSmsTemplateRequest
|
||
req = DeleteSmsTemplateRequest(template_code=args.code)
|
||
resp = client.delete_sms_template(req)
|
||
body = resp.body
|
||
if body.code == 'OK':
|
||
print(f'✅ 模板 {args.code} 已删除')
|
||
else:
|
||
print(f'ERROR: {body.code} - {body.message}')
|
||
|
||
# ──────────────── 发送记录查询 ────────────────
|
||
|
||
def query_send_details(args):
|
||
client = get_client()
|
||
from alibabacloud_dysmsapi20170525.models import QuerySendDetailsRequest
|
||
phone = args.phone.lstrip('+86')
|
||
send_date = args.date or datetime.now().strftime('%Y%m%d')
|
||
req = QuerySendDetailsRequest(
|
||
phone_number=phone,
|
||
send_date=send_date,
|
||
page_size=args.size,
|
||
current_page=args.page,
|
||
)
|
||
resp = client.query_send_details(req)
|
||
body = resp.body
|
||
if body.code != 'OK':
|
||
print(f'ERROR: {body.code} - {body.message}')
|
||
return
|
||
details = body.sms_send_detail_dtos.sms_send_detail_dto if body.sms_send_detail_dtos else []
|
||
print(f'\n📋 发送记录 (手机: {args.phone}, 日期: {send_date}, 共 {body.total_count} 条)\n')
|
||
print(f'{"发送时间":<22} {"状态":<8} {"模板Code":<18} {"内容"}')
|
||
print('-' * 90)
|
||
status_map = {1: '等待', 2: '失败', 3: '✅ 成功'}
|
||
for d in details:
|
||
st = status_map.get(d.send_status, str(d.send_status))
|
||
content = (d.content or '')[:40]
|
||
print(f'{d.send_date:<22} {st:<8} {d.template_code:<18} {content}')
|
||
|
||
# ──────────────── 发送统计 ────────────────
|
||
|
||
def query_send_statistics(args):
|
||
client = get_client()
|
||
from alibabacloud_dysmsapi20170525.models import QuerySendStatisticsRequest
|
||
today = datetime.now()
|
||
start = args.start or (today - timedelta(days=args.days)).strftime('%Y%m%d')
|
||
end = args.end or today.strftime('%Y%m%d')
|
||
req = QuerySendStatisticsRequest(
|
||
is_globe=1,
|
||
start_date=start,
|
||
end_date=end,
|
||
page_index=1,
|
||
page_size=50,
|
||
)
|
||
resp = client.query_send_statistics(req)
|
||
body = resp.body
|
||
if body.code != 'OK':
|
||
print(f'ERROR: {body.code} - {body.message}')
|
||
return
|
||
data = body.data
|
||
items = data.target_list or [] if data else []
|
||
total_sent = sum(i.total_count for i in items)
|
||
total_success = sum(i.responded_success_count for i in items)
|
||
total_fail = sum(i.responded_fail_count for i in items)
|
||
print(f'\n📋 SMS 发送统计 ({start} ~ {end})\n')
|
||
print(f'{"日期":<12} {"发送":<8} {"成功":<8} {"失败":<8}')
|
||
print('-' * 40)
|
||
for i in items:
|
||
print(f'{i.send_date:<12} {i.total_count:<8} {i.responded_success_count:<8} {i.responded_fail_count:<8}')
|
||
print('-' * 40)
|
||
print(f'{"合计":<12} {total_sent:<8} {total_success:<8} {total_fail:<8}')
|
||
print(f'\n 注: SMS 为按量付费，无资源包余额概念')
|
||
|
||
# ──────────────── 发送测试 ────────────────
|
||
|
||
def send_sms(args):
|
||
client = get_client()
|
||
from alibabacloud_dysmsapi20170525.models import SendSmsRequest
|
||
from alibabacloud_tea_util.models import RuntimeOptions
|
||
phone = args.phone
|
||
if not phone.startswith('+'):
|
||
phone = '+86' + phone.lstrip('0')
|
||
req = SendSmsRequest(
|
||
phone_numbers=phone,
|
||
sign_name=args.sign or os.environ.get('ALIYUN_SMS_SIGN_NAME', ''),
|
||
template_code=args.template or os.environ.get('ALIYUN_SMS_TEMPLATE_CODE', ''),
|
||
template_param=json.dumps({'code': args.code}) if args.code else None,
|
||
)
|
||
runtime = RuntimeOptions(connect_timeout=15000, read_timeout=15000)
|
||
resp = client.send_sms_with_options(req, runtime)
|
||
body = resp.body
|
||
if body.code == 'OK':
|
||
print(f'✅ 短信发送成功: phone={phone} bizId={body.biz_id}')
|
||
else:
|
||
print(f'ERROR: {body.code} - {body.message}')
|
||
|
||
# ──────────────── CLI ────────────────
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description='阿里云 SMS 管理工具')
|
||
sub = parser.add_subparsers(dest='command')
|
||
|
||
# 签名
|
||
p = sub.add_parser('list-signs', help='列出所有短信签名')
|
||
p.add_argument('--page', type=int, default=1)
|
||
p.add_argument('--size', type=int, default=50)
|
||
|
||
p = sub.add_parser('create-sign', help='创建短信签名')
|
||
p.add_argument('--name', required=True, help='签名名称')
|
||
p.add_argument('--source', type=int, default=0, help='来源: 0=企事业 1=工信部 2=商标 3=APP 4=网站 5=公众号 6=小程序 7=电商')
|
||
p.add_argument('--remark', default='', help='备注说明')
|
||
|
||
p = sub.add_parser('delete-sign', help='删除短信签名')
|
||
p.add_argument('--name', required=True, help='签名名称')
|
||
|
||
# 模板
|
||
p = sub.add_parser('list-templates', help='列出所有短信模板')
|
||
p.add_argument('--page', type=int, default=1)
|
||
p.add_argument('--size', type=int, default=50)
|
||
|
||
p = sub.add_parser('create-template', help='创建短信模板')
|
||
p.add_argument('--name', required=True, help='模板名称')
|
||
p.add_argument('--content', required=True, help='模板内容，如: 您的验证码为${code}')
|
||
p.add_argument('--type', type=int, default=0, help='类型: 0=验证码 1=通知 2=推广')
|
||
p.add_argument('--remark', default='', help='备注说明')
|
||
p.add_argument('--sign', default='', help='关联签名名称（新API必填，默认读ALIYUN_SMS_SIGN_NAME）')
|
||
p.add_argument('--rule', default='', help='变量规则JSON，如: {"code":"numberCaptcha"}')
|
||
|
||
p = sub.add_parser('delete-template', help='删除短信模板')
|
||
p.add_argument('--code', required=True, help='模板Code')
|
||
|
||
# 发送记录
|
||
p = sub.add_parser('query', help='查询发送记录')
|
||
p.add_argument('--phone', required=True, help='手机号')
|
||
p.add_argument('--date', help='日期 YYYYMMDD (默认今天)')
|
||
p.add_argument('--page', type=int, default=1)
|
||
p.add_argument('--size', type=int, default=20)
|
||
|
||
# 发送统计
|
||
p = sub.add_parser('stats', help='查询发送统计')
|
||
p.add_argument('--days', type=int, default=30, help='最近N天 (默认30)')
|
||
p.add_argument('--start', help='起始日期 YYYYMMDD')
|
||
p.add_argument('--end', help='结束日期 YYYYMMDD')
|
||
|
||
# 发送测试
|
||
p = sub.add_parser('send', help='发送短信 (测试)')
|
||
p.add_argument('--phone', required=True, help='手机号')
|
||
p.add_argument('--sign', help='签名 (默认取环境变量)')
|
||
p.add_argument('--template', help='模板Code (默认取环境变量)')
|
||
p.add_argument('--code', help='验证码内容')
|
||
|
||
args = parser.parse_args()
|
||
if not args.command:
|
||
parser.print_help()
|
||
return
|
||
|
||
cmds = {
|
||
'list-signs': list_signs,
|
||
'create-sign': create_sign,
|
||
'delete-sign': delete_sign,
|
||
'list-templates': list_templates,
|
||
'create-template': create_template,
|
||
'delete-template': delete_template,
|
||
'query': query_send_details,
|
||
'stats': query_send_statistics,
|
||
'send': send_sms,
|
||
}
|
||
try:
|
||
cmds[args.command](args)
|
||
except Exception as e:
|
||
err = str(e)
|
||
if 'NoPermission' in err or '403' in err:
|
||
print(f'\n❌ 权限不足: RAM 子账号缺少对应 API 权限')
|
||
print(f' 请在阿里云 RAM 控制台添加 AliyunDysmsFullAccess 策略')
|
||
print(f' 控制台: https://ram.console.aliyun.com')
|
||
else:
|
||
print(f'ERROR: {e}')
|
||
|
||
if __name__ == '__main__':
|
||
main()
|