TikTok API get user videos: any public account, with two limits

import csv
import json
import os
import sys
import time
from datetime import datetime, timezone

import requests

BASE = "https://api.toolzerhub.com/v2/tiktok/app"
HEADERS = {"x-api-key": os.environ.get("TOOLZERHUB_API_KEY", "")}
FIELDS = ["account", "video_id", "url", "listed", "posted_at", "pinned", "seconds",
          "plays", "likes", "comments", "shares", "saves", "caption"]
calls = 0


def cell(value):
    # Spreadsheets run text starting with = + - or @ as a formula; an apostrophe keeps it text.
    if isinstance(value, str) and value[:1] in ("=", "+", "-", "@", "\t", "\r"):
        return "'" + value
    return value


def call(path, params=None, ids=None, tries=4):
    # 429 and 503 are retried; the wait comes from the error body or the Retry-After header.
    global calls
    for attempt in range(tries):
        if ids:
            r = requests.post(BASE + path, headers=HEADERS, json=ids, timeout=90)
        else:
            r = requests.get(BASE + path, headers=HEADERS, params=params, timeout=90)
        calls += 1
        if r.status_code == 400:  # malformed input: skip this account
            print("bad request:", str(r.json()["error"].get("message")).split("\n")[0])
            return None
        if r.status_code not in (429, 503) or attempt == tries - 1:
            if r.status_code >= 500:  # any 5xx, or a 503 past the retries: skip
                print(f"HTTP {r.status_code}, skipping:", r.text[:120].replace("\n", " "))
                return None
            r.raise_for_status()
            return r.json().get("data")
        try:
            wait = r.json()["error"]["details"]["retryAfter"]
        except (ValueError, KeyError, TypeError):
            wait = None
        try:
            wait = int(wait or r.headers.get("Retry-After") or 5 * (attempt + 1))
        except ValueError:
            wait = 5 * (attempt + 1)
        time.sleep(min(wait, 60))


def write(writer, account, v, listed):
    author, stats = v.get("author") or {}, v.get("statistics") or {}
    writer.writerow({
        "account": account, "video_id": v["aweme_id"], "listed": listed,
        "url": f"https://www.tiktok.com/@{author.get('unique_id') or account}/video/{v['aweme_id']}",
        "posted_at": datetime.fromtimestamp(v["create_time"], timezone.utc).isoformat()
        if v.get("create_time") else "",
        "pinned": "yes" if v.get("is_top") else "",  # pinned videos come first, out of date order
        "seconds": round(((v.get("video") or {}).get("duration") or 0) / 1000) or "",
        "plays": stats.get("play_count"), "likes": stats.get("digg_count"),
        "comments": stats.get("comment_count"), "shares": stats.get("share_count"),
        "saves": stats.get("collect_count"),
        "caption": cell((v.get("desc") or "").replace("\r", " ").replace("\n", " ")),
    })


def export_account(writer, account, max_pages, fetch_hidden):
    params, seen, hidden, pages = {"unique_id": account, "max_cursor": 0}, set(), set(), 0
    status = "page cap reached, more available"
    while pages < max_pages:
        data = call("/user/posts", params)
        videos = (data or {}).get("aweme_list") or []
        if not videos and not (data or {}).get("has_more"):  # unknown users: 200 with no list
            status = "stopped early: failed or empty page" if pages else "nothing returned"
            break
        pages += 1
        for v in videos:  # a page can be empty when every video on it is geofenced
            if v.get("aweme_id") and v["aweme_id"] not in seen:  # pages can repeat a video
                seen.add(v["aweme_id"])
                write(writer, account, v, "yes")
        # Videos TikTok won't show in this region are left out, but their IDs are listed here.
        reasons = data.get("lite_filter_reasons") or {}
        hidden.update((json.loads(reasons) if isinstance(reasons, str) else reasons) or {})
        if not (data.get("has_more") and data.get("max_cursor")):
            status = "end"
            break
        params["max_cursor"] = data["max_cursor"]
        time.sleep(0.3)
    hidden -= seen
    got = 0
    if fetch_hidden:
        todo = sorted(hidden)
        for i in range(0, len(todo), 25):  # the batch route takes up to 25 IDs a call
            for v in call("/video/multi", ids=todo[i:i + 25]) or []:
                if v.get("aweme_id") and v["aweme_id"] not in seen:
                    seen.add(v["aweme_id"])
                    write(writer, account, v, "no, fetched by ID")
                    got += 1
    print(f"{account}: {len(seen) - got} listed, {len(hidden)} hidden, {got} fetched, {pages} pages ({status})")
    return len(seen)


def main(accounts, max_pages="50", fetch_hidden="no", out="tiktok_videos.csv"):
    total = 0
    with open(out, "w", newline="", encoding="utf-8-sig") as f:  # BOM: Excel reads emoji
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        writer.writeheader()
        for account in [a.strip().lstrip("@") for a in accounts.split(",") if a.strip()]:
            total += export_account(writer, account, int(max_pages), fetch_hidden == "yes")
    print(f"wrote {total} videos to {out} in {calls} calls")


if __name__ == "__main__":
    if len(sys.argv) < 2 or (len(sys.argv) > 3 and sys.argv[3] not in ("yes", "no")):
        sys.exit("usage: python tiktok_user_videos.py nasa,nba [max_pages] [yes|no fetch hidden] [file.csv]")
    if not HEADERS["x-api-key"]:
        sys.exit("set TOOLZERHUB_API_KEY first")
    main(*sys.argv[1:])

export TOOLZERHUB_API_KEY=your_key
python tiktok_user_videos.py "nasa, @natgeo ,,zzzznotauser9q8z7x" 70 yes tiktok_videos.csv

nasa: 57 listed, 0 hidden, 0 fetched, 3 pages (end)
natgeo: 307 listed, 1232 hidden, 1231 fetched, 34 pages (end)
zzzznotauser9q8z7x: 0 listed, 0 hidden, 0 fetched, 0 pages (nothing returned)
wrote 1595 videos to tiktok_videos.csv in 88 calls