feat(scripts): 阿里云管理工具 (SMS/CloudAuth/Domain)
- aliyun_sms_manager.py: 签名/模板管理、发送记录查询、发送统计、发送测试 - aliyun_cloudauth_manager.py: 实人认证场景、人脸活体检测、身份二要素核验、套餐余额查询 - aliyun_domain_manager.py: 域名列表/详情/续费/转出/信息模板管理 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
34b85f68ae
commit
8af65a3a48
|
|
@ -0,0 +1,236 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
阿里云实人认证 (CloudAuth) 管理工具
|
||||
功能: 认证场景管理、发起认证、查询结果、套餐余额查询
|
||||
依赖: pip install alibabacloud-cloudauth20190307 alibabacloud-bssopenapi20171214
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
import os
|
||||
import uuid
|
||||
|
||||
def get_client():
|
||||
from alibabacloud_cloudauth20190307.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', '')
|
||||
endpoint = os.environ.get('ALIYUN_KYC_ENDPOINT', 'cloudauth.aliyuncs.com')
|
||||
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=endpoint,
|
||||
)
|
||||
return Client(config)
|
||||
|
||||
# ──────────────── 认证场景 ────────────────
|
||||
|
||||
def describe_verify_setting(args):
|
||||
"""查询认证场景配置 (通过探测 API 检查服务状态)"""
|
||||
client = get_client()
|
||||
from alibabacloud_cloudauth20190307.models import InitFaceVerifyRequest
|
||||
scene_id = args.scene_id or os.environ.get('ALIYUN_KYC_SCENE_ID', '')
|
||||
try:
|
||||
# Probe service via InitFaceVerify with dummy data
|
||||
req = InitFaceVerifyRequest(
|
||||
scene_id=int(scene_id) if scene_id else 0,
|
||||
outer_order_no=str(uuid.uuid4()),
|
||||
product_code='ID_PRO',
|
||||
cert_type='IDENTITY_CARD',
|
||||
cert_name='probe',
|
||||
cert_no='000000000000000000',
|
||||
)
|
||||
resp = client.init_face_verify(req)
|
||||
body = resp.body
|
||||
print(f'\n📋 实人认证服务状态\n')
|
||||
print(f' 场景ID: {scene_id or "(未设置)"}')
|
||||
print(f' 产品码: ID_PRO (人脸活体检测)')
|
||||
print(f' 请求ID: {body.request_id}')
|
||||
print(f' 状态: ✅ 服务可用 (权限正常)')
|
||||
if body.code and body.code != '200':
|
||||
print(f' 探测响应: {body.code} - {body.message or ""}')
|
||||
except Exception as e:
|
||||
err = str(e)
|
||||
if 'NoPermission' in err or '403' in err:
|
||||
print(f'\n❌ 权限不足: RAM 子账号缺少 CloudAuth 权限')
|
||||
print(f' 请添加 AliyunYundunCloudAuthFullAccess 策略')
|
||||
else:
|
||||
print(f'ERROR: {e}')
|
||||
|
||||
# ──────────────── 发起认证 ────────────────
|
||||
|
||||
def init_face_verify(args):
|
||||
"""发起人脸活体检测认证"""
|
||||
client = get_client()
|
||||
from alibabacloud_cloudauth20190307.models import InitFaceVerifyRequest
|
||||
scene_id = args.scene_id or os.environ.get('ALIYUN_KYC_SCENE_ID', '')
|
||||
outer_order_no = args.order_no or str(uuid.uuid4())
|
||||
req = InitFaceVerifyRequest(
|
||||
scene_id=int(scene_id),
|
||||
outer_order_no=outer_order_no,
|
||||
product_code='ID_PRO',
|
||||
cert_type='IDENTITY_CARD',
|
||||
cert_name=args.name,
|
||||
cert_no=args.id_number,
|
||||
)
|
||||
try:
|
||||
resp = client.init_face_verify(req)
|
||||
body = resp.body
|
||||
if body.code == '200' or body.code == 200:
|
||||
result = body.result_object
|
||||
print(f'✅ 认证已发起')
|
||||
print(f' CertifyId: {result.certify_id}')
|
||||
print(f' CertifyUrl: {result.certify_url}')
|
||||
print(f' OrderNo: {outer_order_no}')
|
||||
else:
|
||||
print(f'ERROR: {body.code} - {body.message}')
|
||||
except Exception as e:
|
||||
print(f'ERROR: {e}')
|
||||
|
||||
def describe_face_verify(args):
|
||||
"""查询认证结果"""
|
||||
client = get_client()
|
||||
from alibabacloud_cloudauth20190307.models import DescribeFaceVerifyRequest
|
||||
req = DescribeFaceVerifyRequest(
|
||||
scene_id=int(args.scene_id or os.environ.get('ALIYUN_KYC_SCENE_ID', '')),
|
||||
certify_id=args.certify_id,
|
||||
)
|
||||
try:
|
||||
resp = client.describe_face_verify(req)
|
||||
body = resp.body
|
||||
if body.code == '200' or body.code == 200:
|
||||
result = body.result_object
|
||||
passed = result.passed == 'T'
|
||||
print(f'\n📋 认证结果')
|
||||
print(f' CertifyId: {args.certify_id}')
|
||||
print(f' 通过: {"✅ 是" if passed else "❌ 否"}')
|
||||
print(f' SubCode: {result.sub_code}')
|
||||
if result.material_info:
|
||||
print(f' 材料信息: {result.material_info}')
|
||||
else:
|
||||
print(f'ERROR: {body.code} - {body.message}')
|
||||
except Exception as e:
|
||||
print(f'ERROR: {e}')
|
||||
|
||||
# ──────────────── 身份二要素核验 ────────────────
|
||||
|
||||
def verify_identity(args):
|
||||
"""身份证二要素核验 (姓名+身份证号)"""
|
||||
client = get_client()
|
||||
from alibabacloud_cloudauth20190307.models import Id2MetaVerifyRequest
|
||||
req = Id2MetaVerifyRequest(
|
||||
param_type='normal',
|
||||
user_name=args.name,
|
||||
identity_card_number=args.id_number,
|
||||
)
|
||||
try:
|
||||
resp = client.id_2meta_verify(req)
|
||||
body = resp.body
|
||||
if body.code == '200' or body.code == 200:
|
||||
result = body.result_object
|
||||
bizCode = result.biz_code if result else 'UNKNOWN'
|
||||
passed = bizCode == '1'
|
||||
print(f'\n📋 身份二要素核验')
|
||||
print(f' 姓名: {args.name}')
|
||||
print(f' 身份证: {args.id_number[:6]}****{args.id_number[-4:]}')
|
||||
print(f' 结果: {"✅ 一致" if passed else "❌ 不一致"}')
|
||||
print(f' BizCode: {bizCode}')
|
||||
else:
|
||||
print(f'ERROR: {body.code} - {body.message}')
|
||||
except Exception as e:
|
||||
print(f'ERROR: {e}')
|
||||
|
||||
# ──────────────── 套餐余额查询 ────────────────
|
||||
|
||||
def query_quota(args):
|
||||
"""查询实人认证套餐余额 (通过 BSS API)"""
|
||||
from alibabacloud_bssopenapi20171214.client import Client as BssClient
|
||||
from alibabacloud_bssopenapi20171214.models import QueryResourcePackageInstancesRequest
|
||||
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', '')
|
||||
config = Config(
|
||||
access_key_id=ak_id,
|
||||
access_key_secret=ak_secret,
|
||||
endpoint='business.aliyuncs.com',
|
||||
)
|
||||
client = BssClient(config)
|
||||
try:
|
||||
req = QueryResourcePackageInstancesRequest(page_num=1, page_size=100)
|
||||
resp = client.query_resource_package_instances(req)
|
||||
body = resp.body
|
||||
instances = body.data.instances.instance if body.data and body.data.instances else []
|
||||
# Filter cloudauth packages
|
||||
auth_pkgs = [i for i in instances if 'cloudauth' in (i.to_map().get('CommodityCode', '') or '').lower()]
|
||||
if not auth_pkgs:
|
||||
print('\n📋 未找到实人认证相关套餐')
|
||||
return
|
||||
print(f'\n📋 实人认证套餐余额 (共 {len(auth_pkgs)} 个)\n')
|
||||
print(f'{"套餐名称":<30} {"总量":<10} {"剩余":<10} {"到期日":<22} {"状态"}')
|
||||
print('-' * 90)
|
||||
for pkg in auth_pkgs:
|
||||
m = pkg.to_map()
|
||||
name = m.get('Remark', '') or m.get('PackageType', '')
|
||||
total = m.get('TotalAmount', '0')
|
||||
remain = m.get('RemainingAmount', '0')
|
||||
unit = m.get('TotalAmountUnit', '次')
|
||||
expiry = (m.get('ExpiryTime', '') or '')[:10]
|
||||
status = m.get('Status', '')
|
||||
status_str = '✅ 可用' if status == 'Available' else f'❌ {status}'
|
||||
print(f'{name:<30} {total}{unit:<8} {remain}{unit:<8} {expiry:<22} {status_str}')
|
||||
except Exception as e:
|
||||
err = str(e)
|
||||
if 'NotAuthorized' in err:
|
||||
print(f'\n❌ 权限不足: 需要 AliyunBSSReadOnlyAccess 策略')
|
||||
else:
|
||||
print(f'ERROR: {e}')
|
||||
|
||||
# ──────────────── CLI ────────────────
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description='阿里云实人认证管理工具')
|
||||
sub = parser.add_subparsers(dest='command')
|
||||
|
||||
# 场景查询
|
||||
p = sub.add_parser('list-scenes', help='查询认证场景配置')
|
||||
p.add_argument('--scene-id', help='场景ID (留空查全部)')
|
||||
|
||||
# 发起人脸认证
|
||||
p = sub.add_parser('init-face', help='发起人脸活体检测')
|
||||
p.add_argument('--name', required=True, help='真实姓名')
|
||||
p.add_argument('--id-number', required=True, help='身份证号')
|
||||
p.add_argument('--scene-id', help='场景ID (默认取环境变量)')
|
||||
p.add_argument('--order-no', help='业务订单号 (默认自动生成)')
|
||||
|
||||
# 查询认证结果
|
||||
p = sub.add_parser('query-face', help='查询人脸认证结果')
|
||||
p.add_argument('--certify-id', required=True, help='认证ID')
|
||||
p.add_argument('--scene-id', help='场景ID')
|
||||
|
||||
# 套餐余额
|
||||
p = sub.add_parser('quota', help='查询套餐余额')
|
||||
|
||||
# 身份二要素核验
|
||||
p = sub.add_parser('verify-id', help='身份证二要素核验')
|
||||
p.add_argument('--name', required=True, help='真实姓名')
|
||||
p.add_argument('--id-number', required=True, help='身份证号')
|
||||
|
||||
args = parser.parse_args()
|
||||
if not args.command:
|
||||
parser.print_help()
|
||||
return
|
||||
|
||||
cmds = {
|
||||
'list-scenes': describe_verify_setting,
|
||||
'init-face': init_face_verify,
|
||||
'query-face': describe_face_verify,
|
||||
'verify-id': verify_identity,
|
||||
'quota': query_quota,
|
||||
}
|
||||
cmds[args.command](args)
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
|
@ -0,0 +1,206 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
阿里云域名管理工具
|
||||
功能: 域名列表、详情、续费、转出、过户、信息模板管理
|
||||
依赖: pip install alibabacloud-domain20180129
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
import os
|
||||
|
||||
def get_client():
|
||||
from alibabacloud_domain20180129.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='domain.aliyuncs.com',
|
||||
)
|
||||
return Client(config)
|
||||
|
||||
# ──────────────── 域名列表 ────────────────
|
||||
|
||||
def list_domains(args):
|
||||
client = get_client()
|
||||
from alibabacloud_domain20180129.models import QueryDomainListRequest
|
||||
req = QueryDomainListRequest(
|
||||
page_num=args.page,
|
||||
page_size=args.size,
|
||||
)
|
||||
try:
|
||||
resp = client.query_domain_list(req)
|
||||
body = resp.body
|
||||
domains = body.data.domain if body.data and body.data.domain else []
|
||||
print(f'\n📋 域名列表 (共 {body.total_item_num} 个)\n')
|
||||
print(f'{"域名":<30} {"到期日":<14} {"注册日":<14} {"状态"}')
|
||||
print('-' * 75)
|
||||
for d in domains:
|
||||
exp = d.expiration_date[:10] if d.expiration_date else '-'
|
||||
reg = d.registration_date[:10] if d.registration_date else '-'
|
||||
# 域名状态
|
||||
status = '正常'
|
||||
if hasattr(d, 'domain_status') and d.domain_status:
|
||||
status = d.domain_status
|
||||
print(f'{d.domain_name:<30} {exp:<14} {reg:<14} {status}')
|
||||
except Exception as e:
|
||||
print(f'ERROR: {e}')
|
||||
|
||||
# ──────────────── 域名详情 ────────────────
|
||||
|
||||
def domain_info(args):
|
||||
client = get_client()
|
||||
from alibabacloud_domain20180129.models import QueryDomainByDomainNameRequest
|
||||
req = QueryDomainByDomainNameRequest(domain_name=args.domain)
|
||||
try:
|
||||
resp = client.query_domain_by_domain_name(req)
|
||||
body = resp.body
|
||||
print(f'\n📋 域名详情: {args.domain}\n')
|
||||
print(f' 注册商: {body.registrant_organization or body.registrant_name or "-"}')
|
||||
print(f' 注册日: {body.registration_date or "-"}')
|
||||
print(f' 到期日: {body.expiration_date or "-"}')
|
||||
print(f' DNS: {", ".join(body.dns_list.dns) if body.dns_list else "-"}')
|
||||
print(f' 状态: {body.domain_status or "-"}')
|
||||
print(f' 实名认证: {"✅ 已认证" if body.zh_registrant_organization else "❌ 未认证"}')
|
||||
print(f' 转移锁: {"🔒 已锁定" if body.transfer_out_status else "🔓 未锁定"}')
|
||||
if body.email:
|
||||
print(f' 联系邮箱: {body.email}')
|
||||
except Exception as e:
|
||||
print(f'ERROR: {e}')
|
||||
|
||||
# ──────────────── 域名续费 ────────────────
|
||||
|
||||
def renew_domain(args):
|
||||
client = get_client()
|
||||
from alibabacloud_domain20180129.models import SaveSingleTaskForCreatingOrderRenewRequest
|
||||
req = SaveSingleTaskForCreatingOrderRenewRequest(
|
||||
domain_name=args.domain,
|
||||
subscription_duration=args.years,
|
||||
current_expiration_date=0, # will be auto-detected
|
||||
)
|
||||
try:
|
||||
resp = client.save_single_task_for_creating_order_renew(req)
|
||||
body = resp.body
|
||||
print(f'✅ 续费任务已提交: {args.domain} +{args.years}年')
|
||||
print(f' TaskNo: {body.task_no}')
|
||||
except Exception as e:
|
||||
print(f'ERROR: {e}')
|
||||
|
||||
# ──────────────── 域名转出 ────────────────
|
||||
|
||||
def transfer_out(args):
|
||||
"""获取域名转移密码"""
|
||||
client = get_client()
|
||||
from alibabacloud_domain20180129.models import TransferOutDomainRequest
|
||||
req = TransferOutDomainRequest(domain_name=args.domain)
|
||||
try:
|
||||
resp = client.transfer_out_domain(req)
|
||||
body = resp.body
|
||||
print(f'✅ 域名转出已发起: {args.domain}')
|
||||
print(f' 转移密码将发送至域名联系人邮箱')
|
||||
except Exception as e:
|
||||
print(f'ERROR: {e}')
|
||||
|
||||
# ──────────────── 信息模板管理 ────────────────
|
||||
|
||||
def list_templates(args):
|
||||
"""查询信息模板"""
|
||||
client = get_client()
|
||||
from alibabacloud_domain20180129.models import QueryRegistrantProfilesRequest
|
||||
req = QueryRegistrantProfilesRequest(
|
||||
page_num=args.page,
|
||||
page_size=args.size,
|
||||
)
|
||||
try:
|
||||
resp = client.query_registrant_profiles(req)
|
||||
body = resp.body
|
||||
profiles = body.registrant_profiles.registrant_profile if body.registrant_profiles else []
|
||||
print(f'\n📋 信息模板列表 (共 {body.total_item_num} 个)\n')
|
||||
print(f'{"模板ID":<12} {"联系人":<20} {"实名认证":<10} {"邮箱":<30} {"类型"}')
|
||||
print('-' * 90)
|
||||
for p in profiles:
|
||||
verified = '✅ 已认证' if p.email_verification_status == 1 else '❌ 未认证'
|
||||
rtype = '企业' if p.registrant_type == 'e' else '个人'
|
||||
name = p.zh_registrant_organization or p.registrant_name or '-'
|
||||
print(f'{p.registrant_profile_id:<12} {name:<20} {verified:<10} {p.email or "-":<30} {rtype}')
|
||||
except Exception as e:
|
||||
print(f'ERROR: {e}')
|
||||
|
||||
# ──────────────── 任务查询 ────────────────
|
||||
|
||||
def query_task(args):
|
||||
"""查询任务详情"""
|
||||
client = get_client()
|
||||
from alibabacloud_domain20180129.models import QueryTaskDetailListRequest
|
||||
req = QueryTaskDetailListRequest(
|
||||
task_no=args.task_no,
|
||||
page_num=1,
|
||||
page_size=20,
|
||||
)
|
||||
try:
|
||||
resp = client.query_task_detail_list(req)
|
||||
body = resp.body
|
||||
items = body.data.task_detail if body.data else []
|
||||
print(f'\n📋 任务详情: {args.task_no}\n')
|
||||
for item in items:
|
||||
print(f' 域名: {item.domain_name}')
|
||||
print(f' 状态: {item.task_status}')
|
||||
print(f' 结果: {item.task_result or "-"}')
|
||||
except Exception as e:
|
||||
print(f'ERROR: {e}')
|
||||
|
||||
# ──────────────── CLI ────────────────
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description='阿里云域名管理工具')
|
||||
sub = parser.add_subparsers(dest='command')
|
||||
|
||||
# 域名列表
|
||||
p = sub.add_parser('list', help='列出所有域名')
|
||||
p.add_argument('--page', type=int, default=1)
|
||||
p.add_argument('--size', type=int, default=20)
|
||||
|
||||
# 域名详情
|
||||
p = sub.add_parser('info', help='查询域名详情')
|
||||
p.add_argument('--domain', required=True, help='域名')
|
||||
|
||||
# 续费
|
||||
p = sub.add_parser('renew', help='域名续费')
|
||||
p.add_argument('--domain', required=True, help='域名')
|
||||
p.add_argument('--years', type=int, default=1, help='续费年数')
|
||||
|
||||
# 转出
|
||||
p = sub.add_parser('transfer-out', help='域名转出 (获取转移密码)')
|
||||
p.add_argument('--domain', 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=20)
|
||||
|
||||
# 任务查询
|
||||
p = sub.add_parser('query-task', help='查询任务状态')
|
||||
p.add_argument('--task-no', required=True, help='任务编号')
|
||||
|
||||
args = parser.parse_args()
|
||||
if not args.command:
|
||||
parser.print_help()
|
||||
return
|
||||
|
||||
cmds = {
|
||||
'list': list_domains,
|
||||
'info': domain_info,
|
||||
'renew': renew_domain,
|
||||
'transfer-out': transfer_out,
|
||||
'list-templates': list_templates,
|
||||
'query-task': query_task,
|
||||
}
|
||||
cmds[args.command](args)
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
|
@ -0,0 +1,301 @@
|
|||
#!/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()
|
||||
from alibabacloud_dysmsapi20170525.models import AddSmsTemplateRequest
|
||||
req = AddSmsTemplateRequest(
|
||||
template_type=args.type,
|
||||
template_name=args.name,
|
||||
template_content=args.content,
|
||||
remark=args.remark,
|
||||
)
|
||||
resp = client.add_sms_template(req)
|
||||
body = resp.body
|
||||
if body.code == 'OK':
|
||||
print(f'✅ 模板 "{args.name}" 提交成功 (Code: {body.template_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 = 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()
|
||||
Loading…
Reference in New Issue