140 lines
4.7 KiB
Python
140 lines
4.7 KiB
Python
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 {}}
|
|
)
|