feat():learning后台管理项目初始化
This commit is contained in:
yuj
174
scripts/test_email.py
Executable file
174
scripts/test_email.py
Executable file
@@ -0,0 +1,174 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user