import os import subprocess import requests import whisper from datetime import datetime, time, timedelta from zoneinfo import ZoneInfo import json # --------------------------- # Configuration # --------------------------- TWITCH_CLIENT_ID = os.environ.get("TWITCH_CLIENT_ID", "") TWITCH_CLIENT_SECRET = os.environ.get("TWITCH_CLIENT_SECRET", "") CHANNEL_NAME = os.environ.get("CHANNEL_NAME", "madmonq") CHANNEL_LANGUAGE = os.environ.get("CHANNEL_LANGUAGE", "en") SEARCH_KEYWORDS = ["madmonq", "madmonge", "madmong", "medmong", "medmonk", "madmonk"] MODEL_NAME = "turbo" # Whisper model # Define base directories for each file category under a folder named after the channel. base_dirs = { "vods": os.path.join("vods", CHANNEL_NAME), "audio": os.path.join("audio", CHANNEL_NAME), "transcripts": os.path.join("transcripts", CHANNEL_NAME), "chat": os.path.join("chat", CHANNEL_NAME), "clips": os.path.join("clips", CHANNEL_NAME) } # Create directories if they do not exist. for path in base_dirs.values(): os.makedirs(path, exist_ok=True) # --------------------------- # Twitch API Helper Functions # --------------------------- def get_access_token(): url = "https://id.twitch.tv/oauth2/token" payload = { "client_id": TWITCH_CLIENT_ID, "client_secret": TWITCH_CLIENT_SECRET, "grant_type": "client_credentials" } response = requests.post(url, data=payload) response.raise_for_status() data = response.json() return data["access_token"] def get_channel_id(channel_name, token): headers = { "Client-ID": TWITCH_CLIENT_ID, "Authorization": f"Bearer {token}" } url = f"https://api.twitch.tv/helix/users?login={channel_name}" response = requests.get(url, headers=headers) response.raise_for_status() data = response.json() if data.get("data"): return data["data"][0]["id"] else: print("Channel not found.") return None def get_vods_from_yesterday(channel_id, token): headers = { "Client-ID": TWITCH_CLIENT_ID, "Authorization": f"Bearer {token}" } prague_tz = ZoneInfo("Europe/Prague") today_prague = datetime.now(prague_tz).date() yesterday = today_prague - timedelta(days=1) # Change days as needed start_time = datetime.combine(yesterday, time.min).replace(tzinfo=prague_tz) end_time = datetime.combine(yesterday, time.max).replace(tzinfo=prague_tz) url = f"https://api.twitch.tv/helix/videos?user_id={channel_id}&type=archive&first=100" response = requests.get(url, headers=headers) response.raise_for_status() vods = [] for vod in response.json().get("data", []): published_at = datetime.fromisoformat(vod["published_at"].replace("Z", "+00:00")) published_at_prague = published_at.astimezone(prague_tz) if start_time <= published_at_prague <= end_time: vods.append(vod) return vods # --------------------------- # VOD Processing Functions # --------------------------- def download_vod(vod_url, output_filename): if os.path.exists(output_filename): print(f"{output_filename} already exists. Skipping download.") return command = ["yt-dlp", "-o", output_filename, vod_url] subprocess.run(command, check=True) print(f"Downloaded VOD to {output_filename}") def extract_audio(video_file, audio_file): if os.path.exists(audio_file): print(f"{audio_file} already exists. Skipping audio extraction.") return command = ["ffmpeg", "-i", video_file, "-vn", "-acodec", "mp3", audio_file, "-y"] subprocess.run(command, check=True) print(f"Extracted audio to {audio_file}") def transcribe_audio(audio_file, model_name): model = whisper.load_model(model_name, download_root="/app/models") result = model.transcribe(audio_file, language=CHANNEL_LANGUAGE) return result def search_transcription(result, keywords): matches = [] if "segments" in result: for segment in result["segments"]: segment_text = segment["text"].lower() for keyword in keywords: if keyword.lower() in segment_text: matches.append(segment) break # Stop checking further keywords for this segment return matches def scrape_chat_log(vod_id, output_filename): if os.path.exists(output_filename): print(f"{output_filename} already exists. Skipping chat log scrape.") return headers = { "Client-ID": TWITCH_CLIENT_ID, "Accept": "application/vnd.twitchtv.v5+json" } base_url = f"https://api.twitch.tv/v5/videos/{vod_id}/comments" comments = [] cursor = None while True: params = {} if cursor: params["cursor"] = cursor response = requests.get(base_url, headers=headers, params=params) if response.status_code != 200: print(f"Error fetching chat comments for VOD {vod_id}: {response.text}") break data = response.json() comments.extend(data.get("comments", [])) cursor = data.get("_next") if not cursor: break with open(output_filename, "w", encoding="utf-8") as f: json.dump(comments, f, ensure_ascii=False, indent=4) print(f"Chat log saved to {output_filename}") def create_clip_from_vod(video_file, match_start, vod_id): clip_start = max(match_start - 15, 0) clip_duration = 60 # seconds clip_dir = base_dirs["clips"] os.makedirs(clip_dir, exist_ok=True) clip_filename = os.path.join(clip_dir, f"clip_{vod_id}_{int(match_start)}.mp4") command = [ "ffmpeg", "-ss", str(clip_start), "-i", video_file, "-t", str(clip_duration), "-c", "copy", clip_filename, "-y" ] subprocess.run(command, check=True) print(f"Clip created: {clip_filename}") return clip_filename def find_comments_by_keywords(chat_log, keywords): matching_comments = [] # Ensure chat_log is a list of dictionaries. for comment in chat_log: if not isinstance(comment, dict): continue message = comment.get("message", {}) if not isinstance(message, dict): continue text = message.get("body", "").lower() for keyword in keywords: if keyword.lower() in text: matching_comments.append(comment) break return matching_comments def create_clip_from_comment_timestamp(video_file, comment_timestamp, vod_id): clip_start = max(comment_timestamp - 15, 0) clip_duration = 60 # seconds clip_dir = base_dirs["clips"] os.makedirs(clip_dir, exist_ok=True) clip_filename = os.path.join(clip_dir, f"clip_{vod_id}_{int(comment_timestamp)}.mp4") command = [ "ffmpeg", "-ss", str(clip_start), "-i", video_file, "-t", str(clip_duration), "-c", "copy", clip_filename, "-y" ] subprocess.run(command, check=True) print(f"Clip created: {clip_filename}") return clip_filename # --------------------------- # Main Processing Pipeline # --------------------------- def main(): print("Obtaining access token...") token = get_access_token() print("Access token obtained.") channel_id = get_channel_id(CHANNEL_NAME, token) if not channel_id: return vods = get_vods_from_yesterday(channel_id, token) if not vods: print("No VODs from yesterday found.") return for vod in vods: vod_url = vod["url"] vod_id = vod["id"] # Define file paths in the respective directories video_filename = os.path.join(base_dirs["vods"], f"vod_{vod_id}.mp4") audio_filename = os.path.join(base_dirs["audio"], f"vod_{vod_id}.mp3") transcript_filename = os.path.join(base_dirs["transcripts"], f"transcript_{vod_id}.json") chat_log_filename = os.path.join(base_dirs["chat"], f"chat_{vod_id}.json") print(f"\nProcessing VOD: {vod_url}") download_vod(vod_url, video_filename) extract_audio(video_filename, audio_filename) # Check if transcript already exists; if yes, load it, otherwise transcribe and save. if os.path.exists(transcript_filename): print(f"{transcript_filename} already exists. Skipping transcription.") with open(transcript_filename, "r", encoding="utf-8") as f: result = json.load(f) else: print("Transcribing audio. This may take some time...") result = transcribe_audio(audio_filename, MODEL_NAME) with open(transcript_filename, "w", encoding="utf-8") as f: json.dump(result, f, ensure_ascii=False, indent=4) print(f"Transcript saved to {transcript_filename}") scrape_chat_log(vod_id, chat_log_filename) # Search transcript for keywords matches = search_transcription(result, SEARCH_KEYWORDS) if matches: print(f"Found {len(matches)} mention(s) of {SEARCH_KEYWORDS} in VOD {vod_id}:") for match in matches: start = match["start"] text = match["text"] print(f" - At {start:.2f}s: {text}") create_clip_from_vod(video_filename, start, vod_id) else: print(f"No mentions of {SEARCH_KEYWORDS} found in VOD {vod_id}.") # Load chat log from file try: with open(chat_log_filename, "r", encoding="utf-8") as f: chat_log = json.load(f) except Exception as e: print(f"Error loading chat log: {e}") chat_log = [] # Search chat log using an array of keywords (using the same keywords as for transcript) comment_matches = find_comments_by_keywords(chat_log, SEARCH_KEYWORDS) if comment_matches: for comment in comment_matches: timestamp = comment.get("content_offset_seconds") print(f"Found a matching comment at {timestamp} seconds.") create_clip_from_comment_timestamp(video_filename, timestamp, vod_id) else: print("No matching comments found.") if __name__ == "__main__": main()