goalfylearning-admin/scripts/test_email.py

#!/usr/bin/env python3
"""
邮件发送功能测试脚本

测试步骤：
1. 添加 language 字段到数据库
2. 创建中文和英文测试申请
3. 通过 Go 服务审批并发送邮件
"""

import os
import sys
import time
import pymysql
from datetime import datetime

# 数据库配置
DB_USER = os.getenv("DB_USER", "goalfymax_prod")
DB_PASSWORD = os.getenv("DB_PASSWORD", "X6cQDaOLOifFBOMq")
DB_HOST = os.getenv("DB_HOST", "goalfyagent-aurora-mysql-staging.cb2sq6y2mg93.us-west-2.rds.amazonaws.com")
DB_PORT = int(os.getenv("DB_PORT", "3306"))
DB_NAME = os.getenv("DB_NAME", "goalfymax_prod")

def get_db_connection():
    """获取数据库连接"""
    return pymysql.connect(
        host=DB_HOST,
        port=DB_PORT,
        user=DB_USER,
        password=DB_PASSWORD,
        database=DB_NAME,
        charset='utf8mb4',
        cursorclass=pymysql.cursors.DictCursor
    )

def add_language_column():
    """添加 language 字段"""
    print("\n=== 步骤 1: 添加 language 字段 ===")

    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            # 检查字段是否已存在
            cursor.execute("""
                SELECT COLUMN_NAME
                FROM INFORMATION_SCHEMA.COLUMNS
                WHERE TABLE_SCHEMA = %s
                AND TABLE_NAME = 'admin_invite_code_applications'
                AND COLUMN_NAME = 'language'
            """, (DB_NAME,))

            if cursor.fetchone():
                print("✓ language 字段已存在")
            else:
                # 添加字段
                cursor.execute("""
                    ALTER TABLE admin_invite_code_applications
                    ADD COLUMN language VARCHAR(10) DEFAULT 'zh' COMMENT '语言:zh-中文,en-英文' AFTER reason
                """)
                conn.commit()
                print("✓ language 字段添加成功")
    except Exception as e:
        print(f"✗ 添加字段失败: {e}")
        return False
    finally:
        conn.close()

    return True

def create_test_applications():
    """创建测试申请"""
    print("\n=== 步骤 2: 创建测试申请 ===")

    conn = get_db_connection()
    test_emails = [
        ("test_zh@example.com", "测试中文邮件", "zh"),
        ("test_en@example.com", "Testing English email", "en")
    ]

    created_ids = []

    try:
        with conn.cursor() as cursor:
            for email, reason, language in test_emails:
                # 删除旧的测试数据
                cursor.execute("""
                    DELETE FROM admin_invite_code_applications
                    WHERE email = %s
                """, (email,))

                # 创建新申请
                cursor.execute("""
                    INSERT INTO admin_invite_code_applications
                    (email, reason, language, status, created_at, updated_at)
                    VALUES (%s, %s, %s, 'pending', NOW(), NOW())
                """, (email, reason, language))

                created_ids.append(cursor.lastrowid)
                print(f"✓ 创建申请成功: {email} (ID: {cursor.lastrowid}, Language: {language})")

            conn.commit()
    except Exception as e:
        print(f"✗ 创建申请失败: {e}")
        return []
    finally:
        conn.close()

    return created_ids

def check_applications():
    """检查申请状态"""
    print("\n=== 查看测试申请 ===")

    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute("""
                SELECT id, email, reason, language, status, created_at
                FROM admin_invite_code_applications
                WHERE email LIKE 'test_%@example.com'
                ORDER BY id DESC
                LIMIT 10
            """)

            applications = cursor.fetchall()
            if applications:
                print("\n当前测试申请:")
                for app in applications:
                    print(f"  ID: {app['id']}, Email: {app['email']}, Language: {app['language']}, Status: {app['status']}")
            else:
                print("没有找到测试申请")

            return applications
    finally:
        conn.close()

def main():
    print("=" * 60)
    print("邮件发送功能测试")
    print("=" * 60)

    # 步骤 1: 添加 language 字段
    if not add_language_column():
        print("\n❌ 测试失败：无法添加 language 字段")
        sys.exit(1)

    # 步骤 2: 创建测试申请
    application_ids = create_test_applications()
    if not application_ids:
        print("\n❌ 测试失败：无法创建测试申请")
        sys.exit(1)

    # 步骤 3: 显示测试申请
    check_applications()

    print("\n" + "=" * 60)
    print("✓ 测试准备完成！")
    print("=" * 60)
    print("\n下一步操作：")
    print("1. 启动 Go 服务: ./scripts/start.sh")
    print("2. 登录后台管理系统")
    print("3. 在邀请码申请管理中审批这些测试申请")
    print("4. 检查邮件发送情况（中文和英文）")
    print("\n测试邮箱:")
    print("  - test_zh@example.com (中文)")
    print("  - test_en@example.com (英文)")
    print("\n或者使用 API 测试:")
    for app_id in application_ids:
        print(f"  curl -X POST http://localhost:8087/api/admin/invite-code/applications/approve \\")
        print(f"    -H 'Content-Type: application/json' \\")
        print(f"    -d '{{\"application_id\": {app_id}, \"valid_days\": 7}}'")

if __name__ == "__main__":
    main()