Tolan/sendmsg/send_sms_gn.py

"""发送国内短信验证码服务（准实时）

从数据库读取待发送的验证码（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())