feat: 添加邮件功能第三阶段 - 密码重置

实现通过邮件自助重置密码功能： - 新增发送密码重置邮件API (/api/forgot-password) - 新增密码重置页面路由 (/reset-password/<token>) - 新增确认密码重置API (/api/reset-password-confirm) 新增文件： - templates/email/reset_password.html - 密码重置邮件模板 - templates/reset_password.html - 密码重置页面修改文件： - email_service.py - 添加密码重置相关函数 - send_password_reset_email() - verify_password_reset_token() - confirm_password_reset() - app.py - 添加密码重置相关API - templates/login.html - 忘记密码支持两种方式： - 启用邮件功能：通过邮件自助重置 - 未启用邮件：提交申请等待管理员审核 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-11 21:58:49 +08:00
parent 648cf0adf0
commit 0c0a5a7770
5 changed files with 691 additions and 4 deletions
--- a/email_service.py
+++ b/email_service.py
@@ -1225,6 +1225,166 @@ def resend_register_verification_email(user_id: int, email: str, username: str)
    return send_register_verification_email(email, username, user_id)


+# ============ 密码重置邮件 ============
+
+def send_password_reset_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}
+    """
+    # 检查发送频率限制（密码重置限制5分钟）
+    if not check_rate_limit(email, EMAIL_TYPE_RESET):
+        return {
+            'success': False,
+            'error': '发送太频繁，请5分钟后再试',
+            'token': None
+        }
+
+    # 使旧的重置token失效
+    with db_pool.get_db() as conn:
+        cursor = conn.cursor()
+        cursor.execute("""
+            UPDATE email_tokens SET used = 1
+            WHERE user_id = ? AND token_type = ? AND used = 0
+        """, (user_id, EMAIL_TYPE_RESET))
+        conn.commit()
+
+    # 生成新的验证Token
+    token = generate_email_token(email, EMAIL_TYPE_RESET, 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'
+
+    # 生成重置链接
+    reset_url = f"{base_url.rstrip('/')}/reset-password/{token}"
+
+    # 读取邮件模板
+    template_path = os.path.join(os.path.dirname(__file__), 'templates', 'email', 'reset_password.html')
+    try:
+        with open(template_path, 'r', encoding='utf-8') as f:
+            html_template = f.read()
+    except FileNotFoundError:
+        html_template = """
+        <html>
+        <body>
+            <h1>密码重置</h1>
+            <p>您好，{{ username }}！</p>
+            <p>请点击下面的链接重置您的密码：</p>
+            <p><a href="{{ reset_url }}">{{ reset_url }}</a></p>
+            <p>此链接30分钟内有效。</p>
+        </body>
+        </html>
+        """
+
+    # 替换模板变量
+    html_body = html_template.replace('{{ username }}', username)
+    html_body = html_body.replace('{{ reset_url }}', reset_url)
+
+    # 纯文本版本
+    text_body = f"""
+您好，{username}！
+
+我们收到了您的密码重置请求。请点击下面的链接重置您的密码：
+
+{reset_url}
+
+此链接30分钟内有效。
+
+如果您没有申请过密码重置，请忽略此邮件。
+"""
+
+    # 发送邮件
+    result = send_email(
+        to_email=email,
+        subject='【知识管理平台】密码重置',
+        body=text_body,
+        html_body=html_body,
+        email_type=EMAIL_TYPE_RESET,
+        user_id=user_id
+    )
+
+    if result['success']:
+        return {'success': True, 'error': '', 'token': token}
+    else:
+        return {'success': False, 'error': result['error'], 'token': None}
+
+
+def verify_password_reset_token(token: str) -> Optional[Dict[str, Any]]:
+    """
+    验证密码重置Token（不标记为已使用）
+
+    Returns:
+        成功返回 {'user_id': int, 'email': str}，失败返回 None
+    """
+    with db_pool.get_db() as conn:
+        cursor = conn.cursor()
+        cursor.execute("""
+            SELECT id, user_id, email, expires_at, used
+            FROM email_tokens
+            WHERE token = ? AND token_type = ?
+        """, (token, EMAIL_TYPE_RESET))
+
+        row = cursor.fetchone()
+        if not row:
+            return None
+
+        token_id, user_id, email, expires_at, used = row
+
+        # 检查是否已使用
+        if used:
+            return None
+
+        # 检查是否过期
+        if datetime.strptime(expires_at, '%Y-%m-%d %H:%M:%S') < datetime.now():
+            return None
+
+        return {'user_id': user_id, 'email': email, 'token_id': token_id}
+
+
+def confirm_password_reset(token: str) -> Optional[Dict[str, Any]]:
+    """
+    确认密码重置（标记Token为已使用）
+
+    Returns:
+        成功返回 {'user_id': int, 'email': str}，失败返回 None
+    """
+    result = verify_password_reset_token(token)
+    if not result:
+        return None
+
+    # 标记为已使用
+    with db_pool.get_db() as conn:
+        cursor = conn.cursor()
+        cursor.execute("""
+            UPDATE email_tokens SET used = 1 WHERE id = ?
+        """, (result['token_id'],))
+        conn.commit()
+
+    return {'user_id': result['user_id'], 'email': result['email']}
+
+
 # ============ 异步发送队列 ============

 class EmailQueue: