refactor（kinit-api）：发送阿里云短信功能模块重构，涉及登录短信，重置密码短信调用都有更新

kinit-api/apps/vadmin/auth/crud.py

@@ -10,7 +10,6 @@ from typing import List, Any
 from aioredis import Redis
 from fastapi import UploadFile
 from sqlalchemy.orm import joinedload
-
 from core.exception import CustomException
 from fastapi.encoders import jsonable_encoder
 from sqlalchemy import select
@@ -18,10 +17,10 @@ from core.crud import DalBase
 from sqlalchemy.ext.asyncio import AsyncSession
 from core.validator import vali_telephone
 from utils.file.aliyun_oss import AliyunOSS, BucketConf
-from utils.aliyun_sms import AliyunSMS
 from utils.excel.import_manage import ImportManage, FieldType
 from utils.excel.write_xlsx import WriteXlsx
 from utils.send_email import EmailSender
+from utils.sms.reset_passwd import ResetPasswordSMS
 from .params import UserParams
 from utils.tools import test_password
 from . import models, schemas
@@ -239,9 +238,9 @@ class UserDal(DalBase):
                 user["send_sms_msg"] = "重置密码失败"
                 continue
             password = user.pop("password")
-            sms = AliyunSMS(rd, user.get("telephone"))
+            sms = ResetPasswordSMS([user.get("telephone")], rd)
             try:
-                send_result = await sms.main_async(AliyunSMS.Scene.reset_password, password=password)
+                send_result = (await sms.main_async(password=password))[0]
                 user["send_sms_status"] = send_result
                 user["send_sms_msg"] = "" if send_result else "发送失败，请联系管理员"
             except CustomException as e:

kinit-api/apps/vadmin/auth/utils/login_manage.py

@@ -12,8 +12,8 @@ from application import settings
 import jwt
 from apps.vadmin.auth import models
 from core.database import redis_getter
+from utils.sms.code import CodeSMS
 from .validation import LoginValidation, LoginForm, LoginResult
-from utils.aliyun_sms import AliyunSMS
 
 
 class LoginManage:
@@ -37,7 +37,7 @@ class LoginManage:
         验证用户短信验证码
         """
         rd = redis_getter(request)
-        sms = AliyunSMS(rd, data.telephone)
+        sms = CodeSMS(data.telephone, rd)
         result = await sms.check_sms_code(data.password)
         if result:
             return LoginResult(status=True, msg="验证成功")

kinit-api/apps/vadmin/system/views.py

@@ -8,19 +8,20 @@
 # UploadFile 库依赖：pip install python-multipart
 from typing import List
 from aioredis import Redis
-from fastapi import APIRouter, Depends, Body, UploadFile, Request, Form
+from fastapi import APIRouter, Depends, Body, UploadFile, Form
 from sqlalchemy.ext.asyncio import AsyncSession
 from application.settings import ALIYUN_OSS
 from core.database import db_getter, redis_getter
 from utils.file.aliyun_oss import AliyunOSS, BucketConf
-from utils.aliyun_sms import AliyunSMS
 from utils.file.file_manage import FileManage
 from utils.response import SuccessResponse, ErrorResponse
+from utils.sms.code import CodeSMS
 from . import schemas, crud
 from core.dependencies import IdList
-from apps.vadmin.auth.utils.current import AllUserAuth, FullAdminAuth
+from apps.vadmin.auth.utils.current import AllUserAuth, FullAdminAuth, OpenAuth
 from apps.vadmin.auth.utils.validation.auth import Auth
 from .params import DictTypeParams, DictDetailParams
+from apps.vadmin.auth import crud as vadminAuthCRUD
 
 app = APIRouter()
 
@@ -127,9 +128,12 @@ async def upload_image_to_local(file: UploadFile, path: str = Form(...)):
 #    短信服务管理
 ###########################################################
 @app.post("/sms/send", summary="发送短信验证码（阿里云服务）")
-async def sms_send(telephone: str, rd: Redis = Depends(redis_getter)):
+async def sms_send(telephone: str, rd: Redis = Depends(redis_getter), auth: Auth = Depends(OpenAuth())):
-    sms = AliyunSMS(rd, telephone)
+    user = await vadminAuthCRUD.UserDal(auth.db).get_data(telephone=telephone, v_return_none=True)
-    return SuccessResponse(await sms.main_async(AliyunSMS.Scene.login))
+    if not user:
+        return ErrorResponse("手机号不存在！")
+    sms = CodeSMS(telephone, rd)
+    return SuccessResponse(await sms.main_async())
 
 
 ###########################################################

kinit-api/utils/aliyun_sms.py

@@ -1,194 +0,0 @@
-# -*- coding: utf-8 -*-
-# @version        : 1.3
-# @Create Time    : 2021/3/15
-# @Author         : LZK
-# @File           : aliyun_sms.py
-# @Software       : PyCharm
-# @Update Time    : 2022/11/08
-# @Title          : 最新版阿里云短信服务发送程序（Python版本）2022-11-08
-
-
-"""
-短信 API 官方文档：https://help.aliyun.com/document_detail/419298.html?spm=5176.25163407.help.dexternal.6ff2bb6ercg9x3
-发送短信 官方文档：https://help.aliyun.com/document_detail/419273.htm?spm=a2c4g.11186623.0.0.36855d7aRBSwtP
-Python SDK 官方文档：https://help.aliyun.com/document_detail/215764.html?spm=a2c4g.11186623.0.0.6a0c4198XsBJNW
-
-环境要求
-Python 3
-安装 SDK 核心库 OpenAPI ，使用pip安装包依赖:
-pip install alibabacloud_tea_openapi
-pip install alibabacloud_dysmsapi20170525
-"""
-import random
-import re
-from enum import Enum, unique
-from core.exception import CustomException
-from alibabacloud_dysmsapi20170525.client import Client as Dysmsapi20170525Client
-from alibabacloud_tea_openapi import models as open_api_models
-from alibabacloud_dysmsapi20170525 import models as dysmsapi_20170525_models
-from alibabacloud_tea_util import models as util_models
-from core.logger import logger
-import datetime
-from aioredis.client import Redis
-from utils.cache import Cache
-from utils import status
-
-
-class AliyunSMS:
-
-    # 返回错误码对应：
-    doc = "https://help.aliyun.com/document_detail/101346.html"
-
-    @unique
-    class Scene(Enum):
-        login = "sms_template_code_1"
-        reset_password = "sms_template_code_2"
-
-    def __init__(self, rd: Redis, telephone: str):
-        self.check_telephone_format(telephone)
-        self.telephone = telephone
-        self.rd = rd
-        self.code = None
-        self.scene = None
-
-    async def __get_settings(self, retry: int = 3):
-        """
-        获取配置信息
-        """
-        aliyun_sms = await Cache(self.rd).get_tab_name("aliyun_sms", retry)
-        self.access_key = aliyun_sms.get("sms_access_key")
-        self.access_key_secret = aliyun_sms.get("sms_access_key_secret")
-        self.send_interval = int(aliyun_sms.get("sms_send_interval"))
-        self.valid_time = int(aliyun_sms.get("sms_valid_time"))
-        if self.scene == self.Scene.login:
-            self.sign_name = aliyun_sms.get("sms_sign_name_1")
-        else:
-            self.sign_name = aliyun_sms.get("sms_sign_name_2")
-        self.template_code = aliyun_sms.get(self.scene.value)
-
-    async def main_async(self, scene: Scene, **kwargs) -> bool:
-        """
-        主程序入口，异步方式
-        """
-        send_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-        if await self.rd.get(self.telephone + "_flag_"):
-            logger.error(f'{send_time} {self.telephone} 短信发送失败，短信发送过于频繁')
-            print(f"{self.telephone} 短信发送频繁")
-            raise CustomException(msg="短信发送频繁", code=status.HTTP_ERROR)
-        self.scene = scene
-        await self.__get_settings()
-        return await self.__send(**kwargs)
-
-    async def __send(self, **kwargs) -> bool:
-        """
-        发送短信
-        """
-        client = self.create_client(self.access_key, self.access_key_secret)
-        send_sms_request = dysmsapi_20170525_models.SendSmsRequest(
-            phone_numbers=self.telephone,
-            sign_name=self.sign_name,
-            template_code=self.template_code,
-            template_param=self.__get_template_param(**kwargs)
-        )
-        runtime = util_models.RuntimeOptions()
-        try:
-            # 复制代码运行请自行打印 API 的返回值
-            resp = await client.send_sms_with_options_async(send_sms_request, runtime)
-            return await self.__validation(resp)
-        except Exception as e:
-            print(e.__str__())
-            return False
-
-    def __get_template_param(self, **kwargs) -> str:
-        """
-        获取模板参数
-        """
-        if self.scene == self.Scene.login:
-            self.code = kwargs.get("code", self.get_code())
-            template_param = '{"code":"%s"}' % self.code
-        elif self.scene == self.Scene.reset_password:
-            self.code = kwargs.get("password")
-            template_param = '{"password":"%s"}' % self.code
-        else:
-            raise CustomException(msg="获取发送场景失败", code=status.HTTP_ERROR)
-        return template_param
-
-    async def __validation(self, resp: dysmsapi_20170525_models.SendSmsResponse) -> bool:
-        """
-        验证结果并返回
-        """
-        send_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-        if resp.body.code == "OK":
-            logger.info(f'{send_time} {self.telephone} 短信发送成功，返回code：{resp.body.code}')
-            await self.rd.set(self.telephone, self.code, self.valid_time)
-            await self.rd.set(self.telephone + "_flag_", self.code, self.send_interval)
-            return True
-        else:
-            logger.error(f'{send_time} {self.telephone} 短信发送失败，返回code：{resp.body.code}，请参考文档：{self.doc}')
-            return False
-
-    async def check_sms_code(self, code: str) -> bool:
-        """
-        检查短信验证码是否正确
-        """
-        if code and code == await self.rd.get(self.telephone):
-            await self.rd.delete(self.telephone)
-            await self.rd.delete(self.telephone + "_flag_")
-            return True
-        return False
-
-    @staticmethod
-    def get_code(length: int = 6, blend: bool = False) -> str:
-        """
-        随机获取短信验证码
-        短信验证码只支持数字，不支持字母及其他符号
-
-        :param length: 验证码长度
-        :param blend: 是否 字母+数字 混合
-        """
-        code = ""  # 创建字符串变量,存储生成的验证码
-        for i in range(length):  # 通过for循环控制验证码位数
-            num = random.randint(0, 9)  # 生成随机数字0-9
-            if blend:  # 需要字母验证码,不用传参,如果不需要字母的,关键字alpha=False
-                upper_alpha = chr(random.randint(65, 90))
-                lower_alpha = chr(random.randint(97, 122))
-                # 随机选择其中一位
-                num = random.choice([num, upper_alpha, lower_alpha])
-            code = code + str(num)
-        return code
-
-    @staticmethod
-    def check_telephone_format(telephone: str) -> bool:
-        """
-        检查手机号格式是否合法
-        """
-        REGEX_TELEPHONE = r'^1(3\d|4[4-9]|5[0-35-9]|6[67]|7[013-8]|8[0-9]|9[0-9])\d{8}$'
-
-        if not telephone:
-            raise CustomException(msg="手机号码不能为空", code=400)
-        elif not re.match(REGEX_TELEPHONE, telephone):
-            raise CustomException(msg="手机号码格式不正确", code=400)
-        return True
-
-    @staticmethod
-    def create_client(
-            access_key_id: str,
-            access_key_secret: str,
-    ) -> Dysmsapi20170525Client:
-        """
-        使用AK&SK初始化账号Client
-        :param access_key_id:
-        :param access_key_secret:
-        :return: Client
-        :throws Exception
-        """
-        config = open_api_models.Config(
-            # 您的 AccessKey ID,
-            access_key_id=access_key_id,
-            # 您的 AccessKey Secret,
-            access_key_secret=access_key_secret
-        )
-        # 访问的域名
-        config.endpoint = f'dysmsapi.aliyuncs.com'
-        return Dysmsapi20170525Client(config)
-

kinit-api/utils/db_getter.py

@@ -0,0 +1,66 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# @version        : 1.0
+# @Create Time    : 2023/5/6 9:29
+# @File           : task.py
+# @IDE            : PyCharm
+# @desc           : 任务基础类
+
+import re
+import pymysql
+from application.settings import SQLALCHEMY_DATABASE_URL
+from core.logger import logger
+
+
+class DBGetter:
+
+    def __init__(self):
+        self.mysql_cursor = None
+        self.mysql_conn = None
+
+    def conn_mysql(self) -> None:
+        """
+        连接系统中配置的 mysql 数据库
+        """
+        try:
+            connection_string = SQLALCHEMY_DATABASE_URL.split("//")[1]
+            pattern = r'^(?P<username>[^:]+):(?P<password>[^@]+)@(?P<host>[^:/]+):(?P<port>\d+)/(?P<database>[^/]+)$'
+            match = re.match(pattern, connection_string)
+
+            username = match.group('username')
+            password = match.group('password')
+            host = match.group('host')
+            port = int(match.group('port'))
+            database = match.group('database')
+
+            self.mysql_conn = pymysql.connect(
+                host=host,
+                port=port,
+                user=username,
+                password=password,
+                database=database
+            )
+            self.mysql_cursor = self.mysql_conn.cursor()
+        except pymysql.err.OperationalError as e:
+            logger.error(f"数据库连接失败，{e}")
+            raise ValueError("数据库连接失败！")
+        except AttributeError as e:
+            logger.error(f"数据库链接解析失败，{e}")
+            raise ValueError("数据库链接解析失败！")
+
+    def close_mysql(self) -> None:
+        """
+        关闭 mysql 链接
+        """
+        try:
+            self.mysql_cursor.close()
+            self.mysql_conn.close()
+        except AttributeError as e:
+            logger.error(f"未连接数据库，无需关闭！，{e}")
+            raise ValueError("未连接数据库，无需关闭！")
+
+
+if __name__ == '__main__':
+    t = DBGetter()
+    t.conn_mysql()
+    t.close_mysql()

kinit-api/utils/sms/__init__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# @version        : 1.0
+# @Create Time    : 2023/6/14 15:26 
+# @File           : __init__.py
+# @IDE            : PyCharm
+# @desc           : 简要说明

kinit-api/utils/sms/aliyun.py

@@ -0,0 +1,245 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# @version        : 1.0
+# @Create Time    : 2023/6/14 15:26 
+# @File           : aliyun.py
+# @IDE            : PyCharm
+# @desc           : 最新版阿里云短信服务发送程序（Python版本）2022-11-08
+
+"""
+短信 API 官方文档：https://help.aliyun.com/document_detail/419298.html?spm=5176.25163407.help.dexternal.6ff2bb6ercg9x3
+发送短信 官方文档：https://help.aliyun.com/document_detail/419273.htm?spm=a2c4g.11186623.0.0.36855d7aRBSwtP
+Python SDK 官方文档：https://help.aliyun.com/document_detail/215764.html?spm=a2c4g.11186623.0.0.6a0c4198XsBJNW
+
+环境要求
+Python 3
+安装 SDK 核心库 OpenAPI ，使用pip安装包依赖:
+pip install alibabacloud_tea_openapi
+pip install alibabacloud_dysmsapi20170525
+"""
+import random
+import re
+from typing import List
+from core.exception import CustomException
+from alibabacloud_dysmsapi20170525.client import Client as Dysmsapi20170525Client
+from alibabacloud_tea_openapi import models as open_api_models
+from alibabacloud_dysmsapi20170525 import models as dysmsapi_20170525_models
+from alibabacloud_tea_util import models as util_models
+from core.logger import logger
+import datetime
+from aioredis.client import Redis
+from utils.cache import Cache
+from utils.db_getter import DBGetter
+
+
+class AliyunSMS(DBGetter):
+    # 返回错误码对应：
+    doc = "https://help.aliyun.com/document_detail/101346.html"
+
+    def __init__(self, telephones: List[str], rd: Redis = None):
+        super().__init__()
+        self.check_telephones_format(telephones)
+        self.telephones = telephones
+        self.rd = rd
+
+        self.sign_conf = None  # 数据库中 sms_sign_name_* 的配置
+        self.template_code_conf = None  # 数据库中 sms_template_code_* 的配置
+        # 以上两个配置项的好处在于可以灵活配置短信信息，不需要改代码
+
+    async def main_async(self, **kwargs) -> List[bool]:
+        """
+        主程序入口，异步方式
+
+        redis 对象必填
+        """
+        result = []
+        await self._get_settings_async()
+        for telephone in self.telephones:
+            result.append(await self._send_async(telephone, **kwargs))
+        return result
+
+    async def _send_async(self, telephone: str, **kwargs) -> bool:
+        """
+        发送短信
+        """
+        client = self.create_client(self.access_key, self.access_key_secret)
+        send_sms_request = dysmsapi_20170525_models.SendSmsRequest(
+            phone_numbers=telephone,
+            sign_name=self.sign_name,
+            template_code=self.template_code,
+            template_param=self._get_template_param(**kwargs)
+        )
+        runtime = util_models.RuntimeOptions()
+        try:
+            # 复制代码运行请自行打印 API 的返回值
+            resp = await client.send_sms_with_options_async(send_sms_request, runtime)
+            return self._validation(telephone, resp)
+        except Exception as e:
+            print(e.__str__())
+            return False
+
+    async def _get_settings_async(self, retry: int = 3):
+        """
+        获取配置信息
+        """
+        if not self.rd:
+            raise ValueError("缺少 redis 对象参数！")
+        elif not self.sign_conf or not self.template_code_conf:
+            raise ValueError("缺少短信签名信息和短信模板ID！")
+        aliyun_sms = await Cache(self.rd).get_tab_name("aliyun_sms", retry)
+        self.access_key = aliyun_sms.get("sms_access_key")
+        self.access_key_secret = aliyun_sms.get("sms_access_key_secret")
+        self.send_interval = int(aliyun_sms.get("sms_send_interval"))
+        self.valid_time = int(aliyun_sms.get("sms_valid_time"))
+        self.sign_name = aliyun_sms.get(self.sign_conf)
+        self.template_code = aliyun_sms.get(self.template_code_conf)
+
+    def main(self, **kwargs) -> List[bool]:
+        """
+        主程序入口，同步方式
+        """
+        result = []
+        self._get_settings()
+        for telephone in self.telephones:
+            result.append(self._send(telephone, **kwargs))
+        return result
+
+    def _send(self, telephone: str, **kwargs) -> bool:
+        """
+        同步方式发送短信
+        """
+        client = self.create_client(self.access_key, self.access_key_secret)
+        send_sms_request = dysmsapi_20170525_models.SendSmsRequest(
+            phone_numbers=telephone,
+            sign_name=self.sign_name,
+            template_code=self.template_code,
+            template_param=self._get_template_param(**kwargs)
+        )
+        runtime = util_models.RuntimeOptions()
+        try:
+            # 复制代码运行请自行打印 API 的返回值
+            resp = client.send_sms_with_options(send_sms_request, runtime)
+            return self._validation(telephone, resp)
+        except Exception as e:
+            print(e.__str__())
+            return False
+
+    def _get_settings(self):
+        """
+        同步方式获取配置信息
+        """
+        if not self.sign_conf or not self.template_code_conf:
+            raise ValueError("缺少短信签名信息和短信模板ID！")
+        self.conn_mysql()
+        sql = f"""
+        SELECT
+            config_value 
+        FROM
+            `vadmin_system_settings` 
+        WHERE
+            config_key IN (
+                'sms_access_key',
+                'sms_access_key_secret',
+                'sms_send_interval',
+                'sms_valid_time',
+                '{self.sign_conf}',
+                '{self.template_code_conf}'
+            )
+        """
+        self.mysql_cursor.execute(sql)
+        result = self.mysql_cursor.fetchall()
+        self.close_mysql()
+        self.access_key = result[0][0]
+        self.access_key_secret = result[1][0]
+        self.send_interval = result[2][0]
+        self.valid_time = result[3][0]
+        self.sign_name = result[4][0]
+        self.template_code = result[5][0]
+
+    def _get_template_param(self, **kwargs) -> str:
+        """
+        获取模板参数
+
+        可以被子类继承的受保护的私有方法
+        """
+        raise NotImplementedError("该方法应该被重写！")
+
+    def _validation(self, telephone: str, resp: dysmsapi_20170525_models.SendSmsResponse) -> bool:
+        """
+        验证结果并返回
+        """
+        send_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        if resp.body.code == "OK":
+            logger.info(f'{send_time} {telephone} 短信发送成功，返回code：{resp.body.code}')
+            return True
+        else:
+            logger.error(f'{send_time} {telephone} 短信发送失败，返回code：{resp.body.code}，请参考文档：{self.doc}')
+            return False
+
+    @staticmethod
+    def get_code(length: int = 6, blend: bool = False) -> str:
+        """
+        随机获取短信验证码
+        短信验证码只支持数字，不支持字母及其他符号
+
+        :param length: 验证码长度
+        :param blend: 是否 字母+数字 混合
+        """
+        code = ""  # 创建字符串变量,存储生成的验证码
+        for i in range(length):  # 通过for循环控制验证码位数
+            num = random.randint(0, 9)  # 生成随机数字0-9
+            if blend:  # 需要字母验证码,不用传参,如果不需要字母的,关键字alpha=False
+                upper_alpha = chr(random.randint(65, 90))
+                lower_alpha = chr(random.randint(97, 122))
+                # 随机选择其中一位
+                num = random.choice([num, upper_alpha, lower_alpha])
+            code = code + str(num)
+        return code
+
+    @classmethod
+    def check_telephones_format(cls, telephones: List[str]) -> bool:
+        """
+        同时检查多个手机号格式是否合法
+
+        不合法就会抛出异常
+        """
+        for telephone in telephones:
+            cls.check_telephone_format(telephone)
+        return True
+
+    @staticmethod
+    def check_telephone_format(telephone: str) -> bool:
+        """
+        检查手机号格式是否合法
+
+        不合法就会抛出异常
+        """
+        REGEX_TELEPHONE = r'^1(3\d|4[4-9]|5[0-35-9]|6[67]|7[013-8]|8[0-9]|9[0-9])\d{8}$'
+
+        if not telephone:
+            raise CustomException(msg=f"手机号码（{telephone}）不能为空", code=400)
+        elif not re.match(REGEX_TELEPHONE, telephone):
+            raise CustomException(msg=f"手机号码（{telephone}）格式不正确", code=400)
+        return True
+
+    @staticmethod
+    def create_client(
+            access_key_id: str,
+            access_key_secret: str,
+    ) -> Dysmsapi20170525Client:
+        """
+        使用AK&SK初始化账号Client
+        :param access_key_id:
+        :param access_key_secret:
+        :return: Client
+        :throws Exception
+        """
+        config = open_api_models.Config(
+            # 您的 AccessKey ID,
+            access_key_id=access_key_id,
+            # 您的 AccessKey Secret,
+            access_key_secret=access_key_secret
+        )
+        # 访问的域名
+        config.endpoint = f'dysmsapi.aliyuncs.com'
+        return Dysmsapi20170525Client(config)

kinit-api/utils/sms/code.py

@@ -0,0 +1,69 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# @version        : 1.0
+# @Create Time    : 2023/6/14 15:55 
+# @File           : code.py
+# @IDE            : PyCharm
+# @desc           : 发送验证码短信
+
+import datetime
+import warnings
+from aioredis import Redis
+from .aliyun import AliyunSMS
+from core.logger import logger
+from core.exception import CustomException
+
+
+class CodeSMS(AliyunSMS):
+
+    def __init__(self, telephone: str, rd: Redis):
+        super().__init__([telephone], rd)
+
+        self.telephone = telephone
+        self.sign_conf = "sms_sign_name_1"
+        self.template_code_conf = "sms_template_code_1"
+
+    async def main_async(self) -> bool:
+        """
+        主程序入口，异步方式
+
+        redis 对象必填
+        """
+
+        send_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        if await self.rd.get(self.telephone + "_flag_"):
+            logger.error(f'{send_time} {self.telephone} 短信发送失败，短信发送过于频繁')
+            raise CustomException(msg="短信发送频繁", code=400)
+        await self._get_settings_async()
+        result = await self._send_async(self.telephone)
+        if result:
+            await self.rd.set(self.telephone, self.code, self.valid_time)
+            await self.rd.set(self.telephone + "_flag_", self.code, self.send_interval)
+        return result
+
+    async def main(self) -> None:
+        """
+        主程序入口，同步方式
+        """
+        warnings.warn("此方法已废弃，如需要请重写该方法", DeprecationWarning)
+
+    async def check_sms_code(self, code: str) -> bool:
+        """
+        检查短信验证码是否正确
+        """
+        if code and code == await self.rd.get(self.telephone):
+            await self.rd.delete(self.telephone)
+            await self.rd.delete(self.telephone + "_flag_")
+            return True
+        return False
+
+    def _get_template_param(self, **kwargs) -> str:
+        """
+        获取模板参数
+
+        可以被子类继承的受保护的私有方法
+        """
+        self.code = kwargs.get("code", self.get_code())
+        template_param = '{"code":"%s"}' % self.code
+        return template_param
+

kinit-api/utils/sms/reset_passwd.py

@@ -0,0 +1,47 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# @version        : 1.0
+# @Create Time    : 2023/6/14 16:58 
+# @File           : reset_passwd.py
+# @IDE            : PyCharm
+# @desc           : 重置密码
+
+from typing import List
+from aioredis import Redis
+from .aliyun import AliyunSMS
+
+
+class ResetPasswordSMS(AliyunSMS):
+
+    def __init__(self, telephones: List[str], rd: Redis = None):
+        super().__init__(telephones, rd)
+
+        self.sign_conf = "sms_sign_name_2"
+        self.template_code_conf = "sms_template_code_2"
+
+    async def main_async(self, password: str) -> List[bool]:
+        """
+        主程序入口，异步方式
+
+        redis 对象必填
+        @params password: 新密码
+        """
+        return await super().main_async(password=password)
+
+    def main(self, password: str) -> List[bool]:
+        """
+        主程序入口，同步方式
+
+        @params password: 新密码
+        """
+        return super().main(password=password)
+
+    def _get_template_param(self, **kwargs) -> str:
+        """
+        获取模板参数
+
+        可以被子类继承的受保护的私有方法
+        """
+        password = kwargs.get("password")
+        template_param = '{"password":"%s"}' % password
+        return template_param