import json import os import sys import urllib.request import urllib.error import datetime import time # 設定 CONFIG_FILE = '../config.json' DATA_FILE = '../data.json' SETTINGS_FILE = '../settings.json' def load_json(filepath, default_value): if os.path.exists(filepath): with open(filepath, 'r', encoding='utf-8') as f: try: return json.load(f) except json.JSONDecodeError: return default_value return default_value def save_json(filepath, data): with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) def fetch_twitch_data(twitch_id): """TwitchTracker APIを利用してTwitchのデータを取得""" if not twitch_id: return None url = f"https://twitchtracker.com/api/channels/summary/{twitch_id}" req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'}) try: res = urllib.request.urlopen(req) data = json.loads(res.read()) return { "followers": data.get('followers_total', 0), "stream_time_hours": round(data.get('minutes_streamed', 0) / 60), "avg_viewers": data.get('avg_viewers', 0), "peak_viewers": data.get('max_viewers', 0) } except Exception as e: print(f" [Twitch] Error fetching data for {twitch_id}: {e}") return None def fetch_youtube_data(youtube_id, api_key): """YouTubeの登録者数と配信時間をAPIで取得""" if not youtube_id or not api_key: return None import urllib.parse followers = 0 stream_time_hours = 0 try: # 1. Channel Stats (Subscriber Count) url_ch = f"https://www.googleapis.com/youtube/v3/channels?part=statistics&id={youtube_id}&key={api_key}" req_ch = urllib.request.Request(url_ch) res_ch = urllib.request.urlopen(req_ch) data_ch = json.loads(res_ch.read()) if data_ch.get("items"): followers = int(data_ch["items"][0]["statistics"].get("subscriberCount", 0)) # 2. Search recent live streams this month now = datetime.datetime.now(datetime.timezone.utc) start_of_month = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0) # YouTube APIはRFC 3339形式を要求するため、'Z'サフィックスに統一 published_after = start_of_month.strftime('%Y-%m-%dT%H:%M:%SZ') url_search = f"https://www.googleapis.com/youtube/v3/search?part=id&channelId={youtube_id}&eventType=completed&type=video&maxResults=50&publishedAfter={urllib.parse.quote(published_after)}&key={api_key}" req_search = urllib.request.Request(url_search) res_search = urllib.request.urlopen(req_search) data_search = json.loads(res_search.read()) video_ids = [item['id']['videoId'] for item in data_search.get('items', [])] # 3. Get duration for these streams stream_time_seconds = 0 if video_ids: ids_str = ",".join(video_ids) url_vid = f"https://www.googleapis.com/youtube/v3/videos?part=liveStreamingDetails&id={ids_str}&key={api_key}" req_vid = urllib.request.Request(url_vid) res_vid = urllib.request.urlopen(req_vid) data_vid = json.loads(res_vid.read()) for item in data_vid.get('items', []): lsd = item.get('liveStreamingDetails', {}) start_str = lsd.get('actualStartTime') end_str = lsd.get('actualEndTime') if start_str and end_str: try: start_dt = datetime.datetime.fromisoformat(start_str.replace('Z', '+00:00')) end_dt = datetime.datetime.fromisoformat(end_str.replace('Z', '+00:00')) duration = (end_dt - start_dt).total_seconds() stream_time_seconds += duration except Exception as e: print(f" Warning: parsing date failed: {e}") stream_time_hours = round(stream_time_seconds / 3600) return { "followers": followers, "stream_time_hours": stream_time_hours, "avg_viewers": 0, # YouTube API does not readily provide historical avg viewers without Analytics API "peak_viewers": 0 } except Exception as e: print(f" [YouTube API] Error fetching data for {youtube_id}: {e}") return None def fetch_x_data(x_id): """X(Twitter)のフォロワー数を取得""" if not x_id: return None url = f"https://api.vxtwitter.com/{x_id}" req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'}) try: res = urllib.request.urlopen(req) data = json.loads(res.read()) followers = data.get('followers_count', 0) return { "followers": followers } except Exception as e: print(f" [X] Error fetching data for {x_id}: {e}") return None def main(): # --force フラグで既存データを上書き再取得 force_update = '--force' in sys.argv print("--- 収集開始 ---") if force_update: print("[強制更新モード] 既存データを上書きします。") config = load_json(CONFIG_FILE, {"influencers": []}) influencers = config.get("influencers", []) settings = load_json(SETTINGS_FILE, {}) youtube_api_key = settings.get("youtube_api_key") # 既存データのロード { "YYYY-MM": { "shaka": { "twitch": {...}, "youtube": {...}, "x": {...} } } } historical_data = load_json(DATA_FILE, {}) current_month = datetime.datetime.now().strftime('%Y-%m') if current_month not in historical_data: historical_data[current_month] = {} month_data = historical_data[current_month] for inf in influencers: inf_id = inf.get("id") name = inf.get("name") platforms = inf.get("platforms", {}) print(f"[{name}] のデータを取得中...") if inf_id not in month_data: month_data[inf_id] = {} inf_data = month_data[inf_id] # Twitch if force_update or "twitch" not in inf_data or not inf_data["twitch"]: print(f" > Twitch ({platforms.get('twitch')}) を取得...") t_data = fetch_twitch_data(platforms.get("twitch")) if t_data: inf_data["twitch"] = t_data time.sleep(1) # API制限回避 # YouTube if force_update or "youtube" not in inf_data or not inf_data["youtube"]: print(f" > YouTube ({platforms.get('youtube')}) を取得...") y_data = fetch_youtube_data(platforms.get("youtube"), youtube_api_key) if y_data: inf_data["youtube"] = y_data time.sleep(1) # X if force_update or "x" not in inf_data or not inf_data["x"]: print(f" > X ({platforms.get('x')}) を取得...") x_data = fetch_x_data(platforms.get("x")) if x_data: inf_data["x"] = x_data time.sleep(1) # データの保存 save_json(DATA_FILE, historical_data) print("--- 収集完了 ---") print(f"データは {DATA_FILE} に保存されました。") if __name__ == "__main__": main()