237899745/zsglpt
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 添加邮件功能第二阶段 - 注册邮箱验证

Browse Source 
实现注册时的邮箱验证功能：
- 修改注册API支持邮箱验证流程
- 新增邮箱验证API (/api/verify-email/<token>)
- 新增重发验证邮件API (/api/resend-verify-email)
- 新增邮箱验证状态查询API (/api/email/verify-status)

新增文件：
- templates/email/register.html - 注册验证邮件模板
- templates/verify_success.html - 验证成功页面
- templates/verify_failed.html - 验证失败页面

修改文件：
- email_service.py - 添加发送注册验证邮件函数
- app.py - 添加邮箱验证相关API
- database.py - 添加get_user_by_email函数
- app_config.py - 添加BASE_URL配置
- templates/register.html - 支持邮箱必填切换
- templates/login.html - 添加重发验证邮件功能
- templates/admin.html - 添加注册验证开关和BASE_URL设置

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
yuyx
2025-12-11 21:51:07 +08:00
parent 966572cc94
commit de8edcb3a6
10 changed files with 794 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

182
email_service.py
View File

@@ -107,6 +107,8 @@ def init_email_tables():
id INTEGER PRIMARY KEY DEFAULT 1,
enabled INTEGER DEFAULT 0,
failover_enabled INTEGER DEFAULT 1,
register_verify_enabled INTEGER DEFAULT 0,
base_url TEXT DEFAULT '',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
@@ -188,26 +190,67 @@ def get_email_settings() -> Dict[str, Any]:
"""获取全局邮件设置"""
with db_pool.get_db() as conn:
cursor = conn.cursor()
cursor.execute("SELECT enabled, failover_enabled, updated_at FROM email_settings WHERE id = 1")
# 先检查表结构，添加新字段（兼容旧版本数据库）
try:
cursor.execute("SELECT register_verify_enabled FROM email_settings LIMIT 1")
except:
cursor.execute("ALTER TABLE email_settings ADD COLUMN register_verify_enabled INTEGER DEFAULT 0")
conn.commit()
try:
cursor.execute("SELECT base_url FROM email_settings LIMIT 1")
except:
cursor.execute("ALTER TABLE email_settings ADD COLUMN base_url TEXT DEFAULT ''")
conn.commit()
cursor.execute("""
SELECT enabled, failover_enabled, register_verify_enabled, base_url, updated_at
FROM email_settings WHERE id = 1
""")
row = cursor.fetchone()
if row:
return {
'enabled': bool(row[0]),
'failover_enabled': bool(row[1]),
'updated_at': row[2]
'register_verify_enabled': bool(row[2]) if row[2] is not None else False,
'base_url': row[3] or '',
'updated_at': row[4]
}
return {'enabled': False, 'failover_enabled': True, 'updated_at': None}
return {
'enabled': False,
'failover_enabled': True,
'register_verify_enabled': False,
'base_url': '',
'updated_at': None
}
def update_email_settings(enabled: bool, failover_enabled: bool) -> bool:
def update_email_settings(
enabled: bool,
failover_enabled: bool,
register_verify_enabled: bool = None,
base_url: str = None
) -> bool:
"""更新全局邮件设置"""
with db_pool.get_db() as conn:
cursor = conn.cursor()
cursor.execute("""
# 构建动态更新语句
updates = ['enabled = ?', 'failover_enabled = ?', 'updated_at = CURRENT_TIMESTAMP']
params = [int(enabled), int(failover_enabled)]
if register_verify_enabled is not None:
updates.append('register_verify_enabled = ?')
params.append(int(register_verify_enabled))
if base_url is not None:
updates.append('base_url = ?')
params.append(base_url)
cursor.execute(f"""
UPDATE email_settings
SET enabled = ?, failover_enabled = ?, updated_at = CURRENT_TIMESTAMP
SET {', '.join(updates)}
WHERE id = 1
""", (int(enabled), int(failover_enabled)))
""", params)
conn.commit()
return True
@@ -1057,6 +1100,131 @@ def cleanup_expired_tokens() -> int:
return deleted
# ============ 注册验证邮件 ============
def send_register_verification_email(
email: str,
username: str,
user_id: int,
base_url: str = None
) -> Dict[str, Any]:
"""
发送注册验证邮件
Args:
email: 用户邮箱
username: 用户名
user_id: 用户ID
base_url: 网站基础URL
Returns:
{'success': bool, 'error': str, 'token': str}
"""
# 检查发送频率限制
if not check_rate_limit(email, EMAIL_TYPE_REGISTER):
return {
'success': False,
'error': '发送太频繁，请稍后再试',
'token': None
}
# 生成验证Token
token = generate_email_token(email, EMAIL_TYPE_REGISTER, user_id)
# 获取base_url
if not base_url:
settings = get_email_settings()
base_url = settings.get('base_url', '')
if not base_url:
# 尝试从配置获取
try:
from app_config import Config
base_url = Config.BASE_URL
except:
base_url = 'http://localhost:51233'
# 生成验证链接
verify_url = f"{base_url.rstrip('/')}/api/verify-email/{token}"
# 读取邮件模板
template_path = os.path.join(os.path.dirname(__file__), 'templates', 'email', 'register.html')
try:
with open(template_path, 'r', encoding='utf-8') as f:
html_template = f.read()
except FileNotFoundError:
# 使用简单的HTML模板
html_template = """
<html>
<body>
<h1>邮箱验证</h1>
<p>您好，{{ username }}！</p>
<p>请点击下面的链接验证您的邮箱地址：</p>
<p><a href="{{ verify_url }}">{{ verify_url }}</a></p>
<p>此链接24小时内有效。</p>
</body>
</html>
"""
# 替换模板变量
html_body = html_template.replace('{{ username }}', username)
html_body = html_body.replace('{{ verify_url }}', verify_url)
# 纯文本版本
text_body = f"""
您好，{username}！
感谢您注册知识管理平台。请点击下面的链接验证您的邮箱地址：
{verify_url}
此链接24小时内有效。
如果您没有注册过账号，请忽略此邮件。
"""
# 发送邮件
result = send_email(
to_email=email,
subject='【知识管理平台】邮箱验证',
body=text_body,
html_body=html_body,
email_type=EMAIL_TYPE_REGISTER,
user_id=user_id
)
if result['success']:
return {'success': True, 'error': '', 'token': token}
else:
return {'success': False, 'error': result['error'], 'token': None}
def resend_register_verification_email(user_id: int, email: str, username: str) -> Dict[str, Any]:
"""
重发注册验证邮件
Args:
user_id: 用户ID
email: 用户邮箱
username: 用户名
Returns:
{'success': bool, 'error': str}
"""
# 检查是否有未过期的token
with db_pool.get_db() as conn:
cursor = conn.cursor()
# 先使旧token失效
cursor.execute("""
UPDATE email_tokens SET used = 1
WHERE user_id = ? AND token_type = ? AND used = 0
""", (user_id, EMAIL_TYPE_REGISTER))
conn.commit()
# 发送新的验证邮件
return send_register_verification_email(email, username, user_id)
# ============ 异步发送队列 ============
class EmailQueue: