#!/usr/bin/env python3 """youtube_analytics.py — YouTube の分析の数字を、API で取って残す. これまでは YouTube Studio の画面や書き出した zip から読んでいた数字を、自動で取る。 議論係（debate_runner.py）に渡して、AI が想像ではなく実際の数字で企画を議論できるようにする。 取るもの（期間は既定で直近90日）: - 動画ごとの表示回数・クリック率・視聴回数・平均視聴時間・平均視聴率・登録の増加 - 本編ごとの視聴者維持率（動画の長さを100に分けた点ごと）と、30秒の時点の残り - 本編ごとの来た道（YouTube 検索・ブラウジング・関連動画・ショートなど） 置き場所: iCloud の PixelstoMotors/sns_analytics/youtube/api/ {開始}_{終了}.json 取ったままの数字 {開始}_{終了}.md 議論に渡す要約（本編の一覧・30秒の残り・来た道） 許可: ~/.config/pixelstomotors/client_secret.json youtube_schedule.py と同じもの（pixelstomotors-studio） ~/.config/pixelstomotors/youtube_analytics_token.json 分析を読む許可（このスクリプトが作る。ほかの許可とは別のファイル） 分析を読むだけ。動画を変えたり上げたりする権限はない 使い方（Mac のターミナルで）: python3 youtube_analytics.py --login 最初に1回。ブラウザで Pixels to Motors を選んで許可 python3 youtube_analytics.py 直近90日を取る python3 youtube_analytics.py --days 28 直近28日 python3 youtube_analytics.py --start 2026-10-01 --end 2026-11-13 """ import argparse, json, os, sys, urllib.error, urllib.parse, urllib.request from datetime import date, datetime, timedelta, timezone HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) import youtube_schedule as ys # ログインの仕組みは youtube_schedule.py のものを使う # ブランドアカウントにはメールがないので openid・email は入れない（youtube_upload.py と同じ理由） ys.SCOPE = "https://www.googleapis.com/auth/yt-analytics.readonly https://www.googleapis.com/auth/youtube.readonly" ys.TOKEN = os.path.join(ys.CONF, "youtube_analytics_token.json") JST = timezone(timedelta(hours=9)) AN = "https://youtubeanalytics.googleapis.com/v2/reports" VIDEOS_JSON = os.path.join(HERE, "youtube_videos.json") # youtube_schedule.py が作る動画の一覧 PTM = os.path.dirname(os.path.dirname(os.path.dirname(HERE))) # PixelstoMotors/ OUTDIR = os.path.join(PTM, "sns_analytics", "youtube", "api") BASE = "views,estimatedMinutesWatched,averageViewDuration,averageViewPercentage,subscribersGained" REACH = "videoThumbnailImpressions,videoThumbnailImpressionsClickRate" TRAFFIC_JA = { "YT_SEARCH": "YouTube 検索", "SUBSCRIBER": "ブラウジング（登録者）", "RELATED_VIDEO": "関連動画", "YT_CHANNEL": "チャンネルページ", "EXT_URL": "外部", "SHORTS": "ショートのフィード", "NO_LINK_OTHER": "直接・不明", "PLAYLIST": "再生リスト", "END_SCREEN": "終了画面", "NOTIFICATION": "通知", "YT_OTHER_PAGE": "YouTube のほかのページ", "HASHTAGS": "ハッシュタグ", "BROWSE": "ブラウジング", "SHORTS_CONTENT_LINKS": "関連のショート動画", "ANNOTATION": "カード", "YT_PLAYLIST_PAGE": "再生リストのページ", "SOUND_PAGE": "サウンドのページ", "CAMPAIGN_CARD": "キャンペーン", } def get(tok, **q): url = AN + "?" + urllib.parse.urlencode({"ids": "channel==MINE", **q}) req = urllib.request.Request(url, headers={"Authorization": "Bearer " + tok}) try: with urllib.request.urlopen(req, timeout=120) as r: return json.loads(r.read().decode()) except urllib.error.HTTPError as e: raise RuntimeError(f"分析の API がエラーを返しました（{e.code}）: {e.read().decode('utf-8', 'replace')[:600]}") def rows(d): cols = [c["name"] for c in d.get("columnHeaders", [])] return [dict(zip(cols, r)) for r in d.get("rows", []) or []] def video_list(): """youtube_videos.json から、題名・長さ・ショートかどうかを引く""" try: d = json.load(open(VIDEOS_JSON, encoding="utf-8")) return {v["id"]: v for v in d.get("動画", [])} except Exception: return {} def fmt_sec(s): s = int(round(s or 0)) return f"{s // 60}分{s % 60:02d}秒" def main(): ap = argparse.ArgumentParser(description="YouTube の分析の数字を API で取る") ap.add_argument("--login", action="store_true") ap.add_argument("--days", type=int, default=90) ap.add_argument("--start") ap.add_argument("--end") a = ap.parse_args() if a.login: ys.login() tok = ys.access_token() me = ys.call(f"{ys.API}/channels?part=snippet&mine=true", headers={"Authorization": "Bearer " + tok}) if ys.CHANNEL_ID not in [i["id"] for i in me.get("items", [])]: raise SystemExit("許可したアカウントのチャンネルが Pixels to Motors ではありません。--login で選び直してください") print("チャンネル Pixels to Motors の分析を読める許可です") return end = date.fromisoformat(a.end) if a.end else date.today() - timedelta(days=1) start = date.fromisoformat(a.start) if a.start else end - timedelta(days=a.days - 1) tok = ys.access_token() vids = video_list() print(f"期間: {start}〜{end}") # 1. 動画ごとの数字（表示回数とクリック率が取れないときは、それ抜きで取り直す） note = "" try: per = rows(get(tok, startDate=start, endDate=end, dimensions="video", metrics=BASE + "," + REACH, sort="-views", maxResults=200)) except RuntimeError as e: note = f"表示回数とクリック率は取れませんでした（{str(e)[:160]}）" print("注意:", note) per = rows(get(tok, startDate=start, endDate=end, dimensions="video", metrics=BASE, sort="-views", maxResults=200)) for r in per: v = vids.get(r["video"], {}) r["title"] = v.get("title", "（一覧にない動画）") r["seconds"] = v.get("seconds") r["is_short"] = v.get("is_short") r["published"] = v.get("published") longs = [r for r in per if r.get("is_short") is False] # 2. 本編ごとの視聴者維持率と来た道 for r in longs: vid = r["video"] try: ret = rows(get(tok, startDate=start, endDate=end, dimensions="elapsedVideoTimeRatio", metrics="audienceWatchRatio,relativeRetentionPerformance", filters=f"video=={vid};audienceType==ORGANIC", sort="elapsedVideoTimeRatio")) r["retention"] = ret if ret and r.get("seconds"): x = 30 / r["seconds"] near = min(ret, key=lambda p: abs(p["elapsedVideoTimeRatio"] - x)) r["retention_30s"] = near["audienceWatchRatio"] except RuntimeError as e: r["retention_error"] = str(e)[:200] try: r["traffic"] = rows(get(tok, startDate=start, endDate=end, dimensions="insightTrafficSourceType", metrics="views,averageViewDuration", filters=f"video=={vid}", sort="-views")) except RuntimeError as e: r["traffic_error"] = str(e)[:200] os.makedirs(OUTDIR, exist_ok=True) stem = f"{start}_{end}" with open(os.path.join(OUTDIR, stem + ".json"), "w", encoding="utf-8") as f: json.dump({"取った日時": datetime.now(JST).isoformat(timespec="minutes"), "期間": [str(start), str(end)], "注意": note, "動画": per}, f, ensure_ascii=False, indent=1) # 3. 議論に渡す要約 L = [f"# YouTube の数字（{start}〜{end}・API で取得 {datetime.now(JST).strftime('%Y-%m-%d %H:%M')}）", ""] if note: L += [f"- 注意: {note}", ""] tv = sum(r["views"] for r in per) lv = sum(r["views"] for r in longs) L += [f"- 視聴の合計 {tv:,} 回（本編 {lv:,} 回・ショート {tv - lv:,} 回）", "", "## 本編（視聴の多い順）", "", "| 題名 | 長さ | 表示 | クリック率 | 視聴 | 平均視聴 | 平均視聴率 | 30秒の残り | 登録 |", "|---|---|---|---|---|---|---|---|---|"] for r in sorted(longs, key=lambda r: -r["views"]): imp = r.get("videoThumbnailImpressions") ctr = r.get("videoThumbnailImpressionsClickRate") r30 = r.get("retention_30s") L.append(f"| {r['title'][:60]} | {fmt_sec(r.get('seconds'))} | {imp if imp is not None else '—'} | " f"{f'{ctr * 100:.1f}%' if ctr is not None else '—'} | {r['views']} | {fmt_sec(r['averageViewDuration'])} | " f"{r['averageViewPercentage']:.0f}% | {f'{r30 * 100:.0f}%' if r30 is not None else '—'} | {r['subscribersGained']} |") L += ["", "## 本編の来た道（上位3つ）", ""] for r in sorted(longs, key=lambda r: -r["views"]): t = r.get("traffic") or [] if t: parts = [f"{TRAFFIC_JA.get(x['insightTrafficSourceType'], x['insightTrafficSourceType'])} {x['views']}" for x in t[:3]] L.append(f"- {r['title'][:50]}：{'・'.join(parts)}") L += ["", "## ショート（視聴の多い順・上位10）", "", "| 題名 | 視聴 | 平均視聴率 | 登録 |", "|---|---|---|---|"] for r in sorted([r for r in per if r.get("is_short")], key=lambda r: -r["views"])[:10]: L.append(f"| {r['title'][:60]} | {r['views']} | {r['averageViewPercentage']:.0f}% | {r['subscribersGained']} |") L += ["", "- 30秒の残り：維持率のグラフで、30秒の時点にいちばん近い点の値（自然な流入だけ。再生を始めた人に対する割合）", "- 平均視聴率：1回の再生で、動画の長さの何%を見たか", ""] with open(os.path.join(OUTDIR, stem + ".md"), "w", encoding="utf-8") as f: f.write("\n".join(L)) print(f"書き出しました: sns_analytics/youtube/api/{stem}.json と .md（本編 {len(longs)} 本・全部で {len(per)} 本）") if __name__ == "__main__": try: main() except RuntimeError as e: raise SystemExit(f"× {e}")