disknext/routers/controllers/user.py

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy import and_
from webauthn import generate_registration_options
from webauthn.helpers import options_to_json_dict

import models
import service
from middleware.auth import AuthRequired
from middleware.dependencies import SessionDep

user_router = APIRouter(
    prefix="/user",
    tags=["user"],
)

user_settings_router = APIRouter(
    prefix='/user/settings',
    tags=["user", "user_settings"],
    dependencies=[Depends(AuthRequired)],
)

@user_router.post(
    path='/session',
    summary='用户登录',
    description='User login endpoint.',
)
async def router_user_session(
    session: SessionDep,
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> models.response.TokenModel:
    username = form_data.username
    password = form_data.password

    result = await service.user.Login(
        session,
        models.request.LoginRequest(username=username, password=password),
    )

    if isinstance(result, models.response.TokenModel):
        return result
    elif result is None:
        raise HTTPException(status_code=401, detail="Invalid username or password")
    elif result is False:
        raise HTTPException(status_code=403, detail="User account is banned or not fully registered")
    else:
        raise HTTPException(status_code=500, detail="Internal server error during login")

@user_router.post(
    path='/',
    summary='用户注册',
    description='User registration endpoint.',
)
def router_user_register() -> models.response.ResponseModel:
    """
    User registration endpoint.
    
    Returns:
        dict: A dictionary containing user registration information.
    """
    pass

@user_router.post(
    path='/2fa',
    summary='用两步验证登录',
    description='Two-factor authentication login endpoint.',
)
def router_user_2fa() -> models.response.ResponseModel:
    """
    Two-factor authentication login endpoint.
    
    Returns:
        dict: A dictionary containing two-factor authentication information.
    """
    pass

@user_router.post(
    path='/code',
    summary='发送验证码邮件',
    description='Send a verification code email.',
)
def router_user_email_code() -> models.response.ResponseModel:
    """
    Send a verification code email.
    
    Returns:
        dict: A dictionary containing information about the password reset email.
    """
    pass

@user_router.patch(
    path='/reset',
    summary='通过邮件里的链接重设密码',
    description='Reset password via email link.',
    deprecated=True,
)
def router_user_reset_patch() -> models.response.ResponseModel:
    """
    Reset password via email link.
    
    Returns:
        dict: A dictionary containing information about the password reset.
    """
    pass

@user_router.get(
    path='/qq',
    summary='初始化QQ登录',
    description='Initialize QQ login for a user.',
)
def router_user_qq() -> models.response.ResponseModel: 
    """
    Initialize QQ login for a user.
    
    Returns:
        dict: A dictionary containing QQ login initialization information.
    """
    pass

@user_router.get(
    path='authn/{username}',
    summary='WebAuthn登录初始化',
    description='Initialize WebAuthn login for a user.',
)
async def router_user_authn(username: str) -> models.response.ResponseModel:
    
    pass

@user_router.post(
    path='authn/finish/{username}',
    summary='WebAuthn登录',
    description='Finish WebAuthn login for a user.',
)
def router_user_authn_finish(username: str) -> models.response.ResponseModel:
    """
    Finish WebAuthn login for a user.
    
    Args:
        username (str): The username of the user.
    
    Returns:
        dict: A dictionary containing WebAuthn login information.
    """
    pass

@user_router.get(
    path='/profile/{id}',
    summary='获取用户主页展示用分享',
    description='Get user profile for display.',
)
def router_user_profile(id: str) -> models.response.ResponseModel:
    """
    Get user profile for display.
    
    Args:
        id (str): The user ID.
    
    Returns:
        dict: A dictionary containing user profile information.
    """
    pass

@user_router.get(
    path='/avatar/{id}/{size}',
    summary='获取用户头像',
    description='Get user avatar by ID and size.',
)
def router_user_avatar(id: str, size: int = 128) -> models.response.ResponseModel:
    """
    Get user avatar by ID and size.
    
    Args:
        id (str): The user ID.
        size (int): The size of the avatar image.
    
    Returns:
        str: A Base64 encoded string of the user avatar image.
    """
    pass

#####################
# 需要登录的接口
#####################

@user_router.get(
    path='/me',
    summary='获取用户信息',
    description='Get user information.',
    dependencies=[Depends(dependency=AuthRequired)],
    response_model=models.response.ResponseModel,
)
async def router_user_me(
    session: SessionDep,
    user: Annotated[models.user.User, Depends(AuthRequired)],
) -> models.response.ResponseModel:
    """
    获取用户信息.

    :return: response.ResponseModel containing user information.
    :rtype: response.ResponseModel
    """
    group = await models.Group.get(session, models.Group.id == user.group_id)

    user_group = models.response.GroupModel(
        id=group.id,
        name=group.name,
        allowShare=group.share_enabled,
    )

    users = models.response.UserModel(
        id=user.id,
        username=user.username,
        nickname=user.nick,
        status=user.status,
        created_at=user.created_at,
        score=user.score,
        group=user_group,
    ).model_dump()

    return models.response.ResponseModel(data=users)

@user_router.get(
    path='/storage',
    summary='存储信息',
    description='Get user storage information.',
    dependencies=[Depends(AuthRequired)],
)
def router_user_storage(
    user: Annotated[models.user.User, Depends(AuthRequired)],
) -> models.response.ResponseModel:
    """
    Get user storage information.
    
    Returns:
        dict: A dictionary containing user storage information.
    """
    return models.response.ResponseModel(
        data={
            "used": 0,
            "free": 0,
            "total": 0,
        }
    )

@user_router.put(
    path='/authn/start',
    summary='WebAuthn登录初始化',
    description='Initialize WebAuthn login for a user.',
    dependencies=[Depends(AuthRequired)],
)
async def router_user_authn_start(
    session: SessionDep,
    user: Annotated[models.user.User, Depends(AuthRequired)],
) -> models.response.ResponseModel:
    """
    Initialize WebAuthn login for a user.

    Returns:
        dict: A dictionary containing WebAuthn initialization information.
    """
    # TODO: 检查 WebAuthn 是否开启，用户是否有注册过 WebAuthn 设备等
    authn_setting = await models.Setting.get(
        session,
        and_(models.Setting.type == "authn", models.Setting.name == "authn_enabled")
    )
    if not authn_setting or authn_setting.value != "1":
        raise HTTPException(status_code=400, detail="WebAuthn is not enabled")

    site_url_setting = await models.Setting.get(
        session,
        and_(models.Setting.type == "basic", models.Setting.name == "siteURL")
    )
    site_title_setting = await models.Setting.get(
        session,
        and_(models.Setting.type == "basic", models.Setting.name == "siteTitle")
    )

    options = generate_registration_options(
        rp_id=site_url_setting.value if site_url_setting else "",
        rp_name=site_title_setting.value if site_title_setting else "",
        user_name=user.username,
        user_display_name=user.nick or user.username,
    )

    return models.response.ResponseModel(data=options_to_json_dict(options))

@user_router.put(
    path='/authn/finish',
    summary='WebAuthn登录',
    description='Finish WebAuthn login for a user.',
    dependencies=[Depends(AuthRequired)],
)
def router_user_authn_finish() -> models.response.ResponseModel:
    """
    Finish WebAuthn login for a user.
    
    Returns:
        dict: A dictionary containing WebAuthn login information.
    """
    pass

@user_settings_router.get(
    path='/policies',
    summary='获取用户可选存储策略',
    description='Get user selectable storage policies.',
)
def router_user_settings_policies() -> models.response.ResponseModel:
    """
    Get user selectable storage policies.
    
    Returns:
        dict: A dictionary containing available storage policies for the user.
    """
    pass

@user_settings_router.get(
    path='/nodes',
    summary='获取用户可选节点',
    description='Get user selectable nodes.',
    dependencies=[Depends(AuthRequired)],
)
def router_user_settings_nodes() -> models.response.ResponseModel:
    """
    Get user selectable nodes.
    
    Returns:
        dict: A dictionary containing available nodes for the user.
    """
    pass

@user_settings_router.get(
    path='/tasks',
    summary='任务队列',
    description='Get user task queue.',
    dependencies=[Depends(AuthRequired)],
)
def router_user_settings_tasks() -> models.response.ResponseModel:
    """
    Get user task queue.
    
    Returns:
        dict: A dictionary containing the user's task queue information.
    """
    pass

@user_settings_router.get(
    path='/',
    summary='获取当前用户设定',
    description='Get current user settings.',
    dependencies=[Depends(AuthRequired)],
)
def router_user_settings() -> models.response.ResponseModel:
    """
    Get current user settings.
    
    Returns:
        dict: A dictionary containing the current user settings.
    """
    return models.response.ResponseModel(data=models.response.UserSettingModel().model_dump())

@user_settings_router.post(
    path='/avatar',
    summary='从文件上传头像',
    description='Upload user avatar from file.',
    dependencies=[Depends(AuthRequired)],
)
def router_user_settings_avatar() -> models.response.ResponseModel:
    """
    Upload user avatar from file.
    
    Returns:
        dict: A dictionary containing the result of the avatar upload.
    """
    pass

@user_settings_router.put(
    path='/avatar',
    summary='设定为Gravatar头像',
    description='Set user avatar to Gravatar.',
    dependencies=[Depends(AuthRequired)],
)
def router_user_settings_avatar_gravatar() -> models.response.ResponseModel:
    """
    Set user avatar to Gravatar.
    
    Returns:
        dict: A dictionary containing the result of setting the Gravatar avatar.
    """
    pass

@user_settings_router.patch(
    path='/{option}',
    summary='更新用户设定',
    description='Update user settings.',
    dependencies=[Depends(AuthRequired)],
)
def router_user_settings_patch(option: str) -> models.response.ResponseModel:
    """
    Update user settings.
    
    Args:
        option (str): The setting option to update.
    
    Returns:
        dict: A dictionary containing the result of the settings update.
    """
    pass

@user_settings_router.get(
    path='/2fa',
    summary='获取两步验证初始化信息',
    description='Get two-factor authentication initialization information.',
    dependencies=[Depends(AuthRequired)],
)
def router_user_settings_2fa() -> models.response.ResponseModel:
    """
    Get two-factor authentication initialization information.
    
    Returns:
        dict: A dictionary containing two-factor authentication setup information.
    """
    pass