TikTok live API: find live streams with viewer counts

export TOOLZERHUB_API_KEY=your_key
curl -H "x-api-key: $TOOLZERHUB_API_KEY" \
  "https://api.toolzerhub.com/v2/tiktok/app/search/live?keyword=music"

{
  "type": 1,
  "lives": {
    "author": {"unique_id": "troystarrmusic", "room_id_str": "7704408845073877773"},
    "rawdata": {
      "id_str": "7704408845073877773",
      "title": "Lets Go LIVE!",
      "user_count": 62,
      "stats": {"total_user": 1609},
      "like_count": 60041,
      "start_time": 1793822566,
      "hashtag": {"title": "Music"},
      "owner": {"follow_info": {"follower_count": 17264}}
    }
  }
}

import csv
import json
import os
import statistics
import sys
import time

import requests

API = "https://api.toolzerhub.com/v2/tiktok/app/search/live"
HEADERS = {"x-api-key": os.environ.get("TOOLZERHUB_API_KEY", "")}
FIELDS = ["keyword", "room_id", "host", "followers", "title", "category", "viewers_now", "viewers_total", "likes",
          "minutes_live", "url"]


def get(params, tries=3):
    # Retry rate limits, server errors and dropped connections.
    for attempt in range(tries):
        last = attempt == tries - 1
        try:
            r = requests.get(API, headers=HEADERS, params=params, timeout=120)
        except requests.RequestException:
            if last:
                raise
            time.sleep(5 * (attempt + 1))
            continue
        if r.status_code in (429, 500, 502, 503, 504) and not last:
            time.sleep(5 * (attempt + 1))
            continue
        r.raise_for_status()
        return r.json()["data"]


def main(keywords, pages="2", out="tiktok_live.csv"):
    with open(out, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        writer.writeheader()
        for keyword in filter(None, (k.strip() for k in keywords.split(","))):
            rooms, seen, offset = [], set(), 0
            for _ in range(int(pages)):
                data = get({"keyword": keyword, "offset": offset})
                # Results mix live rooms (type 1) with accounts; each room's details are a JSON string.
                for item in data.get("data") or []:
                    if item.get("type") != 1:
                        continue
                    raw = item["lives"]["rawdata"]
                    room = json.loads(raw) if isinstance(raw, str) else raw
                    if room["id_str"] not in seen:
                        seen.add(room["id_str"])
                        rooms.append((item["lives"]["author"], room))
                if not data.get("has_more"):
                    break
                offset = data["cursor"]
            now = time.time()
            for author, room in rooms:
                writer.writerow({
                    "keyword": keyword, "room_id": room["id_str"], "host": author["unique_id"],
                    "followers": (room.get("owner") or {}).get("follow_info", {}).get("follower_count"),
                    "title": room.get("title"), "category": (room.get("hashtag") or {}).get("title"),
                    "viewers_now": room.get("user_count"), "viewers_total": (room.get("stats") or {}).get("total_user"),
                    "likes": room.get("like_count"), "minutes_live": round((now - room["start_time"]) / 60),
                    "url": f"https://www.tiktok.com/@{author['unique_id']}/live",
                })
            viewers = [room.get("user_count") or 0 for _, room in rooms]
            hours = [(now - room["start_time"]) / 3600 for _, room in rooms]
            print(f"{keyword}: {len(rooms)} live rooms, median {statistics.median(viewers or [0]):,.0f} viewers now, "
                  f"{sum(v < 10 for v in viewers)} under 10, longest live {max(hours or [0]):.1f} hours")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        sys.exit('usage: python tiktok_live.py "music,cooking" [pages] [out.csv]')
    if not HEADERS["x-api-key"]:
        sys.exit("set TOOLZERHUB_API_KEY first")
    main(*sys.argv[1:])