dy/backend/auth/scopes.py

from typing import Optional

from fastapi import HTTPException, status
from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from models.models import Account, AutoReplyRule, MessageLog, ReceivedMessageLog, SystemLog, User
from .permissions import (
    ACCOUNTS_UPDATE,
    ACCOUNTS_WRITE_GRANULAR,
    RULES_WRITE,
)
from .roles import has_global_scope, has_permission, is_admin


async def get_owned_account(
    db: AsyncSession,
    user: User,
    account_id: int,
    *,
    write: bool = False,
    write_permission: str | None = None,
) -> Account:
    result = await db.execute(select(Account).where(Account.id == account_id))
    account = result.scalar_one_or_none()
    if not account:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="账号不存在")
    if has_global_scope(user.role):
        if write:
            needed = write_permission or ACCOUNTS_UPDATE
            if not has_permission(user.role, needed):
                raise HTTPException(
                    status_code=status.HTTP_403_FORBIDDEN,
                    detail=f"缺少权限：{needed}",
                )
        return account
    if account.owner_id != user.id:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="无权访问该账号")
    if write:
        needed = write_permission or ACCOUNTS_UPDATE
        if not has_permission(user.role, needed):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"缺少权限：{needed}",
            )
    return account


def accounts_for_user(user: User):
    stmt = select(Account)
    if not has_global_scope(user.role):
        stmt = stmt.where(Account.owner_id == user.id)
    return stmt


async def owned_account_ids(db: AsyncSession, user: User) -> Optional[set[int]]:
    if has_global_scope(user.role):
        return None
    result = await db.execute(select(Account.id).where(Account.owner_id == user.id))
    return {row[0] for row in result.all()}


def logs_for_user(user: User, account_id: Optional[int] = None):
    stmt = select(MessageLog)
    if account_id is not None:
        stmt = stmt.where(MessageLog.account_id == account_id)
    if not has_global_scope(user.role):
        owned = select(Account.id).where(Account.owner_id == user.id)
        stmt = stmt.where(MessageLog.account_id.in_(owned))
    return stmt


def received_logs_for_user(user: User, account_id: Optional[int] = None):
    stmt = select(ReceivedMessageLog)
    if account_id is not None:
        stmt = stmt.where(ReceivedMessageLog.account_id == account_id)
    if not has_global_scope(user.role):
        owned = select(Account.id).where(Account.owner_id == user.id)
        stmt = stmt.where(ReceivedMessageLog.account_id.in_(owned))
    return stmt


def rules_for_user(user: User, account_id: Optional[int] = None):
    stmt = select(AutoReplyRule)
    if account_id is not None:
        stmt = stmt.where(AutoReplyRule.account_id == account_id)
    if has_global_scope(user.role):
        return stmt
    owned = select(Account.id).where(Account.owner_id == user.id)
    return stmt.where(AutoReplyRule.account_id.in_(owned))


async def get_accessible_rule(
    db: AsyncSession, user: User, rule_id: int, *, write: bool = False
) -> AutoReplyRule:
    result = await db.execute(select(AutoReplyRule).where(AutoReplyRule.id == rule_id))
    rule = result.scalar_one_or_none()
    if not rule:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="规则不存在")

    if has_global_scope(user.role):
        if write and not has_permission(user.role, RULES_WRITE):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="缺少权限：rules.write")
        return rule

    if rule.account_id is None:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="无权访问全局规则")

    account = await get_owned_account(db, user, rule.account_id, write=False)
    if rule.owner_id and rule.owner_id != user.id and account.owner_id != user.id:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="无权访问该规则")
    if write and not has_permission(user.role, RULES_WRITE):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="缺少权限：rules.write")
    return rule


def system_logs_for_user(user: User, account_id: Optional[int] = None):
    stmt = select(SystemLog)
    if account_id is not None:
        stmt = stmt.where(SystemLog.account_id == account_id)
    if not has_global_scope(user.role):
        owned = select(Account.id).where(Account.owner_id == user.id)
        stmt = stmt.where(
            or_(
                SystemLog.account_id.in_(owned),
                SystemLog.account_id.is_(None),
            )
        )
    return stmt


def users_for_manager(manager: User):
    """Admins see all users; others only see themselves and users they created."""
    stmt = select(User)
    if is_admin(manager.role):
        return stmt
    return stmt.where(or_(User.created_by == manager.id, User.id == manager.id))


def ensure_user_manageable(manager: User, target: User) -> None:
    """Raise 403 unless manager may edit/delete target user."""
    if is_admin(manager.role):
        return
    if target.id == manager.id:
        return
    if target.created_by == manager.id:
        return
    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="只能管理自己创建的用户",
    )