YouTubeデータ取得 データ取得

YouTube APIで取得した動画一覧を分析用データセットに整える方法|動画種別・カテゴリ付与まで

2026年6月29日

YouTubeチャンネルの動画一覧を分析したいとき、まずはYouTube Data APIで動画ID、タイトル、投稿日、再生数などを取得します。

自分のデータに合う実行経路を選ぶ

以下はWindows PowerShellでの手順です。Python 3.10以上を用意し、必要なファイルを保存したフォルダーをエクスプローラーで開き、アドレス欄に powershell と入力してEnterを押します。Get-Location で作業フォルダーを確認してください。コード全文は後ろに掲載しています。選んだ経路だけを実行し、別の経路のコードを同じファイルへつなげないでください。

本記事は「動画ID→詳細と補助分類のCSV」を作る段階です。保存した動画IDを使っても詳細・公式カテゴリ名のAPI通信とキーが必要です。完成済みの分析用CSVを使う経路だけは通信不要です。

一覧CSVがある/チャンネルから作る:どちらもAPI通信あり

youtube_dataset.py に次の定義コード10個を掲載順に保存します:①uploads playlist(import含む)②動画ID一覧③動画詳細(chunked含む)④動画時間⑤動画種別⑥独自カテゴリ⑦1行へ整形⑧公式カテゴリ名⑨CSV保存(import hashlibから)⑩「保存済み動画IDを再利用する」のif __name__実行部。pip・.env・列名例・aggregate_youtube_csv.pyは連結しません。.envは同じフォルダーへ置きます。

python -m pip install requests python-dotenv pandas isodate
python youtube_dataset.py

保存済み一覧を使う既定値は MODE="csv"。入力は outputs/youtube/channel_videos.csv です。一覧は再取得せず動画詳細をAPIで追加取得します。直接チャンネルから作る場合だけMODEをchannelにしてCHANNEL_IDを実値へ変更します。2経路は代替で、同じ実行部を2回追加しません。CSV経路は元一覧の範囲を引き継ぐため、1ページ分をチャンネル全件と呼びません。

分析用CSVがある:キーなしで集計する

独立集計コードのみを aggregate_youtube_csv.py として保存します。入力は outputs/youtube/channel_video_analysis_dataset.csv。取得用のyoutube_dataset.pyを実行する必要はありません。

python -m pip install pandas
python aggregate_youtube_csv.py

相対入出力は作業フォルダー基準です。取得結果は channel_video_analysis_dataset.csv と同名.metadata.json。未応答を調べるならdetail_statusとmetadataを確認します。独立集計ではinput/dated/outputが表示され、outputs/youtube/aggregates/monthly_publications.csv のmonth・video_countで月別本数を確認します。範囲は取得CSV内、月境界はUTCです。

グラフも必要なら 可視化記事の実用版へ進みます。実用版は集計関数を呼ぶため、上の集計を先に実行する必要はありません。保存済み集計CSVだけから図を再描画する独立コマンドは、このプログラムにはありません。APIキーなしの練習は 既設の架空6行を使います。

掲載コードと詳しい条件

ただ、取得したデータをそのまま集計しようとすると、少し困ることがあります。通常動画なのか、ライブ配信アーカイブなのか、Shortsなのか。あるいは、ゲーム、雑談、音楽、解説など、どのような内容の動画なのか。こうした分析で見たい分類は、APIのレスポンスに完成した形で入っているわけではありません。

この記事では、YouTube Data APIで取得した動画一覧に、分析用の補助ラベルを付けてCSV化する流れを整理します。VTuberや配信者チャンネルの活動傾向を見る場合にも使えますが、内容としてはYouTubeチャンネル全般に使える形を目指します。

この記事でやること

  • YouTube APIでチャンネルの動画一覧を取得する
  • videos.listで再生数、動画時間、公式カテゴリなどを追加取得する
  • 通常動画、ライブ配信アーカイブ、Shortsなどの動画種別を付与する
  • 公式カテゴリとは別に、分析用の独自カテゴリを付与する
  • 後続の集計や可視化で使いやすいCSVとして保存する

この記事では、動画種別や独自カテゴリを「正解データ」として扱うのではなく、分析しやすくするための補助ラベルとして扱います。分類の精度を上げたい場合は、最終的に目視確認や手動修正を組み合わせる前提です。

前提記事

この記事は、YouTube Data APIの基本的な使い方をある程度理解している前提で進めます。APIキーの発行やチャンネル動画一覧の取得から確認したい場合は、先に以下の記事を見ると流れを追いやすいです。

YouTube APIで取得できる情報

YouTube Data APIのvideos.listでは、動画IDを指定して、タイトル、説明文、タグ、公式カテゴリID、動画時間、再生数、コメント数、ライブ配信関連の情報などを取得できます。

項目API上の主な場所分析での使い方
タイトルsnippet.titleキーワード分類、動画内容の確認
説明文snippet.descriptionハッシュタグ、企画名、リンクの確認
タグsnippet.tags内容分類の補助
公式カテゴリIDsnippet.categoryId大まかな公式分類
ライブ状態snippet.liveBroadcastContentライブ中・予定配信の確認
動画時間contentDetails.durationShorts候補、長時間配信の判定
再生数statistics.viewCount人気動画や再生傾向の確認
コメント数statistics.commentCount反応量の参考
ライブ配信詳細liveStreamingDetailsライブ配信アーカイブ候補の判定

一方で、分析でよく使いたくなる「通常動画」「ライブ配信アーカイブ」「Shorts」といった動画種別は、そのままの列として用意されているわけではありません。そのため、複数の項目を組み合わせて自分で補助ラベルを作ります。

動画種別は公式の正解ラベルではない

動画種別を付けるときに注意したいのは、API上に「これはShorts」「これは配信アーカイブ」といった分析用の完全な正解ラベルがあるわけではないことです。

ライブ配信についてはliveBroadcastContentliveStreamingDetailsが参考になります。ただし、公開済みの配信アーカイブでは状態がnoneになっていることもあるため、ライブ配信詳細の有無や動画時間なども見ながら判断します。

Shortsについても、#shortsがタイトルや説明文に入っている場合は判定しやすいですが、すべてのShortsに必ず付いているとは限りません。また、180秒以内の通常動画もあるため、動画時間だけで断定するのも危険です。

この記事では、以下のような考え方で動画種別を付けます。

動画種別判定に使う情報注意点
live_archiveliveStreamingDetails、長い動画時間、ライブ関連情報公開済みアーカイブではliveBroadcastContentだけでは判定しきれない
shorts_candidate#shorts、短い動画時間短い通常動画と混ざる可能性がある
other_video上記に該当しない公開済み動画プレミア公開や編集動画も含まれることがある
upcoming_or_liveliveBroadcastContentliveまたはupcoming分析時点によって状態が変わる
unknown情報が不足しているもの手動確認の候補にする

公式カテゴリは粒度が粗い

YouTube APIにはcategoryIdがあります。これはYouTube側の公式カテゴリで、videoCategories.listを使うとカテゴリ名も取得できます。

公式カテゴリは、動画内容をかなり大きな単位で分けるための分類です。たとえば、以下のようなカテゴリがあります。

公式カテゴリ例含まれやすい動画の例分析時に困りやすい点
Gamingゲーム実況、ゲーム配信アーカイブ、ゲーム紹介、攻略動画ゲーム名、配信形式、雑談多めのゲーム配信などは分からない
Music歌ってみた、MV、歌枠、演奏動画オリジナル曲、カバー、歌枠アーカイブなどの違いは分からない
Entertainment企画動画、バラエティ寄りの動画、配信者系コンテンツ企画、雑談、コラボ、切り抜きなどの細かい違いは分からない
People & Blogs日常系、雑談、個人発信、Vlog系の動画雑談、近況報告、作業配信などの目的までは分からない
Education解説、講座、学習系コンテンツプログラミング、語学、資格、考察などのテーマは別途見たい
Science & Technology技術解説、ガジェット、サイエンス系動画AI、Python、PC、アプリ紹介などの細分類はできない

ただし、チャンネル分析で見たい分類とは粒度が合わないことがあります。たとえば配信者やVTuberの活動傾向を見たい場合、「ゲーム」「雑談」「歌」「ASMR」「企画」「コラボ」のような分類が欲しくなります。しかし公式カテゴリだけでは、そこまで細かく分けられません。

たとえば同じGamingカテゴリでも、ゲーム実況の編集動画、長時間のライブ配信アーカイブ、参加型配信、ゲーム内イベントの視点配信では、分析で見たい意味がかなり違います。公式カテゴリは残しつつ、分析目的に合わせた独自カテゴリを追加する理由はここにあります。

そのため、この記事では公式カテゴリは公式データとして残しつつ、タイトル・説明文・タグを使って分析用の独自カテゴリも付与します。

完成イメージ

最終的には、以下のような列を持つCSVを作ります。

video_id
url
channel_id
channel_title
title
description
published_at
published_date
published_month
duration
duration_sec
view_count
like_count
comment_count
official_category_id
official_category_name
live_broadcast_content
has_live_streaming_details
video_type
content_category
days_since_published
views_per_day

official_category_iddurationはAPI由来の情報です。一方で、video_typecontent_categoryは、分析しやすくするためにこちらで加工して追加する列です。

事前準備

必要なライブラリをインストールします。

APIキーは.envに保存しておきます。

YOUTUBE_API_KEY=ここにAPIキーを書く

APIキーをコードに直接書かない理由や、.envの使い方は以下の記事で整理しています。

APIキーを外部ファイルで安全に管理する方法|.env・config.iniの使い分け

チャンネルのuploads playlistを取得する

チャンネルに投稿された動画一覧を取得する場合、まずchannels.listuploads playlist IDを取得します。

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

import isodate
import pandas as pd
import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("YOUTUBE_API_KEY")

if not API_KEY:
    raise RuntimeError("YOUTUBE_API_KEY が設定されていません")


def request_youtube_api(url: str, params: dict) -> dict:
    response = requests.get(url, params=params, timeout=30)
    response.raise_for_status()
    return response.json()


def get_uploads_playlist_id(channel_id: str) -> str:
    url = "https://www.googleapis.com/youtube/v3/channels"
    params = {
        "part": "contentDetails",
        "id": channel_id,
        "key": API_KEY,
    }
    data = request_youtube_api(url, params)

    if not data.get("items"):
        raise RuntimeError("チャンネルが見つかりませんでした")

    return data["items"][0]["contentDetails"]["relatedPlaylists"]["uploads"]

動画ID一覧を取得する

playlistItems.listで、uploads playlistに含まれる動画IDを取得します。1回のリクエストで取得できる件数には上限があるため、nextPageTokenを使ってページングします。

def get_upload_video_ids(playlist_id: str, max_pages: int | None = None) -> list[str]:
    url = "https://www.googleapis.com/youtube/v3/playlistItems"
    video_ids = []
    page_token = None
    page_count = 0

    while True:
        params = {
            "part": "contentDetails",
            "playlistId": playlist_id,
            "maxResults": 50,
            "key": API_KEY,
        }

        if page_token:
            params["pageToken"] = page_token

        data = request_youtube_api(url, params)

        for item in data.get("items", []):
            video_id = item.get("contentDetails", {}).get("videoId")
            if video_id:
                video_ids.append(video_id)

        page_count += 1
        if max_pages is not None and page_count >= max_pages:
            break

        page_token = data.get("nextPageToken")
        if not page_token:
            break

    return video_ids

最初から全件取得すると確認に時間がかかることがあります。動作確認ではmax_pages=1にして、問題なければNoneに変えると扱いやすいです。

動画詳細を取得する

動画ID一覧を取得したら、videos.listで動画詳細を取得します。ここではsnippetcontentDetailsstatisticsliveStreamingDetailsを指定します。

def chunked(values: list[str], size: int) -> list[list[str]]:
    return [values[i:i + size] for i in range(0, len(values), size)]


def get_video_details(video_ids: list[str]) -> list[dict]:
    url = "https://www.googleapis.com/youtube/v3/videos"
    videos = []

    for chunk in chunked(video_ids, 50):
        params = {
            "part": "snippet,contentDetails,statistics,liveStreamingDetails",
            "id": ",".join(chunk),
            "key": API_KEY,
        }
        data = request_youtube_api(url, params)
        videos.extend(data.get("items", []))

    return videos

動画時間を秒に変換する

YouTube APIの動画時間は、PT1H23M45SのようなISO 8601 duration形式で返ってきます。集計しやすいように秒数へ変換します。

def parse_duration_seconds(duration: str | None) -> int | None:
    if not duration:
        return None

    try:
        parsed = isodate.parse_duration(duration)
        return int(parsed.total_seconds())
    except Exception:
        return None

動画種別を付与する

ここでは、ライブ関連情報、動画時間、#shortsの有無を使って、分析用のvideo_typeを付与します。

以下の区分は分析用の推定で、YouTube公式の確定区分ではありません。ハッシュタグあり、または0秒より長く180秒以内の動画を shorts_candidate とします。公式条件には正方形・縦長という縦横比とアップロード日も含まれ、標準チャンネルは2024年10月15日、公式アーティストチャンネルは2025年12月8日からの条件があります。この例では縦横比・チャンネル種別・正確なアップロード時点を検証できないため、候補をShortsと断定しません。公開日時だけでアップロード日を代用しません。既存の分析データは変更しません。

def contains_shorts_marker(title: str, description: str, tags: list[str]) -> bool:
    text = " ".join([title, description, " ".join(tags)]).lower()
    return "#shorts" in text or "#short" in text


def classify_video_type(
    live_broadcast_content: str | None,
    live_streaming_details: dict,
    duration_sec: int | None,
    title: str,
    description: str,
    tags: list[str],
) -> str:
    if live_broadcast_content in {"live", "upcoming"}:
        return "upcoming_or_live"

    if live_streaming_details:
        return "live_archive"

    if contains_shorts_marker(title, description, tags):
        return "shorts_candidate"

    if duration_sec is not None and 0 < duration_sec <= 180:
        return "shorts_candidate"

    if duration_sec is None:
        return "unknown"

    return "other_video"

shorts_candidateは、ハッシュタグまたは動画時間からShorts候補としたものです。実際にShortsかどうかを厳密に確認したい場合は、URLや画面表示、手動確認を組み合わせるのが安全です。

分析用カテゴリを付与する

次に、タイトル・説明文・タグから、分析用のcontent_categoryを付与します。ここではシンプルなキーワードルールで分類します。

CATEGORY_KEYWORDS = {
    "game": [
        "ゲーム", "実況", "gta", "minecraft", "マイクラ", "apex", "valorant",
        "原神", "崩壊", "モンハン", "steam",
    ],
    "talk": [
        "雑談", "朝活", "おはよう", "作業", "近況", "振り返り",
    ],
    "music": [
        "歌", "歌枠", "歌ってみた", "cover", "music", "karaoke",
    ],
    "asmr": [
        "asmr", "睡眠", "耳かき", "囁き",
    ],
    "collaboration": [
        "コラボ", "collab", "with",
    ],
    "announcement": [
        "告知", "お知らせ", "発表", "重大発表",
    ],
}


def classify_content_category(title: str, description: str, tags: list[str]) -> str:
    text = " ".join([title, description, " ".join(tags)]).lower()

    for category, keywords in CATEGORY_KEYWORDS.items():
        for keyword in keywords:
            if keyword.lower() in text:
                return category

    return "other"

この分類は、あくまで最初の整理用です。たとえばゲーム配信の中にも雑談が多い枠はありますし、歌枠の告知動画など複数の意味を持つ動画もあります。厳密に分類したい場合は、複数ラベルにする、手動確認列を追加する、チャンネルごとにキーワードを調整する、といった対応が必要です。

動画データを1行に整える

APIレスポンスを、CSVに保存しやすい1行の辞書へ変換します。

def safe_int(value) -> int | None:
    if value is None:
        return None
    try:
        return int(value)
    except ValueError:
        return None


def normalize_video_item(item: dict, category_map: dict[str, str], reference_at: str | None) -> dict:
    snippet = item.get("snippet", {})
    content_details = item.get("contentDetails", {})
    statistics = item.get("statistics", {})
    live_details = item.get("liveStreamingDetails", {})

    title = snippet.get("title", "")
    description = snippet.get("description", "")
    tags = snippet.get("tags", [])
    duration = content_details.get("duration")
    duration_sec = parse_duration_seconds(duration)
    live_broadcast_content = snippet.get("liveBroadcastContent")
    category_id = snippet.get("categoryId")
    published_at = snippet.get("publishedAt", "")

    video_type = classify_video_type(
        live_broadcast_content=live_broadcast_content,
        live_streaming_details=live_details,
        duration_sec=duration_sec,
        title=title,
        description=description,
        tags=tags,
    )

    content_category = classify_content_category(title, description, tags)

    published_dt = pd.to_datetime(published_at, errors="coerce", utc=True)
    reference = pd.to_datetime(reference_at, errors="coerce", utc=True)
    days_since_published = None
    if pd.notna(published_dt) and pd.notna(reference):
        age = (reference - published_dt).total_seconds() / 86400
        if age >= 1:
            days_since_published = age
    view_count = safe_int(statistics.get("viewCount"))
    views_per_day = None
    if view_count is not None and view_count >= 0 and days_since_published is not None:
        views_per_day = round(view_count / days_since_published, 2)

    return {
        "video_id": item.get("id"),
        "url": f"https://www.youtube.com/watch?v={item.get('id')}",
        "channel_id": snippet.get("channelId"),
        "channel_title": snippet.get("channelTitle"),
        "title": title,
        "description": description,
        "published_at": published_at,
        "published_date": published_dt.date().isoformat() if pd.notna(published_dt) else "",
        "published_month": published_dt.strftime("%Y-%m") if pd.notna(published_dt) else "",
        "duration": duration,
        "duration_sec": duration_sec,
        "view_count": view_count,
        "like_count": safe_int(statistics.get("likeCount")),
        "comment_count": safe_int(statistics.get("commentCount")),
        "official_category_id": category_id,
        "official_category_name": category_map.get(category_id, ""),
        "live_broadcast_content": live_broadcast_content,
        "has_live_streaming_details": bool(live_details),
        "video_type": video_type,
        "content_category": content_category,
        "days_since_published": days_since_published,
        "views_per_day": views_per_day,
        "tags": "|".join(tags),
    }

公式カテゴリ名を取得する

categoryIdだけだと読みにくいため、videoCategories.listでカテゴリ名も取得しておきます。

def get_video_category_map(region_code: str = "JP") -> dict[str, str]:
    url = "https://www.googleapis.com/youtube/v3/videoCategories"
    params = {
        "part": "snippet",
        "regionCode": region_code,
        "key": API_KEY,
    }
    data = request_youtube_api(url, params)

    return {
        item["id"]: item["snippet"]["title"]
        for item in data.get("items", [])
    }

公式カテゴリは、YouTube側が持っている大分類として残します。独自カテゴリとは目的が違うため、どちらか一方に寄せるより、両方の列を残しておく方が後で見返しやすいです。

CSVに保存する

ここまでの処理をまとめて、分析用CSVを作ります。

分割コードは上から同じPython実行内で定義し、最後に build_analysis_dataset を呼びます。詳細0件でも固定列を保存するため、後続の列選択と件数確認ができます。同名CSV・metadataがある場合は停止します。改訂前のファイルを残し、出力先を分けてください。全件取得は別の追記実行ではなく、最初の呼び出しの max_pages=1None に変更する選択肢です。

import hashlib
import json
from collections import Counter

DATASET_COLUMNS = ['video_id', 'url', 'channel_id', 'channel_title', 'title', 'description', 'published_at', 'published_date', 'published_month', 'duration', 'duration_sec', 'view_count', 'like_count', 'comment_count', 'official_category_id', 'official_category_name', 'live_broadcast_content', 'has_live_streaming_details', 'video_type', 'content_category', 'days_since_published', 'views_per_day', 'tags', 'detail_status']


def build_analysis_dataset(channel_id, output_csv, max_pages=1, input_csv=None):
    started = datetime.now(timezone.utc).isoformat()
    meta_path = output_csv.with_suffix(".metadata.json")
    if output_csv.exists() or meta_path.exists():
        raise FileExistsError("保存先が存在します。新しい保存先へ変更してください。")
    source_meta = {}
    source_hash = None
    if input_csv is not None:
        source = pd.read_csv(input_csv, dtype=str, keep_default_na=False)
        if "video_id" not in source.columns or source["video_id"].str.strip().eq("").any():
            raise ValueError("入力CSVのvideo_id列・空欄を確認してください。")
        video_ids = source["video_id"].str.strip().tolist()
        source_hash = hashlib.sha256(input_csv.read_bytes()).hexdigest()
        sidecar = input_csv.with_suffix(".metadata.json")
        if sidecar.exists():
            source_meta = json.loads(sidecar.read_text(encoding="utf-8"))
            if source_meta.get("csv_sha256") != source_hash:
                raise ValueError("入力CSVとmetadataのhashが一致しません。")
        channels = sorted(set(source.get("channel_id", pd.Series(dtype=str))) - {""})
        scope = source_meta.get("scope", "UNKNOWN: input CSV rows only")
        list_started = source_meta.get("list_started_at")
        list_finished = source_meta.get("list_finished_at")
    else:
        if not channel_id:
            raise ValueError("直接取得ではchannel_idが必要です。")
        list_started = datetime.now(timezone.utc).isoformat()
        playlist_id = get_uploads_playlist_id(channel_id)
        video_ids = get_upload_video_ids(playlist_id, max_pages=max_pages)
        list_finished = datetime.now(timezone.utc).isoformat()
        channels = [channel_id]
        scope = {"source": "uploads playlist", "max_pages": max_pages,
                 "coverage": "API pagination exhausted" if max_pages is None else "at most specified pages"}
    counts = Counter(video_ids)
    ids = list(counts)
    details_started = datetime.now(timezone.utc).isoformat()
    videos = get_video_details(ids) if ids else []
    details_finished = datetime.now(timezone.utc).isoformat()
    by_id = {}
    for item in videos:
        key = item.get("id")
        if key not in counts or key in by_id:
            raise ValueError("要求と応答の動画IDが不整合です。保存を中止します。")
        by_id[key] = item
    category_map = get_video_category_map("JP") if videos else {}
    reference_at = details_finished  # One explicit reference for this acquisition batch.
    rows = []
    for key in ids:
        if key in by_id:
            row = normalize_video_item(by_id[key], category_map, reference_at)
            row["detail_status"] = "returned"
        else:
            row = {"video_id": key, "video_type": "unknown", "content_category": "unknown", "detail_status": "not_returned"}
        rows.append(row)
    df = pd.DataFrame(rows, columns=DATASET_COLUMNS)
    output_csv.parent.mkdir(parents=True, exist_ok=True)
    df.to_csv(output_csv, index=False, encoding="utf-8-sig")
    metadata = {"csv_file": output_csv.name, "csv_sha256": hashlib.sha256(output_csv.read_bytes()).hexdigest(),
        "fetch_started_at": started, "fetch_finished_at": datetime.now(timezone.utc).isoformat(),
        "list_started_at": list_started, "list_finished_at": list_finished,
        "details_started_at": details_started, "details_finished_at": details_finished,
        "reference_at": reference_at, "channel_ids": channels,
        "input_file": str(input_csv) if input_csv else None, "input_sha256": source_hash,
        "source_metadata": source_meta, "scope": scope, "input_rows": len(video_ids), "saved_rows": len(df),
        "duplicate_ids": {k: v for k, v in counts.items() if v > 1},
        "not_returned_ids": [key for key in ids if key not in by_id], "month_timezone": "UTC",
        "note": "IDを一意化し、詳細未応答の行もunknownとして保持。未応答理由は未確認。"}
    meta_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"saved={output_csv.resolve()} / input={len(video_ids)} / unique={len(df)} / returned={len(by_id)}")
    if not ids:
        print("0件のためAPI詳細取得を省略し、列名だけを保存しました。入力範囲を確認してください。")
    return df

保存済み動画IDを再利用する:2つの代替経路

ここまでのPythonの定義ブロックを掲載順に1つの youtube_dataset.py へ保存します(pipと.envの設定例はPythonファイルに入れません)。最後に以下の実行部を1つだけ追加します。aggregate_youtube_csv.py は別ファイルです。

すでに outputs/youtube/channel_videos.csv がある場合は既定のCSV経路を使います。動画一覧は取り直さず、重複IDを一意化して動画詳細・公式カテゴリ名だけを取得します。入力一覧の行数、重複ID、詳細未応答IDはmetadataに残り、未応答行は不明として保持します。直接取得したい場合だけMODEをchannelへ変更してください。両方を続けて実行する必要はありません。

if __name__ == "__main__":
    MODE = "csv"  # Alternative: "channel"
    CHANNEL_ID = "UCxxxxxxxxxxxxxxxxxxxxxx"
    try:
        df = build_analysis_dataset(
            channel_id=CHANNEL_ID if MODE == "channel" else None,
            input_csv=Path("outputs/youtube/channel_videos.csv") if MODE == "csv" else None,
            output_csv=Path("outputs/youtube/channel_video_analysis_dataset.csv"),
            max_pages=1,
        )
        print(df[["video_id", "title", "video_type", "view_count", "detail_status"]].head())
    except (requests.RequestException, ValueError, OSError):
        raise SystemExit("保存未完了。入力列・出力先・APIキー・通信を確認してください。APIキーを含むエラーURLは共有しないでください。") from None

直接取得で全ページをたどる場合は、この呼び出しの max_pages=1None へ変更します。CSV経路ではmax_pagesは使わず、元CSVに含まれるIDだけが対象です。1ページ分のCSVをチャンネル全件と呼ばないでください。

CSVとmetadataで取得範囲・時点を残す

CSVの横に channel_video_analysis_dataset.metadata.json を保存します。CSVのSHA-256、保存件数、入力ファイルとhash、チャンネルID、取得開始・終了、一覧と詳細それぞれの時点、取得範囲を確認できます。APIキーは保存しません。元CSVのmetadataがない場合、一覧の取得時点・範囲は不明のままです。ファイルの更新日時や現在日時で過去の取得日時を補いません。

公開日時は動画が公開された時刻、取得日時はAPI応答を保存した時刻、集計実行日時はCSVを計算した時刻です。詳細取得終了をそのバッチ共通の基準時刻とし、取得中の時間差も開始・終了で記録します。views_per_dayは「取得時点の累積再生数を公開後の経過日数で割った参考値」です。実際の日別再生数・直近の伸び率・人気の公平な比較指標ではありません。未来の公開日時、日付欠損、公開後24時間未満では計算しません。過去CSVの取得時点が不明なら、その値を現在時刻で再計算せず、時点が必要な指標を省略します。

2026年9月8日に確認したYouTube公式仕様のviewCountでは、2026年8月24日以降、長尺・ライブ・Shortsを含む全形式で再生開始時にカウントする定義(自動再生、ポインターを重ねた再生、クリック/タップを含む)が示されています。定義変更前後の累積値を同じ条件の伸び率として比較しないでください。このコードは日別視聴履歴や視聴時間を取得しません。

保存CSVを別のPython実行で集計する

取得直後なら print(df["video_type"].value_counts(dropna=False)) で確認できます。保存後は次を aggregate_youtube_csv.py として保存し、新しいPython実行で読み直します。必要なのはpandasだけで、APIキー・API通信・取得時の変数は不要です。入力は outputs/youtube/channel_video_analysis_dataset.csv、必要列はvideo_id、published_at、video_type、content_categoryです。動画IDは文字列として保持します。

"""Offline aggregation: no API imports, credentials or network calls."""
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path

import pandas as pd

REQUIRED = {"video_id", "published_at", "video_type", "content_category"}


def load_dataset(path):
    df = pd.read_csv(path, dtype=str, keep_default_na=False)
    missing = REQUIRED - set(df.columns)
    if missing:
        raise ValueError(f"必要列がありません: {sorted(missing)}。データセット記事の出力を選んでください。")
    if df["video_id"].str.strip().eq("").any() or df["video_id"].duplicated().any():
        raise ValueError("動画IDの空欄・重複があります。取得時の記録で確認してください。")
    meta_path = path.with_suffix(".metadata.json")
    meta = {"scope": "UNKNOWN", "reference_at": None}
    if meta_path.exists():
        meta = json.loads(meta_path.read_text(encoding="utf-8"))
        if meta.get("csv_sha256") != hashlib.sha256(path.read_bytes()).hexdigest():
            raise ValueError("CSVとmetadataが別の取得回です。")
        if meta.get("saved_rows") != len(df):
            raise ValueError("metadataの保存件数とCSVが一致しません。")
    for col in ["video_type", "content_category"]:
        df[col] = df[col].str.strip().replace("", "unknown")
    df["published_utc"] = pd.to_datetime(df["published_at"], utc=True, errors="coerce", format="mixed")
    return df, meta


def aggregate(source, output_dir):
    df, meta = load_dataset(source)
    output_dir.mkdir(parents=True, exist_ok=False)
    dated = df.loc[df["published_utc"].notna()].copy()
    dated["month"] = dated["published_utc"].dt.strftime("%Y-%m")
    monthly = dated.groupby("month").size().reset_index(name="video_count")
    types = df.groupby("video_type").size().reset_index(name="video_count")
    categories = df.groupby("content_category").size().reset_index(name="video_count")
    tables = {"monthly_publications": monthly, "video_types": types, "content_categories": categories}
    hashes = {}
    for name, table in tables.items():
        path = output_dir / f"{name}.csv"
        table.to_csv(path, index=False, encoding="utf-8-sig")
        hashes[path.name] = hashlib.sha256(path.read_bytes()).hexdigest()
    report = {"aggregation_at": datetime.now(timezone.utc).isoformat(), "month_timezone": "UTC",
        "input_file": str(source), "input_sha256": hashlib.sha256(source.read_bytes()).hexdigest(),
        "scope": meta.get("scope", "UNKNOWN"), "source_metadata": meta,
        "input_rows": len(df), "date_rows": len(dated), "missing_date_rows": len(df) - len(dated),
        "output_hashes": hashes, "note": "取得範囲外の月は0件として補完していません。"}
    (output_dir / "aggregation.metadata.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"input={len(df)} / dated={len(dated)} / output={output_dir.resolve()}")
    if df.empty:
        print("0件です。列名のみを保存しました。元CSVと取得範囲を確認してください。")
    return df, tables, report


if __name__ == "__main__":
    try:
        aggregate(Path("outputs/youtube/channel_video_analysis_dataset.csv"), Path("outputs/youtube/aggregates"))
    except (ValueError, OSError) as exc:
        raise SystemExit(str(exc)) from None

出力先は outputs/youtube/aggregates/。monthly_publications.csv、video_types.csv、content_categories.csvと集計条件のmetadataができます。既存フォルダーは上書きしません。日付欠損を除くのは月別集計だけです。不明・未分類を種別やカテゴリの分母から除外しません。月境界はUTCで、JSTへ変換していません。集計範囲外の月を投稿0件として埋めず、0件入力では列名だけを保存して取得範囲の確認を案内します。

グラフHTMLまで作る手順はYouTube動画一覧CSVの集計・可視化へ進んでください。この集計ファイルを可視化コードと同じフォルダーに置いて再利用します。

このデータでできる分析

  • 月別の投稿本数を見る
  • その他の動画、配信アーカイブ候補、Shorts候補、不明の構成を見る
  • ゲーム、雑談、音楽などのカテゴリ別に投稿傾向を見る
  • 再生数の多い動画を動画種別ごとに確認する
  • 取得時点が分かる場合だけviews_per_dayを参考値として確認する
  • VTuberや配信者チャンネルの活動スタイルを整理する

たとえばVTuberチャンネルを見る場合、ライブ配信中心なのか、Shortsも活用しているのか、ゲーム配信が多いのか、歌や雑談の比率が高いのか、といった入口を作れます。数字だけで魅力を判断するのではなく、チャンネルを知るための地図を作るイメージです。

注意点

分類は完全ではない

動画種別や独自カテゴリは、APIから返ってくる公式の正解ラベルではありません。タイトルや説明文に依存するため、表記ゆれやチャンネルごとの文化に影響されます。

Shorts判定は特に揺れやすい

#shortsが付いていないShortsや、180秒以内の通常動画もあります。この記事の判定は、分析のための近似的な分類として使います。

公式カテゴリと独自カテゴリを混同しない

公式カテゴリはYouTube側の大きな分類です。独自カテゴリは、自分が分析したい粒度に合わせて後から付ける分類です。記事やレポートで使うときは、どちらの分類なのかを分けて説明した方が誤解が少なくなります。

公開データの範囲で扱う

この記事で扱うのは、YouTube APIから取得できる公開メタデータです。非公開動画、削除済み動画、取得できない統計情報などは分析対象に含められません。

APIクォータに注意する

YouTube Data APIにはクォータがあります。投稿動画一覧を作るだけならuploads playlistを使う方が扱いやすく、検索目的でない場合にsearch.listを多用する必要はあまりありません。取得したCSVを保存しておき、同じデータを何度も取り直さないようにすると安心です。

公式ドキュメント

まとめ

YouTube APIで取得した動画一覧は、そのままでも基本的な集計に使えます。ただ、チャンネルの活動傾向を見る場合は、通常動画、ライブ配信アーカイブ、Shortsといった動画種別や、ゲーム、雑談、音楽などの独自カテゴリを付けておくと見通しがよくなります。

一方で、これらの分類は公式の正解データではありません。APIで取得できる情報の制約を理解したうえで、分析用の補助ラベルとして使うのが大切です。

この形でCSVを整えておくと、投稿頻度、配信形式の比率、人気動画の傾向、カテゴリ別の活動スタイルなどを確認しやすくなります。保存済みCSVから月別公開本数や動画種別の件数を集計し、可視化記事でHTMLへ保存できます。

-YouTubeデータ取得, データ取得