139 lines
4.9 KiB
Python
139 lines
4.9 KiB
Python
# social/services/tiktok.py
|
|
import requests
|
|
import datetime
|
|
import time
|
|
import logging
|
|
from django.conf import settings
|
|
from django.utils import timezone
|
|
from apps.social.utils.tiktok import TikTokConstants
|
|
from apps.social.models import SocialAccount
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class TikTokAPIError(Exception):
|
|
pass
|
|
|
|
class TikTokService:
|
|
|
|
@staticmethod
|
|
def get_auth_url():
|
|
state = "random_secure_string" # Ideally generate dynamically
|
|
return (
|
|
f"{TikTokConstants.ENDPOINTS['AUTH']}"
|
|
f"?app_id={settings.TIKTOK_CLIENT_KEY}"
|
|
f"&state={state}"
|
|
f"&scope={TikTokConstants.SCOPES}"
|
|
f"&redirect_uri={settings.TIKTOK_REDIRECT_URI}"
|
|
)
|
|
|
|
@staticmethod
|
|
def exchange_code_for_token(code):
|
|
url = f"{TikTokConstants.BASE_URL}{TikTokConstants.ENDPOINTS['TOKEN']}"
|
|
payload = {
|
|
"app_id": settings.TIKTOK_CLIENT_KEY,
|
|
"secret": settings.TIKTOK_CLIENT_SECRET,
|
|
"auth_code": code,
|
|
}
|
|
resp = requests.post(url, json=payload)
|
|
data = resp.json()
|
|
|
|
if data.get('code') != 0:
|
|
logger.error(f"TikTok Token Error: {data}")
|
|
raise TikTokAPIError(data.get('message', 'Token exchange failed'))
|
|
|
|
return data.get('data', {})
|
|
|
|
@staticmethod
|
|
def refresh_tokens(account):
|
|
url = f"{TikTokConstants.BASE_URL}{TikTokConstants.ENDPOINTS['TOKEN']}"
|
|
payload = {
|
|
"app_id": settings.TIKTOK_CLIENT_KEY,
|
|
"secret": settings.TIKTOK_CLIENT_SECRET,
|
|
"refresh_token": account.refresh_token,
|
|
}
|
|
resp = requests.post(url, json=payload)
|
|
data = resp.json()
|
|
|
|
if data.get('code') != 0:
|
|
account.is_active = False
|
|
account.save()
|
|
raise TikTokAPIError("Refresh Failed")
|
|
|
|
token_data = data.get('data', {})
|
|
account.access_token = token_data['access_token']
|
|
account.refresh_token = token_data['refresh_token']
|
|
account.expires_at = timezone.now() + datetime.timedelta(seconds=token_data.get('expires_in', 86400))
|
|
account.save()
|
|
return account
|
|
|
|
@staticmethod
|
|
def get_valid_token(account):
|
|
if account.expires_at <= timezone.now() + datetime.timedelta(minutes=5):
|
|
account = TikTokService.refresh_tokens(account)
|
|
return account.access_token
|
|
|
|
@staticmethod
|
|
def make_api_request(endpoint, account, payload=None, method="POST"):
|
|
token = TikTokService.get_valid_token(account)
|
|
headers = {
|
|
"Access-Token": token, # Business API specific header
|
|
"Content-Type": "application/json"
|
|
}
|
|
url = f"{TikTokConstants.BASE_URL}{endpoint}"
|
|
|
|
if payload is None:
|
|
payload = {}
|
|
|
|
# CRITICAL: Business API requires advertiser_id in the body
|
|
if not account.platform_id:
|
|
raise TikTokAPIError("Advertiser ID is missing from account configuration.")
|
|
|
|
payload['advertiser_id'] = account.platform_id
|
|
|
|
try:
|
|
# Business API primarily uses POST for data retrieval
|
|
if method == "POST":
|
|
response = requests.post(url, headers=headers, json=payload)
|
|
else:
|
|
response = requests.get(url, headers=headers, params=payload)
|
|
|
|
data = response.json()
|
|
|
|
if data.get('code') != 0:
|
|
raise TikTokAPIError(f"API Error {data.get('code')}: {data.get('message')}")
|
|
|
|
return data.get('data', {})
|
|
|
|
except Exception as e:
|
|
raise TikTokAPIError(str(e))
|
|
|
|
@staticmethod
|
|
def fetch_ads(account, page=1, page_size=20):
|
|
"""Fetch Ads to act as 'Content' containers for comments."""
|
|
payload = {
|
|
"page": page,
|
|
"page_size": page_size,
|
|
# Filtering could be added here, e.g., filtering by active status
|
|
}
|
|
return TikTokService.make_api_request(TikTokConstants.ENDPOINTS['AD_LIST'], account, payload)
|
|
|
|
@staticmethod
|
|
def fetch_comments_for_ad(account, ad_id, page=1, page_size=20):
|
|
"""Fetch comments for a specific Advertisement."""
|
|
payload = {
|
|
# 'item_id' in Business API context is the Ad ID
|
|
"item_id": ad_id,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
}
|
|
return TikTokService.make_api_request(TikTokConstants.ENDPOINTS['COMMENT_LIST'], account, payload)
|
|
|
|
@staticmethod
|
|
def reply_to_comment(account, ad_id, comment_id, text):
|
|
"""Reply to a comment on an Advertisement."""
|
|
payload = {
|
|
"item_id": ad_id, # Ad ID required
|
|
"comment_id": comment_id,
|
|
"text": text
|
|
}
|
|
return TikTokService.make_api_request(TikTokConstants.ENDPOINTS['COMMENT_REPLY'], account, payload) |