巨量星图 API 教程：达人搜索、受众画像与传播表现

/api/douyin-xingtu/gw/api/gsearch/search_for_author_square/v1

python3 -m pip install requests
export JUSTONEAPI_TOKEN='your_token'

import json
import os
from datetime import datetime, timezone
from pathlib import Path

import requests

BASE_URL = "https://api.justoneapi.com"
TOKEN = os.environ.get("JUSTONEAPI_TOKEN", "")


def call_api(path, **params):
    try:
        response = requests.get(
            BASE_URL + path,
            params={"token": TOKEN, **params},
            timeout=120, allow_redirects=False,
        )
    except requests.RequestException:
        raise RuntimeError("API network error or timeout") from None
    if response.status_code != 200:
        raise RuntimeError(f"API HTTP {response.status_code}")
    try:
        result = response.json()
    except ValueError:
        raise RuntimeError("API response is not JSON") from None
    if not isinstance(result, dict) or result.get("code") != 0:
        code = result.get("code") if isinstance(result, dict) else None
        raise RuntimeError(f"API failed: code={code}; check the docs")
    data = result.get("data")
    if not isinstance(data, dict):
        raise RuntimeError("Unexpected data shape; inspect the docs")
    inner = data.get("base_resp")
    if isinstance(inner, dict) and inner.get("status_code") not in (0, "0"):
        raise RuntimeError("Platform response was unsuccessful")
    return result


def save_json(folder, name, value):
    (folder / name).write_text(
        json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8"
    )


def require_list(data, key):
    value = data.get(key)
    if not isinstance(value, list):
        raise RuntimeError(f"Expected a list at {key}; inspect saved JSON")
    return value


def output_folder(topic):
    if not TOKEN or TOKEN == "your_token":
        raise RuntimeError("Set JUSTONEAPI_TOKEN before running")
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    folder = Path(topic + "-output") / run_id
    folder.mkdir(parents=True, exist_ok=False)
    return folder


MAX_PAGES = 2
MAX_CREATORS = 3
PREFIX = "/api/douyin-xingtu/gw/api"
PLATFORM = "SHORT_VIDEO"


def main():
    folder = output_folder("xingtu")
    ids, seen = [], set()
    for page in range(1, MAX_PAGES + 1):
        result = call_api(
            PREFIX + "/gsearch/search_for_author_square/v1",
            keyword="露营", searchType="CONTENT", page=page, sort="FANS",
        )
        save_json(folder, f"search-{page}.json", result)
        data = result["data"]
        rows = require_list(data, "authors")
        if not rows:
            break
        previous_count = len(ids)
        for row in rows:
            uid = row.get("star_id") if isinstance(row, dict) else None
            if uid and str(uid) not in seen:
                seen.add(str(uid))
                ids.append(str(uid))
        if len(ids) == previous_count:
            break
        pagination = data.get("pagination")
        if not isinstance(pagination, dict):
            raise RuntimeError("Missing pagination; inspect saved JSON")
        if pagination.get("has_more") not in (True, 1, "1"):
            break

    selected = ids[:MAX_CREATORS]
    save_json(folder, "selected-creators.json", selected)
    scope = {
        "platform": PLATFORM, "range": "DAY_30",
        "type": "PERSONAL_VIDEO", "onlyAssign": "false",
        "flowType": "EXCLUDE",
    }
    save_json(folder, "analysis-scope.json", {**scope, "linkType": "CONNECTED"})
    for index, uid in enumerate(selected, start=1):
        profile = call_api(
            PREFIX + "/author/get_author_base_info/v1",
            oAuthorId=uid, platform=PLATFORM,
        )
        save_json(folder, f"creator-{index}-profile.json", profile)
        audience = call_api(
            PREFIX + "/data_sp/author_audience_distribution/v1",
            oAuthorId=uid, platform=PLATFORM, linkType="CONNECTED",
        )
        save_json(folder, f"creator-{index}-audience.json", audience)
        spread = call_api(
            PREFIX + "/data_sp/get_author_spread_info/v1",
            oAuthorId=uid, **scope,
        )
        save_json(folder, f"creator-{index}-spread.json", spread)
    print(f"Saved {len(selected)} creator samples to {folder}")


if __name__ == "__main__":
    try:
        main()
    except RuntimeError as error:
        raise SystemExit(str(error)) from None