mpv-youtube-queue-server/server.py

#!/usr/bin/env python3
import logging
import os
import socket
import time
import urllib.parse
from http.server import BaseHTTPRequestHandler, HTTPServer

# Configuration
MPV_SOCKET = os.getenv("MPV_SOCKET", "/tmp/mpvsocket")
HOST_NAME = os.getenv("HOST_NAME", "0.0.0.0")
PORT_NUMBER = int(os.getenv("PORT_NUMBER", "8080"))
SOCKET_RETRY_DELAY = 5  # Time in seconds between retries to connect to the socket
MAX_RETRIES = 10  # Maximum number of retries to connect to the socket

# Set up basic logging
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)


def send_to_mpv(command):
    """Send a command to the mpv socket, retrying up to MAX_RETRIES times if the socket is not available."""
    attempts = 0
    while attempts < MAX_RETRIES:
        try:
            with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client_socket:
                client_socket.connect(MPV_SOCKET)
                client_socket.sendall(command.encode("utf-8"))
                logging.info("Command sent to mpv successfully.")
                return True
        except socket.error as e:
            attempts += 1
            logging.error(
                f"Failed to connect to socket (attempt {attempts}/{MAX_RETRIES}): {e}. Retrying in {SOCKET_RETRY_DELAY} seconds..."
            )
            time.sleep(SOCKET_RETRY_DELAY)

    logging.error(f"Exceeded maximum retries ({MAX_RETRIES}). Ignoring the request.")
    return False


class MyHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        # Parse the URL and extract the "url" parameter
        query_components = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
        video_url = query_components.get("url", [None])[0]

        if video_url:
            video_url = urllib.parse.unquote(video_url)  # Decode the URL
            logging.info(f"Received URL: {video_url}")

            # Create the command to send to mpv
            command = f'{{"command": ["script-message", "add_to_youtube_queue", "{video_url}"]}}\n'

            # Try to send the command to mpv
            if send_to_mpv(command):
                self.send_response(200)
                self.end_headers()
                self.wfile.write(b"URL added to mpv queue")
            else:
                self.send_response(500)
                self.end_headers()
                self.wfile.write(b"Failed to add URL to mpv queue after max retries")
        else:
            logging.error("Missing 'url' parameter")
            self.send_response(400)
            self.end_headers()
            self.wfile.write(b"Missing 'url' parameter")

    def log_message(self, format, *args):
        # Override default log_message method to avoid duplicate logging from BaseHTTPRequestHandler
        logging.info(f"{self.address_string()} - {format % args}")


if __name__ == "__main__":
    logging.info(f"Starting server on {HOST_NAME}:{PORT_NUMBER}...")

    try:
        httpd = HTTPServer((HOST_NAME, PORT_NUMBER), MyHandler)
        logging.info(f"Server running on port {PORT_NUMBER}...")
        httpd.serve_forever()
    except Exception as e:
        logging.exception(f"Error occurred: {e}")
    except KeyboardInterrupt:
        logging.info("Server is shutting down...")
        httpd.server_close()
        logging.info("Server stopped.")