shanxia/bx/bx-backend/app/routers/user.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import or_
from sqlalchemy.orm import Session

from app.database import get_db
from app.core.deps import require_roles
from app.core.security import hash_password
from app.models.user import User
from app.schemas.common import ApiResponse, PagedData
from app.schemas.user import UserCreate, UserUpdate, UserOut, UserBriefOut, UserStatsSummaryOut
from app.services.user import UserService

DEFAULT_PASSWORD = "123456"

router = APIRouter(prefix="/api/users", tags=["用户管理"])


@router.get("/", response_model=ApiResponse[PagedData[UserOut]])
def list_users(
    page: int = 1,
    pageSize: int = 10,
    keyword: str = "",
    status: int | None = None,
    db: Session = Depends(get_db),
    _: User = Depends(require_roles("ADMIN")),
):
    filters = [User.del_flag == 0]
    if status is not None:
        filters.append(User.status == status)
    if keyword:
        filters.append(or_(User.username.contains(keyword), User.real_name.contains(keyword)))
    rows, total = UserService.get_paged(db, page=page, page_size=pageSize, filters=filters)
    out_rows = [UserOut.model_validate(r) for r in rows]
    paged_data = PagedData(rows=out_rows, total=total, page=page, page_size=pageSize)
    return ApiResponse(data=paged_data)


@router.get("/dropdown", response_model=ApiResponse[list[UserBriefOut]])
def list_users_dropdown(db: Session = Depends(get_db), _: User = Depends(require_roles("ADMIN"))):
    rows = db.query(User).filter(User.del_flag == 0, User.status == 1).order_by(User.id).all()
    return ApiResponse(data=[UserBriefOut.model_validate(r) for r in rows])


@router.get("/stats/summary", response_model=ApiResponse[UserStatsSummaryOut])
def users_stats_summary(db: Session = Depends(get_db), _: User = Depends(require_roles("ADMIN"))):
    data = UserService.summary_stats(db)
    return ApiResponse(data=UserStatsSummaryOut.model_validate(data))


@router.get("/{user_id}", response_model=ApiResponse[UserOut])
def get_user(user_id: int, db: Session = Depends(get_db), _: User = Depends(require_roles("ADMIN"))):
    user = UserService.get_by_id(db, user_id)
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在")
    return ApiResponse(data=UserOut.model_validate(user))


@router.post("/", response_model=ApiResponse[UserOut])
def create_user(body: UserCreate, db: Session = Depends(get_db), _: User = Depends(require_roles("ADMIN"))):
    data = body.model_dump(exclude_unset=True)
    role_ids = data.pop("role_ids", None) or data.pop("roleIds", None) or []
    for key in list(data.keys()):
        if key in ("role_ids", "roleIds"):
            del data[key]
    user = UserService.create(db, data)
    if role_ids:
        from app.models.role import Role
        for rid in role_ids:
            role = db.query(Role).filter(Role.id == rid, Role.del_flag == 0).first()
            if role:
                user.roles.append(role)
        db.flush()
    return ApiResponse(data=UserOut.model_validate(user))


@router.put("/{user_id}", response_model=ApiResponse[UserOut])
def update_user(user_id: int, body: UserUpdate, db: Session = Depends(get_db), _: User = Depends(require_roles("ADMIN"))):
    data = body.model_dump(exclude_unset=True)
    role_ids = data.pop("role_ids", None) or data.pop("roleIds", None)
    for key in list(data.keys()):
        if key in ("role_ids", "roleIds"):
            del data[key]
    if data.get("password") == "":
        del data["password"]
    if "username" in data:
        name = (data.get("username") or "").strip()
        if not name:
            del data["username"]
        else:
            data["username"] = name
            taken = (
                db.query(User)
                .filter(User.username == name, User.del_flag == 0, User.id != user_id)
                .first()
            )
            if taken:
                raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名已被占用")
    user = UserService.update(db, user_id, data)
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在")
    if role_ids is not None:
        from app.models.role import Role
        user.roles.clear()
        for rid in role_ids:
            role = db.query(Role).filter(Role.id == rid, Role.del_flag == 0).first()
            if role:
                user.roles.append(role)
        db.flush()
    return ApiResponse(data=UserOut.model_validate(user))




@router.put("/{user_id}/reset-password", response_model=ApiResponse[None])
def reset_password(user_id: int, db: Session = Depends(get_db), _: User = Depends(require_roles("ADMIN"))):
    user = UserService.get_by_id(db, user_id)
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在")
    user.password = hash_password(DEFAULT_PASSWORD)
    db.commit()
    return ApiResponse(data=None)


@router.delete("/{user_id}", response_model=ApiResponse[None])
def delete_user(user_id: int, db: Session = Depends(get_db), _: User = Depends(require_roles("ADMIN"))):
    UserService.soft_delete(db, user_id)
    return ApiResponse(data=None)