""" YouTube Live リアルタイムモニター（最適化版） 5分ごとに実行し、配信中チャンネルのconcurrentViewersを記録。 配信終了時にavg/peakを計算してdata.jsonに書き込む。 APIコスト: - channels一括確認: 3ユニット/回（50チャンネルまで1リクエスト） - videoId取得(初回のみ): 100ユニット/ストリーム - 視聴者数取得: 1ユニット/ライブチャンネル/回 """ import json import os import datetime import urllib.request import urllib.parse import time BASE_DIR = os.path.dirname(os.path.abspath(__file__)) CONFIG_FILE = os.path.join(BASE_DIR, '../config.json') DATA_FILE = os.path.join(BASE_DIR, '../data.json') SETTINGS_FILE = os.path.join(BASE_DIR, '../settings.json') # 進行中の配信セッションを管理するキャッシュ SESSIONS_FILE = os.path.join(BASE_DIR, '../live_sessions.json') def load_json(path, default): if os.path.exists(path): try: with open(path, 'r', encoding='utf-8') as f: return json.load(f) except Exception: pass return default def save_json(path, data): with open(path, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) def api_get(url): """YouTube APIへのGETリクエスト。エラー時はNoneを返す。""" try: req = urllib.request.Request(url, headers={'Accept': 'application/json'}) res = urllib.request.urlopen(req, timeout=10) return json.loads(res.read()) except Exception as e: print(f" [API Error] {e}") return None def check_channels_live(channel_ids, api_key): """ channels?part=snippetで最大50チャンネルを一括確認。 snippet.liveBroadcastContent == 'live' のチャンネルIDセットを返す。 コスト: 3ユニット/コール（50チャンネル一括） """ live_set = set() # 50件ずつ分割 for i in range(0, len(channel_ids), 50): batch = channel_ids[i:i+50] ids_str = ','.join(batch) url = (f"https://www.googleapis.com/youtube/v3/channels" f"?part=snippet&id={ids_str}&key={api_key}") data = api_get(url) if not data: continue for item in data.get('items', []): ch_id = item.get('id') live_status = item.get('snippet', {}).get('liveBroadcastContent', 'none') if live_status == 'live': live_set.add(ch_id) return live_set def get_live_video_id(channel_id, api_key): """ 配信中のvideoIdを取得。初回のみ呼ぶ（100ユニット）。 """ url = (f"https://www.googleapis.com/youtube/v3/search" f"?part=id&channelId={channel_id}&eventType=live&type=video&key={api_key}") data = api_get(url) if not data: return None items = data.get('items', []) if items: return items[0]['id']['videoId'] return None def get_concurrent_viewers(video_id, api_key): """ videos?part=liveStreamingDetailsで同時視聴者数を取得（1ユニット）。 """ url = (f"https://www.googleapis.com/youtube/v3/videos" f"?part=liveStreamingDetails&id={video_id}&key={api_key}") data = api_get(url) if not data: return None items = data.get('items', []) if not items: return None lsd = items[0].get('liveStreamingDetails', {}) cv = lsd.get('concurrentViewers') if cv is not None: return int(cv) return None def finalize_session(inf_id, session, data_json): """配信終了時にavg/peakを計算してdata.jsonへ書き込む。""" samples = session.get('samples', []) if not samples: return avg_viewers = round(sum(samples) / len(samples)) peak_viewers = max(samples) month = session.get('month', datetime.datetime.now().strftime('%Y-%m')) if month not in data_json: data_json[month] = {} if inf_id not in data_json[month]: data_json[month][inf_id] = {} if 'youtube' not in data_json[month][inf_id]: data_json[month][inf_id]['youtube'] = {} # avg/peakのみ上書き（followersやstream_time_hoursは保持） data_json[month][inf_id]['youtube']['avg_viewers'] = avg_viewers data_json[month][inf_id]['youtube']['peak_viewers'] = peak_viewers print(f" [{inf_id}] YouTube配信終了 → avg={avg_viewers:,} peak={peak_viewers:,} ({len(samples)}サンプル)") def main(): print(f"--- YouTube Live モニター実行 {datetime.datetime.now().strftime('%Y-%m-%d %H:%M')} ---") settings = load_json(SETTINGS_FILE, {}) api_key = settings.get('youtube_api_key') if not api_key: print(" [ERROR] youtube_api_key が settings.json に設定されていません。") return config = load_json(CONFIG_FILE, {"influencers": []}) influencers = config.get('influencers', []) # YouTubeチャンネルIDを持つインフルエンサーのみ対象 yt_influencers = { inf['id']: inf['platforms'].get('youtube') for inf in influencers if inf.get('platforms', {}).get('youtube') } if not yt_influencers: print(" YouTubeチャンネルが登録されていません。") return channel_ids = list(yt_influencers.values()) # channel_id → inf_id の逆引きマップ ch_to_inf = {v: k for k, v in yt_influencers.items()} # 進行中セッションのロード { channel_id: { video_id, samples, month } } sessions = load_json(SESSIONS_FILE, {}) data_json = load_json(DATA_FILE, {}) current_month = datetime.datetime.now().strftime('%Y-%m') # ① channels一括確認（3ユニット/50チャンネル） live_channel_ids = check_channels_live(channel_ids, api_key) print(f" ライブ中: {len(live_channel_ids)}/{len(channel_ids)} チャンネル") updated_sessions = {} for ch_id in channel_ids: inf_id = ch_to_inf[ch_id] is_live = ch_id in live_channel_ids if is_live: # ② videoIdを取得（セッションにキャッシュ済みなら不要） session = sessions.get(ch_id, {}) video_id = session.get('video_id') if not video_id: # 初回: search APIでvideoIdを取得（100ユニット） video_id = get_live_video_id(ch_id, api_key) if not video_id: print(f" [{inf_id}] videoId取得失敗 → スキップ") continue session = { 'video_id': video_id, 'samples': [], 'month': current_month } print(f" [{inf_id}] 配信開始検知 → videoId={video_id}") # ③ 同時視聴者数取得（1ユニット） viewers = get_concurrent_viewers(video_id, api_key) if viewers is not None: session['samples'].append(viewers) print(f" [{inf_id}] {viewers:,} 視聴者 (累計 {len(session['samples'])} サンプル)") else: print(f" [{inf_id}] 視聴者数取得失敗") updated_sessions[ch_id] = session time.sleep(0.5) # API制限回避 else: # 配信終了: セッションがあれば集計 if ch_id in sessions: print(f" [{inf_id}] 配信終了を検知 → 集計中...") finalize_session(inf_id, sessions[ch_id], data_json) # セッションは削除（updated_sessionsに追加しない） # 保存 save_json(SESSIONS_FILE, updated_sessions) save_json(DATA_FILE, data_json) print("--- 完了 ---") if __name__ == '__main__': main()