import logging import random import secrets import string import time from app.core.config import settings logger = logging.getLogger(__name__) SMS_CODE_TTL_SECONDS = 300 # 5分钟有效期 _sms_codes: dict[str, tuple[str, float]] = {} def _send_sms_aliyun(phone: str, code: str) -> bool: """通过阿里云发送短信""" try: from alibabacloud_dysmsapi20170525.client import Client from alibabacloud_dysmsapi20170525 import models as dysmsapi_models from alibabacloud_tea_openapi import models as open_api_models config = open_api_models.Config( access_key_id=settings.aliyun_access_key_id, access_key_secret=settings.aliyun_access_key_secret, ) config.endpoint = "dysmsapi.aliyuncs.com" client = Client(config) request = dysmsapi_models.SendSmsRequest( phone_numbers=phone, sign_name=settings.aliyun_sms_sign_name, template_code=settings.aliyun_sms_template_code, template_param=f'{{"code":"{code}"}}', ) response = client.send_sms(request) if response.body.code == "OK": logger.info(f"[阿里云短信] 发送成功: {phone}") return True else: logger.error(f"[阿里云短信] 发送失败: {phone}, 错误: {response.body.message}") return False except ImportError: logger.error("[阿里云短信] 未安装依赖,请执行: pip install alibabacloud-dysmsapi20170525") return False except Exception as e: logger.error(f"[阿里云短信] 发送异常: {phone}, 异常: {str(e)}") return False def _send_sms_local(phone: str, code: str) -> bool: """本地测试模式:将验证码打印到终端""" print(f"\n{'='*50}") print(f"[短信验证码] 手机号: {phone}") print(f"[短信验证码] 验证码: {code}") print(f"[短信验证码] 有效期: {SMS_CODE_TTL_SECONDS}秒") print(f"{'='*50}\n") logger.info(f"[本地短信] 验证码已生成: {phone} -> {code}") return True def send_sms(phone: str, code: str) -> bool: """发送短信验证码(根据配置选择发送方式)""" if settings.sms_mode == "aliyun": return _send_sms_aliyun(phone, code) else: return _send_sms_local(phone, code) def generate_sms_code(phone: str) -> str: """生成短信验证码""" clean_expired_codes() code = "".join(random.choices(string.digits, k=6)) sms_id = secrets.token_urlsafe(16) _sms_codes[sms_id] = (code, time.time() + SMS_CODE_TTL_SECONDS) # 发送短信 send_sms(phone, code) return sms_id def verify_sms_code(sms_id: str, code: str) -> bool: """验证短信验证码""" clean_expired_codes() record = _sms_codes.pop(sms_id, None) if not record: return False expected, expires_at = record if expires_at < time.time(): return False return expected == code.strip() def clean_expired_codes() -> None: """清理过期的验证码""" now = time.time() expired = [sms_id for sms_id, (_, expires_at) in _sms_codes.items() if expires_at < now] for sms_id in expired: _sms_codes.pop(sms_id, None)