抖音视频数据怎么采集？用 API 获取视频搜索结果、详情和评论

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", "")
KEYWORD = "露营装备"
MAX_SEARCH_PAGES = 2
MAX_VIDEOS = 3


def call_api(path, **params):
    try:
        response = requests.get(
            BASE_URL + path,
            params={"token": TOKEN, **params},
            timeout=120,
        )
    except requests.RequestException:
        raise RuntimeError("Network error or timeout") from None
    if response.status_code != 200:
        raise RuntimeError(f"HTTP {response.status_code}")
    try:
        result = response.json()
    except ValueError:
        raise RuntimeError("Response is not valid 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")
    if not isinstance(result.get("data"), dict):
        raise RuntimeError("Unexpected data shape; check the response")
    return result


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


def main():
    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("douyin-output") / run_id
    folder.mkdir(parents=True, exist_ok=False)
    video_ids = []
    seen = set()
    search_id = None

    for page in range(1, MAX_SEARCH_PAGES + 1):
        params = dict(
            keyword=KEYWORD, sortType="_0", publishTime="_0",
            duration="_0", page=page,
        )
        if page > 1:
            params["searchId"] = search_id
        result = call_api("/api/douyin/search-video/v4", **params)
        save_json(folder, f"search-{page}.json", result)
        data = result["data"]
        for card in data.get("business_data") or []:
            info = (card.get("data") or {}).get("aweme_info") or {}
            video_id = info.get("aweme_id")
            if video_id and str(video_id) not in seen:
                video_id = str(video_id)
                seen.add(video_id)
                video_ids.append(video_id)
        config = data.get("business_config") or {}
        if config.get("has_more") not in (1, "1"):
            break
        search_id = (config.get("next_page") or {}).get("search_id")
        if page < MAX_SEARCH_PAGES and not search_id:
            raise RuntimeError("Missing next search_id; inspect saved JSON")

    selected = video_ids[:MAX_VIDEOS]
    save_json(folder, "selected-video-ids.json", selected)
    for index, video_id in enumerate(selected, start=1):
        detail = call_api(
            "/api/douyin/get-video-detail/v2", videoId=video_id,
        )
        save_json(folder, f"video-{index}-detail.json", detail)
        comments = call_api(
            "/api/douyin/get-video-comment/v1", awemeId=video_id, page=1,
        )
        save_json(folder, f"video-{index}-comments-1.json", comments)

    print(f"Saved {len(selected)} video samples to {folder}")


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