特別講義DS 補足C YouTube Data APIによる動画・チャンネルデータの取得

uv add google-api-python-client pandas

from googleapiclient.discovery import build
import pandas as pd

# API キー (環境変数や外部ファイルから読み込むことを推奨)
API_KEY = 'YOUR_API_KEY'

# YouTube Data API クライアントの作成
youtube = build('youtube', 'v3', developerKey=API_KEY)

def get_channel_stats(channel_id: str) -> dict:
    """
    チャンネル ID を受け取り, 統計情報 (登録者数・総再生数・動画本数) を辞書で返す.

    Parameters
    ----------
    channel_id : str
        YouTube チャンネル ID (例: "UCxxxxxx")

    Returns
    -------
    dict
        チャンネル名, 登録者数, 総再生数, 動画本数を含む辞書
    """
    request = youtube.channels().list(
        part='snippet,statistics',
        id=channel_id
    )
    response = request.execute()

    if not response.get('items'):
        raise ValueError(f"チャンネルが見つかりませんでした: {channel_id}")

    item = response['items'][0]
    stats = item['statistics']
    snippet = item['snippet']

    return {
        'channel_id'       : channel_id,
        'channel_name'     : snippet['title'],
        'subscriber_count' : int(stats.get('subscriberCount', 0)),
        'view_count'       : int(stats.get('viewCount', 0)),
        'video_count'      : int(stats.get('videoCount', 0)),
    }

# 例: NHK のチャンネル (channel_id は実際の ID に置き換えてください)
channel_id = 'UCXXXXXXXXXXXXXXXXXXXXXXX'  # 実際のチャンネル ID に置き換える (調べ方は下の note を参照)

stats = get_channel_stats(channel_id)
df_channel = pd.DataFrame([stats])
print(df_channel)

def search_videos(query: str, max_results: int = 10) -> pd.DataFrame:
    """
    キーワードで YouTube 動画を検索し, 動画 ID・タイトル・投稿日の DataFrame を返す.

    Parameters
    ----------
    query : str
        検索キーワード
    max_results : int
        取得する動画の最大件数 (関数の既定値 10, API の上限は 50)

    Returns
    -------
    pd.DataFrame
        video_id, title, published_at を列に持つ DataFrame
    """
    request = youtube.search().list(
        part='snippet',
        q=query,
        type='video',         # 動画のみ (チャンネル・再生リストを除外)
        maxResults=max_results,
        relevanceLanguage='ja'  # 日本語優先
    )
    response = request.execute()

    rows = []
    for item in response.get('items', []):
        rows.append({
            'video_id'    : item['id']['videoId'],
            'title'       : item['snippet']['title'],
            'published_at': item['snippet']['publishedAt'],
        })

    return pd.DataFrame(rows)

# 例: 「データサイエンス 入門」で検索
df_search = search_videos('データサイエンス 入門', max_results=10)
print(df_search)

def get_video_stats(video_ids: list[str]) -> pd.DataFrame:
    """
    動画 ID のリストを受け取り, 各動画の統計情報を DataFrame で返す.
    50 件を超える場合は自動的に複数回に分けてリクエストする.

    Parameters
    ----------
    video_ids : list[str]
        動画 ID のリスト

    Returns
    -------
    pd.DataFrame
        video_id, title, view_count, like_count, comment_count, published_at を列に持つ DataFrame
    """
    rows = []
    # YouTube API は 1 リクエストあたり最大 50 件
    chunk_size = 50
    for i in range(0, len(video_ids), chunk_size):
        chunk = video_ids[i:i + chunk_size]
        request = youtube.videos().list(
            part='snippet,statistics',
            id=','.join(chunk)
        )
        response = request.execute()

        for item in response.get('items', []):
            stats   = item.get('statistics', {})
            snippet = item.get('snippet', {})
            rows.append({
                'video_id'     : item['id'],
                'title'        : snippet.get('title', ''),
                'published_at' : snippet.get('publishedAt', ''),
                'view_count'   : int(stats.get('viewCount', 0)),
                'like_count'   : int(stats.get('likeCount', 0)),
                'comment_count': int(stats.get('commentCount', 0)),
            })

    return pd.DataFrame(rows)

# (b) の検索結果を使って動画統計を取得
video_ids = df_search['video_id'].tolist()
df_stats = get_video_stats(video_ids)

# 再生数の多い順に並べ替え
df_stats_sorted = df_stats.sort_values('view_count', ascending=False)
print(df_stats_sorted[['title', 'view_count', 'like_count', 'comment_count']])

def get_video_comments(video_id: str, max_comments: int = 100) -> pd.DataFrame:
    """
    動画 ID を受け取り, トップレベルコメントを DataFrame で返す.

    Returns
    -------
    pd.DataFrame
        video_id, comment, comment_like_count, comment_published_at を列に持つ DataFrame
    """
    rows = []
    request = youtube.commentThreads().list(
        part='snippet',
        videoId=video_id,
        maxResults=100,        # 1 ページの上限
        textFormat='plainText',
        order='relevance',
    )
    while request is not None and len(rows) < max_comments:
        response = request.execute()
        for item in response.get('items', []):
            c = item['snippet']['topLevelComment']['snippet']
            rows.append({
                'video_id'            : video_id,
                'comment'             : c['textDisplay'],
                'comment_like_count'  : int(c.get('likeCount', 0)),
                'comment_published_at': c['publishedAt'],
            })
        request = youtube.commentThreads().list_next(request, response)

    return pd.DataFrame(rows[:max_comments])

{
  "kind": "youtube#commentThread",
  "id": "UgwwJ4D-3oTROYQub454AaABAg",
  "snippet": {
    "channelId": "UCpdtHm6VFP_Qc-IDIsyLh1A",
    "videoId": "T1r_LQCmeT8",
    "topLevelComment": {
      "kind": "youtube#comment",
      "snippet": {
        "videoId": "T1r_LQCmeT8",
        "textDisplay": "長時間おつかれさまでした。\nありがとうございました。",
        "authorDisplayName": "@example_user",
        "authorChannelId": { "value": "UCxxxxxxxxxxxxxxxxxxxxxx" },
        "likeCount": 14,
        "publishedAt": "2025-07-20T22:49:19Z",
        "updatedAt": "2025-07-20T22:49:19Z"
      }
    },
    "canReply": true,
    "totalReplyCount": 0,
    "isPublic": true
  }
}

import os
import pandas as pd
from googleapiclient.discovery import build

API_KEY    = os.environ.get('YOUTUBE_API_KEY', 'YOUR_API_KEY')
QUERY      = '市川市'              # 取得したいキーワード
MAX_VIDEOS = 10
OUTPUT_CSV = 'ichikawa_youtube.csv'

youtube = build('youtube', 'v3', developerKey=API_KEY)

# (b) 検索で動画 ID を集める
df_search = search_videos(QUERY, max_results=MAX_VIDEOS)   # (b) の関数
video_ids = df_search['video_id'].tolist()

# (c) 動画統計 (タイトル・再生数・高評価数・コメント数)
df_stats  = get_video_stats(video_ids)                     # (c) の関数 (DataFrame を返す)

# (d) 各動画のコメントを取得し, 動画メタを各コメント行に付与
frames = []
for _, video in df_stats.iterrows():
    try:
        cdf = get_video_comments(video['video_id'], max_comments=100)
    except Exception as e:
        print(f"skip {video['video_id']}: {e}")   # コメント無効化などはスキップ
        continue
    if cdf.empty:
        continue
    cdf['title']         = video['title']
    cdf['view_count']    = video['view_count']
    cdf['like_count']    = video['like_count']
    cdf['comment_count'] = video['comment_count']
    frames.append(cdf)

df = pd.concat(frames, ignore_index=True)
df = df[['video_id', 'title', 'view_count', 'like_count', 'comment_count',
         'comment', 'comment_like_count', 'comment_published_at']]
df.to_csv(OUTPUT_CSV, index=False, encoding='utf-8-sig')
print(f'{OUTPUT_CSV} に {len(df)} 行 ({df["video_id"].nunique()} 動画) を保存しました.')

uv add transformers fugashi unidic-lite torch tqdm matplotlib-fontja

import torch
from transformers import pipeline, AutoModelForSequenceClassification, BertJapaneseTokenizer
from tqdm import tqdm

MODEL_NAME = 'christian-phu/bert-finetuned-japanese-sentiment'

def load_classifier():
    """感情分析の pipeline を作成する. GPU があれば利用する."""
    if torch.backends.mps.is_available():
        device = torch.device('mps')     # Mac の GPU
    elif torch.cuda.is_available():
        device = torch.device('cuda:0')  # NVIDIA GPU
    else:
        device = torch.device('cpu')
    model = AutoModelForSequenceClassification.from_pretrained(MODEL_NAME)
    tokenizer = BertJapaneseTokenizer.from_pretrained(MODEL_NAME)
    return pipeline('sentiment-analysis', model=model, tokenizer=tokenizer, device=device)


def classify_comments(df: pd.DataFrame, classifier) -> pd.DataFrame:
    """comment 列を分類し, Label と Score を付与した DataFrame を返す."""
    df = df[df['comment'].apply(lambda x: isinstance(x, str))].copy()
    labels, scores = [], []
    for text in tqdm(df['comment']):
        try:
            result = classifier(text, truncation=True)[0]   # 長文はモデル最大長で切り詰め
            labels.append(result['label'].upper())           # POSITIVE / NEGATIVE / NEUTRAL に正規化
            scores.append(round(result['score'], 4))
        except Exception:
            labels.append('ERROR')
            scores.append(0.0)
    df['Label'] = labels
    df['Score'] = scores
    return df

import pandas as pd
import matplotlib.pyplot as plt
import matplotlib_fontja   # 日本語ラベルの文字化けを防ぐ

df = pd.read_csv('ichikawa_youtube.csv')
classifier = load_classifier()
df = classify_comments(df, classifier)
print(df[['title', 'comment', 'Label', 'Score']].head())

import re

# 任意の 1 動画を取り出す (ここでは最初の動画)
one = df[df['video_id'] == df['video_id'].iloc[0]]

counts = one['Label'].value_counts()
counts.plot.pie(autopct='%1.1f%%', ylabel='')
title = re.sub(r'[^\w\s]', '', one['title'].iloc[0])[:18]   # 絵文字・記号を除いて短縮
plt.title(f'コメントの感情比率（{title}）')
plt.show()

LABEL_TO_SCORE = {'POSITIVE': 1, 'NEUTRAL': 0, 'NEGATIVE': -1}

df['signed'] = df['Label'].map(LABEL_TO_SCORE).fillna(0) * df['Score']

# 動画ごとに集約 (動画統計は動画内で一定なので first を取る)
agg = df.groupby('video_id').agg(
    title          = ('title', 'first'),
    n_comments     = ('comment', 'size'),
    positive_ratio = ('Label', lambda s: (s == 'POSITIVE').mean()),
    mean_score     = ('signed', 'mean'),
    view_count     = ('view_count', 'first'),
    like_count     = ('like_count', 'first'),
    comment_count  = ('comment_count', 'first'),
).reset_index()

# 相関
cols = ['positive_ratio', 'mean_score', 'view_count', 'like_count', 'comment_count']
print(agg[cols].corr())

# 散布図: ポジ率 vs 高評価数
plt.scatter(agg['positive_ratio'], agg['like_count'])
plt.xlabel('ポジティブコメント比率')
plt.ylabel('高評価数')
plt.title('コメントのポジ率と高評価数の関係')
plt.show()