Michael Johnson特定のTikTok動画への反応が増えているかを知るには、その動画の公開指標を、以前に記録した値と比較します。1回の検索でわかるのは、それまでに蓄積された再生数や反応の数です。同じ検索で見つかった動画を繰り返し取得すれば、実行間の変化を確認できます。以下では、Pythonで検索し、指標をCSVに保存して差分を計算する、最小限の実用的な手順を紹介します。
このモニタリングで測定できること
このスクリプトは、あるテーマについて検索で見つかった動画の一覧と、検証可能な観測履歴が必要なマーケター、アナリスト、開発者に適しています。たとえば、チームが製品の発売や特定の視聴者層に関する動画を定期的に追跡し、前回の確認後に再生数、いいね、コメント、シェアが増えた動画を把握したい場合に使えます。
比較の基準となる単位は、aweme_idで識別される動画です。各レコードには取得時刻、ID、説明文、URL、投稿者名、および取得可能な各種カウンターを保存します。その後、2つのスナップショットをIDで照合します。これにより、ある動画の再生数の増加を、別の動画の累計再生数と混同せずに済みます。
結果が示すのは、特定の検索クエリを特定の実行時に送って返された動画だけです。検索結果は変動する可能性があり、統計情報の一部が欠けていることもあります。そのため、スナップショットと取得時刻を保存し、レポートはテーマに該当する投稿の完全なカタログではなく、すでに見つかった動画の観測結果として扱うとよいでしょう。
APIを使った取得の仕組み
TikTok Scraper APIのドキュメントには、GETルート/searchVideoが記載されています。検索にはkeyword文字列が必須です。追加パラメーターにはcount、region、sort_by、date_postedがあります。APIキーはlicense_keyとして渡します。リファレンスに記載されたベースURLはhttps://tiktok.evelode.comです。
レスポンスには外側のステータスと、tiktokオブジェクトが含まれます。検索結果の動画は、tiktok.aweme_list、またはtiktok.search_item_list[].aweme_info内に格納される場合があります。以下の処理コードは、どちらの形式にも対応しています。指標はstatisticsから取得し、play_count、digg_count、comment_count、share_countを読み取ります。値がない場合はゼロに置き換えず、空欄として保存します。
ルートのリファレンスには、検索結果をページ送りするためのcursorと、選択可能な値とともにdate_postedおよびsort_byが記載されています。シンプルで再現可能なモニタリングであれば、まず結果の1ページを取得してレスポンスを記録し、必要に応じてページネーションを追加すれば十分です。ページネーションが必要な場合は、取得したカーソルを変換せずに渡してください。
Pythonでの実用例
python -m pip install requestsを実行してrequestsライブラリをインストールし、環境変数JSONSCRAPER_TIKTOK_LICENSE_KEYにキーを設定します。このスクリプトは検索語句を引数に取り、見つかったスナップショットを1つのCSVファイルに追記します。Python 3.8以降で動作します。
import csv
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
import requests
BASE_URL = "https://tiktok.evelode.com/searchVideo"
CSV_PATH = Path("tiktok_video_snapshots.csv")
FIELDS = [
"collected_at", "aweme_id", "author", "description", "share_url",
"play_count", "digg_count", "comment_count", "share_count",
]
def extract_items(payload):
"""Support the observed search response layouts."""
data = payload.get("tiktok") or {}
items = data.get("aweme_list") or []
if items:
return items
found = []
for result in data.get("search_item_list") or []:
info = result.get("aweme_info")
if info:
found.append(info)
return found
def main():
key = os.environ.get("JSONSCRAPER_TIKTOK_LICENSE_KEY")
if not key:
raise SystemExit("Set JSONSCRAPER_TIKTOK_LICENSE_KEY first")
if len(sys.argv) < 2:
raise SystemExit('Usage: python collect.py "search phrase"')
params = {
"license_key": key,
"keyword": sys.argv[1],
"count": 20,
"region": "US",
"sort_by": "relevance",
"date_posted": "this-week",
}
response = requests.get(BASE_URL, params=params, timeout=30)
response.raise_for_status()
payload = response.json()
if payload.get("status") != "ok":
raise RuntimeError("API returned a non-ok outer status")
data = payload.get("tiktok") or {}
items = extract_items(payload)
collected_at = datetime.now(timezone.utc).isoformat()
rows = []
for item in items:
stats = item.get("statistics") or {}
author = item.get("author") or {}
rows.append({
"collected_at": collected_at,
"aweme_id": item.get("aweme_id"),
"author": author.get("unique_id") or author.get("nickname"),
"description": item.get("desc"),
"share_url": item.get("share_url"),
"play_count": stats.get("play_count"),
"digg_count": stats.get("digg_count"),
"comment_count": stats.get("comment_count"),
"share_count": stats.get("share_count"),
})
write_header = not CSV_PATH.exists()
with CSV_PATH.open("a", newline="", encoding="utf-8") as file:
writer = csv.DictWriter(file, fieldnames=FIELDS)
if write_header:
writer.writeheader()
writer.writerows(rows)
print("Saved", len(rows), "returned videos")
print("has_more:", data.get("has_more"), "cursor:", data.get("cursor"))
if __name__ == "__main__":
main()たとえば、python collect.py "製品に関する検索語句"を実行します。キーはライブラリによってURLパラメーターに設定され、スクリプトからターミナルに出力されることはありません。リクエストURLにはキーが含まれるため、URL全体を公開ログに保存しないでください。本番環境ではシークレット管理ツールを使用し、スナップショットファイルへのアクセスを制限してください。
実行間の増加分を計算する
このスクリプトは前回の値を上書きせず、すべての観測結果を記録します。次の短い分析コードは、各動画について最新の2つのスナップショットを照合し、カウンターの差を出力します。2つのレコードのいずれかで指標の値が欠けている場合、結果は取得不可として扱います。
import csv
from collections import defaultdict
from pathlib import Path
rows_by_video = defaultdict(list)
with Path("tiktok_video_snapshots.csv").open(encoding="utf-8", newline="") as file:
for row in csv.DictReader(file):
if row["aweme_id"]:
rows_by_video[row["aweme_id"]].append(row)
metrics = ("play_count", "digg_count", "comment_count", "share_count")
for video_id, snapshots in rows_by_video.items():
snapshots.sort(key=lambda row: row["collected_at"])
if len(snapshots) < 2:
continue
previous, latest = snapshots[-2], snapshots[-1]
changes = {}
for metric in metrics:
old, new = previous[metric], latest[metric]
changes[metric] = int(new) - int(old) if old and new else None
print(video_id, changes, latest["share_url"])差分は、保存した2つの値の引き算であり、一定の周期が保証された増加速度ではありません。実行間隔が異なる場合は、collected_atの差から経過時間を計算し、実際の期間に応じた増加分を比較してください。観測の間にカウンターが変化することがあります。保存したタイムスタンプを見れば、使用した値をいつ取得したか確認できます。
レポートを自動化して読み解く
日次レポートを作成するには、サーバーやCIのタスクスケジューラーに取得処理を登録し、検索語句を固定して、CSVに保存するか、行データをデータベースに読み込みます。検索結果に重複がある場合は、同じスナップショット内でaweme_idを使って重複を除去してください。履歴を管理する際は、正規化した列とともに元のJSONも保存すると、レポートのスキーマが変わったときに再処理しやすくなります。
表には動画のIDとURL、投稿者、説明文、取得時刻、指標ごとの差分を表示すると便利です。再生数の増加順に並べると拡散状況がわかり、コメント数やシェア数の増加を見ると、別の種類の反応を把握できます。各カウンターの意味は異なるため、明確なルールなしに単一のランキングへ集約しないでください。
スケジューラーからメール、表計算シート、社内チャンネルにレポートを送信することもできますが、その処理は利用者側のインフラで行います。上記の例が担うのはデータの取得と分析だけです。より多くの結果を取得するには、結果ページを順にたどり、カーソルを保存し、APIが次のページがないと通知した時点でループを終了します。検索カーソルはレスポンスに含まれる不透明な値なので、自分で計算しないでください。ページをすべてたどった場合でも、得られるのは返された検索結果であり、該当する投稿の完全な一覧が保証されるわけではありません。
実用上の成果は、再現可能な観測記録と、見つかった動画の変化をまとめた表です。確認の間に公開カウンターが変化した投稿を見つけ、内容を詳しく分析する担当者に個別のリンクを共有できます。製品の詳細とAPIへのアクセスについては、TikTok Scraper APIのページをご覧ください。
定期実行の前に確認すること
tiktokを処理する前に、外側のレスポンスステータスがokであることを確認してください。- 検索項目の2種類の構造に対応しているか確認してください。リストがない場合はエラーではなく、取得件数ゼロとして処理します。
- 元の値とタイムスタンプを保存してください。指標が空の場合は、空のままにします。
- スナップショットをわかりやすく比較できるよう、同一の検索語句とパラメーターを使用してください。設定を変更した場合は、レポートと併せて記録します。
- 実行頻度、HTTPエラー、履歴のサイズを監視し、適用されるプラットフォームのルールとデータ処理要件を遵守してください。
つまり、観測結果を得るには、同じ検索を繰り返し、aweme_idで動画を識別し、保存した指標を時系列で比較すれば十分です。この方法で答えられるのは、「すでに見つかった動画のうち、2回の確認の間に公開カウンターが変化したものはどれか」という限定的で実務的な問いです。元データも、監査や追加処理のために引き続き利用できます。
