From 7e6318742c722da336b3bcb99daa7aaf48c759eb Mon Sep 17 00:00:00 2001
From: 18610128193 <10574456+chenweiqiang-123@user.noreply.gitee.com>
Date: Fri, 3 Jul 2026 13:54:22 +0800
Subject: [PATCH] =?UTF-8?q?=E5=BF=AB=E6=89=8B=E6=8E=88=E6=9D=83=E4=B8=8D?=
 =?UTF-8?q?=E5=AE=8C=E5=96=84?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 video-gen-api/app/utils/kuaishouApi.py     | 139 +++++++++++
 video-gen-api/app/utils/kuaishouRequest.py | 276 +++++++++++++++++++++
 2 files changed, 415 insertions(+)
 create mode 100644 video-gen-api/app/utils/kuaishouApi.py
 create mode 100644 video-gen-api/app/utils/kuaishouRequest.py

diff --git a/video-gen-api/app/utils/kuaishouApi.py b/video-gen-api/app/utils/kuaishouApi.py
new file mode 100644
index 00000000..0d24fb18
--- /dev/null
+++ b/video-gen-api/app/utils/kuaishouApi.py
@@ -0,0 +1,139 @@
+from typing import Any, Dict, Optional
+
+from app.utils.kuaishouRequest import KuaishouRequest
+from app.models.user_oauth import UserOAuth
+
+
+class KuaishouApi:
+    def __init__(self):
+        self.request = KuaishouRequest()
+
+    # 获取广告主信息
+    async def get_advertiser_info(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 = "https://ad.e.kuaishou.com/rest/openapi/v1/advertiser/info"
+        return await self.request.request_with_token_with_context(
+            oauth_id,
+            url,
+            'GET',
+            {'params': params or {}}
+        )
+
+    # 上传图片素材
+    async def upload_image_material(self, oauth_id: str, data: Optional[Dict[str, Any]] = None, files: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        if not oauth_id:
+            raise RuntimeError('OAuth ID is not set.')
+
+        url = "https://ad.e.kuaishou.com/rest/openapi/v1/file/ad/image/upload"
+        options: Dict[str, Any] = {}
+        if data:
+            options['data'] = data
+        if files:
+            options['files'] = files
+        return await self.request.request_with_token_with_context(
+            oauth_id,
+            url,
+            'POST',
+            options,
+            request_count=3
+        )
+
+    # 上传视频素材
+    async def upload_video_material(self, oauth_id: str, data: Optional[Dict[str, Any]] = None, files: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        if not oauth_id:
+            raise RuntimeError('OAuth ID is not set.')
+
+        url = "https://ad.e.kuaishou.com/rest/openapi/v1/file/ad/video/upload"
+        options: Dict[str, Any] = {}
+        if data:
+            options['data'] = data
+        if files:
+            options['files'] = files
+        return await self.request.request_with_token_with_context(
+            oauth_id,
+            url,
+            'POST',
+            options,
+            request_count=3
+        )
+
+    # 获取地域信息
+    async def get_area(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 = "https://ad.e.kuaishou.com/rest/openapi/v1/tools/admin/info"
+        return await self.request.request_with_token_with_context(
+            oauth_id,
+            url,
+            'GET',
+            {'params': params or {}}
+        )
+
+    # 获取素材消耗（报表查询）
+    async def get_material_cost(self, oauth_id: str, params: any, request_count: int = 3) -> Dict[str, Any]:
+        if not oauth_id:
+            raise RuntimeError('OAuth ID is not set.')
+
+        url = "https://ad.e.kuaishou.com/rest/openapi/v1/report/get"
+        return await self.request.request_with_token_with_context(
+            oauth_id,
+            url,
+            'GET',
+            {'params': params or {}},
+            request_count=request_count
+        )
+
+    # 获取账户信息
+    async def get_account_info(self, oauth_id: str, params: any) -> Dict[str, Any]:
+        if not oauth_id:
+            raise RuntimeError('OAuth ID is not set.')
+
+        url = "https://ad.e.kuaishou.com/rest/openapi/v1/advertiser/info"
+        return await self.request.request_with_token_with_context(
+            oauth_id,
+            url,
+            'GET',
+            {'params': params or {}}
+        )
+
+    # 获取广告计划列表
+    async def get_ad_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 = "https://ad.e.kuaishou.com/rest/openapi/v1/ad/list"
+        return await self.request.request_with_token_with_context(
+            oauth_id,
+            url,
+            'GET',
+            {'params': params or {}}
+        )
+
+    # 获取广告组列表
+    async def get_ad_unit_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 = "https://ad.e.kuaishou.com/rest/openapi/v2/ad_unit/list"
+        return await self.request.request_with_token_with_context(
+            oauth_id,
+            url,
+            'GET',
+            {'params': params or {}}
+        )
+
+    # 获取账户余额
+    async def get_account_balance(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 = "https://ad.e.kuaishou.com/rest/openapi/v1/advertiser/balance"
+        return await self.request.request_with_token_with_context(
+            oauth_id,
+            url,
+            'GET',
+            {'params': params or {}}
+        )
diff --git a/video-gen-api/app/utils/kuaishouRequest.py b/video-gen-api/app/utils/kuaishouRequest.py
new file mode 100644
index 00000000..239b9b59
--- /dev/null
+++ b/video-gen-api/app/utils/kuaishouRequest.py
@@ -0,0 +1,276 @@
+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
+from app.utils.logger import get_logger
+
+logger = get_logger("kuaishou_request", "kuaishou_request")
+
+
+class KuaishouRequest:
+    def __init__(self, platform: str = "kuaishou"):
+        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).replace(tzinfo=timezone.utc)
+                    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("应用已被删除或禁用")
+
+        # 快手刷新token接口
+        response = await self.client.request(
+            'POST',
+            'https://ad.e.kuaishou.com/rest/openapi/oauth2/authorize/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('access_token_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: any = None,
+        request_count: int = 3,
+    ) -> Any:
+        options = options or {}
+        token = await self.get_access_token(oauth_id)
+
+        for i in range(1, request_count+1):
+            try:
+                headers = options.get('headers', {}).copy()
+                headers['Access-Token'] = token
+
+                has_files = 'files' in options
+                if has_files:
+                    headers.pop('Content-Type', None)
+                else:
+                    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, 'msg': 'JSON解析失败'}
+
+                # 如果没有code，直接返回数据，快手触发接口频次会返回空
+                if 'code' not in data:
+                    await asyncio.sleep(i * 5)
+                    continue
+
+                code = data.get('code', 0)
+
+                # 检查是否需要刷新令牌
+                if code in [402000, 400003, 402007, 402005, 402004, 401000]:
+                    # 如果message包含 '该账户不在您的代理商下'，说明账户已经转走了，不需要刷新token，直接返回
+                    message = data.get('message', '')
+                    if '该账户不在您的代理商下' in message:
+                        return data
+
+                    await asyncio.sleep(i * 5)
+                    token = await self.get_access_token(oauth_id, force_refresh=True)
+                    continue
+
+                # 检查是否触发接口频次
+                if code in [400001, 402007, 402008, 410000, 410001]:
+                    await asyncio.sleep(i * 5)
+                    continue
+
+                # 服务端错误
+                if code >= 500000:
+                    await asyncio.sleep(i * 5)
+                    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
+
+        options_log = {}
+        if options:
+            for key, value in options.items():
+                if key == 'files':
+                    options_log[key] = {k: (v[0], 'bytes_content', v[2]) for k, v in value.items()}
+                else:
+                    options_log[key] = value
+
+        res = json.dumps(data, ensure_ascii=False) if 'data' in locals() else ''
+
+        logger.error(
+            f'KuaiShou API request failed after {request_count} retries. '
+            f'url:{url};method:{method};oauth_id:{oauth_id};options:{json.dumps(options_log, ensure_ascii=False)};response:{res}'
+        )
+
+        if 'data' in locals() and data.get('code', 0) != 0:
+            raise ValueError(f'接口返回错误[code:{data.get("code", "接口编码")}]{data.get("message", "接口返回错误")}')
+        else:
+            raise ValueError('网络错误，稍后重试。')
+