from fastapi import APIRouter, Depends, UploadFile, File from typing import Optional from app.models.models import UserInfo, PasswordChangeRequest, UserListResponse, CreateUserRequest, UpdateUserRequest, RoleInfo, UserMcpInfo from app.core.database import get_db_connection from app.core.auth import get_current_user from app.core.response import create_api_response from app.services.system_config_service import SystemConfigService import app.core.config as config_module from app.core.config import UPLOAD_DIR, AVATAR_DIR import hashlib import datetime import re import os import shutil import uuid import secrets from pathlib import Path router = APIRouter() def validate_email(email: str) -> bool: """Basic email validation""" pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' return re.match(pattern, email) is not None def hash_password(password: str) -> str: return hashlib.sha256(password.encode()).hexdigest() def _generate_mcp_bot_id() -> str: return f"nexbot_{secrets.token_hex(8)}" def _generate_mcp_bot_secret() -> str: random_part = secrets.token_urlsafe(24).replace('-', '').replace('_', '') return f"nxbotsec_{random_part}" def _get_user_mcp_record(cursor, user_id: int): cursor.execute( """ SELECT id, user_id, bot_id, bot_secret, status, last_used_at, created_at, updated_at FROM sys_user_mcp WHERE user_id = %s """, (user_id,), ) return cursor.fetchone() def _ensure_user_exists(cursor, user_id: int) -> bool: cursor.execute("SELECT user_id FROM sys_users WHERE user_id = %s", (user_id,)) return cursor.fetchone() is not None def _serialize_user_mcp(record: dict) -> dict: return UserMcpInfo(**record).dict() def _ensure_user_mcp_record(connection, cursor, user_id: int): record = _get_user_mcp_record(cursor, user_id) if record: return record bot_id = _generate_mcp_bot_id() while True: cursor.execute("SELECT id FROM sys_user_mcp WHERE bot_id = %s", (bot_id,)) if not cursor.fetchone(): break bot_id = _generate_mcp_bot_id() cursor.execute( """ INSERT INTO sys_user_mcp (user_id, bot_id, bot_secret, status, last_used_at, created_at, updated_at) VALUES (%s, %s, %s, 1, NULL, NOW(), NOW()) """, (user_id, bot_id, _generate_mcp_bot_secret()), ) connection.commit() return _get_user_mcp_record(cursor, user_id) @router.get("/roles") def get_all_roles(current_user: dict = Depends(get_current_user)): """获取所有角色列表""" if current_user['role_id'] != 1: # 1 is admin return create_api_response(code="403", message="仅管理员有权限查看角色列表") with get_db_connection() as connection: cursor = connection.cursor(dictionary=True) cursor.execute("SELECT role_id, role_name FROM sys_roles ORDER BY role_id") roles = cursor.fetchall() return create_api_response(code="200", message="获取角色列表成功", data=[RoleInfo(**role).dict() for role in roles]) @router.post("/users") def create_user(request: CreateUserRequest, current_user: dict = Depends(get_current_user)): if current_user['role_id'] != 1: # 1 is admin return create_api_response(code="403", message="仅管理员有权限创建用户") if not validate_email(request.email): return create_api_response(code="400", message="邮箱格式不正确") with get_db_connection() as connection: cursor = connection.cursor(dictionary=True) cursor.execute("SELECT user_id FROM sys_users WHERE username = %s", (request.username,)) if cursor.fetchone(): return create_api_response(code="400", message="用户名已存在") password = request.password if request.password else SystemConfigService.get_default_reset_password() hashed_password = hash_password(password) query = "INSERT INTO sys_users (username, password_hash, caption, email, avatar_url, role_id, created_at) VALUES (%s, %s, %s, %s, %s, %s, %s)" created_at = datetime.datetime.utcnow() cursor.execute(query, (request.username, hashed_password, request.caption, request.email, request.avatar_url, request.role_id, created_at)) connection.commit() return create_api_response(code="200", message="用户创建成功") @router.put("/users/{user_id}") def update_user(user_id: int, request: UpdateUserRequest, current_user: dict = Depends(get_current_user)): # Allow admin (role_id=1) or self if current_user['role_id'] != 1 and current_user['user_id'] != user_id: return create_api_response(code="403", message="没有权限修改此用户信息") if request.email and not validate_email(request.email): return create_api_response(code="400", message="邮箱格式不正确") with get_db_connection() as connection: cursor = connection.cursor(dictionary=True) cursor.execute("SELECT user_id, username, caption, email, avatar_url, role_id FROM sys_users WHERE user_id = %s", (user_id,)) existing_user = cursor.fetchone() if not existing_user: return create_api_response(code="404", message="用户不存在") if request.username and request.username != existing_user['username']: cursor.execute("SELECT user_id FROM sys_users WHERE username = %s AND user_id != %s", (request.username, user_id)) if cursor.fetchone(): return create_api_response(code="400", message="用户名已存在") # Restrict role_id update to admins only target_role_id = existing_user['role_id'] if current_user['role_id'] == 1 and request.role_id is not None: target_role_id = request.role_id update_data = { 'username': request.username if request.username else existing_user['username'], 'caption': request.caption if request.caption else existing_user['caption'], 'email': request.email if request.email else existing_user['email'], 'avatar_url': request.avatar_url if request.avatar_url is not None else existing_user.get('avatar_url'), 'role_id': target_role_id } query = "UPDATE sys_users SET username = %s, caption = %s, email = %s, avatar_url = %s, role_id = %s WHERE user_id = %s" cursor.execute(query, (update_data['username'], update_data['caption'], update_data['email'], update_data['avatar_url'], update_data['role_id'], user_id)) connection.commit() cursor.execute(''' SELECT u.user_id, u.username, u.caption, u.email, u.avatar_url, u.created_at, u.role_id, r.role_name FROM sys_users u LEFT JOIN sys_roles r ON u.role_id = r.role_id WHERE u.user_id = %s ''', (user_id,)) updated_user = cursor.fetchone() user_info = UserInfo( user_id=updated_user['user_id'], username=updated_user['username'], caption=updated_user['caption'], email=updated_user['email'], avatar_url=updated_user['avatar_url'], created_at=updated_user['created_at'], role_id=updated_user['role_id'], role_name=updated_user['role_name'] or '普通用户' ) return create_api_response(code="200", message="用户信息更新成功", data=user_info.dict()) @router.delete("/users/{user_id}") def delete_user(user_id: int, current_user: dict = Depends(get_current_user)): if current_user['role_id'] != 1: # 1 is admin return create_api_response(code="403", message="仅管理员有权限删除用户") with get_db_connection() as connection: cursor = connection.cursor(dictionary=True) cursor.execute("SELECT user_id FROM sys_users WHERE user_id = %s", (user_id,)) if not cursor.fetchone(): return create_api_response(code="404", message="用户不存在") cursor.execute("DELETE FROM sys_users WHERE user_id = %s", (user_id,)) connection.commit() return create_api_response(code="200", message="用户删除成功") @router.post("/users/{user_id}/reset-password") def reset_password(user_id: int, current_user: dict = Depends(get_current_user)): if current_user['role_id'] != 1: # 1 is admin return create_api_response(code="403", message="仅管理员有权限重置密码") with get_db_connection() as connection: cursor = connection.cursor(dictionary=True) cursor.execute("SELECT user_id FROM sys_users WHERE user_id = %s", (user_id,)) if not cursor.fetchone(): return create_api_response(code="404", message="用户不存在") hashed_password = hash_password(SystemConfigService.get_default_reset_password()) query = "UPDATE sys_users SET password_hash = %s WHERE user_id = %s" cursor.execute(query, (hashed_password, user_id)) connection.commit() return create_api_response(code="200", message=f"用户 {user_id} 的密码已重置") @router.get("/users") def get_all_users( page: int = 1, size: int = 10, role_id: Optional[int] = None, search: Optional[str] = None, current_user: dict = Depends(get_current_user) ): with get_db_connection() as connection: cursor = connection.cursor(dictionary=True) # 构建WHERE条件 where_conditions = [] count_params = [] if role_id is not None: where_conditions.append("u.role_id = %s") count_params.append(role_id) if search: search_pattern = f"%{search}%" where_conditions.append("(username LIKE %s OR caption LIKE %s)") count_params.extend([search_pattern, search_pattern]) # 统计查询 count_query = "SELECT COUNT(*) as total FROM sys_users u" if where_conditions: count_query += " WHERE " + " AND ".join(where_conditions) cursor.execute(count_query, tuple(count_params)) total = cursor.fetchone()['total'] offset = (page - 1) * size # 主查询 query = ''' SELECT u.user_id, u.username, u.caption, u.email, u.avatar_url, u.created_at, u.role_id, COALESCE(r.role_name, '普通用户') AS role_name FROM sys_users u LEFT JOIN sys_roles r ON u.role_id = r.role_id ''' query_params = [] if where_conditions: query += " WHERE " + " AND ".join(where_conditions) query_params.extend(count_params) query += ''' ORDER BY u.user_id ASC LIMIT %s OFFSET %s ''' query_params.extend([size, offset]) cursor.execute(query, tuple(query_params)) users = cursor.fetchall() user_list = [UserInfo(**user) for user in users] response_data = UserListResponse(users=user_list, total=total) return create_api_response(code="200", message="获取用户列表成功", data=response_data.dict()) @router.get("/users/{user_id}") def get_user_info(user_id: int, current_user: dict = Depends(get_current_user)): with get_db_connection() as connection: cursor = connection.cursor(dictionary=True) user_query = ''' SELECT u.user_id, u.username, u.caption, u.email, u.avatar_url, u.created_at, u.role_id, COALESCE(r.role_name, '普通用户') AS role_name FROM sys_users u LEFT JOIN sys_roles r ON u.role_id = r.role_id WHERE u.user_id = %s ''' cursor.execute(user_query, (user_id,)) user = cursor.fetchone() if not user: return create_api_response(code="404", message="用户不存在") user_info = UserInfo( user_id=user['user_id'], username=user['username'], caption=user['caption'], email=user['email'], avatar_url=user['avatar_url'], created_at=user['created_at'], role_id=user['role_id'], role_name=user['role_name'] ) return create_api_response(code="200", message="获取用户信息成功", data=user_info.dict()) @router.put("/users/{user_id}/password") def update_password(user_id: int, request: PasswordChangeRequest, current_user: dict = Depends(get_current_user)): if user_id != current_user['user_id'] and current_user['role_id'] != 1: return create_api_response(code="403", message="没有权限修改其他用户的密码") with get_db_connection() as connection: cursor = connection.cursor(dictionary=True) cursor.execute("SELECT password_hash FROM sys_users WHERE user_id = %s", (user_id,)) user = cursor.fetchone() if not user: return create_api_response(code="404", message="用户不存在") if current_user['role_id'] != 1: if user['password_hash'] != hash_password(request.old_password): return create_api_response(code="400", message="旧密码错误") new_password_hash = hash_password(request.new_password) cursor.execute("UPDATE sys_users SET password_hash = %s WHERE user_id = %s", (new_password_hash, user_id)) connection.commit() return create_api_response(code="200", message="密码修改成功") @router.post("/users/{user_id}/avatar") def upload_user_avatar( user_id: int, file: UploadFile = File(...), current_user: dict = Depends(get_current_user) ): # Allow admin or self if current_user['role_id'] != 1 and current_user['user_id'] != user_id: return create_api_response(code="403", message="没有权限上传此用户头像") # Validate file type ALLOWED_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.gif', '.webp'} file_ext = os.path.splitext(file.filename)[1].lower() if file_ext not in ALLOWED_EXTENSIONS: return create_api_response(code="400", message="不支持的文件类型") # Ensure upload directory exists: AVATAR_DIR / str(user_id) user_avatar_dir = config_module.get_user_avatar_dir(user_id) if not user_avatar_dir.exists(): os.makedirs(user_avatar_dir) # Generate unique filename unique_filename = f"{uuid.uuid4()}{file_ext}" file_path = user_avatar_dir / unique_filename # Save file with open(file_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) # Generate URL (relative) # AVATAR_DIR is uploads/user/avatar # file path is uploads/user/avatar/{user_id}/{filename} # URL should be /uploads/user/avatar/{user_id}/{filename} avatar_url = f"/uploads/user/{user_id}/avatar/{unique_filename}" # Update database with get_db_connection() as connection: cursor = connection.cursor(dictionary=True) cursor.execute("UPDATE sys_users SET avatar_url = %s WHERE user_id = %s", (avatar_url, user_id)) connection.commit() return create_api_response(code="200", message="头像上传成功", data={"avatar_url": avatar_url}) @router.get("/users/{user_id}/mcp-config") def get_user_mcp_config(user_id: int, current_user: dict = Depends(get_current_user)): if current_user['role_id'] != 1 and current_user['user_id'] != user_id: return create_api_response(code="403", message="没有权限查看该用户的MCP配置") with get_db_connection() as connection: cursor = connection.cursor(dictionary=True) if not _ensure_user_exists(cursor, user_id): return create_api_response(code="404", message="用户不存在") record = _ensure_user_mcp_record(connection, cursor, user_id) return create_api_response(code="200", message="获取MCP配置成功", data=_serialize_user_mcp(record)) @router.post("/users/{user_id}/mcp-config/regenerate") def regenerate_user_mcp_secret(user_id: int, current_user: dict = Depends(get_current_user)): if current_user['role_id'] != 1 and current_user['user_id'] != user_id: return create_api_response(code="403", message="没有权限更新该用户的MCP配置") with get_db_connection() as connection: cursor = connection.cursor(dictionary=True) if not _ensure_user_exists(cursor, user_id): return create_api_response(code="404", message="用户不存在") record = _get_user_mcp_record(cursor, user_id) if not record: record = _ensure_user_mcp_record(connection, cursor, user_id) else: cursor.execute( """ UPDATE sys_user_mcp SET bot_secret = %s, status = 1, updated_at = NOW() WHERE user_id = %s """, (_generate_mcp_bot_secret(), user_id), ) connection.commit() record = _get_user_mcp_record(cursor, user_id) return create_api_response(code="200", message="MCP Secret 已重新生成", data=_serialize_user_mcp(record))