stream-shot/main.py

import cv2
import yt_dlp
import numpy as np
import json
import time
import os
import base64

# Simple in-memory cache
# Format: {url: {"time": timestamp, "buffer": base64_string}}
cache = {}
CACHE_DURATION = 10  # seconds

def capture_stream_snapshot(youtube_url, output_file="snapshot.jpg", toBuffer=False):
    # Check cache first
    if youtube_url in cache:
        cached_item = cache[youtube_url]
        if time.time() - cached_item["time"] < CACHE_DURATION:
            print(f"Returning cached snapshot for {youtube_url}")
            return base64.b64decode(cached_item["buffer"]) if toBuffer else True

    # 1. Configure yt-dlp to get the direct stream URL
    ydl_opts = {
        "format": "best",
        "quiet": True,
        "no_warnings": True,
    }

    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        try:
            info = ydl.extract_info(youtube_url, download=False)
            stream_url = info["url"]
        except Exception as e:
            print(f"Error extracting info: {e}")
            return None

    # 2. Use OpenCV to capture a frame from the stream URL
    try:
        cap = cv2.VideoCapture(stream_url)
    except Exception as e:
        print(f"Error opening VideoCapture: {e}")
        return None

    if not cap.isOpened():
        print("Error: Could not open video stream.")
        return None

    # Read a single frame
    success, frame = cap.read()

    result = None
    if success:
        if toBuffer:
            # Encode frame as JPEG to memory buffer
            ret, buf = cv2.imencode(".jpg", frame)
            if ret:
                print("Snapshot captured to buffer.")
                result = buf.tobytes()
            else:
                print("Error: Could not encode frame to buffer.")
        else:
            # 3. Save the frame as an image
            cv2.imwrite(output_file, frame)
            print(f"Snapshot saved to {output_file}")
            result = True
            
        # Update cache if successful
        if result:
            if toBuffer:
                buf_to_cache = result
            else:
                with open(output_file, "rb") as f:
                    buf_to_cache = f.read()
            
            cache[youtube_url] = {
                "time": time.time(),
                "buffer": base64.b64encode(buf_to_cache).decode('utf-8')
            }
    else:
        print("Error: Could not read frame from stream.")

    # Cleanup
    cap.release()
    return result


# Shsf Handler
def main(args):
    route = args.get("route")
    if route == "snapshot":
        url_input = args.get("body", "")
        try:
            body_json = json.loads(url_input)
        except json.JSONDecodeError:
            print("Error: Invalid JSON input.")
            return {
                "_shsf": "v2",
                "_code": 400,
                "_res": {"message": "Invalid JSON input."},
                "_headers": {
                    "Content-Type": "application/json",
                },
            }
        url = body_json.get("url", "")
        if not url:
            print("Error: URL not provided.")
            return {
                "_shsf": "v2",
                "_code": 400,
                "_res": {"message": "URL not provided."},
                "_headers": {
                    "Content-Type": "application/json",
                },
            }
        
        # Check cache explicitly in main to avoid redundant toBuffer logic if possible
        if url in cache:
            cached_item = cache[url]
            if time.time() - cached_item["time"] < CACHE_DURATION:
                print(f"Serving {url} from cache.")
                return {
                    "_shsf": "v2",
                    "_code": 200,
                    "_res": {
                        "message": "done (cached)",
                        "buffer": cached_item["buffer"]
                    },
                    "_headers": {
                        "Content-Type": "application/json",
                    },
                }

        try:
            # Use toBuffer=True to get the bytes directly
            content = capture_stream_snapshot(url, toBuffer=True)
        except Exception as e:
            print(f"Error: {e}")
            return {
                "_shsf": "v2",
                "_code": 500,
                "_res": {"message": f"Could not capture snapshot: {str(e)}"},
                "_headers": {
                    "Content-Type": "application/json",
                },
            }
        
        if content is None:
            print("Error: Could not capture snapshot.")
            return {
                "_shsf": "v2",
                "_code": 500,
                "_res": {"message": "Could not capture snapshot."},
                "_headers": {
                    "Content-Type": "application/json",
                },
            }
        
        # The content returned is already bytes if not from cache
        encoded_content = base64.b64encode(content).decode('utf-8')
        
        return {
            "_shsf": "v2",
            "_code": 200,
            "_res": {
                "message": "done",
                "buffer": encoded_content
            },
            "_headers": {
                "Content-Type": "application/json",
            },
        }
    elif route == "health":
        return {
            "_shsf": "v2",
            "_code": 200,
            "_res": {"message": "Service is healthy."},
            "_headers": {
                "Content-Type": "application/json",
            },
        }
    elif route == "default": # UI Route
        try:
            content = ""
            with open("/app/ui.html", "r") as f:
                content = f.read()
        except FileNotFoundError:
            print("Error: /app/ui.html not found.")
            return {
                "_shsf": "v2",
                "_code": 404,
                "_res": {"message": "UI file not found."},
                "_headers": {
                    "Content-Type": "application/json",
                },
            }
        except Exception as e:
            print(f"Error loading UI: {e}")
            return {
                "_shsf": "v2",
                "_code": 500,
                "_res": {"message": f"Server error: {str(e)}"},
                "_headers": {
                    "Content-Type": "application/json",
                },
            }
        return {
            "_shsf": "v2",
            "_code": 200,
            "_res": content,
            "_headers": {
                "Content-Type": "text/html",
            },
        }
        
if __name__ == "__main__":
    print("omg im on main")
    capture_stream_snapshot("https://www.youtube.com/watch?v=n15V_fCsl_c", output_file="snapshot.jpg")