yaoyaoai/app/services/admin_user.py

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.roles import ROLE_ADMIN, ROLE_AGENT, ROLE_OEM
from app.core.security import hash_password
from app.models.user import User
from app.schemas.admin_user import AdminUserCreate, AdminUserUpdate

DEFAULT_PAGE_SIZE = 20
MAX_PAGE_SIZE = 200


class AdminUserError(Exception):
    def __init__(self, message: str) -> None:
        self.message = message
        super().__init__(message)


def _list_users_filters(
    *,
    username: str | None = None,
    device_serial: str | None = None,
    oem_id: int | None = None,
    agent_id: int | None = None,
    role_id: int | None = None,
) -> list:
    conditions = []
    if username and username.strip():
        conditions.append(User.username.ilike(f"%{username.strip()}%"))
    if device_serial and device_serial.strip():
        conditions.append(User.device_serial.ilike(f"%{device_serial.strip()}%"))
    if oem_id is not None:
        conditions.append(User.oem_id == oem_id)
    if agent_id is not None:
        conditions.append(User.agent_id == agent_id)
    if role_id is not None:
        conditions.append(User.role_id == role_id)
    return conditions


async def list_users(
    db: AsyncSession,
    *,
    page: int = 1,
    page_size: int = DEFAULT_PAGE_SIZE,
    username: str | None = None,
    device_serial: str | None = None,
    oem_id: int | None = None,
    agent_id: int | None = None,
    role_id: int | None = None,
) -> tuple[list[User], int]:
    page_size = min(max(page_size, 1), MAX_PAGE_SIZE)
    page = max(page, 1)
    offset = (page - 1) * page_size
    conditions = _list_users_filters(
        username=username,
        device_serial=device_serial,
        oem_id=oem_id,
        agent_id=agent_id,
        role_id=role_id,
    )

    count_stmt = select(func.count()).select_from(User)
    list_stmt = select(User).order_by(User.id.desc())
    if conditions:
        count_stmt = count_stmt.where(*conditions)
        list_stmt = list_stmt.where(*conditions)

    total = await db.scalar(count_stmt) or 0
    result = await db.scalars(list_stmt.offset(offset).limit(page_size))
    return list(result.all()), total


async def _validate_owner_ids(
    db: AsyncSession,
    *,
    oem_id: int | None,
    agent_id: int | None,
) -> None:
    if oem_id is not None:
        oem = await db.get(User, oem_id)
        if oem is None or oem.role_id != ROLE_OEM:
            raise AdminUserError("所选 OEM 不存在")
    if agent_id is not None:
        agent = await db.get(User, agent_id)
        if agent is None or agent.role_id != ROLE_AGENT:
            raise AdminUserError("所选代理不存在")


async def create_user(db: AsyncSession, body: AdminUserCreate,oem: User) -> User:
    existing = await db.scalar(select(User).where(User.username == body.username))
    if existing is not None:
        raise AdminUserError("用户名已存在")

    if body.phone:
        phone_taken = await db.scalar(select(User).where(User.phone == body.phone))
        if phone_taken is not None:
            raise AdminUserError("手机号已被使用")

    await _validate_owner_ids(db, oem_id=oem.id, agent_id=body.agent_id)

    user = User(
        username=body.username,
        password_hash=hash_password(body.password),
        phone=body.phone,
        role_id=body.role_id,
        oem_id=oem.id,
        agent_id=body.agent_id,
        vip_end_time=body.vip_end_time,
    )
    db.add(user)
    await db.flush()
    await db.refresh(user)
    return user


async def update_user(
    db: AsyncSession,
    oem: User,
    user_id: int,
    body: AdminUserUpdate,
    *,
    actor_id: int,
) -> User:
    user = await db.get(User, user_id)
    if user is None:
        raise AdminUserError("用户不存在")
    body.oem_id=oem.id
    if body.username is not None and body.username != user.username:
        taken = await db.scalar(select(User).where(User.username == body.username))
        if taken is not None:
            raise AdminUserError("用户名已存在")
        user.username = body.username

    if body.phone is not None:
        if body.phone != user.phone:
            taken = await db.scalar(select(User).where(User.phone == body.phone))
            if taken is not None:
                raise AdminUserError("手机号已被使用")
        user.phone = body.phone

    if body.role_id is not None:
        if user_id == actor_id and body.role_id != ROLE_ADMIN:
            raise AdminUserError("不能修改自己的管理员角色")
        user.role_id = body.role_id

    if body.oem_id is not None or body.agent_id is not None:
        await _validate_owner_ids(
            db,
            oem_id=body.oem_id if body.oem_id is not None else user.oem_id,
            agent_id=body.agent_id if body.agent_id is not None else user.agent_id,
        )
    if body.oem_id is not None:
        user.oem_id = body.oem_id
    if body.agent_id is not None:
        user.agent_id = body.agent_id

    if body.password:
        user.password_hash = hash_password(body.password)

    if body.clear_vip_end_time:
        user.vip_end_time = None
    elif body.vip_end_time is not None:
        user.vip_end_time = body.vip_end_time

    await db.flush()
    await db.refresh(user)
    return user


async def clear_device_serial(db: AsyncSession, user_id: int) -> User:
    """管理员解绑：将用户 device_serial 置空。"""
    user = await db.get(User, user_id)
    if user is None:
        raise AdminUserError("用户不存在")
    if not (user.device_serial or "").strip():
        raise AdminUserError("该用户未绑定设备")
    user.device_serial = ""
    await db.flush()
    await db.refresh(user)
    return user


async def delete_user(db: AsyncSession, oem: User, user_id: int, *, actor_id: int) -> None:
    if user_id == actor_id:
        raise AdminUserError("不能删除当前登录账号")

    user = await db.get(User, user_id)
    if user is None:
        raise AdminUserError("用户不存在")
    if user.oem_id != oem.id:
        raise AdminUserError("无权限删除该用户")
    await db.delete(user)