新增用户授权功能

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

@@ -1,31 +1,34 @@
 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_account_info_by_type,
+    get_token,
-    get_token_by_type,
-    save_oauth_token,
 )
+from app.tasks.user_oauth_tasks import update_oauth_accounts
 
-router = APIRouter(prefix="/user-oauth", tags=["user-oauth"])
+router = APIRouter(prefix="/user-oauth", tags=["oauth"])
 
 
 @router.post(
     "/request_oauth",
     summary="获取授权链接",
-    description="用户提交oauth_type，返回对应的第三方授权链接",
+    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.oauth_type, current_user.id)
+        auth_url = await build_oauth_url(req.open_type, current_user.id, db)
         return {"auth_url": auth_url}
     except ValueError as e:
         raise HTTPException(
@@ -46,25 +49,43 @@ async def juliang_callback(
 ):
     try:
         parts = state.split(":")
-        if len(parts) != 4:
+        if len(parts) != 2 or not parts[0] or not parts[1]:
             raise HTTPException(
                 status_code=status.HTTP_400_BAD_REQUEST,
                 detail="无效的state参数",
             )
 
-        oauth_type = int(parts[0])
+        user_id = parts[0]
-        user_id = parts[1]
+        app_id = parts[1]
-        app_id = parts[2]
-        app_type = parts[3]
 
-        token = await get_token_by_type(auth_code, oauth_type, app_id, app_type)
+        user = await db.execute(
-        account_info = await get_account_info_by_type(token, oauth_type, app_type)
+            select(User).where(User.id == user_id).limit(1)
-        user_oauth = await save_oauth_token(db, user_id, oauth_type, token, account_info, app_id)
+        )
+        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,
-            "data": UserOAuthOut.model_validate(user_oauth),
         }
 
     except ValueError as e:
@@ -72,53 +93,75 @@ async def juliang_callback(
             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(
-    "/callback",
+    "/update_account",
-    summary="通用授权回调",
+    summary="更新权限下的所有账户",
-    description="其他平台授权回调地址，接收code和state参数",
+    description="用户提交授权登录账户id，或者授权id",
 )
-async def oauth_callback(
+async def update_account(
-    code: str = Query(..., description="第三方返回的授权码"),
+    account_id: str | None = Query(None, description="授权账户id"),
-    state: str = Query(..., description="请求时传递的自定义参数"),
+    account_userid: str | None = Query(None, description="授权登录账号id"),
+    current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
     try:
-        parts = state.split(":")
+        if not account_id and not account_userid:
-        if len(parts) != 4:
             raise HTTPException(
                 status_code=status.HTTP_400_BAD_REQUEST,
-                detail="无效的state参数",
+                detail="请提交授权账户id或授权登录账号id",
             )
-
+        #获取单独授权账号    
-        oauth_type = int(parts[0])
+        if account_id:
-        user_id = parts[1]
+            exist = await db.execute(
-        app_id = parts[2]
+                select(UserOAuth).where(
-        app_type = parts[3]
+                    UserOAuth.account_id == account_id,
-
+                    UserOAuth.user_id == current_user.id,
-        token = await get_token_by_type(code, oauth_type, app_id, app_type)
+                    UserOAuth.deleted_at.is_(None),
-        account_info = await get_account_info_by_type(token, oauth_type, app_type)
+                ).limit(1)
-        user_oauth = await save_oauth_token(db, user_id, oauth_type, token, account_info, app_id)
+            )
-
+            exist = exist.scalar_one_or_none()
-        return {
+            if not exist:
-            "message": "授权成功",
+                raise HTTPException(
-            "code": 0,
+                    status_code=status.HTTP_404_NOT_FOUND,
-            "data": UserOAuthOut.model_validate(user_oauth),
+                    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=f"授权失败: {str(e)}",
+            detail=str(e),
         )

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

@@ -24,6 +24,9 @@ class UserOAuth(Base, TimestampMixin, SoftDeleteMixin):
     account_username: Mapped[str | None] = mapped_column(
         String(128), nullable=True, comment="授权账户登录账号"
     )
+    account_userid: Mapped[str | None] = mapped_column(
+        String(128), nullable=True, comment="授权账户登录userid，同一个用户不同的授权账户token不一样"
+    )
     user_id: Mapped[str] = mapped_column(
         String(32), nullable=False, index=True,
         comment="用户id"

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

@@ -4,9 +4,9 @@ from pydantic import BaseModel, Field
 
 
 class RequestOAuthRequest(BaseModel):
-    oauth_type: int = Field(
+    open_type: int = Field(
         ...,
-        description="开户方式（1=巨量广告，2=巨量千川，3=快手，4=腾讯）",
+        description="开户方式（1=千川，2=广告，3=本地推，4=星图，5=快手代理商，6=巨量星图，7=巨量服务单，8=腾讯服务单，9=腾讯营销K2，10=腾讯营销K3）",
     )
 
 

video-gen-api/app/services/user_oauth_service.py

@@ -1,5 +1,5 @@
 import random
-from datetime import datetime
+from datetime import datetime, timedelta
 
 import httpx
 from sqlalchemy import select, func
@@ -7,6 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import settings
 from app.models.user_oauth import UserOAuth
+from app.models.user_oauth_app import UserOAuthApp
 from app.utils.id_gen import generate_id
 
 
@@ -23,35 +24,45 @@ OAUTH_TYPE_CONFIG = {
     10: {"port_type": 5, "name": "腾讯营销K3", "app_type": "tencent"},
 }
 
-
+#随机获取一个可用的应用配置
-async def get_available_app(app_type: str, db: AsyncSession) -> dict:
+async def get_available_app(open_type: int, db: AsyncSession) -> dict:
-    if app_type == "juliang_ad":
+    # 从 user_oauth_app 表查询可用应用
-        apps = settings.JULIANG_AD_APPS
+    # 过滤条件：open_type 匹配、status=1（正常）、deleted_at is None
-    elif app_type == "juliang_qianchuan":
+    result = await db.execute(
-        apps = settings.JULIANG_QIANCHUAN_APPS
+        select(UserOAuthApp).where(
-    elif app_type == "kuaishou":
+            UserOAuthApp.open_type == open_type,
-        apps = settings.KUAISHOU_APPS
+            UserOAuthApp.status == 1,
-    elif app_type == "tencent":
+            UserOAuthApp.deleted_at.is_(None),
-        apps = settings.TENCENT_APPS
+        )
-    else:
+    )
-        raise ValueError(f"不支持的应用类型: {app_type}")
+    apps = result.scalars().all()
 
     if not apps:
-        raise ValueError(f"{app_type}未配置应用")
+        raise ValueError("没有找到可用的应用配置")
 
     available_apps = []
     for app in apps:
-        app_id = app.get("app_id")
+        app_id = app.app_id
-        if not app_id:
-            continue
         
-        result = await db.execute(
+        # 查询该应用当前授权数量
-            select(func.count(UserOAuth.id)).where(UserOAuth.appid == app_id)
+        count_result = await db.execute(
+            select(func.count(UserOAuth.id)).where(
+                UserOAuth.appid == app_id,
+                UserOAuth.deleted_at.is_(None),
             )
-        count = result.scalar() or 0
+        )
+        count = count_result.scalar() or 0
 
-        if count < 5000:
+        # 检查是否达到最大授权数
-            available_apps.append(app)
+        max_users = app.count
+        if count < max_users:
+            available_apps.append({
+                "app_id": app.app_id,
+                "secret": app.secret,
+                "open_type": app.open_type,
+                "auth_url": app.auth_url,
+                "company": app.company,
+            })
 
     if not available_apps:
         raise ValueError("所有应用授权已超过最大数量")
@@ -59,175 +70,89 @@ async def get_available_app(app_type: str, db: AsyncSession) -> dict:
     return random.choice(available_apps)
 
 
-async def build_oauth_url(oauth_type: int, user_id: str) -> str:
+async def build_oauth_url(open_type: int, user_id: str, db: AsyncSession) -> str:
-    if oauth_type == 1:
+    if open_type == 1:
-        return await _build_juliang_oauth_url(oauth_type, user_id, app_type)
+        #千川
-    elif oauth_type == 2:
+        return await _build_jl_oauth_url(open_type, user_id, db)
-        return await _build_kuaishou_oauth_url(oauth_type, user_id)
+    elif open_type == 2:
-    elif oauth_type == 3:
+        #广告
-        return await _build_tencent_oauth_url(oauth_type, user_id)
+        return await _build_jl_oauth_url(open_type, user_id, db)
-    elif oauth_type == 4:
+    elif open_type == 5:
-        return await _build_tencent_oauth_url(oauth_type, user_id)    
+        #快手代理商
+        return "无配置"
+    elif open_type == 9:
+        #腾讯营销K2
+        return "无配置"
+    elif open_type == 10:
+        #腾讯营销K3
+        return "无配置"
     else:
         raise ValueError(f"不支持的应用类型: {app_type}")
 
-
+#千川授权链接构建
-async def _build_juliang_oauth_url(oauth_type: int, user_id: str, app_type: str) -> str:
+async def _build_jl_oauth_url(open_type: int, user_id: str, db: AsyncSession) -> str:
-    async with AsyncSession() as db:
+    app = await get_available_app(open_type, db)
-        app = await get_available_app(app_type, db)
     app_id = app.get("app_id")
-
+    auth_url = app.get("auth_url")
-    redirect_uri = "https://open.oceanengine.com/audit/oauth.html"
+    if not auth_url:
-    rid = "ktm0cl7napb"
+        raise ValueError("应用授权链接不能为空")
-    if oauth_type == 1:
-        redirect_uri = "https://qianchuan.jinritemai.com/openapi/qc/audit/oauth.html"
-        rid = "vr7kclvmvs9"
-
     params = {
         "app_id": app_id,
-        "state": f"{oauth_type}:{user_id}:{app_id}:{app_type}",
+        "state": f"{user_id}:{app_id}",
-        "material_auth": 1,
-        "rid": rid,
     }
     query_string = "&".join(f"{k}={v}" for k, v in params.items())
-    return f"{redirect_uri}?{query_string}"
+    return f"{auth_url}&{query_string}"
 
 
-async def _build_kuaishou_oauth_url(oauth_type: int, user_id: str) -> str:
+async def get_token(code: str, user_id: str, app_id: str, db: AsyncSession) -> dict:
-    async with AsyncSession() as db:
+    #1.根据app_id查询应用配置
-        app = await get_available_app("kuaishou", db)
+    result = await db.execute(
-        app_id = app.get("app_id")
+        select(UserOAuthApp).where(UserOAuthApp.app_id == app_id)
-
+    )
-    redirect_uri = f"{settings.BASE_URL}/api/user-oauth/callback"
+    app = result.scalar_one_or_none()
-    params = {
-        "app_id": app_id,
-        "redirect_uri": redirect_uri,
-        "response_type": "code",
-        "scope": "basic",
-        "state": f"{oauth_type}:{user_id}:{app_id}:kuaishou",
-    }
-    query_string = "&".join(f"{k}={v}" for k, v in params.items())
-    return f"https://open.kuaishou.com/oauth2/authorize?{query_string}"
-
-
-async def _build_tencent_oauth_url(oauth_type: int, user_id: str) -> str:
-    async with AsyncSession() as db:
-        app = await get_available_app("tencent", db)
-        app_id = app.get("app_id")
-
-    redirect_uri = f"{settings.BASE_URL}/api/user-oauth/callback"
-    params = {
-        "app_id": app_id,
-        "redirect_uri": redirect_uri,
-        "response_type": "code",
-        "scope": "get_user_info",
-        "state": f"{oauth_type}:{user_id}:{app_id}:tencent",
-    }
-    query_string = "&".join(f"{k}={v}" for k, v in params.items())
-    return f"https://api.e.qq.com/oauth/authorize?{query_string}"
-
-
-async def get_token_by_type(code: str, oauth_type: int, app_id: str, app_type: str) -> dict:
-    if app_type in ("juliang_ad", "juliang_qianchuan"):
-        return await get_juliang_token(code, oauth_type, app_id, app_type)
-    elif app_type == "kuaishou":
-        return await get_kuaishou_token(code, oauth_type, app_id)
-    elif app_type == "tencent":
-        return await get_tencent_token(code, oauth_type, app_id)
-    else:
-        raise ValueError(f"不支持的应用类型: {app_type}")
-
-
-async def get_juliang_token(code: str, oauth_type: int, app_id: str, app_type: str) -> dict:
-    url = "https://api.oceanengine.com/open_api/oauth2/access_token/"
-
-    if app_type == "juliang_ad":
-        apps = settings.JULIANG_AD_APPS
-    else:
-        apps = settings.JULIANG_QIANCHUAN_APPS
-
-    app = next((a for a in apps if a.get("app_id") == app_id), None)
     if not app:
         raise ValueError("应用配置不存在")
    
+    open_type = app.open_type
+    secret = app.secret
+
+    if open_type == 1 or open_type == 2:
+        #千川或广告
+        return await get_juliang_token(app_id, secret, code, open_type, user_id, db)
+    elif open_type == 5:
+        #快手代理商
+        return await get_kuaishou_token(app_id, secret, code, open_type, user_id, db)
+    elif open_type == 9 or open_type == 10:
+        #腾讯营销K2或腾讯营销K3
+        return await get_tencent_token(app_id, secret, code, open_type, user_id, db)
+    else:
+        raise ValueError(f"不支持的应用类型: {open_type}")
+
+
+async def get_juliang_token(app_id: str, secret: str, code: str, open_type: int, user_id: str, db: AsyncSession) -> dict:
+    
     async with httpx.AsyncClient() as client:
+        #1.请求token
+        url = "https://api.oceanengine.com/open_api/oauth2/access_token/"
         response = await client.post(
             url,
-            data={
+            json={
                 "app_id": app_id,
-                "secret": app.get("secret"),
+                "secret": secret,
                 "auth_code": code,
             },
         )
         response.raise_for_status()
-        content = response.json()
+        data = response.json()
-        if content.get("code") != 0:
+        if data.get("code") != 0:
-            raise ValueError(content.get("message", "获取token失败"))
+            raise ValueError(data.get("message", "获取token失败")+f"，错误信息：{data.get('message', '')}")
-        return content.get("data", {})
+        data = data.get("data", {})
-
+        access_token = data.get("access_token", "")
-
+        refresh_token = data.get("refresh_token", "")
-async def get_kuaishou_token(code: str, oauth_type: int, app_id: str) -> dict:
+        expires_in = datetime.now() + timedelta(seconds=data.get("expires_in", 0)) 
-    url = "https://open.kuaishou.com/oauth2/token"
+        refresh_token_expires_in = datetime.now() + timedelta(seconds=data.get("refresh_token_expires_in", 0))
-    redirect_uri = f"{settings.BASE_URL}/api/user-oauth/callback"
+        #2.获取已授权角色账户，一个授权可能有多个角色账户
-
+        url = "https://api.oceanengine.com/open_api/oauth2/advertiser/get/"
-    app = next((a for a in settings.KUAISHOU_APPS if a.get("app_id") == app_id), None)
-    if not app:
-        raise ValueError("应用配置不存在")
-
-    async with httpx.AsyncClient() as client:
-        response = await client.post(
-            url,
-            data={
-                "app_id": app_id,
-                "secret": app.get("secret"),
-                "code": code,
-                "grant_type": "authorization_code",
-                "redirect_uri": redirect_uri,
-            },
-        )
-        response.raise_for_status()
-        return response.json()
-
-
-async def get_tencent_token(code: str, oauth_type: int, app_id: str) -> dict:
-    url = "https://api.e.qq.com/oauth/token"
-    redirect_uri = f"{settings.BASE_URL}/api/user-oauth/callback"
-
-    app = next((a for a in settings.TENCENT_APPS if a.get("app_id") == app_id), None)
-    if not app:
-        raise ValueError("应用配置不存在")
-
-    async with httpx.AsyncClient() as client:
-        response = await client.post(
-            url,
-            data={
-                "app_id": app_id,
-                "secret": app.get("secret"),
-                "code": code,
-                "grant_type": "authorization_code",
-                "redirect_uri": redirect_uri,
-            },
-        )
-        response.raise_for_status()
-        return response.json()
-
-
-async def get_account_info_by_type(token: dict, oauth_type: int, app_type: str) -> dict:
-    if app_type in ("juliang_ad", "juliang_qianchuan"):
-        return await _get_juliang_account_info(token)
-    elif app_type == "kuaishou":
-        return await _get_kuaishou_account_info(token)
-    elif app_type == "tencent":
-        return await _get_tencent_account_info(token)
-    else:
-        raise ValueError(f"不支持的应用类型: {app_type}")
-
-
-async def _get_juliang_account_info(token: dict) -> dict:
-    access_token = token.get("access_token")
-    url = "https://ad.oceanengine.com/openapi/oauth/user/info/"
-
-    async with httpx.AsyncClient() as client:
         response = await client.get(
             url,
             headers={"Access-Token": access_token},
@@ -235,119 +160,111 @@ async def _get_juliang_account_info(token: dict) -> dict:
         response.raise_for_status()
         data = response.json()
         if data.get("code") != 0:
-            raise ValueError(data.get("message", "获取账户信息失败"))
+            raise ValueError(data.get("message", "获取已授权账户失败")+f"，错误信息：{data.get('message', '')}")
         data = data.get("data", {})
-        return {
+        account_list = data.get("list", [])
-            "account_id": data.get("advertiser_id", data.get("account_id", "")),
+        #3.获取已授权登录信息
-            "account_name": data.get("advertiser_name", data.get("account_name", "")),
+        url = "https://api.oceanengine.com/open_api/2/user/info/"
-            "account_role": data.get("role", ""),
-            "account_username": data.get("username", ""),
-        }
-
-
-async def _get_kuaishou_account_info(token: dict) -> dict:
-    access_token = token.get("access_token")
-    url = "https://open.kuaishou.com/api/user/info"
-
-    async with httpx.AsyncClient() as client:
         response = await client.get(
             url,
-            headers={"Authorization": f"Bearer {access_token}"},
+            headers={"Access-Token": access_token},
         )
         response.raise_for_status()
         data = response.json()
-        return {
+        if data.get("code") != 0:
-            "account_id": data.get("account_id", ""),
+            raise ValueError(data.get("message", "获取已授权登录信息失败")+f"，错误信息：{data.get('message', '')}")
-            "account_name": data.get("account_name", ""),
+        data = data.get("data", {})
-            "account_role": data.get("role", ""),
+        account_username = data.get("email", "")
-            "account_username": data.get("username", ""),
+        account_userid = str(data.get("id", ""))
-        }
+        material_auth_status = data.get("material_auth_status", False)
-
+        #4.根据登录信息判断是否新增token或更新token，不同的登录信息对应不同的token，然后更新数据库
-
+        #查询email，appid，user_id是否存在已授权记录
-async def _get_tencent_account_info(token: dict) -> dict:
+        oauth = await db.execute(
-    access_token = token.get("access_token")
+            select(UserOAuth)
-    url = "https://api.e.qq.com/user/info"
+            .where(UserOAuth.account_username == account_username, UserOAuth.account_userid == account_userid, UserOAuth.appid == app_id, UserOAuth.user_id == user_id)
-
+            .limit(1)
-    async with httpx.AsyncClient() as client:
-        response = await client.get(
-            url,
-            headers={"Authorization": f"Bearer {access_token}"},
         )
-        response.raise_for_status()
+        oauth = oauth.scalar_one_or_none()
-        data = response.json()
+        if not oauth:
-        return {
+            for account in account_list:
-            "account_id": data.get("account_id", ""),
+                #新增授权记录
-            "account_name": data.get("account_name", ""),
+                db.add(UserOAuth(
-            "account_role": data.get("role", ""),
-            "account_username": data.get("username", ""),
-        }
-
-
-async def save_oauth_token(
-    db: AsyncSession,
-    user_id: str,
-    oauth_type: int,
-    token: dict,
-    account_info: dict,
-    app_id: str,
-) -> UserOAuth:
-    config = OAUTH_TYPE_CONFIG.get(oauth_type)
-    if not config:
-        raise ValueError(f"不支持的oauth_type: {oauth_type}")
-
-    access_token = token.get("access_token")
-    access_token_expired = token.get("expires_in")
-    refresh_token = token.get("refresh_token")
-    refresh_token_expired = token.get("refresh_token_expires_in")
-
-    expires_at = None
-    if access_token_expired:
-        expires_at = datetime.now().timestamp() + int(access_token_expired)
-        expires_at = datetime.fromtimestamp(expires_at)
-
-    refresh_expires_at = None
-    if refresh_token_expired:
-        refresh_expires_at = datetime.now().timestamp() + int(refresh_token_expired)
-        refresh_expires_at = datetime.fromtimestamp(refresh_expires_at)
-
-    existing = await db.execute(
-        select(UserOAuth).where(
-            UserOAuth.user_id == user_id,
-            UserOAuth.open_type == oauth_type,
-            UserOAuth.account_id == account_info.get("account_id", ""),
-        ).limit(1)
-    )
-    existing_oauth = existing.scalar_one_or_none()
-
-    if existing_oauth:
-        existing_oauth.access_token = access_token
-        existing_oauth.access_token_expired = expires_at
-        existing_oauth.refresh_token = refresh_token
-        existing_oauth.refresh_token_expired = refresh_expires_at
-        existing_oauth.account_name = account_info.get("account_name", "")
-        existing_oauth.account_role = account_info.get("account_role", "")
-        existing_oauth.account_username = account_info.get("account_username", "")
-        existing_oauth.appid = app_id
-        await db.flush()
-        return existing_oauth
-
-    user_oauth = UserOAuth(
                     id=generate_id(),
-        account_id=account_info.get("account_id", ""),
+                    account_id = str(account.get("account_id", "")),
-        account_name=account_info.get("account_name", ""),
+                    account_name = account.get("account_name", ""),
-        account_role=account_info.get("account_role", ""),
+                    account_role = account.get("account_role", ""),
-        account_username=account_info.get("account_username", ""),
+                    account_username = account_username,
+                    account_userid = account_userid,
                     user_id = user_id,
-        open_type=oauth_type,
-        port_type=config["port_type"],
                     appid = app_id,
+                    open_type = open_type,
+                    port_type = 1,
                     access_token = access_token,
-        access_token_expired=expires_at,
+                    access_token_expired = expires_in,
                     refresh_token = refresh_token,
-        refresh_token_expired=refresh_expires_at,
+                    refresh_token_expired = refresh_token_expires_in,
-        material_auth_status=True,
+                    material_auth_status = material_auth_status,
+                ))
+                await db.commit()
+        else: 
+            # 查询现有授权记录（未删除的）
+            existing_accounts = await db.execute(
+                select(UserOAuth).where(
+                    UserOAuth.account_username == account_username,
+                    UserOAuth.account_userid == account_userid,
+                    UserOAuth.appid == app_id,
+                    UserOAuth.user_id == user_id,
+                    UserOAuth.deleted_at.is_(None),
                 )
+            )
+            existing_accounts = {acc.account_id: acc for acc in existing_accounts.scalars().all()}
 
-    db.add(user_oauth)
+            # 新的账户列表
-    await db.flush()
+            new_account_ids = {str(account.get("account_id")) for account in account_list}
-    return user_oauth
+            old_account_ids = set(existing_accounts.keys())
+
+            # 1. 软删除已消失的账户
+            for account_id in old_account_ids - new_account_ids:
+                existing_accounts[account_id].deleted_at = datetime.now()
+
+            # 2. 更新或新增账户
+            for account in account_list:
+                account_id = str(account.get("account_id", ""))
+                if account_id in existing_accounts:
+                    # 更新现有记录
+                    existing_oauth = existing_accounts[account_id]
+                    existing_oauth.account_name = account.get("account_name", "")
+                    existing_oauth.account_role = account.get("account_role", "")
+                    existing_oauth.access_token = access_token
+                    existing_oauth.access_token_expired = expires_in
+                    existing_oauth.refresh_token = refresh_token
+                    existing_oauth.refresh_token_expired = refresh_token_expires_in
+                    existing_oauth.material_auth_status = material_auth_status
+                else:
+                    # 新增记录
+                    db.add(UserOAuth(
+                            id=generate_id(),
+                            account_id=str(account_id),
+                            account_name=account.get("account_name", ""),
+                            account_role=account.get("account_role", ""),
+                            account_username=account_username,
+                            account_userid = account_userid,
+                            user_id=user_id,
+                            appid=app_id,
+                            open_type=open_type,
+                            port_type=1,
+                            access_token = access_token,
+                            access_token_expired=expires_in,
+                            refresh_token=refresh_token,
+                            refresh_token_expired=refresh_token_expires_in,
+                            material_auth_status=material_auth_status,
+                        ))
+            await db.commit()
+
+        #5.返回成功
+        return {"message": "授权成功"}
+
+async def get_kuaishou_token(app_id: str, secret: str, code: str, oauth_type: int, user_id: str, db: AsyncSession) -> dict:
+    return "未配置"
+async def get_tencent_token(app_id: str, secret: str, code: str, oauth_type: int, user_id: str, db: AsyncSession) -> str:
+    return "未配置"

video-gen-api/app/tasks/user_oauth_tasks.py

@@ -0,0 +1,44 @@
+from asyncio import Condition
+
+import httpx
+from datetime import datetime, timedelta
+
+from sqlalchemy import select
+
+from app.models.base import async_session
+from app.models.user_oauth import UserOAuth
+from app.models.user_oauth_app import UserOAuthApp
+from app.tasks.async_runner import run_async
+from app.tasks.celery_app import celery_app
+
+
+async def _update_oauth_accounts(account_id: str, account_userid: str, current_user_id: str, db: async_session):
+    conditions = [UserOAuth.user_id == current_user_id, UserOAuth.deleted_at.is_(None)]
+    if account_id:
+        conditions.append(UserOAuth.account_id == account_id)    
+    if account_userid:
+        conditions.append(UserOAuth.account_userid == account_userid)
+    result = await db.execute(
+        select(UserOAuth).where(
+            *conditions
+        )
+    )
+    oauth_records = result.scalars().all()
+    if not oauth_records:
+        return True
+    for oauth in oauth_records:
+        pass
+
+if celery_app:
+    @celery_app.task(name="user_oauth.update_oauth_accounts", bind=True, max_retries=3, default_retry_delay=60)
+    def update_oauth_accounts(self, account_id: str, account_userid: str, current_user_id: str, db):
+        return run_async(_update_oauth_accounts(account_id, account_userid, current_user_id, db))
+else:
+    class _DisabledTask:
+        def delay(self, *args, **kwargs):
+            pass
+
+        def apply_async(self, *args, **kwargs):
+            pass
+
+    update_oauth_accounts = _DisabledTask()

video-gen-api/app/utils/__init__.py

@@ -0,0 +1,4 @@
+from app.utils.douyinRequest import DouyinRequest
+from app.utils.douyinApi import DouyinApi
+
+__all__ = ["DouyinRequest", "DouyinApi"]

video-gen-api/app/utils/douyinApi.py

@@ -0,0 +1,20 @@
+from typing import Any, Dict, Optional
+
+from app.utils.douyinRequest import DouyinRequest
+from app.models.user_oauth import UserOAuth
+
+class DouyinApi:
+    def __init__(self, request: DouyinRequest):
+        self.request = request
+
+    async def get_advertiser_list(self, oauth_id: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        if not oauth_id:
+            raise RuntimeError('OAuth ID is not set.')
+
+        url = ""
+        return await self.request.request_with_token_with_context(
+            oauth_id,
+            url,
+            'GET',
+            {'params': params or {}}
+        )

video-gen-api/app/utils/douyinRequest.py

@@ -0,0 +1,289 @@
+import json
+import asyncio
+from typing import Any, Dict, Optional, Tuple
+
+import httpx
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+from datetime import datetime, timezone, timedelta
+
+from app.models.user_oauth import UserOAuth
+from app.models.user_oauth_app import UserOAuthApp
+from app.models.base import async_session
+from app.utils.redis import get_redis
+
+
+class DouyinRequest:
+    def __init__(self, platform: str = "douyin"):
+        self._client: Optional[httpx.AsyncClient] = None
+        self._platform = platform
+
+    @property
+    def client(self) -> httpx.AsyncClient:
+        if not self._client:
+            self._client = httpx.AsyncClient(
+                timeout=httpx.Timeout(30.0),
+                follow_redirects=True,
+            )
+        return self._client
+
+    @property
+    def _redis_key(self) -> str:
+        return f"{self._platform}:tokens"
+
+    async def close(self):
+        if self._client:
+            await self._client.aclose()
+            self._client = None
+
+    async def _get_redis_token(self, oauth_id: str) -> Optional[str]:
+        redis = get_redis()
+        if not redis:
+            raise ValueError("Redis连接未配置")
+
+        try:
+            cache_str = await redis.hget(self._redis_key, oauth_id)
+            if cache_str:
+                cache = json.loads(cache_str)
+                token = cache.get("token")
+                expired_at_str = cache.get("expired_at")
+                if token and expired_at_str:
+                    expired_at = datetime.fromisoformat(expired_at_str)
+                    if expired_at > datetime.now(timezone.utc):
+                        return token
+        except Exception as e:
+            raise ValueError(f"获取Redis缓存失败: {e}")
+
+        return None
+
+    async def _set_redis_token(self, oauth_id: str, token: str, expired_at: datetime):
+        redis = get_redis()
+        if not redis:
+            raise ValueError("Redis连接未配置")
+
+        try:
+            cache = {
+                "token": token,
+                "expired_at": expired_at.isoformat(),
+            }
+            await redis.hset(self._redis_key, oauth_id, json.dumps(cache))
+        except Exception as e:
+            raise ValueError(f"设置Redis缓存失败: {e}")
+
+    async def _delete_redis_token(self, oauth_id: str):
+        redis = get_redis()
+        if not redis:
+            raise ValueError("Redis连接未配置")
+
+        try:
+            await redis.hdel(self._redis_key, oauth_id)
+        except Exception as e:
+            raise ValueError(f"删除Redis缓存失败: {e}")
+
+    async def get_access_token(self, oauth_id: str, force_refresh: bool = False) -> str:
+        if not force_refresh:
+            token = await self._get_redis_token(oauth_id)
+            if token:
+                return token
+
+
+        async with async_session() as db:
+            oauth_data = await db.execute(
+                select(UserOAuth).where(
+                    UserOAuth.id == oauth_id,
+                    UserOAuth.deleted_at.is_(None),
+                ).limit(1)
+            )
+            oauth_data = oauth_data.scalar_one_or_none()
+
+            if not oauth_data:
+                raise ValueError("无效的oauth_id")
+
+            if force_refresh:
+                await db.execute(
+                    update(UserOAuth).where(UserOAuth.id == oauth_id).values(
+                        access_token=None,
+                        access_token_expired=None,
+                    )
+                )
+                await db.commit()
+                await self._delete_redis_token(oauth_id)
+
+                new_token, new_expired_at = await self.refresh_access_token(db, oauth_id, oauth_data.appid, oauth_data.refresh_token)
+                await self._set_redis_token(oauth_id, new_token, new_expired_at)
+                return new_token
+
+            if oauth_data.access_token_expired and oauth_data.access_token_expired > datetime.now(timezone.utc):
+                token = oauth_data.access_token
+                expired_at = oauth_data.access_token_expired
+                await self._set_redis_token(oauth_id, token, expired_at)
+                return token
+
+            if oauth_data.refresh_token_expired and oauth_data.refresh_token_expired < datetime.now(timezone.utc):
+                raise ValueError("授权已过期，请重新授权")
+
+            new_token, new_expired_at = await self.refresh_access_token(db, oauth_id, oauth_data.appid, oauth_data.refresh_token)
+            await self._set_redis_token(oauth_id, new_token, new_expired_at)
+            return new_token
+
+    async def refresh_access_token(self, db: AsyncSession, oauth_id: str, appid: str, refresh_token: str) -> Tuple[str, datetime]:
+        result = await db.execute(
+            select(UserOAuthApp.secret).where(
+                UserOAuthApp.app_id == appid,
+                UserOAuthApp.status == 1,
+                UserOAuthApp.deleted_at.is_(None),
+            ).limit(1)
+        )
+        app_secret = result.scalar_one_or_none()
+
+        if not app_secret:
+            raise ValueError("应用已被删除或禁用")
+
+        response = await self.client.request(
+            'POST',
+            'https://api.oceanengine.com/open_api/oauth2/refresh_token/',
+            data={
+                'app_id': appid,
+                'secret': app_secret,
+                'refresh_token': refresh_token,
+            },
+        )
+        response.raise_for_status()
+        data = response.json()
+
+        if 'code' not in data:
+            raise ValueError("刷新access_token失败，接口未返回code")
+
+        code = data.get('code', 0)
+        if code != 0:
+            raise ValueError(f"刷新access_token失败,接口返回：{data}")
+
+        data = data.get('data', {})
+        new_access_token = data.get('access_token', '')
+        new_refresh_token = data.get('refresh_token', '')
+        expires_in = data.get('expires_in', 0)
+        refresh_token_expires_in = data.get('refresh_token_expires_in', 0)
+
+        new_expired_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in)
+
+        await db.execute(
+            update(UserOAuth).where(
+                UserOAuth.id == oauth_id,
+            ).values(
+                access_token=new_access_token,
+                refresh_token=new_refresh_token,
+                access_token_expired=new_expired_at,
+                refresh_token_expired=datetime.now(timezone.utc) + timedelta(seconds=refresh_token_expires_in),
+            )
+        )
+        await db.commit()
+
+        return new_access_token, new_expired_at
+
+    # 有token请求
+    async def request_with_token_with_context(
+        self,
+        oauth_id: str,
+        url: str,
+        method: str = 'GET',
+        options: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        options = options or {}
+        token = await self.get_access_token(oauth_id)
+
+        for i in range(1, 6):
+            try:
+                headers = options.get('headers', {}).copy()
+                headers['Access-Token'] = token
+                headers.setdefault('Content-Type', 'application/json')
+                options['headers'] = headers
+
+                response = await self.client.request(method, url, **options)
+                response.raise_for_status()
+
+                try:
+                    data = response.json()
+                except json.JSONDecodeError:
+                    data = {'code': 0, 'data': response.text}
+
+                if 'code' not in data:
+                    await asyncio.sleep(i * 5)
+                    continue
+
+                code = data.get('code', 0)
+
+                if code in [40102, 40104]:
+                    await asyncio.sleep(i * 5)
+                    token = await self.get_access_token(oauth_id, force_refresh=True)
+                    continue
+
+                if code in [40100, 40110]:
+                    wait_time = min(2 * (2 ** (i - 1)), 10)
+                    await asyncio.sleep(wait_time)
+                    continue
+
+                if code >= 50000:
+                    await asyncio.sleep(i * 10)
+                    continue
+
+                return data
+
+            except httpx.HTTPStatusError as e:
+                await asyncio.sleep(i * 10)
+                continue
+            except httpx.RequestError as e:
+                await asyncio.sleep(i * 10)
+                continue
+
+        res = json.dumps(data) if 'data' in locals() else ''
+        raise RuntimeError(
+            f'DouYin API request failed after 5 retries. '
+            f'url:{url};oauthId:{oauth_id};options:{json.dumps(options)};response:{res}'
+        )
+
+    # 无token请求
+    async def request_with_context(
+        self,
+        url: str,
+        method: str = 'GET',
+        options: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        options = options or {}
+
+        for i in range(1, 6):
+            try:
+                headers = options.get('headers', {}).copy()
+                headers.setdefault('Content-Type', 'application/json')
+                options['headers'] = headers
+
+                response = await self.client.request(method, url, **options)
+                response.raise_for_status()
+
+                try:
+                    data = response.json()
+                except json.JSONDecodeError:
+                    data = {'code': 0, 'data': response.text}
+
+                if 'code' not in data:
+                    await asyncio.sleep(i * 5)
+                    continue
+
+                code = data.get('code', 0)
+                if code >= 50000:
+                    await asyncio.sleep(i * 10)
+                    continue
+
+                return data
+
+            except httpx.HTTPStatusError as e:
+                await asyncio.sleep(i * 10)
+                continue
+            except httpx.RequestError as e:
+                await asyncio.sleep(i * 10)
+                continue
+
+        res = json.dumps(data) if 'data' in locals() else ''
+        raise RuntimeError(
+            f'DouYin API request failed after 5 retries. '
+            f'url:{url};options:{json.dumps(options)};response:{res}'
+        )