Tolan/sendintlmsg/send_international_sms.py

#!/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())