"""发送国内短信验证码服务（准实时） 从数据库读取待发送的验证码（area_code=+86），使用阿里云短信服务发送。 Prerequisites: - Python >= 3.7 - `pip install alibabacloud_dysmsapi20170525==4.3.0` - 配置文件：send.conf（与脚本同目录） - 必需配置：[aliyun_sms] 下的 access_key_id 和 access_key_secret - 必需配置：[database] 下的数据库配置 运行: python send_sms_gn.py """ import asyncio import configparser import os import sys import logging from datetime import datetime from pathlib import Path from typing import Dict, Optional, Tuple from alibabacloud_dysmsapi20170525.client import Client as Dysmsapi20170525Client from alibabacloud_dysmsapi20170525 import models as dysms_models from alibabacloud_tea_openapi import models as open_api_models from alibabacloud_tea_util import models as util_models from alibabacloud_tea_util.client import Client as UtilClient from Tea.exceptions import UnretryableException, TeaException from sqlalchemy import select, update # 加载配置文件 config_file = Path(__file__).parent / "send.conf" config = configparser.ConfigParser() if config_file.exists(): config.read(config_file, encoding='utf-8') else: raise FileNotFoundError(f"配置文件不存在: {config_file}") # 导入数据库相关模块（需要从 backend 目录导入） backend_dir = Path(__file__).parent.parent / "backend" sys.path.insert(0, str(backend_dir)) from database import DatabaseManager from models_db import VerificationCode logger = logging.getLogger(__name__) # ------- 从配置文件读取参数 ------- # 阿里云短信配置 ACCESS_KEY_ID = config.get("aliyun_sms", "access_key_id", fallback="").strip() ACCESS_KEY_SECRET = config.get("aliyun_sms", "access_key_secret", fallback="").strip() SIGN_NAME = config.get("aliyun_sms", "sign_name", fallback="灵语科技") TEMPLATE_CODE = config.get("aliyun_sms", "template_code", fallback="SMS_498720756") ENDPOINT = config.get("aliyun_sms", "endpoint", fallback="dysmsapi.aliyuncs.com") # 数据库配置 MYSQL_HOST = config.get("database", "mysql_host", fallback="").strip() MYSQL_PORT = config.get("database", "mysql_port", fallback="").strip() MYSQL_USER = config.get("database", "mysql_user", fallback="").strip() MYSQL_PASSWORD = config.get("database", "mysql_password", fallback="").strip() MYSQL_DATABASE = config.get("database", "mysql_database", fallback="").strip() # 服务配置 POLL_INTERVAL = float(config.get("service", "poll_interval", fallback="2.0")) def create_client() -> Dysmsapi20170525Client: """初始化阿里云短信客户端（从配置文件读取 AccessKey）""" if not ACCESS_KEY_ID or not ACCESS_KEY_SECRET: raise RuntimeError( "Missing AccessKey. Please set access_key_id and access_key_secret " "in send.conf file." ) client_config = open_api_models.Config( access_key_id=ACCESS_KEY_ID, access_key_secret=ACCESS_KEY_SECRET, endpoint=ENDPOINT, ) return Dysmsapi20170525Client(client_config) async def send_sms_async( client: Dysmsapi20170525Client, phone: str, code: str, ) -> Tuple[bool, Optional[str]]: """ 发送短信验证码（异步包装） Args: client: 阿里云短信客户端 phone: 手机号 code: 验证码 Returns: (是否成功, 错误信息) """ template_params: Dict[str, str] = {"code": code} request = dysms_models.SendSmsRequest( phone_numbers=phone, sign_name=SIGN_NAME, template_code=TEMPLATE_CODE, template_param=UtilClient.to_jsonstring(template_params), ) try: # 使用 asyncio.to_thread 将同步调用转为异步 resp = await asyncio.to_thread( client.send_sms_with_options, request, util_models.RuntimeOptions() ) # 检查响应 if resp.body.code == "OK": logger.info(f"✅ 短信发送成功: phone={phone}, code_id={code[:4]}***") return True, None else: error_msg = f"Code: {resp.body.code}, Message: {resp.body.message}" logger.error(f"❌ 短信发送失败: phone={phone}, {error_msg}") return False, error_msg except UnretryableException as e: error_msg = f"Network error: {e}" logger.error(f"❌ 短信发送网络错误: phone={phone}, {error_msg}") return False, error_msg except TeaException as e: error_msg = f"Business error: {e.message}" if e.data: error_msg += f", Recommend: {e.data.get('Recommend', '')}" logger.error(f"❌ 短信发送业务错误: phone={phone}, {error_msg}") return False, error_msg except Exception as e: error_msg = f"Unexpected error: {e}" logger.error(f"❌ 短信发送异常: phone={phone}, {error_msg}") return False, error_msg async def process_domestic_sms_codes(client: Dysmsapi20170525Client, db_manager: DatabaseManager): """扫描并发送国内短信验证码（area_code=+86）""" async with db_manager.get_session() as session: # 查询未发送的国内短信验证码（area_code=+86） result = await session.execute( select(VerificationCode) .where( VerificationCode.send_type == "domestic_sms", VerificationCode.area_code == "+86", VerificationCode.is_sent == False, VerificationCode.is_valid == True, VerificationCode.phone.isnot(None) ) .order_by(VerificationCode.created_at.asc()) .limit(50) # 每次处理50条 ) codes = result.scalars().all() if not codes: return 0, 0 logger.info(f"📋 找到 {len(codes)} 条待发送的国内短信验证码") success_count = 0 fail_count = 0 for code_record in codes: try: # 发送短信 success, error_msg = await send_sms_async( client, code_record.phone, code_record.code ) # 更新数据库状态（使用显式 UPDATE，避免 ORM 同步问题） now = datetime.now() if success: await session.execute( update(VerificationCode) .where(VerificationCode.id == code_record.id) .values( is_sent=True, sent_at=now, metadata_json={ **(code_record.metadata_json or {}), "send_status": "success", "send_at": now.isoformat(), }, ) ) success_count += 1 else: # 失败：更新 metadata，并将验证码标记为失效，避免重复发送 await session.execute( update(VerificationCode) .where(VerificationCode.id == code_record.id) .values( is_valid=False, metadata_json={ **(code_record.metadata_json or {}), "send_status": "failed", "error": error_msg, "send_at": now.isoformat(), }, ) ) fail_count += 1 await session.commit() except Exception as e: logger.error(f"❌ 处理验证码失败: code_id={code_record.code_id}, error={e}") try: now = datetime.now() await session.execute( update(VerificationCode) .where(VerificationCode.id == code_record.id) .values( is_valid=False, metadata_json={ **(code_record.metadata_json or {}), "send_status": "error", "error": str(e), "send_at": now.isoformat(), }, ) ) await session.commit() except: pass fail_count += 1 if success_count > 0 or fail_count > 0: logger.info(f"📊 处理完成: 成功 {success_count} 条, 失败 {fail_count} 条") return success_count, fail_count async def main_loop(): """主循环：准实时轮询并发送验证码""" # 设置数据库配置到环境变量（DatabaseManager 从环境变量读取） if MYSQL_HOST: os.environ["MYSQL_HOST"] = MYSQL_HOST if MYSQL_PORT: os.environ["MYSQL_PORT"] = MYSQL_PORT if MYSQL_USER: os.environ["MYSQL_USER"] = MYSQL_USER if MYSQL_PASSWORD: os.environ["MYSQL_PASSWORD"] = MYSQL_PASSWORD if MYSQL_DATABASE: os.environ["MYSQL_DATABASE"] = MYSQL_DATABASE db_manager = DatabaseManager() client = None try: # 初始化数据库连接 await db_manager.connect() logger.info("✅ 数据库连接成功") # 初始化短信客户端 client = create_client() logger.info("✅ 阿里云短信客户端初始化成功") logger.info(f"🚀 开始准实时发送服务（轮询间隔: {POLL_INTERVAL}秒）...") # 主循环：持续轮询 while True: try: await process_domestic_sms_codes(client, db_manager) except Exception as e: logger.error(f"❌ 处理验证码时出错: {e}", exc_info=True) # 等待下次轮询 await asyncio.sleep(POLL_INTERVAL) except KeyboardInterrupt: logger.info("🛑 收到中断信号，正在关闭...") except Exception as e: logger.error(f"❌ 服务异常: {e}", exc_info=True) sys.exit(1) finally: await db_manager.disconnect() logger.info("✅ 服务已关闭") def setup_logging(): """配置日志：正常日志输出到 send_sms_gn.log，错误日志输出到 send_sms_gn_error.log""" log_dir = Path(__file__).parent log_file = log_dir / "send_sms_gn.log" error_log_file = log_dir / "send_sms_gn_error.log" # 创建格式化器 formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) # 配置根日志记录器 root_logger = logging.getLogger() root_logger.setLevel(logging.INFO) # 清除已有的处理器 root_logger.handlers.clear() # 1. 控制台处理器（INFO及以上） console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(logging.INFO) console_handler.setFormatter(formatter) root_logger.addHandler(console_handler) # 2. 文件处理器（所有日志，INFO及以上） file_handler = logging.FileHandler(log_file, encoding='utf-8') file_handler.setLevel(logging.INFO) file_handler.setFormatter(formatter) root_logger.addHandler(file_handler) # 3. 错误日志文件处理器（ERROR及以上） error_file_handler = logging.FileHandler(error_log_file, encoding='utf-8') error_file_handler.setLevel(logging.ERROR) error_file_handler.setFormatter(formatter) root_logger.addHandler(error_file_handler) logger.info(f"✅ 日志配置完成：正常日志 -> {log_file}, 错误日志 -> {error_log_file}") if __name__ == "__main__": setup_logging() asyncio.run(main_loop())