TikTok 直播状态 API：按用户名检查账号是否开播

把关注的 TikTok 账号交给 SocQ 批量检查，将用户名、开播状态和检查时间整理成状态表。按自己的业务节奏刷新，再比较前后观察，就能把开播变化接入运营看板和提醒流程。

SocQ TikTok User Live API 按用户名查询直播状态，为成功查询返回状态记录。本教程展示批量请求、状态表导出和变化识别代码。

快速答案： 将非空 usernames 数组提交到 POST /v1/tiktok/user-live，任务成功后，从 data.results.items 读取 is_live 与 status。成功的离线记录为 is_live=false、status=offline；一个任务执行一次检查后便结束。

按用户名检查 TikTok 账号是否开播

设置 SOCQ_API_KEY，提交账号用户名：

curl -X POST "https://api.socq.ai/v1/tiktok/user-live" \
  -H "Authorization: Bearer $SOCQ_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"usernames":["thejustalex"]}'

输入使用用户名，带 @ 和不带 @ 均可。批量查询时，将多个不同用户名放入同一数组：

{"usernames":["thejustalex","@tiktok"]}

User Live 从用户名发起查询。将响应的 data.task_id 保存到 TASK_ID，再读取该任务：

curl "https://api.socq.ai/v1/tasks/$TASK_ID?limit=100" \
  -H "Authorization: Bearer $SOCQ_API_KEY"

queued、running 和 retrying 表示采集任务尚未完成。任务成功后，用记录中的 is_live 判断账号状态；任务失败时查看 data.error_message。客户端超时后，保留任务 ID 继续查询。

区分离线、指标缺失和查询失败

正在直播的记录使用 title、author、media 和 metrics.viewers_count 保存标题、作者、封面及观看人数。来源未提供的字段保留 null，实际返回的数值零保留为 0。

下面是一条离线状态记录的字段示例：

{
  "id": "6742945285876515845",
  "type": "live_status",
  "is_live": false,
  "status": "offline",
  "title": null,
  "author": {"id": "6742945285876515845", "username": "thejustalex"},
  "metrics": null,
  "media": [],
  "collected_at": "2026-10-15T08:12:57.773115Z"
}

这份样本展示了离线状态的字段结构。读取同一个任务会取得该次保存的观察；刷新账号状态时，创建新的 User Live 任务。

用 Python 建立可刷新的状态表

将以下只使用标准库的任务函数保存为 socq_collect.py。它在任务成功后读取全部结果页：

import json
import os
import time
from urllib.parse import urlencode
from urllib.request import Request, urlopen

BASE = "https://api.socq.ai/v1"
PENDING = {"queued", "running", "retrying"}

def api(method, path, payload=None):
    body = json.dumps(payload).encode("utf-8") if payload is not None else None
    request = Request(
        BASE + path, data=body, method=method,
        headers={
            "Authorization": "Bearer " + os.environ["SOCQ_API_KEY"],
            "Content-Type": "application/json",
        },
    )
    with urlopen(request, timeout=60) as response:
        result = json.load(response)
    if result.get("code") != 200:
        raise RuntimeError(result)
    return result["data"]

def collect(endpoint, payload):
    task_id = api("POST", "/" + endpoint, payload)["task_id"]
    print("Task:", task_id, flush=True)
    path = "/tasks/" + task_id
    deadline = time.monotonic() + 600
    data = api("GET", path + "?limit=100")
    while data["status"] in PENDING:
        if time.monotonic() >= deadline:
            raise TimeoutError("Resume GET " + path + "; do not resubmit automatically")
        time.sleep(2)
        data = api("GET", path + "?limit=100")
    if data["status"] != "succeeded":
        raise RuntimeError(data.get("error_message") or data["status"])

    items, seen_cursors = [], set()
    while True:
        if time.monotonic() >= deadline:
            raise TimeoutError("Resume pagination for " + task_id)
        page = data["results"]
        items.extend(page["items"])
        if not page["has_more"]:
            return task_id, items
        cursor = page.get("next_cursor")
        if not cursor or cursor in seen_cursors:
            raise RuntimeError("Missing or repeated result cursor")
        seen_cursors.add(cursor)
        data = api("GET", path + "?" + urlencode({"limit": 100, "cursor": cursor}))
        if data["status"] != "succeeded":
            raise RuntimeError("Task is not ready for result pagination")

在同一目录保存 check_live.py，按需要修改账号列表：

import csv
import json
from datetime import datetime
from pathlib import Path
from socq_collect import collect

usernames = ["thejustalex", "@tiktok"]
requested = {u.strip().lstrip("@").casefold() for u in usernames}
state_path = Path("live-status-state.json")
previous = json.loads(state_path.read_text()) if state_path.exists() else {}
state = dict(previous)
task_id, items = collect("tiktok/user-live", {"usernames": usernames})
with open("live-status-results.json", "w", encoding="utf-8") as f:
    json.dump({"task_id": task_id, "input": {"usernames": usernames}, "items": items},
              f, ensure_ascii=False, indent=2)

def timestamp(value):
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
        return parsed if parsed.tzinfo else None
    except (AttributeError, TypeError, ValueError):
        return None

def csv_cell(value):
    if value is None:
        return ""
    if isinstance(value, str) and value.lstrip().startswith(("=", "+", "-", "@")):
        return "'" + value
    return value

rows, observed, went_live = [], set(), []
for item in items:
    author = item.get("author") or {}
    username = (author.get("username") or "").lstrip("@").casefold()
    live = item.get("is_live")
    checked = timestamp(item.get("collected_at"))
    if username not in requested or type(live) is not bool or checked is None:
        continue
    old = state.get(username) or {}
    old_checked = timestamp(old.get("collected_at"))
    if old_checked is not None and checked <= old_checked:
        continue
    same_account = bool(author.get("id")) and old.get("author_id") == author["id"]
    if same_account and old.get("is_live") is False and live is True:
        went_live.append(username)
    state[username] = {"author_id": author.get("id"), "is_live": live,
                       "collected_at": item["collected_at"]}
    observed.add(username)
    rows.append({"username": username, "status": "live" if live else "offline",
                 "viewers_count": (item.get("metrics") or {}).get("viewers_count"),
                 "collected_at": item["collected_at"]})

state_path.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")
with open("live-status.csv", "w", encoding="utf-8-sig", newline="") as f:
    fields = ["username", "status", "viewers_count", "collected_at"]
    writer = csv.DictWriter(f, fieldnames=fields)
    writer.writeheader()
    writer.writerows({k: csv_cell(v) for k, v in row.items()} for row in rows)
print(json.dumps({"went_live_since_previous_check": went_live,
                  "without_new_usable_status": sorted(requested - observed)}))

python3 check_live.py

每次运行完成一个批次。程序按账号 ID 和采集时间比较状态：首次观察建立基线，同一账号由离线变为开播时记录变化，较早或重复的观察保持原状态。输入账号没有新记录时，保留旧状态及其检查时间。

CSV 保存本轮取得的新状态，状态文件保留各账号最近一次接受的观察。界面同时展示状态与检查时间。任务失败时，程序在写入状态文件前停止。

在应用中增加状态变化提醒

示例将离线到开播的用户名写入 went_live_since_previous_check。应用读取这个列表后执行自己的通知流程。用账号 ID 与观察到的变化时间生成通知标识，供发送重试时去重。

两次检查确定的是状态变化发生的时间区间，collected_at 记录本次观察时间。刷新间隔按业务需要的更新时效与重复查询成本设置；检查之间发生并结束的直播，需要结合直播历史记录查看。

callback_url 接收这一个采集任务的完成通知。持续检查由应用安排新的 User Live 任务，开播提醒由应用保存和比较状态后触发。

直播状态与直播间详情怎么选？

从用户名出发、需要账号状态表时使用 User Live。已知直播间与用户信息、需要房间详情时，使用独立的 TikTok Live Room Info，并遵循该接口自己的输入要求。

本教程输出用户名、状态、检查时间及来源提供的观看人数。账号、视频和评论数据见 TikTok 采集教程，内容发现流程见 TikTok 视频与图片搜索。

把开播检查接入你的业务节奏

SocQ User Live 让一组用户名进入同一个查询流程，并以标准状态字段返回结果。配合本文的状态保存和变化识别代码，你可以安排定时检查、更新运营看板，再由应用发送开播提醒，刷新频率由你掌握。

打开 User Live API，检查你的账号名单。

常见问题

如何判断一个 TikTok 用户是否正在直播？

将用户名放入 User Live 的 usernames，等待任务成功后检查 is_live，同时用 collected_at 展示该状态的观察时间。

可以一次检查多个账号吗？

可以，在 usernames 中提交多个公开账号用户名。按返回的 author.id 或 author.username 匹配结果；输入账号没有记录时，将本次检查标记为“未取得状态”。

离线是否算错误，会不会计费？

离线记录是成功查询结果，按返回状态记录计费。最新计费说明见 User Live API 页面。

接口会一直监控，直到账号开播吗？

一个任务执行一次检查后结束。持续检查由应用定时创建新任务，并比较各次返回的状态。

为什么观看人数和直播标题为空？

离线记录的房间字段可以为空；来源未提供在线房间指标时也保留 null。开播状态以 is_live 和 status 为准。

完成回调会在创作者每次开播时通知我吗？

完成回调对应单次任务。开播提醒需要应用定时查询、保存前后状态，并在观察到离线到开播的变化后发送。

两次检查能确定准确开播时间吗？

两次检查给出状态变化的时间区间。collected_at 是观察时间；准确开播时刻需要对应的直播开始时间数据。