video-gen/video-gen-api/app/api/v1/user_oauth.py

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_current_user, get_db
from app.models.user import User
from app.models.user_oauth import UserOAuth
from app.models.user_oauth_app import UserOAuthApp
from app.schemas.user_oauth import RequestOAuthRequest, RequestOAuthResponse, UserOAuthOut
from app.services.user_oauth_service import (
    build_oauth_url,
    get_token,
)
from app.tasks.user_oauth_tasks import update_oauth_accounts

router = APIRouter(prefix="/user-oauth", tags=["oauth"])


@router.post(
    "/request_oauth",
    summary="获取授权链接",
    description="用户提交开户方式open_type，返回对应的第三方授权链接",
    response_model=RequestOAuthResponse,
)
async def request_oauth(
    req: RequestOAuthRequest,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    try:
        auth_url = await build_oauth_url(req.open_type, current_user.id, db)
        return {"auth_url": auth_url}
    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
        )


@router.get(
    "/juliang_callback",
    summary="巨量授权回调",
    description="巨量引擎授权回调地址，接收code和state参数，获取token并保存",
)
async def juliang_callback(
    auth_code: str = Query(..., description="第三方返回的授权码"),
    state: str = Query(..., description="请求时传递的自定义参数"),
    db: AsyncSession = Depends(get_db),
):
    try:
        parts = state.split(":")
        if len(parts) != 2 or not parts[0] or not parts[1]:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="无效的state参数",
            )

        user_id = parts[0]
        app_id = parts[1]

        user = await db.execute(
            select(User).where(User.id == user_id).limit(1)
        )
        user = user.scalar_one_or_none()
        if not user:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="用户不存在",
            )

        app = await db.execute(
            select(UserOAuthApp).where(
                UserOAuthApp.app_id == app_id,
                UserOAuthApp.status == 1,
                UserOAuthApp.deleted_at.is_(None),
            ).limit(1)
        )
        app = app.scalar_one_or_none()
        if not app:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="应用不存在",
            )
        
        await get_token(auth_code, user_id, app_id, db)
        return {
            "message": "授权成功",
            "code": 0,
        }

    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
        )
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"授权失败: {str(e)}",
        )

@router.get(
    "/update_account",
    summary="更新权限下的所有账户",
    description="用户提交授权登录账户id，或者授权id",
)
async def update_account(
    account_id: str | None = Query(None, description="授权账户id"),
    account_userid: str | None = Query(None, description="授权登录账号id"),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    try:
        if not account_id and not account_userid:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="请提交授权账户id或授权登录账号id",
            )
        #获取单独授权账号    
        if account_id:
            exist = await db.execute(
                select(UserOAuth).where(
                    UserOAuth.account_id == account_id,
                    UserOAuth.user_id == current_user.id,
                    UserOAuth.deleted_at.is_(None),
                ).limit(1)
            )
            exist = exist.scalar_one_or_none()
            if not exist:
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail="授权账户不存在",
                )
        #获取授权登录账号的所有账号        
        if account_userid:
            exist = await db.execute(
                select(UserOAuth).where(
                    UserOAuth.account_userid == account_userid,
                    UserOAuth.user_id == current_user.id,
                    UserOAuth.deleted_at.is_(None),
                ).limit(1)
            )
            exist = exist.scalar_one_or_none()
            if not exist:
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail="授权登录账号不存在",
                )
        await update_oauth_accounts(account_id, account_userid, current_user.id, db)
        return {"message": "提交成功，等待处理", "code": 0}
    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
        )
    except HTTPException as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=str(e),
        )    
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=str(e),
        )