How to Automate TikTok Video Posting using the API

1#!/usr/bin/env python3
2"""TikTok Direct Post Video Publisher."""
3import os
4import math
5import time
6import logging
7import requests
8
9logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
10log = logging.getLogger(__name__)
11
12CLIENT_KEY = os.environ['TIKTOK_CLIENT_KEY']
13CLIENT_SECRET = os.environ['TIKTOK_CLIENT_SECRET']
14REFRESH_TOKEN = os.environ['TIKTOK_REFRESH_TOKEN']
15CHUNK_SIZE = 10 * 1024 * 1024  # 10MB
16
17def get_fresh_token():
18    resp = requests.post('https://open.tiktokapis.com/v2/oauth/token/',
19        data={'client_key': CLIENT_KEY, 'client_secret': CLIENT_SECRET,
20              'grant_type': 'refresh_token', 'refresh_token': REFRESH_TOKEN},
21        headers={'Content-Type': 'application/x-www-form-urlencoded'})
22    resp.raise_for_status()
23    return resp.json()['access_token']
24
25def tt(method, url, token, **kwargs):
26    headers = kwargs.pop('headers', {})
27    headers.update({'Authorization': f'Bearer {token}', 'Content-Type': 'application/json; charset=UTF-8'})
28    resp = getattr(requests, method)(url, headers=headers, **kwargs)
29    resp.raise_for_status()
30    data = resp.json()
31    if data.get('error', {}).get('code', 'ok') != 'ok':
32        raise ValueError(f'TikTok API error: {data["error"]}')
33    return data['data']
34
35def query_creator(token):
36    return tt('post', 'https://open.tiktokapis.com/v2/post/publish/creator_info/query/', token, json={})
37
38def init_post(token, video_path, title, privacy_level):
39    video_size = os.path.getsize(video_path)
40    return tt('post', 'https://open.tiktokapis.com/v2/post/publish/video/init/', token, json={
41        'post_info': {'title': title, 'privacy_level': privacy_level, 'disable_duet': False, 'disable_comment': False, 'disable_stitch': False},
42        'source_info': {'source': 'FILE_UPLOAD', 'video_size': video_size, 'chunk_size': CHUNK_SIZE, 'total_chunk_count': math.ceil(video_size / CHUNK_SIZE)}
43    }), video_size
44
45def upload_chunks(upload_url, video_path, video_size):
46    n = math.ceil(video_size / CHUNK_SIZE)
47    with open(video_path, 'rb') as f:
48        for i in range(n):
49            start = i * CHUNK_SIZE
50            end = min(start + CHUNK_SIZE - 1, video_size - 1)
51            f.seek(start)
52            data = f.read(end - start + 1)
53            resp = requests.put(upload_url, data=data,
54                headers={'Content-Range': f'bytes {start}-{end}/{video_size}', 'Content-Type': 'video/mp4'}, timeout=120)
55            resp.raise_for_status()
56            log.info(f'Chunk {i+1}/{n} uploaded')
57
58def poll_status(publish_id, token, timeout=600):
59    deadline = time.time() + timeout
60    delay = 10
61    while time.time() < deadline:
62        data = tt('post', 'https://open.tiktokapis.com/v2/post/publish/status/fetch/', token, json={'publish_id': publish_id})
63        status = data.get('status')
64        log.info(f'Status: {status}')
65        if status == 'PUBLISH_COMPLETE':
66            if data.get('privacy_level') == 'SELF_ONLY':
67                log.warning('Post is PRIVATE — audit approval needed for public posts')
68            return data
69        if status == 'FAILED':
70            raise ValueError(f'Publish failed: {data}')
71        delay = min(delay * 1.5, 60)
72        time.sleep(delay)
73    raise TimeoutError('Polling timeout')
74
75def publish_video(video_path, title, privacy_level='PUBLIC_TO_EVERYONE'):
76    log.info('Refreshing token...')
77    token = get_fresh_token()
78    log.info('Querying creator info...')
79    creator = query_creator(token)
80    log.info(f'Creator: {creator["creator_nickname"]}')
81    if privacy_level not in creator.get('privacy_level_options', []):
82        log.warning(f'{privacy_level} not available; using SELF_ONLY')
83        privacy_level = 'SELF_ONLY'
84    log.info('Initializing post...')
85    (post_data, video_size) = init_post(token, video_path, title, privacy_level)
86    publish_id = post_data['publish_id']
87    upload_url = post_data['upload_url']
88    log.info(f'Uploading video ({video_size/1024/1024:.1f}MB)...')
89    upload_chunks(upload_url, video_path, video_size)
90    log.info('Polling publish status...')
91    result = poll_status(publish_id, token)
92    log.info(f'Done! Video ID: {result.get("video_id")}')
93    return result
94
95if __name__ == '__main__':
96    result = publish_video('/path/to/video.mp4', 'My TikTok video #fyp')
97    print(f'Published: {result}')