#!/usr/bin/env python # -*- coding: utf-8 -*- """ 国际短信发送服务（准实时） 说明： - 从数据库扫描待发送的验证码（area_code != +86），使用阿里云国际短信 API 发送。 - 使用同步 SDK 调用并通过 asyncio.to_thread 包装为异步以不阻塞事件循环。 - 配置文件：send.conf（与脚本同目录），缺失配置会尝试从环境变量读取。 运行: python send_international_sms.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 sqlalchemy import select, update, or_, and_ try: from alibabacloud_dysmsapi20180501.client import Client as Dysmsapi20180501Client from alibabacloud_dysmsapi20180501 import models as dysms_models from alibabacloud_tea_openapi import models as open_api_models from alibabacloud_tea_util import models as util_models except Exception: print("❌ 缺少依赖包，请先安装:") print(" pip install alibabacloud-dysmsapi20180501 alibabacloud_tea_openapi alibabacloud_tea_util") sys.exit(1) # 加载配置文件 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() SENDER_ID = config.get("aliyun_sms", "sender_id", fallback="Lyrria") TEMPLATE_MESSAGE = config.get( "aliyun_sms", "template_message", fallback="[Lyrria] Your verification code is {code}. It expires in 5 minutes. Do not share this code with anyone." ) ENDPOINT = config.get("aliyun_sms", "endpoint", fallback="dysmsapi.ap-southeast-1.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")) SMS_TYPE = config.get("aliyun_sms", "sms_type", fallback="OTP") # OTP/NOTIFY/MKT def create_client() -> Dysmsapi20180501Client: """初始化阿里云国际短信客户端（从配置文件或环境读取 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 or environment." ) client_config = open_api_models.Config( access_key_id=ACCESS_KEY_ID, access_key_secret=ACCESS_KEY_SECRET, # endpoint for international API is region-specific; keep as configured ) client_config.endpoint = ENDPOINT return Dysmsapi20180501Client(client_config) async def send_sms_async( client: Dysmsapi20180501Client, to: str, message: str, sender_id: str, sms_type: str = "OTP", ) -> Tuple[bool, Optional[str]]: """ 发送国际短信（异步包装同步 SDK） Returns: (success, error_message) """ request = dysms_models.SendMessageToGlobeRequest() # 阿里云国际接口要求不带前导 '+'，去掉可能的 '+' 和空白 normalized_to = (to or "").lstrip("+").strip() request.to = normalized_to request.from_ = sender_id request.message = message request.type = sms_type try: resp = await asyncio.to_thread( client.send_message_to_globe_with_options, request, util_models.RuntimeOptions() ) body = getattr(resp, "body", None) if body and getattr(body, "response_code", None) == "OK": logger.info(f"✅ 国际短信发送成功: to={to}, message_id={getattr(body, 'message_id', '')}") return True, None else: error_desc = getattr(body, "response_description", "") if body else "No response body" error_code = getattr(body, "response_code", "") if body else "" error_msg = f"Code: {error_code}, Desc: {error_desc}" logger.error(f"❌ 国际短信发送失败: to={to}, {error_msg}") return False, error_msg except Exception as e: error_msg = f"Unexpected error: {e}" logger.error(f"❌ 国际短信发送异常: to={to}, {error_msg}", exc_info=True) return False, error_msg async def process_international_sms_codes(client: Dysmsapi20180501Client, db_manager: DatabaseManager): """扫描并发送国际短信验证码（area_code != +86）""" async with db_manager.get_session() as session: result = await session.execute( select(VerificationCode) .where( or_( VerificationCode.send_type == "international_sms", and_( 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) ) 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: area_code = (code_record.area_code or "").strip() phone = (code_record.phone or "").strip() code_value = code_record.code record_id = code_record.id if not phone: raise ValueError("empty phone") if area_code.startswith("+"): to = f"{area_code}{phone}" elif area_code: to = f"+{area_code}{phone}" else: to = phone message = TEMPLATE_MESSAGE.format(code=code_value) success, error_msg = await send_sms_async( client, to, message, SENDER_ID, sms_type=SMS_TYPE ) now = datetime.now() if success: await session.execute( update(VerificationCode) .where(VerificationCode.id == 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: await session.execute( update(VerificationCode) .where(VerificationCode.id == 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"❌ 处理国际验证码失败: id={getattr(code_record, 'id', 'n/a')}, error={e}", exc_info=True) try: now = datetime.now() await session.execute( update(VerificationCode) .where(VerificationCode.id == getattr(code_record, "id", None)) .values( is_valid=False, metadata_json={ **(getattr(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_international_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_international_sms.log，错误日志输出到 send_international_sms_error.log""" log_dir = Path(__file__).parent log_file = log_dir / "send_international_sms.log" error_log_file = log_dir / "send_international_sms_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() console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(logging.INFO) console_handler.setFormatter(formatter) root_logger.addHandler(console_handler) file_handler = logging.FileHandler(log_file, encoding='utf-8') file_handler.setLevel(logging.INFO) file_handler.setFormatter(formatter) root_logger.addHandler(file_handler) 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())