提交初始文件，增加授权应用路由

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

@@ -5,6 +5,6 @@ router = APIRouter(prefix="/test", tags=["test"])
 
 
 @router.get("/index")
-async def test(req: Request):
+async def test(req: Request,current_user: User = Depends(get_current_user)):
     return {"message": "test","code":200}
     

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

@@ -0,0 +1,124 @@
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.dependencies import get_current_user, get_db
+from app.models.user import User
+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_by_type,
+    save_oauth_token,
+)
+
+router = APIRouter(prefix="/user-oauth", tags=["user-oauth"])
+
+
+@router.post(
+    "/request_oauth",
+    summary="获取授权链接",
+    description="用户提交oauth_type，返回对应的第三方授权链接",
+    response_model=RequestOAuthResponse,
+)
+async def request_oauth(
+    req: RequestOAuthRequest,
+    current_user: User = Depends(get_current_user),
+):
+    try:
+        auth_url = await build_oauth_url(req.oauth_type, current_user.id)
+        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) != 4:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="无效的state参数",
+            )
+
+        oauth_type = int(parts[0])
+        user_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)
+        account_info = await get_account_info_by_type(token, oauth_type, app_type)
+        user_oauth = await save_oauth_token(db, user_id, oauth_type, token, account_info, app_id)
+
+        return {
+            "message": "授权成功",
+            "code": 0,
+            "data": UserOAuthOut.model_validate(user_oauth),
+        }
+
+    except ValueError as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(e),
+        )
+    except Exception as e:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"授权失败: {str(e)}",
+        )
+
+
+@router.get(
+    "/callback",
+    summary="通用授权回调",
+    description="其他平台授权回调地址，接收code和state参数",
+)
+async def oauth_callback(
+    code: str = Query(..., description="第三方返回的授权码"),
+    state: str = Query(..., description="请求时传递的自定义参数"),
+    db: AsyncSession = Depends(get_db),
+):
+    try:
+        parts = state.split(":")
+        if len(parts) != 4:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="无效的state参数",
+            )
+
+        oauth_type = int(parts[0])
+        user_id = parts[1]
+        app_id = parts[2]
+        app_type = parts[3]
+
+        token = await get_token_by_type(code, oauth_type, app_id, app_type)
+        account_info = await get_account_info_by_type(token, oauth_type, app_type)
+        user_oauth = await save_oauth_token(db, user_id, oauth_type, token, account_info, app_id)
+
+        return {
+            "message": "授权成功",
+            "code": 0,
+            "data": UserOAuthOut.model_validate(user_oauth),
+        }
+
+    except ValueError as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(e),
+        )
+    except Exception as e:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"授权失败: {str(e)}",
+        )

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

@@ -0,0 +1,58 @@
+from datetime import datetime
+
+from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.models.base import Base, TimestampMixin, SoftDeleteMixin
+
+
+class UserOAuth(Base, TimestampMixin, SoftDeleteMixin):
+    __tablename__ = "user_oauth"
+
+    id: Mapped[str] = mapped_column(
+        String(32), primary_key=True, comment="主键"
+    )
+    account_id: Mapped[str] = mapped_column(
+        String(64), nullable=False, index=True, comment="授权账户id"
+    )
+    account_name: Mapped[str] = mapped_column(
+        String(128), nullable=False, comment="授权账户name"
+    )
+    account_role: Mapped[str | None] = mapped_column(
+        String(64), nullable=True, comment="授权账户角色"
+    )
+    account_username: Mapped[str | None] = mapped_column(
+        String(128), nullable=True, comment="授权账户登录账号"
+    )
+    user_id: Mapped[str] = mapped_column(
+        String(32), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True,
+        comment="用户id"
+    )
+    open_type: Mapped[int] = mapped_column(
+        Integer, nullable=False,
+        comment="开户方式（1=千川，2=广告，3=本地推，4=星图，5=快手代理商，6=巨量星图，7=巨量服务单，8=腾讯服务单，9=腾讯营销K2，10=腾讯营销K3）"
+    )
+    port_type: Mapped[int] = mapped_column(
+        Integer, nullable=False, 
+        comment="平台端口（1=巨量，2=磁力，3=巨量星图，4=服务单，5=腾讯）"
+    )
+    appid: Mapped[str | None] = mapped_column(
+        String(64), nullable=True, comment="授权应用id"
+    )
+    access_token: Mapped[str | None] = mapped_column(
+        Text, nullable=True, comment="授权token"
+    )
+    access_token_expired: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True, 
+        comment="token过期时间"
+    )
+    refresh_token: Mapped[str | None] = mapped_column(
+        Text, nullable=True, comment="授权刷新token"
+    )
+    refresh_token_expired: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True,
+        comment="刷新token过期时间"
+    )
+    material_auth_status: Mapped[bool] = mapped_column(
+        Boolean, default=False, comment="是否敏感物料授权（true=是，false=否）"
+    )

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

@@ -0,0 +1,25 @@
+from sqlalchemy import ForeignKey, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.models.base import Base, TimestampMixin, SoftDeleteMixin
+
+
+class UserOAuthAccount(Base, TimestampMixin, SoftDeleteMixin):
+    __tablename__ = "user_oauth_account"
+
+    id: Mapped[str] = mapped_column(
+        String(32), primary_key=True, comment="主键"
+    )
+    account_id: Mapped[str] = mapped_column(
+        String(64), ForeignKey("user_oauth.account_id", ondelete="CASCADE"),
+        nullable=False, index=True, comment="授权账户id（user_oauth表中同一个）"
+    )
+    advertiser_id: Mapped[str | None] = mapped_column(
+        String(64), nullable=True, index=True, comment="广告账户id"
+    )
+    advertiser_name: Mapped[str | None] = mapped_column(
+        String(128), nullable=True, comment="广告账户名"
+    )
+    advertiser_role: Mapped[str | None] = mapped_column(
+        String(64), nullable=True, comment="广告账户类型"
+    )

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

@@ -0,0 +1,29 @@
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+class RequestOAuthRequest(BaseModel):
+    oauth_type: int = Field(
+        ...,
+        description="开户方式（1=巨量广告，2=巨量千川，3=快手，4=腾讯）",
+    )
+
+
+class RequestOAuthResponse(BaseModel):
+    auth_url: str = Field(..., description="第三方授权链接")
+
+
+class UserOAuthOut(BaseModel):
+    id: str = Field(..., description="主键")
+    account_id: str = Field(..., description="授权账户id")
+    account_name: str = Field(..., description="授权账户name")
+    account_role: str | None = Field(None, description="授权账户角色")
+    account_username: str | None = Field(None, description="授权账户登录账号")
+    user_id: str = Field(..., description="用户id")
+    open_type: int = Field(..., description="开户方式")
+    port_type: int = Field(..., description="平台端口")
+    appid: str | None = Field(None, description="授权应用id")
+    material_auth_status: bool = Field(False, description="是否敏感物料授权")
+    created_at: datetime = Field(..., description="创建时间")
+    updated_at: datetime = Field(..., description="更新时间")

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

@@ -0,0 +1,353 @@
+import random
+from datetime import datetime
+
+import httpx
+from sqlalchemy import select, func
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import settings
+from app.models.user_oauth import UserOAuth
+from app.utils.id_gen import generate_id
+
+
+OAUTH_TYPE_CONFIG = {
+    1: {"port_type": 1, "name": "千川", "app_type": "juliang_qianchuan"},
+    2: {"port_type": 1, "name": "广告", "app_type": "juliang_ad"},
+    3: {"port_type": 1, "name": "本地推", "app_type": "juliang_ad"},
+    4: {"port_type": 1, "name": "星图", "app_type": "juliang_ad"},
+    5: {"port_type": 2, "name": "快手代理商", "app_type": "kuaishou"},
+    6: {"port_type": 3, "name": "巨量星图", "app_type": "juliang_ad"},
+    7: {"port_type": 4, "name": "巨量服务单", "app_type": "juliang_ad"},
+    8: {"port_type": 4, "name": "腾讯服务单", "app_type": "tencent"},
+    9: {"port_type": 5, "name": "腾讯营销K2", "app_type": "tencent"},
+    10: {"port_type": 5, "name": "腾讯营销K3", "app_type": "tencent"},
+}
+
+
+async def get_available_app(app_type: str, db: AsyncSession) -> dict:
+    if app_type == "juliang_ad":
+        apps = settings.JULIANG_AD_APPS
+    elif app_type == "juliang_qianchuan":
+        apps = settings.JULIANG_QIANCHUAN_APPS
+    elif app_type == "kuaishou":
+        apps = settings.KUAISHOU_APPS
+    elif app_type == "tencent":
+        apps = settings.TENCENT_APPS
+    else:
+        raise ValueError(f"不支持的应用类型: {app_type}")
+
+    if not apps:
+        raise ValueError(f"{app_type}未配置应用")
+
+    available_apps = []
+    for app in apps:
+        app_id = app.get("app_id")
+        if not app_id:
+            continue
+
+        result = await db.execute(
+            select(func.count(UserOAuth.id)).where(UserOAuth.appid == app_id)
+        )
+        count = result.scalar() or 0
+
+        if count < 5000:
+            available_apps.append(app)
+
+    if not available_apps:
+        raise ValueError("所有应用授权已超过最大数量")
+
+    return random.choice(available_apps)
+
+
+async def build_oauth_url(oauth_type: int, user_id: str) -> str:
+    if oauth_type == 1:
+        return await _build_juliang_oauth_url(oauth_type, user_id, app_type)
+    elif oauth_type == 2:
+        return await _build_kuaishou_oauth_url(oauth_type, user_id)
+    elif oauth_type == 3:
+        return await _build_tencent_oauth_url(oauth_type, user_id)
+    elif oauth_type == 4:
+        return await _build_tencent_oauth_url(oauth_type, user_id)    
+    else:
+        raise ValueError(f"不支持的应用类型: {app_type}")
+
+
+async def _build_juliang_oauth_url(oauth_type: int, user_id: str, app_type: str) -> str:
+    async with AsyncSession() as db:
+        app = await get_available_app(app_type, db)
+        app_id = app.get("app_id")
+
+    redirect_uri = "https://open.oceanengine.com/audit/oauth.html"
+    rid = "ktm0cl7napb"
+    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}",
+        "material_auth": 1,
+        "rid": rid,
+    }
+    query_string = "&".join(f"{k}={v}" for k, v in params.items())
+    return f"{redirect_uri}?{query_string}"
+
+
+async def _build_kuaishou_oauth_url(oauth_type: int, user_id: str) -> str:
+    async with AsyncSession() as db:
+        app = await get_available_app("kuaishou", 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": "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("应用配置不存在")
+
+    async with httpx.AsyncClient() as client:
+        response = await client.post(
+            url,
+            data={
+                "app_id": app_id,
+                "secret": app.get("secret"),
+                "auth_code": code,
+            },
+        )
+        response.raise_for_status()
+        content = response.json()
+        if content.get("code") != 0:
+            raise ValueError(content.get("message", "获取token失败"))
+        return content.get("data", {})
+
+
+async def get_kuaishou_token(code: str, oauth_type: int, app_id: str) -> dict:
+    url = "https://open.kuaishou.com/oauth2/token"
+    redirect_uri = f"{settings.BASE_URL}/api/user-oauth/callback"
+
+    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},
+        )
+        response.raise_for_status()
+        data = response.json()
+        if data.get("code") != 0:
+            raise ValueError(data.get("message", "获取账户信息失败"))
+        data = data.get("data", {})
+        return {
+            "account_id": data.get("advertiser_id", data.get("account_id", "")),
+            "account_name": data.get("advertiser_name", data.get("account_name", "")),
+            "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}"},
+        )
+        response.raise_for_status()
+        data = response.json()
+        return {
+            "account_id": data.get("account_id", ""),
+            "account_name": data.get("account_name", ""),
+            "account_role": data.get("role", ""),
+            "account_username": data.get("username", ""),
+        }
+
+
+async def _get_tencent_account_info(token: dict) -> dict:
+    access_token = token.get("access_token")
+    url = "https://api.e.qq.com/user/info"
+
+    async with httpx.AsyncClient() as client:
+        response = await client.get(
+            url,
+            headers={"Authorization": f"Bearer {access_token}"},
+        )
+        response.raise_for_status()
+        data = response.json()
+        return {
+            "account_id": data.get("account_id", ""),
+            "account_name": data.get("account_name", ""),
+            "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_name=account_info.get("account_name", ""),
+        account_role=account_info.get("account_role", ""),
+        account_username=account_info.get("account_username", ""),
+        user_id=user_id,
+        open_type=oauth_type,
+        port_type=config["port_type"],
+        appid=app_id,
+        access_token=access_token,
+        access_token_expired=expires_at,
+        refresh_token=refresh_token,
+        refresh_token_expired=refresh_expires_at,
+        material_auth_status=True,
+    )
+
+    db.add(user_oauth)
+    await db.flush()
+    return user_oauth