333 lines
14 KiB
Python
333 lines
14 KiB
Python
|
from fastapi import APIRouter, Depends, UploadFile, File
|
||
|
|
from typing import Optional
|
||
|
|
from app.models.models import UserInfo, PasswordChangeRequest, UserListResponse, CreateUserRequest, UpdateUserRequest, RoleInfo
|
||
|
|
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
|
||
|
|
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()
|
||
|
|
|
||
|
|
@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 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 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 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 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 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 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 users u
|
||
|
|
LEFT JOIN 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'],
|
||
|
|
meetings_created=0,
|
||
|
|
meetings_attended=0
|
||
|
|
)
|
||
|
|
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 users WHERE user_id = %s", (user_id,))
|
||
|
|
if not cursor.fetchone():
|
||
|
|
return create_api_response(code="404", message="用户不存在")
|
||
|
|
|
||
|
|
cursor.execute("DELETE FROM 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 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 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 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,
|
||
|
|
r.role_name,
|
||
|
|
(SELECT COUNT(*) FROM meetings WHERE user_id = u.user_id) as meetings_created,
|
||
|
|
(SELECT COUNT(*) FROM attendees WHERE user_id = u.user_id) as meetings_attended
|
||
|
|
FROM users u
|
||
|
|
LEFT JOIN 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, r.role_name
|
||
|
|
FROM users u
|
||
|
|
LEFT JOIN 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="用户不存在")
|
||
|
|
|
||
|
|
created_query = "SELECT COUNT(*) as count FROM meetings WHERE user_id = %s"
|
||
|
|
cursor.execute(created_query, (user_id,))
|
||
|
|
meetings_created = cursor.fetchone()['count']
|
||
|
|
|
||
|
|
attended_query = "SELECT COUNT(*) as count FROM attendees WHERE user_id = %s"
|
||
|
|
cursor.execute(attended_query, (user_id,))
|
||
|
|
meetings_attended = cursor.fetchone()['count']
|
||
|
|
|
||
|
|
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'],
|
||
|
|
meetings_created=meetings_created,
|
||
|
|
meetings_attended=meetings_attended
|
||
|
|
)
|
||
|
|
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 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 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 = AVATAR_DIR / str(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/avatar/{user_id}/{unique_filename}"
|
||
|
|
|
||
|
|
# Update database
|
||
|
|
with get_db_connection() as connection:
|
||
|
|
cursor = connection.cursor(dictionary=True)
|
||
|
|
|
||
|
|
cursor.execute("UPDATE 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})
|