mpv-youtube-queue-server/server.py

#!/usr/bin/env python3
import logging
import os
import socket
import time
from urllib import parse

from flask import Flask, jsonify, request
from sqlalchemy import Column, DateTime, Integer, String, create_engine, exc
from sqlalchemy.orm import declarative_base, sessionmaker
from sqlalchemy.sql import func

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

# Flask app
app = Flask(__name__)

# Flask logging configuration to use the same logger as the rest of the app
app.logger.handlers = logging.getLogger().handlers
app.logger.setLevel(logging.getLogger().level)

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

# Configuration from environment variables
LISTEN_ADDRESS = os.getenv("LISTEN_ADDRESS", "0.0.0.0")
LISTEN_PORT = int(os.getenv("LISTEN_PORT", "8080"))
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./mpv.db")
MPV_SOCKET: str = os.getenv("MPV_SOCKET", "/tmp/mpvsocket")
LOGLEVEL = os.getenv("LOGLEVEL", "INFO").strip().upper()

if LOGLEVEL == "DEBUG":
    logging.getLogger().setLevel(logging.DEBUG)
elif LOGLEVEL == "WARNING":
    logging.getLogger().setLevel(logging.WARNING)
elif LOGLEVEL == "ERROR":
    logging.getLogger().setLevel(logging.ERROR)
else:
    logging.getLogger().setLevel(logging.INFO)

# Set up SQLAlchemy
Base = declarative_base()
engine = create_engine(DATABASE_URL)
Session = sessionmaker(bind=engine)
session = Session()


class WatchHistory(Base):
    __tablename__ = "watch_history"

    whid = Column(Integer, primary_key=True, autoincrement=True)
    video_url = Column(String(255), nullable=False)
    video_name = Column(String(255), nullable=False)
    channel_url = Column(String(255), nullable=False)
    channel_name = Column(String(255), nullable=False)
    watch_date = Column(DateTime, nullable=False, server_default=func.now())
    created_by = Column(
        String(100), nullable=False, server_default="mpv-youtube-queue-server"
    )


class SavedQueue(Base):
    __tablename__ = "saved_queue"
    sqid = Column(Integer, primary_key=True, autoincrement=True)
    video_url = Column(String(255), nullable=False)
    created_date = Column(DateTime, nullable=False, server_default=func.now())
    created_by = Column(
        String(100), nullable=False, server_default="mpv-youtube-queue-server"
    )


# Ensure tables exist
Base.metadata.create_all(engine)


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


@app.route("/save_queue", methods=["POST"])
def save_queue():
    data = request.get_json()
    if data is None or "urls" not in data:
        logging.error("Invalid JSON data")
        return jsonify(message="Invalid JSON data"), 400
    logging.debug(f"Received data: {data}")
    urls = data.get("urls")
    logging.debug("Truncating saved queue")
    session.query(SavedQueue).delete()
    for url in urls:
        logging.debug(f"Adding {url} to the saved queue")
        new_entry = SavedQueue(video_url=url)
        try:
            session.add(new_entry)
        except exc.SQLAlchemyError as e:
            logging.error(f"Failed to insert data into database: {e}")
            return jsonify(message="Failed to add data to database"), 500
    session.commit()
    return jsonify(message="Data added to saved queue"), 200


@app.route("/load_queue", methods=["GET"])
def load_queue():
    logging.debug("Loading saved queue")
    urls = [entry.video_url for entry in session.query(SavedQueue).all()]
    logging.debug(f"Loaded {len(urls)} URLs from the saved queue")
    return jsonify(urls), 200


@app.route("/add_video", methods=["POST"])
def add_video():
    data = request.get_json()

    if data:
        video_url: str = data.get("video_url")
        video_name: str = data.get("video_name")
        channel_url: str = data.get("channel_url")
        channel_name: str = data.get("channel_name")

        if video_url and video_name and channel_url and channel_name:
            logging.debug(f"Received data: {data}")

            try:
                new_entry = WatchHistory(
                    video_url=video_url,
                    video_name=video_name,
                    channel_url=channel_url,
                    channel_name=channel_name,
                )
                session.add(new_entry)
                session.commit()
                logging.debug(
                    f"{video_name} by {channel_name} inserted into the database successfully"
                )
                return jsonify(message="Data added to mpv queue and database"), 200
            except exc.SQLAlchemyError as e:
                session.rollback()
                logging.error(f"Failed to insert data into database: {e}")
                return jsonify(message="Failed to add data to database"), 500
        else:
            logging.error("Missing required data fields")
            return jsonify(message="Missing required data fields"), 400
    else:
        logging.error("Invalid JSON data")
        return jsonify(message="Invalid JSON data"), 400


@app.route("/", methods=["GET"])
def handle_request():
    video_url = request.args.get("url")
    if video_url:
        video_url = 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):
            return "URL added to mpv queue", 200
        else:
            return "Failed to add URL to mpv queue after max retries", 500
    else:
        logging.error("Missing 'url' parameter")
        return "Missing 'url' parameter", 400


if __name__ == "__main__":
    logging.info(f"Starting server on {LISTEN_ADDRESS}:{LISTEN_PORT}...")
    try:
        app.run(host=LISTEN_ADDRESS, port=LISTEN_PORT)
    except Exception as e:
        logging.exception(f"Error occurred: {e}")
    except KeyboardInterrupt:
        logging.info("Server is shutting down...")
        logging.info("Server stopped.")