charles/smssend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

更新配置

Browse Source
This commit is contained in:
ping 2026-03-19 18:04:20 +08:00
parent 4aecf56050
commit 7d127c873c
7 changed files with 19 additions and 405 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
conf/config.json
View File

@ -1,8 +0,0 @@
{
"smsConfig": {
"baidu_sms_access_key": "ALTAKPk92fX9cgGDax83yNL8mP",
"baidu_sms_access_key_secret": "9b16b8efd4dc463d8bbd5462db1db8a5",
"baidu_sms_host": "sms.bj.baidubce.com",
"baidu_sms_signature_id": "sms-sign-xQYUwp42637"
}
}

37
models/table_create.txt
View File

@ -1,37 +0,0 @@
CREATE TABLE `sms_template` (
`id` varchar(32) NOT NULL COMMENT 'id',
`name` varchar(32) DEFAULT NULL COMMENT '模板名称',
`template_type` varchar(32) DEFAULT NULL COMMENT '模板类型',
`code` varchar(32) DEFAULT NULL COMMENT '模板编码',
`content` varchar(200) DEFAULT NULL COMMENT '模板内容',
`description` varchar(100) DEFAULT NULL COMMENT '场景说明',
`provider` varchar(100) DEFAULT NULL COMMENT '短信供应商名称',
`del_flg` varchar(1) DEFAULT '0' COMMENT '删除标志',
`create_at` timestamp NULL DEFAULT current_timestamp() COMMENT '创建时间戳',
PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_general_ci ROW_FORMAT=DYNAMIC COMMENT='短信模板表';
CREATE TABLE `validatecode` (
`id` varchar(32) NOT NULL COMMENT 'id',
`vcode` varchar(32) DEFAULT NULL COMMENT '验证码',
`expire_time` timestamp NOT NULL DEFAULT current_timestamp() COMMENT '有效期',
`del_flg` varchar(1) DEFAULT '0' COMMENT '删除标志',
`create_at` timestamp NOT NULL DEFAULT current_timestamp() COMMENT '创建时间戳',
PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_general_ci ROW_FORMAT=DYNAMIC COMMENT='验证码表';
CREATE TABLE `sms_record` (
`id` varchar(32) NOT NULL COMMENT 'id',
`customerid` varchar(32) DEFAULT NULL COMMENT '客户id',
`send_type` varchar(32) DEFAULT NULL COMMENT '发送类型',
`mobile` varchar(15) DEFAULT NULL COMMENT '手机号',
`email` varchar(32) DEFAULT NULL COMMENT '邮箱',
`message` varchar(510) DEFAULT NULL COMMENT '发送内容',
`send_time` varchar(32) DEFAULT NULL COMMENT '发送时间',
`send_status` varchar(1) DEFAULT NULL COMMENT '发送状态',
`task_status` varchar(2) DEFAULT NULL COMMENT '任务状态 1:待发送,2:待定时发送,11:发送MQ成功,12:MQ消费成功,21:请求成功,22:请求失败,31:取消发送',
`remark` varchar(200) DEFAULT NULL COMMENT '发送失败备注信息',
`del_flg` varchar(1) DEFAULT NULL COMMENT '删除标志',
`create_at` timestamp NULL DEFAULT current_timestamp() COMMENT '创建时间戳',
PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_general_ci ROW_FORMAT=DYNAMIC COMMENT='短信邮箱记录表';

11
smssend/init.py
View File

@ -7,17 +7,9 @@ SMS Send Module Init
- load_smssend() 加载模块 - load_smssend() 加载模块
- 将功能注册到 ServerEnv - 将功能注册到 ServerEnv
""" """
import sys
sys.path.append('D:/Code/dagflow_demo/')
import asyncio import asyncio
from functools import partial from appPublic.log import info
from appPublic.jsonConfig import getConfig
from appPublic.log import debug, exception, error, info
from ahserver.serverenv import ServerEnv from ahserver.serverenv import ServerEnv
from ahserver.configuredServer import add_cleanupctx
import random
import string
from .smssend import ( from .smssend import (
SMSEngine, SMSEngine,
@ -54,7 +46,6 @@ async def smsbacktask(engine, app):
def load_smssend(): def load_smssend():
config = getConfig()
global _sms_engine_instance global _sms_engine_instance
_sms_engine_instance = SMSEngine() _sms_engine_instance = SMSEngine()
info(f'SMS Send Module loaded, signature_id: {_sms_engine_instance.signature_id}') info(f'SMS Send Module loaded, signature_id: {_sms_engine_instance.signature_id}')

18
smssend/smssend.py
View File

@ -70,11 +70,19 @@ class SMSEngine:
doc = "https://cloud.baidu.com/doc/SMS/s/zjwvxry6e" doc = "https://cloud.baidu.com/doc/SMS/s/zjwvxry6e"
def __init__(self): def __init__(self):
config = getConfig() self.access_key = os.getenv('BAIDU_SMS_ACCESS_KEY')
self.access_key = config.smsConfig.baidu_sms_access_key self.access_key_secret = os.getenv('BAIDU_SMS_ACCESS_KEY_SECRET')
self.access_key_secret = config.smsConfig.baidu_sms_access_key_secret self.host = os.getenv('BAIDU_SMS_HOST')
self.host = config.smsConfig.baidu_sms_host self.signature_id = os.getenv('BAIDU_SMS_SIGNATURE_ID')
self.signature_id = config.smsConfig.baidu_sms_signature_id
# 检查必需的环境变量是否都已设置，若缺失则抛出异常
required_vars = [self.access_key, self.access_key_secret, self.host, self.signature_id]
if not all(required_vars):
missing = [name for name, val in zip(
['BAIDU_SMS_ACCESS_KEY', 'BAIDU_SMS_ACCESS_KEY_SECRET', 'BAIDU_SMS_HOST', 'BAIDU_SMS_SIGNATURE_ID'],
required_vars) if val is None]
raise EnvironmentError(f"Missing required environment variables: {', '.join(missing)}")
self.sms_client = self.create_client() self.sms_client = self.create_client()
self.sms_types = {} self.sms_types = {}

7
test/pwd_encry.py
View File

@ -1,7 +0,0 @@
from appPublic.aes import aes_encode_b64
import sys
encry_key = '!@#$%^&*(*&^%$QWERTYUIqwertyui234567'
pwd = 'lima2018'
code = aes_encode_b64(encry_key, pwd)
print(code)

14
test/test_done.py → test/test.py
View File

@ -14,12 +14,8 @@ SMS Send Module Test
import asyncio import asyncio
from smssend import ( from smssend import (
load_smssend, load_smssend,
generate_sms_code,
check_sms_code, check_sms_code,
send_vcode, send_sms
send_sms,
get_sms_engine,
SMSEngine
) )
async def test_check_sms_code(): async def test_check_sms_code():
@ -37,8 +33,8 @@ async def test_check_sms_code():
return False return False
async def test_send_sms(): async def test_send_sms():
phone = "18053191417" phone = ""
stype = "用户注册登录验证" # 充值提醒/欠费短信通知等 stype = "注册登录" # 充值提醒/欠费短信通知等
result = await send_sms(phone, stype) # 需要引用的参数 phone和stype必填 result = await send_sms(phone, stype) # 需要引用的参数 phone和stype必填
print(f"发送结果: {result}") print(f"发送结果: {result}")
@ -77,14 +73,14 @@ async def main():
from ahserver.serverenv import ServerEnv from ahserver.serverenv import ServerEnv
env = ServerEnv() env = ServerEnv()
def get_db_name(s=None): return 'kboss' def get_db_name(s=None): return 'database_name'
env.get_module_dbname = get_db_name env.get_module_dbname = get_db_name
if __name__ == '__main__': if __name__ == '__main__':
from sqlor.dbpools import DBPools from sqlor.dbpools import DBPools
from appPublic.jsonConfig import getConfig from appPublic.jsonConfig import getConfig
p = 'D:/Code/backend_code/test_/' p = ''
config = getConfig(p) config = getConfig(p)
DBPools(config.databases) DBPools(config.databases)
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()

329
test/test_smssend.py
View File

@ -1,329 +0,0 @@
# -*- coding: utf-8 -*-
"""
SMS Send Module Test
====================
测试用例覆盖所有功能接口：
1. generate_sms_code - 生成验证码入库并返回codeid
2. check_sms_code - 验证码校验
3. send_vcode - 发送验证码短信
4. send_sms - 发送短信（整合生成和发送）
5. get_sms_engine - 获取引擎实例
6. SMSEngine - 引擎类方法
"""
import sys
sys.path.append('D:/Code/dagflow_demo/smssend')
import asyncio
from smssend import (
load_smssend,
generate_sms_code,
check_sms_code,
send_vcode,
send_sms,
get_sms_engine,
SMSEngine
)
async def get_vcode_from_db(codeid: str) -> dict:
from sqlor.dbpools import DBPools
db = DBPools()
async with db.sqlorContext('kboss') as sor:
result = await sor.R('validatecode', {'id': codeid, 'del_flg': '0'})
if result:
return result[0]
return None
async def test_get_sms_engine():
print("=" * 50)
print("测试1: get_sms_engine() - 获取引擎实例")
print("=" * 50)
engine = get_sms_engine()
print(f"引擎实例: {engine}")
print(f"引擎类型: {type(engine)}")
if engine is not None and isinstance(engine, SMSEngine):
print("✅ 获取引擎实例成功")
return True
else:
print("❌ 获取引擎实例失败")
return False
async def test_generate_sms_code():
print("\n" + "=" * 50)
print("测试2: generate_sms_code() - 生成验证码入库")
print("=" * 50)
codeid = await generate_sms_code()
print(f"生成的验证码ID: {codeid}")
if not codeid:
print("❌ 生成验证码失败")
return False, None
print("✅ 生成验证码成功")
return True, codeid
async def test_generate_sms_code_with_params():
print("\n" + "=" * 50)
print("测试3: generate_sms_code(自定义长度和有效期) - 生成验证码")
print("=" * 50)
codeid = await generate_sms_code(length=4, expire_minutes=10)
print(f"生成的验证码ID: {codeid}")
print(f"验证码长度: 4, 有效期: 10分钟")
if not codeid:
print("❌ 生成验证码失败")
return False, None
db_result = await get_vcode_from_db(codeid)
if db_result:
vcode = db_result['vcode']
print(f"数据库中的验证码: {vcode}")
if len(vcode) == 4:
print("✅ 自定义参数生成验证码成功")
return True, codeid
print("❌ 自定义参数验证失败")
return False, None
async def test_check_sms_code_wrong():
print("\n" + "=" * 50)
print("测试4: check_sms_code() - 验证码校验(错误验证码)")
print("=" * 50)
codeid = await generate_sms_code()
if not codeid:
print("❌ 生成验证码失败")
return False
wrong_code = "000000"
result = await check_sms_code(codeid, wrong_code)
print(f"校验结果 (错误验证码): {result}")
if not result:
print("✅ 正确返回 False")
return True
else:
print("❌ 应该返回 False")
return False
async def test_check_sms_code_correct():
print("\n" + "=" * 50)
print("测试5: check_sms_code() - 验证码校验(正确验证码)")
print("=" * 50)
codeid = await generate_sms_code()
codeid = 'ltBojwOXD4VvC73lFTeKb'
if not codeid:
print("❌ 生成验证码失败")
return False
db_result = await get_vcode_from_db(codeid)
if not db_result:
print("❌ 获取验证码失败")
return False
real_vcode = db_result['vcode']
real_vcode = '577948'
print(f"真实验证码: {real_vcode}")
result = await check_sms_code(codeid, real_vcode)
print(f"校验结果 (正确验证码): {result}")
if result:
print("✅ 验证码校验成功")
return True
else:
print("❌ 验证码校验失败")
return False
async def test_check_sms_code_invalid_codeid():
print("\n" + "=" * 50)
print("测试6: check_sms_code() - 验证码校验(无效codeid)")
print("=" * 50)
invalid_codeid = "invalid_codeid_12345"
result = await check_sms_code(invalid_codeid, "123456")
print(f"校验结果: {result}")
if not result:
print("✅ 正确返回 False")
return True
else:
print("❌ 应该返回 False")
return False
async def test_check_sms_code_expired():
print("\n" + "=" * 50)
print("测试7: check_sms_code() - 验证码校验(已过期)")
print("=" * 50)
from sqlor.dbpools import DBPools
import datetime
from appPublic.uniqueID import getID as uuid
expired_codeid = uuid()
expired_time = datetime.datetime.now() - datetime.timedelta(minutes=10)
db = DBPools()
async with db.sqlorContext('kboss') as sor:
await sor.C('validatecode', {
'id': expired_codeid,
'vcode': '888888',
'expire_time': expired_time,
'del_flg': '0',
'create_at': datetime.datetime.now()
})
result = await check_sms_code(expired_codeid, "888888")
print(f"校验结果 (已过期验证码): {result}")
if not result:
print("✅ 正确返回 False(已过期)")
return True
else:
print("❌ 应该返回 False")
return False
async def test_send_vcode():
print("\n" + "=" * 50)
print("测试8: send_vcode() - 发送验证码短信")
print("=" * 50)
phone = "18053191417"
stype = "用户注册登录验证"
vcode = {"SMSvCode": "123456"}
result = await send_vcode(phone, stype, vcode)
print(f"发送结果: {result}")
if result.get('status') == False:
if "模板未配置" in result.get('msg', ''):
print("⚠️ 模板未配置，这是预期行为（需先配置sms_template表）")
return True
print("✅ 发送功能正常（可能因配置问题失败）")
return True
elif result.get('status') == True:
print("✅ 发送成功")
return True
else:
print("❌ 发送失败")
return False
async def test_send_sms():
print("\n" + "=" * 50)
print("测试9: send_sms() - 发送短信(整合生成和发送)")
print("=" * 50)
phone = "18053191417"
stype = "用户注册登录验证"
result = await send_sms(phone, stype)
print(f"发送结果: {result}")
if result.get('status') == 'error':
if "模板未配置" in result.get('data', {}).get('message', ''):
print("⚠️ 模板未配置，这是预期行为（需先配置sms_template表）")
return True
print("✅ 发送功能正常（可能因配置问题失败）")
return True
elif result.get('status') == 'ok':
codeid = result.get('data', {}).get('codeid')
print(f"✅ 发送成功，验证码ID: {codeid}")
return True
else:
print("❌ 发送失败")
return False
async def test_sms_engine_attributes():
print("\n" + "=" * 50)
print("测试10: SMSEngine 属性检查")
print("=" * 50)
engine = get_sms_engine()
print(f"access_key: {engine.access_key}")
print(f"host: {engine.host}")
print(f"signature_id: {engine.signature_id}")
print(f"sms_types: {engine.sms_types}")
print(f"doc: {engine.doc}")
if engine.access_key and engine.host and engine.signature_id:
print("✅ 引擎属性正常")
return True
else:
print("❌ 引擎属性异常")
return False
async def main():
print("=" * 50)
print("SMS Send Module 完整测试")
print("=" * 50)
load_smssend()
results = []
# results.append(("get_sms_engine", await test_get_sms_engine()))
# r, codeid = await test_generate_sms_code()
# results.append(("generate_sms_code", r))
# r, codeid2 = await test_generate_sms_code_with_params()
# results.append(("generate_sms_code(自定义参数)", r))
# results.append(("check_sms_code(错误验证码)", await test_check_sms_code_wrong()))
results.append(("check_sms_code(正确验证码)", await test_check_sms_code_correct()))
# results.append(("check_sms_code(无效codeid)", await test_check_sms_code_invalid_codeid()))
# results.append(("check_sms_code(已过期)", await test_check_sms_code_expired()))
# results.append(("send_vcode", await test_send_vcode()))
# results.append(("send_sms", await test_send_sms()))
# results.append(("SMSEngine属性", await test_sms_engine_attributes()))
print("\n" + "=" * 50)
print("测试结果汇总")
print("=" * 50)
passed = 0
failed = 0
for name, result in results:
status = "✅ 通过" if result else "❌ 失败"
print(f"{name}: {status}")
if result:
passed += 1
else:
failed += 1
print("-" * 50)
print(f"总计: {passed} 通过, {failed} 失败")
from ahserver.serverenv import ServerEnv
env = ServerEnv()
def get_db_name(s=None): return 'kboss'
env.get_module_dbname = get_db_name
if __name__ == '__main__':
from sqlor.dbpools import DBPools
from appPublic.jsonConfig import getConfig
p = 'D:/Code/backend_code/test_/'
config = getConfig(p)
DBPools(config.databases)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())