chongming-plugins/auth/router/login.py

import time
from datetime import datetime
import orjson

from sqlalchemy import select

from chongming_worker import rpc_method
from chongming_password import verify_password

from ..models.pydantic import (
    LoginPostRequest,
    LoginPostResponse
)
from ..models.auth import Users


@rpc_method(
    subject="auth.login",
    method="POST",
    summary="登录",
    response_model=LoginPostResponse,
)
async def login(login_post_request: LoginPostRequest) -> LoginPostResponse:
    from ..global_var import jwt, db

    if db is None:
        raise RuntimeError("Database not initialized")
    async with db.get_session() as session:
        statement = select(Users).filter_by(
            username=login_post_request.username,
        )
        result = await session.execute(statement)
        user: Users | None = result.scalar_one_or_none()

        if user is None:
            return LoginPostResponse(
                result="fail",
                message="用户名错误",
                token="",
                timestamp=time.time()
            )

        if not verify_password(login_post_request.password, user.hashed_password):
            return LoginPostResponse(
                result="fail",
                message="密码错误",
                token="",
                timestamp=time.time()
            )

        if not user.is_active:
            return LoginPostResponse(
                result="fail",
                message="用户已经注销或被禁用",
                token="",
                timestamp=time.time()
            )

        user.last_login_at = datetime.now()

        payload = {
            "sub": user.id,
            "username": user.username,
            "email": user.email,
            "full_name": user.full_name,
            "avatar_url": user.avatar_url,
            "phone": user.phone,
            "is_verified": user.is_verified,
            "is_superuser": user.is_superuser,
            "created_at": user.created_at.isoformat(),
            "updated_at": user.updated_at.isoformat(),
            "last_login_at": user.last_login_at.isoformat()
        }

        token = jwt.encode(
            orjson.dumps(payload).decode(), 
            expires_in_seconds=10800   # 3小时有效期
        )

        return LoginPostResponse(
            result="success", 
            message="登录成功", 
            token=token, 
            timestamp=time.time()
        )