kevin/life-echo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 新增短信验证码服务

Browse Source 
- 新增sms_service.py短信验证服务
- 新增test_sms_verification.py短信验证测试
- 更新requirements.txt添加短信服务依赖
- 更新.env.production添加短信服务配置
This commit is contained in:
iammm0
2026-01-27 11:35:54 +08:00
parent 8f4a68f40f
commit 9c39df62bc
4 changed files with 707 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
api/.env.production
View File

@@ -29,4 +29,17 @@ SECRET_KEY=cf47555c7ecbe5ddb7fd2113c59e08a8bcb110810c42f7c644e06a5acc898608
# JWT 算法（可选，默认: HS256）
ALGORITHM=HS256
# 访问令牌过期时间（可选，默认: 120 分钟，即 2 小时）
ACCESS_TOKEN_EXPIRE_MINUTES=120
ACCESS_TOKEN_EXPIRE_MINUTES=120
# =============================================================================
# 腾讯云短信服务配置（必需）
# =============================================================================
# 腾讯云 SecretId 和 SecretKey
TENCENT_SMS_SECRET_ID=AKIDmXEoYQCypUyFI9nH7tWzGOJG7e3ING0Y
TENCENT_SMS_SECRET_KEY=Vq22tb7UghhV6WnskRyQjroY5QDqHUXO
# 短信应用 SDK AppID
TENCENT_SMS_SDK_APP_ID=1319381411
# 短信签名内容（不包含【】符号）
TENCENT_SMS_SIGN_NAME=上海华嘎科技有限公司
# 短信模板 ID
TENCENT_SMS_TEMPLATE_ID=2592163

3
api/requirements.txt
View File

@@ -45,4 +45,7 @@ bcrypt>=4.0.0
# Image Processing
Pillow>=10.0.0
# Tencent Cloud SMS
tencentcloud-sdk-python>=3.0.1000
openai

265
api/services/sms_service.py Normal file
View File

@@ -0,0 +1,265 @@
"""
腾讯云短信服务
"""
import os
import random
import uuid
from datetime import timedelta
from typing import Optional, Tuple
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from tencentcloud.common import credential
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
from tencentcloud.sms.v20210111 import sms_client, models as sms_models
from database.models import SmsVerificationCode, utc_now
# 腾讯云短信配置
TENCENT_SMS_SECRET_ID = os.getenv("TENCENT_SMS_SECRET_ID", "")
TENCENT_SMS_SECRET_KEY = os.getenv("TENCENT_SMS_SECRET_KEY", "")
TENCENT_SMS_SDK_APP_ID = os.getenv("TENCENT_SMS_SDK_APP_ID", "")
TENCENT_SMS_SIGN_NAME = os.getenv("TENCENT_SMS_SIGN_NAME", "")
# 统一使用一个短信模板ID（所有场景共用）
TENCENT_SMS_TEMPLATE_ID = os.getenv("TENCENT_SMS_TEMPLATE_ID", "")
# 验证码配置
CODE_LENGTH = 6 # 验证码长度
CODE_EXPIRE_MINUTES = 5 # 验证码过期时间（分钟）
RATE_LIMIT_SECONDS = 60 # 发送频率限制（秒）
def generate_verification_code() -> str:
"""生成6位随机数字验证码"""
return ''.join([str(random.randint(0, 9)) for _ in range(CODE_LENGTH)])
def get_template_id_by_purpose(purpose: str) -> str:
"""获取短信模板ID（所有场景共用同一个模板）"""
return TENCENT_SMS_TEMPLATE_ID
def send_sms_via_tencent(phone: str, code: str, purpose: str) -> bool:
"""
通过腾讯云发送短信验证码
Args:
phone: 手机号
code: 验证码
purpose: 用途
Returns:
bool: 是否发送成功
"""
try:
# 创建认证对象
cred = credential.Credential(TENCENT_SMS_SECRET_ID, TENCENT_SMS_SECRET_KEY)
# 创建SMS客户端
client = sms_client.SmsClient(cred, "ap-guangzhou")
# 创建请求对象
req = sms_models.SendSmsRequest()
req.SmsSdkAppId = TENCENT_SMS_SDK_APP_ID
req.SignName = TENCENT_SMS_SIGN_NAME
req.TemplateId = get_template_id_by_purpose(purpose)
req.TemplateParamSet = [code, str(CODE_EXPIRE_MINUTES)] # 验证码和过期时间
req.PhoneNumberSet = [f"+86{phone}"]
# 发送短信
resp = client.SendSms(req)
# 检查发送结果
if resp.SendStatusSet and len(resp.SendStatusSet) > 0:
status = resp.SendStatusSet[0]
if status.Code == "Ok":
return True
else:
print(f"短信发送失败: {status.Code} - {status.Message}")
return False
return False
except TencentCloudSDKException as e:
print(f"腾讯云SDK异常: {e}")
return False
except Exception as e:
print(f"发送短信异常: {e}")
return False
async def check_rate_limit(db: AsyncSession, phone: str) -> Tuple[bool, int]:
"""
检查发送频率限制
Args:
db: 数据库会话
phone: 手机号
Returns:
Tuple[bool, int]: (是否允许发送, 剩余等待秒数)
"""
# 查询最近的验证码记录
stmt = select(SmsVerificationCode).where(
SmsVerificationCode.phone == phone
).order_by(
SmsVerificationCode.created_at.desc()
)
result = await db.execute(stmt)
recent_code = result.scalar_one_or_none()
if not recent_code:
return True, 0
# 计算距离上次发送的时间
now = utc_now()
time_diff = (now - recent_code.created_at).total_seconds()
if time_diff < RATE_LIMIT_SECONDS:
remaining = int(RATE_LIMIT_SECONDS - time_diff)
return False, remaining
return True, 0
async def send_verification_code(
db: AsyncSession,
phone: str,
purpose: str,
ip_address: Optional[str] = None
) -> Tuple[bool, str, int]:
"""
发送验证码
Args:
db: 数据库会话
phone: 手机号
purpose: 用途 (register/login/reset_password/change_phone)
ip_address: 请求IP地址
Returns:
Tuple[bool, str, int]: (是否成功, 消息, 过期时间秒数)
"""
# 检查频率限制
can_send, remaining = await check_rate_limit(db, phone)
if not can_send:
return False, f"发送过于频繁，请{remaining}秒后再试", 0
# 生成验证码
code = generate_verification_code()
# 发送短信
success = send_sms_via_tencent(phone, code, purpose)
if not success:
return False, "短信发送失败，请稍后重试", 0
# 保存到数据库
expires_at = utc_now() + timedelta(minutes=CODE_EXPIRE_MINUTES)
verification_code = SmsVerificationCode(
id=str(uuid.uuid4()),
phone=phone,
code=code,
purpose=purpose,
expires_at=expires_at,
ip_address=ip_address
)
db.add(verification_code)
await db.commit()
return True, "验证码已发送", CODE_EXPIRE_MINUTES * 60
async def verify_code(
db: AsyncSession,
phone: str,
code: str,
purpose: str
) -> Tuple[bool, str]:
"""
验证验证码
Args:
db: 数据库会话
phone: 手机号
code: 验证码
purpose: 用途
Returns:
Tuple[bool, str]: (是否验证成功, 消息)
"""
# 查询最新的未使用验证码
stmt = select(SmsVerificationCode).where(
SmsVerificationCode.phone == phone,
SmsVerificationCode.purpose == purpose,
SmsVerificationCode.is_used == False,
SmsVerificationCode.is_expired == False
).order_by(
SmsVerificationCode.created_at.desc()
)
result = await db.execute(stmt)
verification = result.scalar_one_or_none()
if not verification:
return False, "验证码不存在或已使用"
# 检查是否过期
now = utc_now()
if now > verification.expires_at:
verification.is_expired = True
await db.commit()
return False, "验证码已过期"
# 验证验证码
if verification.code != code:
return False, "验证码错误"
# 标记为已使用
verification.is_used = True
verification.verified_at = now
await db.commit()
return True, "验证成功"
async def cleanup_expired_codes(db: AsyncSession) -> int:
"""
清理过期的验证码
Args:
db: 数据库会话
Returns:
int: 清理的记录数
"""
from sqlalchemy import delete
now = utc_now()
# 标记过期的验证码
stmt = select(SmsVerificationCode).where(
SmsVerificationCode.expires_at < now,
SmsVerificationCode.is_expired == False
)
result = await db.execute(stmt)
expired_codes = result.scalars().all()
count = 0
for code in expired_codes:
code.is_expired = True
count += 1
await db.commit()
# 删除7天前的记录
seven_days_ago = now - timedelta(days=7)
delete_stmt = delete(SmsVerificationCode).where(
SmsVerificationCode.created_at < seven_days_ago
)
result = await db.execute(delete_stmt)
old_codes_count = result.rowcount
await db.commit()
return count + old_codes_count

425
api/test_sms_verification.py Executable file
View File

@@ -0,0 +1,425 @@
#!/usr/bin/env python3
"""
短信验证码功能测试脚本
测试用例：
1. 发送验证码（成功/频率限制）
2. 验证码验证（成功/过期/错误）
3. 验证码注册流程
4. 验证码登录流程
5. 密码重置流程
6. 修改密码（已登录）
7. 修改手机号
8. 登出所有设备
"""
import requests
import time
import json
from typing import Optional, Dict, Any
# 配置
BASE_URL = "http://localhost:8000"
API_PREFIX = "/api"
# 测试用户数据
TEST_PHONE = "13800138000"
TEST_PASSWORD = "test123456"
TEST_NICKNAME = "测试用户"
TEST_EMAIL = "test@example.com"
NEW_PHONE = "13900139000"
NEW_PASSWORD = "newpass123456"
# 全局变量
access_token: Optional[str] = None
refresh_token: Optional[str] = None
class Colors:
"""终端颜色"""
HEADER = '\033[95m'
OKBLUE = '\033[94m'
OKCYAN = '\033[96m'
OKGREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
ENDC = '\033[0m'
BOLD = '\033[1m'
UNDERLINE = '\033[4m'
def print_header(text: str):
"""打印测试标题"""
print(f"\n{Colors.HEADER}{Colors.BOLD}{'='*60}{Colors.ENDC}")
print(f"{Colors.HEADER}{Colors.BOLD}{text}{Colors.ENDC}")
print(f"{Colors.HEADER}{Colors.BOLD}{'='*60}{Colors.ENDC}\n")
def print_success(text: str):
"""打印成功信息"""
print(f"{Colors.OKGREEN}✓ {text}{Colors.ENDC}")
def print_error(text: str):
"""打印错误信息"""
print(f"{Colors.FAIL}✗ {text}{Colors.ENDC}")
def print_info(text: str):
"""打印信息"""
print(f"{Colors.OKCYAN}ℹ {text}{Colors.ENDC}")
def print_warning(text: str):
"""打印警告"""
print(f"{Colors.WARNING}⚠ {text}{Colors.ENDC}")
def make_request(
method: str,
endpoint: str,
data: Optional[Dict[str, Any]] = None,
headers: Optional[Dict[str, str]] = None,
expected_status: int = 200
) -> Optional[Dict[str, Any]]:
"""发送HTTP请求"""
url = f"{BASE_URL}{API_PREFIX}{endpoint}"
try:
if method.upper() == "GET":
response = requests.get(url, headers=headers)
elif method.upper() == "POST":
response = requests.post(url, json=data, headers=headers)
elif method.upper() == "PUT":
response = requests.put(url, json=data, headers=headers)
elif method.upper() == "DELETE":
response = requests.delete(url, headers=headers)
else:
print_error(f"不支持的HTTP方法: {method}")
return None
print_info(f"{method.upper()} {endpoint} - Status: {response.status_code}")
if response.status_code == expected_status:
print_success(f"请求成功 (状态码: {response.status_code})")
try:
return response.json()
except:
return {"status": "success"}
else:
print_error(f"请求失败 (期望: {expected_status}, 实际: {response.status_code})")
try:
error_data = response.json()
print_error(f"错误信息: {json.dumps(error_data, ensure_ascii=False, indent=2)}")
except:
print_error(f"响应内容: {response.text}")
return None
except requests.exceptions.ConnectionError:
print_error(f"连接失败: 无法连接到 {BASE_URL}")
print_warning("请确保后端服务正在运行")
return None
except Exception as e:
print_error(f"请求异常: {str(e)}")
return None
def test_send_verification_code(phone: str, purpose: str) -> bool:
"""测试发送验证码"""
print_header(f"测试发送验证码 - {purpose}")
data = {
"phone": phone,
"purpose": purpose
}
result = make_request("POST", "/auth/sms/send", data=data)
if result:
print_success(f"验证码已发送: {result.get('message', '')}")
print_info(f"有效期: {result.get('expires_in', 0)} 秒")
return True
return False
def test_rate_limit(phone: str) -> bool:
"""测试频率限制"""
print_header("测试频率限制")
# 第一次发送应该成功
if not test_send_verification_code(phone, "register"):
return False
print_info("等待1秒后再次发送...")
time.sleep(1)
# 第二次发送应该被限制
data = {
"phone": phone,
"purpose": "register"
}
result = make_request("POST", "/auth/sms/send", data=data, expected_status=429)
if result is None:
print_success("频率限制生效")
return True
else:
print_error("频率限制未生效")
return False
def test_register_with_sms(phone: str, code: str) -> bool:
"""测试验证码注册"""
print_header("测试验证码注册")
data = {
"phone": phone,
"code": code,
"password": TEST_PASSWORD,
"nickname": TEST_NICKNAME,
"email": TEST_EMAIL
}
result = make_request("POST", "/auth/register/sms", data=data, expected_status=201)
if result:
global access_token, refresh_token
access_token = result.get("access_token")
refresh_token = result.get("refresh_token")
print_success("注册成功")
print_info(f"Access Token: {access_token[:20]}...")
print_info(f"Refresh Token: {refresh_token[:20]}...")
return True
return False
def test_login_with_sms(phone: str, code: str) -> bool:
"""测试验证码登录"""
print_header("测试验证码登录")
data = {
"phone": phone,
"code": code
}
result = make_request("POST", "/auth/login/sms", data=data)
if result:
global access_token, refresh_token
access_token = result.get("access_token")
refresh_token = result.get("refresh_token")
print_success("登录成功")
print_info(f"Access Token: {access_token[:20]}...")
return True
return False
def test_reset_password(phone: str, code: str, new_password: str) -> bool:
"""测试重置密码"""
print_header("测试重置密码")
data = {
"phone": phone,
"code": code,
"new_password": new_password
}
result = make_request("POST", "/auth/password/reset", data=data)
if result:
print_success(f"密码重置成功: {result.get('message', '')}")
return True
return False
def test_change_password(old_password: str, new_password: str) -> bool:
"""测试修改密码（已登录）"""
print_header("测试修改密码")
if not access_token:
print_error("未登录，无法测试")
return False
data = {
"old_password": old_password,
"new_password": new_password
}
headers = {
"Authorization": f"Bearer {access_token}"
}
result = make_request("POST", "/auth/password/change", data=data, headers=headers)
if result:
print_success(f"密码修改成功: {result.get('message', '')}")
return True
return False
def test_change_phone(new_phone: str, code: str) -> bool:
"""测试修改手机号"""
print_header("测试修改手机号")
if not access_token:
print_error("未登录，无法测试")
return False
data = {
"new_phone": new_phone,
"code": code
}
headers = {
"Authorization": f"Bearer {access_token}"
}
result = make_request("POST", "/auth/phone/change", data=data, headers=headers)
if result:
print_success(f"手机号修改成功")
print_info(f"新手机号: {result.get('phone', '')}")
return True
return False
def test_logout_all() -> bool:
"""测试登出所有设备"""
print_header("测试登出所有设备")
if not access_token:
print_error("未登录，无法测试")
return False
headers = {
"Authorization": f"Bearer {access_token}"
}
result = make_request("POST", "/auth/logout/all", headers=headers)
if result:
print_success(f"登出成功: {result.get('message', '')}")
return True
return False
def test_get_current_user() -> bool:
"""测试获取当前用户信息"""
print_header("测试获取当前用户信息")
if not access_token:
print_error("未登录，无法测试")
return False
headers = {
"Authorization": f"Bearer {access_token}"
}
result = make_request("GET", "/auth/me", headers=headers)
if result:
print_success("获取用户信息成功")
print_info(f"用户信息: {json.dumps(result, ensure_ascii=False, indent=2)}")
return True
return False
def interactive_test():
"""交互式测试"""
print_header("短信验证码功能交互式测试")
print_info("此模式需要您手动输入收到的验证码")
print_warning("请确保已配置腾讯云短信服务")
phone = input(f"\n请输入测试手机号 (默认: {TEST_PHONE}): ").strip() or TEST_PHONE
# 1. 测试发送验证码
if not test_send_verification_code(phone, "register"):
print_error("发送验证码失败，测试终止")
return
code = input("\n请输入收到的验证码: ").strip()
if not code or len(code) != 6:
print_error("验证码格式错误")
return
# 2. 测试注册
if test_register_with_sms(phone, code):
print_success("注册测试通过")
# 3. 测试获取用户信息
test_get_current_user()
# 4. 测试修改密码
if input("\n是否测试修改密码? (y/n): ").lower() == 'y':
test_change_password(TEST_PASSWORD, NEW_PASSWORD)
# 5. 测试修改手机号
if input("\n是否测试修改手机号? (y/n): ").lower() == 'y':
new_phone = input(f"请输入新手机号 (默认: {NEW_PHONE}): ").strip() or NEW_PHONE
if test_send_verification_code(new_phone, "change_phone"):
code = input("请输入收到的验证码: ").strip()
test_change_phone(new_phone, code)
# 6. 测试登出所有设备
if input("\n是否测试登出所有设备? (y/n): ").lower() == 'y':
test_logout_all()
def automated_test():
"""自动化测试（需要mock验证码）"""
print_header("短信验证码功能自动化测试")
print_warning("此模式需要后端支持测试验证码（如：123456）")
# 测试发送验证码
test_send_verification_code(TEST_PHONE, "register")
# 等待一段时间
print_info("等待60秒以测试频率限制...")
time.sleep(60)
# 测试频率限制
test_rate_limit(TEST_PHONE)
if __name__ == "__main__":
print(f"{Colors.BOLD}{Colors.OKBLUE}")
print("=" * 60)
print("短信验证码功能测试脚本")
print("=" * 60)
print(f"{Colors.ENDC}")
print("\n请选择测试模式:")
print("1. 交互式测试（需要真实短信验证码）")
print("2. 自动化测试（需要测试验证码支持）")
print("3. 仅测试API连接")
choice = input("\n请输入选项 (1/2/3): ").strip()
if choice == "1":
interactive_test()
elif choice == "2":
automated_test()
elif choice == "3":
print_header("测试API连接")
result = make_request("GET", "/health", expected_status=200)
if result:
print_success("API连接正常")
else:
print_error("API连接失败")
else:
print_error("无效的选项")
print(f"\n{Colors.BOLD}{Colors.OKBLUE}测试完成{Colors.ENDC}\n")