from typing import Any from fansifter_common.httpclient import HTTPClient from .constants import AUDIENCE_CALCULATE_TYPE from .exceptions import TikTokClientError from .types import Audience, AudienceFile, AudienceFileResponse, AudienceResponse class TikTokClient(HTTPClient): exception_class = TikTokClientError base_url = "https://business-api.tiktok.com" sensitive_params = { "secret", "auth_token", "access_token", } def __init__(self, client_options: dict[str, Any] | None = None) -> None: client_options = client_options or {} client_options["base_url"] = self.base_url super().__init__(client_options=client_options) def upload_audience_file( self, access_token: str, ad_account_external_id: str, file: bytes, file_signature: str, ) -> AudienceFile: """Upload an audience file to the TikTok""" response = self.request( "POST", "/open_api/v1.3/dmp/custom_audience/file/upload/", headers={ "Access-Token": access_token, }, data={ "advertiser_id": ad_account_external_id, "calculate_type": AUDIENCE_CALCULATE_TYPE, "file_signature": file_signature, }, files={"file": ("file.csv", file)}, type=AudienceFileResponse, timeout=60.0, ) return response.data def create_custom_audience( self, access_token: str, ad_account_external_id: str, audience_name: str, file_paths: list[str], ) -> Audience: response = self.request( "POST", "/open_api/v1.3/dmp/custom_audience/create/", headers={ "Access-Token": access_token, }, json={ "advertiser_id": ad_account_external_id, "custom_audience_name": audience_name, "calculate_type": AUDIENCE_CALCULATE_TYPE, "file_paths": file_paths, }, type=AudienceResponse, ) return response.data