python_scripts/twitch_chat_vod.py

#!/usr/bin/env python3
import  os
import cv2
import linux

def get_video_height(video_path: str) -> int:
    # Open the video file
    video = cv2.VideoCapture(video_path)
    
    # Get the height property
    height = int(video.get(cv2.CAP_PROP_FRAME_HEIGHT))
    
    # Always release the video object
    video.release()
    
    return height

def combine_twitch_vod_and_chat(vod_path: str, layout: str = "side-by-side", force: bool = False) -> bool:
    """
    Renders Twitch chat JSON to video and combines it with the source VOD.
    Provides real-time terminal feedback for all processing steps.
    """
    threads = 8

    video_height = 1080

    video_path = vod_path
    video_chat = ""
    chat_path = video_path.replace(".mp4", "_chat.json")
    temp_with_chat = video_path.replace(".mp4", "_temp_with_chat.mp4")
    temp_chat = video_path.replace(".mp4", "_temp_chat.mp4")
    video_chat_SS = video_path.replace(".mp4", "_SS_with_chat.mp4")
    video_chat_over = video_path.replace(".mp4", "_over_with_chat.mp4")
    mask_path = video_path.replace(".mp4", "_temp_chat_mask.mp4")

    # Step 1: Pre-flight checks
    if not os.path.exists(video_path):
        print(f"❌ Error: Source video not found: {video_path}")
        return False
    
    if not os.path.exists(chat_path):
        print(f"❌ Error: Chat file not found: {chat_path}")
        return False

    if os.path.exists(temp_with_chat):
        os.remove(temp_with_chat)

    if layout == "side-by-side":
        video_chat = video_chat_SS
    elif layout == "overlay":
        video_chat = video_chat_over
    else:
        return False

    if os.path.exists(video_chat):
        if not force:
            print(f"❌ Error: Chat video already exists: {video_chat}")
            return False
        else:
            os.remove(video_chat)

    if os.path.exists(temp_chat):
        os.remove(temp_chat)
    if os.path.exists(mask_path):
        os.remove(mask_path)
    
    # Step 2: Render Chat to Video
    print("====================================================")
    print("🚀 STEP 1: Rendering Chat JSON to Video Layer")
    print("====================================================")

    video_height = get_video_height(vod_path)
    
    chat_cmd = (
        f"TwitchDownloaderCLI chatrender "
        f"-i {chat_path} "
        f"-w 400 -h {video_height} "
        f"--collision Overwrite "
        f"--temp-path download/temp "
        f"--font-size 20 "           
        f"--background-color #00000000 "
        f"-o {temp_chat}"
    )

    if layout == "overlay":
        chat_cmd = (f"{chat_cmd} --generate-mask ")

    if not os.path.exists(temp_chat):
        chat_success = linux.run_command(chat_cmd, look_for=["[STATUS]"])
        if chat_success:
            print("✅ Success: TwitchDownloaderCLI render chat video.")
        else:
            print("❌ Error: TwitchDownloaderCLI failed to render chat video.")
            os.remove(temp_chat)
            os.remove(mask_path)
            return False

    # Step 3: Combine Video and Chat using FFmpeg
    print("====================================================")
    print(f"🚀 STEP 2: Merging VOD and Chat Layout ({layout})")
    print("====================================================")
    
    # -preset superfast speeds up the 3+ hour encoding process significantly
    # -map 0:a? safely includes audio if it exists, without breaking on silent VODs
    if layout == "side-by-side":
        ffmpeg_cmd = (
            f"ffmpeg -y -i {video_path} -i {temp_chat} "
            f"-filter_complex '[1:v]scale=-1:ih[scaled_chat];[0:v][scaled_chat]hstack=inputs=2[v]' "
            f"-map '[v]' -map 0:a? -c:v libx264 -crf 18 -preset slow -c:a copy "
            f"-threads {threads} {temp_with_chat}"
        )
    elif layout == "overlay":
        ffmpeg_cmd = (
            f"ffmpeg -y -i {video_path} -i {temp_chat} -i {mask_path} "
            f"-filter_complex '[1:v][2:v]alphamerge[masked_chat];[0:v][masked_chat]overlay=x=0:y=10[v]' "
            f"-map '[v]' -map 0:a? -c:v libx264 -crf 18 -preset slow -c:a copy "
            f"-threads {threads} {temp_with_chat}"
        )
    else:
        raise ValueError("Invalid layout choice. Choose 'side-by-side' or 'overlay'.")

    ffmpeg_success = linux.run_command(ffmpeg_cmd, progress_prefix="FFmpeg Merge")

    # Step 4: Final verification and cleanup
    if ffmpeg_success and os.path.exists(temp_with_chat):
        os.rename(temp_with_chat, video_chat)
        print("====================================================")
        print(f"🎉 SUCCESS: Video processing complete!")
        print(f"📁 Output Saved: {video_chat}")
        print("====================================================")
        
        # Clean up the massive temporary chat video to save storage space
        if os.path.exists(temp_chat):
            print("🧹 Cleaning up temporary chat render video...")
            os.remove(temp_chat)
        if os.path.exists(mask_path):
            os.remove(mask_path)
        if os.path.exists(temp_with_chat):
            os.remove(temp_with_chat)
        return True
    else:
        print("❌ Error: FFmpeg failed to merge the video streams.")
        if os.path.exists(video_chat):
            os.remove(video_chat)
        if os.path.exists(temp_with_chat):
            os.remove(temp_with_chat)
        return False

if __name__ == "__main__":
    combine_twitch_vod_and_chat("download/videos/2813112936/2813112936.mp4", "overlay", True)