import json import random import time from utils.config import get_config_value try: from aliyunsdkdysmsapi.request.v20170525 import SendSmsRequest from aliyunsdkcore.client import AcsClient ALIBABA_SDK_AVAILABLE = True except ImportError: ALIBABA_SDK_AVAILABLE = False sms_code_store = {} def generate_sms_code(): return ''.join(random.choice('0123456789') for _ in range(4)) def send_sms(phone): use_real_sms = get_config_value('use_real_sms', 'false').lower() == 'true' if not use_real_sms: code = generate_sms_code() sms_code_store[phone] = { 'code': code, 'created_at': time.time(), 'expire_at': time.time() + 300 } return {'success': True, 'message': '验证码已发送(模拟)', 'code': code} if not ALIBABA_SDK_AVAILABLE: print(f'[短信发送失败] SDK未安装 - 手机号: {phone}') return {'success': False, 'message': '阿里云短信SDK未安装,请安装aliyun-python-sdk-dysmsapi'} access_key_id = get_config_value('aliyun_sms_access_key_id', '') access_key_secret = get_config_value('aliyun_sms_access_key_secret', '') sign_name = get_config_value('aliyun_sms_sign_name', '') template_code = get_config_value('aliyun_sms_template_code', '') region_id = get_config_value('aliyun_sms_region_id', 'cn-hangzhou') if not access_key_id or not access_key_secret: print(f'[短信发送失败] AccessKey未配置 - 手机号: {phone}, AccessKey ID: {access_key_id}, AccessKey Secret: {access_key_secret}') return {'success': False, 'message': '请配置阿里云短信AccessKey'} if not sign_name or not template_code: print(f'[短信发送失败] 签名或模板未配置 - 手机号: {phone}, 签名: {sign_name}, 模板: {template_code}') return {'success': False, 'message': '请配置短信签名和模板'} code = generate_sms_code() try: client = AcsClient(access_key_id, access_key_secret, region_id) request = SendSmsRequest.SendSmsRequest() request.set_accept_format('json') request.set_PhoneNumbers(phone) request.set_SignName(sign_name) request.set_TemplateCode(template_code) request.set_TemplateParam(json.dumps({'code': code})) print(f'[短信发送尝试] 手机号: {phone}, AccessKey ID: {access_key_id}, AccessKey Secret: {access_key_secret}, 签名: {sign_name}, 模板: {template_code}') response = client.do_action_with_exception(request) result = json.loads(response.decode('utf-8')) if result.get('Code') == 'OK': sms_code_store[phone] = { 'code': code, 'created_at': time.time(), 'expire_at': time.time() + 300 } print(f'[短信发送成功] 手机号: {phone}, 验证码: {code}') return {'success': True, 'message': '验证码发送成功'} else: print(f'[短信发送失败] 阿里云返回错误 - 手机号: {phone}, AccessKey ID: {access_key_id}, AccessKey Secret: {access_key_secret}, 错误码: {result.get("Code")}, 错误信息: {result.get("Message")}') return {'success': False, 'message': result.get('Message', '发送失败')} except Exception as e: print(f'[短信发送异常] - 手机号: {phone}, AccessKey ID: {access_key_id}, AccessKey Secret: {access_key_secret}, 异常信息: {str(e)}') return {'success': False, 'message': str(e)} def verify_sms_code(phone, code): use_real_sms = get_config_value('use_real_sms', 'false').lower() == 'true' if not use_real_sms: stored = sms_code_store.get(phone) if stored and stored['code'] == code and time.time() < stored['expire_at']: del sms_code_store[phone] return True return False stored = sms_code_store.get(phone) if stored and stored['code'] == code and time.time() < stored['expire_at']: del sms_code_store[phone] return True return False def get_sms_config(): return { 'use_real_sms': get_config_value('use_real_sms', 'false'), 'aliyun_sms_access_key_id': get_config_value('aliyun_sms_access_key_id', ''), 'aliyun_sms_sign_name': get_config_value('aliyun_sms_sign_name', ''), 'aliyun_sms_template_code': get_config_value('aliyun_sms_template_code', ''), 'aliyun_sms_region_id': get_config_value('aliyun_sms_region_id', 'cn-hangzhou') }