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

from typing import Any, Dict, Optional

from app.utils.douyinRequest import DouyinRequest
from app.models.user_oauth import UserOAuth


class DouyinApi:
    def __init__(self):
        self.request = DouyinRequest()

    async def get_advertiser_by_agent(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://api.oceanengine.com/open_api/2/agent/advertiser/select/"
        return await self.request.request_with_token_with_context(
            oauth_id,
            url,
            'GET',
            {'params': params or {}}
        )
    #上传本地推图片
    async def upload_local_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://api.oceanengine.com/open_api/v3.0/local/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_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://api.oceanengine.com/open_api/2/file/image/ad/"
        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://api.oceanengine.com/open_api/2/file/video/ad/"
        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_local_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://api.oceanengine.com/open_api/v3.0/local/file/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://api.oceanengine.com/open_api/2/tools/admin/info/"
        return await self.request.request_with_token_with_context(
            oauth_id,
            url,
            'GET',
            {'params': params or {}}
        )
    
    #前测素材,Adv创建前测任务
    async def pre_test_material(self, oauth_id: str, params: any) -> Dict[str, Any]:
        if not oauth_id:
            raise RuntimeError('OAuth ID is not set.')
        url = "https://api.oceanengine.com/open_api/2/diagnosis_task/adv/create/"
        return await self.request.request_with_token_with_context(
            oauth_id,
            url,
            'POST',
            {"json": 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://api.oceanengine.com/open_api/v3.0/report/custom/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://api.oceanengine.com/open_api/2/agent/advertiser_info/query/"
        return await self.request.request_with_token_with_context(
            oauth_id,
            url,
            'GET',
            {'params': params or {}}
        )

    #获取素材前测结果
    async def get_material_pre_test_result(self, oauth_id: str, params: any) -> Dict[str, Any]:
        if not oauth_id:
            raise RuntimeError('OAuth ID is not set.')
        url = "https://api.oceanengine.com/open_api/2/diagnosis_task/adv/get/"
        return await self.request.request_with_token_with_context(
            oauth_id,
            url,
            'GET',
            {'params': params or {}}
        )