From b24c176b884f5bfa8a8b7b813704307b04197c5d Mon Sep 17 00:00:00 2001 From: 18610128193 <10574456+chenweiqiang-123@user.noreply.gitee.com> Date: Wed, 10 Jun 2026 10:47:18 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8F=90=E4=BA=A4=E5=88=9D=E5=A7=8B=E6=96=87?= =?UTF-8?q?=E4=BB=B6=EF=BC=8C=E5=A2=9E=E5=8A=A0=E6=8E=88=E6=9D=83=E5=BA=94?= =?UTF-8?q?=E7=94=A8=E8=B7=AF=E7=94=B1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- video-gen-api/app/api/v1/test.py | 2 +- video-gen-api/app/api/v1/user_oauth.py | 124 ++++++ video-gen-api/app/models/user_oauth.py | 58 +++ .../app/models/user_oauth_account.py | 25 ++ video-gen-api/app/schemas/user_oauth.py | 29 ++ .../app/services/user_oauth_service.py | 353 ++++++++++++++++++ 6 files changed, 590 insertions(+), 1 deletion(-) create mode 100644 video-gen-api/app/api/v1/user_oauth.py create mode 100644 video-gen-api/app/models/user_oauth.py create mode 100644 video-gen-api/app/models/user_oauth_account.py create mode 100644 video-gen-api/app/schemas/user_oauth.py create mode 100644 video-gen-api/app/services/user_oauth_service.py diff --git a/video-gen-api/app/api/v1/test.py b/video-gen-api/app/api/v1/test.py index c00efa3e..507cfd95 100644 --- a/video-gen-api/app/api/v1/test.py +++ b/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} diff --git a/video-gen-api/app/api/v1/user_oauth.py b/video-gen-api/app/api/v1/user_oauth.py new file mode 100644 index 00000000..d1582a05 --- /dev/null +++ b/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)}", + ) \ No newline at end of file diff --git a/video-gen-api/app/models/user_oauth.py b/video-gen-api/app/models/user_oauth.py new file mode 100644 index 00000000..245b447b --- /dev/null +++ b/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=否）" + ) \ No newline at end of file diff --git a/video-gen-api/app/models/user_oauth_account.py b/video-gen-api/app/models/user_oauth_account.py new file mode 100644 index 00000000..7bb53d50 --- /dev/null +++ b/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="广告账户类型" + ) \ No newline at end of file diff --git a/video-gen-api/app/schemas/user_oauth.py b/video-gen-api/app/schemas/user_oauth.py new file mode 100644 index 00000000..d43d6ff6 --- /dev/null +++ b/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="更新时间") \ No newline at end of file diff --git a/video-gen-api/app/services/user_oauth_service.py b/video-gen-api/app/services/user_oauth_service.py new file mode 100644 index 00000000..fefcb541 --- /dev/null +++ b/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 \ No newline at end of file