discord-yt-remote/bot.py

from selenium.webdriver.firefox.options import Options
from selenium import webdriver
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.by import By
from selenium.webdriver.support.wait import WebDriverWait
from selenium.common.exceptions import NoSuchElementException, ElementNotVisibleException, TimeoutException
from time import sleep
import time
import disnake
import random
from urllib.parse import urlparse
from disnake.ext import commands
from disnake import TextInputStyle
import asyncio
from dotenv import load_dotenv
import os
import logging
import math

logger = logging.getLogger('disnake')
logger.setLevel(logging.DEBUG)
handler = logging.FileHandler(filename='disnake.log', encoding='utf-8', mode='w')
handler.setFormatter(logging.Formatter('%(asctime)s:%(levelname)s:%(name)s: %(message)s'))
logger.addHandler(handler)

queue = []
user_queue = []
skip_list = []
shuffle = False
load_dotenv()
options = Options()
options.profile = webdriver.FirefoxProfile(os.getenv("PROFILE_PATH"))
driver = webdriver.Firefox(options=options)
intents = disnake.Intents.all()
intents.message_content = False
bot = commands.Bot(intents=intents, command_prefix=".", test_guilds=[int(os.getenv("GUILD_ID"))])

@bot.event
async def on_ready():
    global queuetask
    queuetask = asyncio.create_task(queuehandler()) #this will set the waiting for videos image, then exit

def play_video(videourl):
    driver.get(videourl)
    try:
        elem = WebDriverWait(driver, 8, 0.2, None).until(lambda x: x.find_element(By.CLASS_NAME, "ytp-fullscreen-button")) 
        sleep(1)
        elem.send_keys(Keys.RETURN)
    except (NoSuchElementException,TimeoutException) as e:
        print(e)
        return #if this errors there is no fullscreen options, such as playlists, so skip the link
    try:
        elem = driver.find_element(By.XPATH, '//button[@data-tooltip-target-id="ytp-autonav-toggle-button"][@aria-label="Autoplay is on"]')
        elem.send_keys(Keys.RETURN)
    except (NoSuchElementException,ElementNotVisibleException,TimeoutException):
        pass
    try:
        elem = WebDriverWait(driver, 5, 0.2, (NoSuchElementException,ElementNotVisibleException)).until(lambda x: x.find_element(By.XPATH, '//button[@aria-label="Dismiss"]'))
        elem.send_keys(Keys.RETURN)
    except (NoSuchElementException,ElementNotVisibleException,TimeoutException) as e:
        print(e)
    try:
        elem = WebDriverWait(driver, (float(os.getenv("MAX_MIN")) * 60), 1, None).until(lambda x: x.find_element(By.CLASS_NAME, "html5-endscreen").is_displayed())
    except (NoSuchElementException,ElementNotVisibleException,TimeoutException) as e:
        print(e)
        sleep(1)
        return #same as above
    sleep(1)
    return

@bot.slash_command(
    name="play",
    description="adds a video to the queue",
)
async def play(inter: disnake.AppCmdInter, link: str):
    await inter.response.defer(ephemeral=True)
    if user_queue.count(inter.user.id) >= (int(os.getenv("MAX_QUEUE"))):
        await inter.edit_original_response(f"You have reached the queue limit of {os.getenv('MAX_QUEUE')}, try again after one of your videos has played.")
        return
    if urlparse(link).netloc == 'youtube.com' or urlparse(link).netloc == 'www.youtube.com' or urlparse(link).netloc == 'youtu.be':
        queue.append(link)
        user_queue.append(inter.user.id)
        await inter.edit_original_response(f"added to queue!")
        global queuetask
        if queuetask.done():
            queuetask = asyncio.create_task(queuehandler())
        return
    else:
        await inter.edit_original_response(f"This bot only accepts youtube links")
        return

@bot.slash_command(
    name="shuffle",
    description="toggles shuffle on or off, the queue cannot be unshuffled once it is shuffled",
)
async def shuffleplay(inter: disnake.AppCmdInter, toggle: str = commands.Param(choices=["on", "off"])):
    await inter.response.defer(ephemeral=True)
    global shuffle
    if toggle == "on":
        shuffle = True
        await inter.edit_original_response(f"shuffle enabled")
        return
    else:
        shuffle = False
        await inter.edit_original_response(f"shuffle disabled")
        return

@bot.slash_command(
    name="queue",
    description="list the videos in queue",
)
async def getqueue(inter: disnake.AppCmdInter):
    await inter.response.defer(ephemeral=True)
    if not queue:
        await inter.edit_original_response("There are no items in queue")
        return
    message = f"Now Playing: <{queue[0]}>\n"
    message = message + f"Shuffle is currently " + ("off\n" if not shuffle else "on!\n")
    for i in range(11):
        if i == 0:
            continue
        try:
            message = message + f"{i}. <{queue[i]}>\n"
        except IndexError:
            break
    message = message + f"1 of {math.ceil((len(queue)-1)/10) if (len(queue)-1)/10 > 1 else 1}"
    if math.ceil((len(queue)-1)/10) > 1:
        await inter.edit_original_response(message, components=[disnake.ui.Button(label=">>", style=disnake.ButtonStyle.primary, custom_id="Forward"),])
    else: 
        await inter.edit_original_response(message)

@bot.listen("on_button_click")
async def button_listener(inter: disnake.MessageInteraction):
    if not queue:
        await inter.response.edit_message("There are no items in queue")
        return
    ogmsg = inter.message.content
    page = ogmsg.split("\n")
    page = page[-1].split(" of ")
    if inter.component.custom_id == "Forward":
        message = f"Now Playing: <{queue[0]}>\n"
        message = message + f"Shuffle is currently " + ("off\n" if not shuffle else "on!\n")
        offset = int(int(page[0]) * 10)
        for i in range(11):
            if i == 0:
                continue
            try:
                message = message + f"{int(i+offset)}. <{queue[int(i+offset)]}>\n"
            except IndexError:
                break
        message = message + f"{int(page[0])+1} of {math.ceil((len(queue)-1)/10) if (len(queue)-1)/10 > 1 else 1}"
        if (int(page[0])+1) >= int(page[1]):
            await inter.response.edit_message(message, components=[disnake.ui.Button(label="<<", style=disnake.ButtonStyle.primary, custom_id="Backward"),])
        else:
            await inter.response.edit_message(message, components=[disnake.ui.Button(label="<<", style=disnake.ButtonStyle.primary, custom_id="Backward"),
            disnake.ui.Button(label=">>", style=disnake.ButtonStyle.primary, custom_id="Forward"),])
        return
    if inter.component.custom_id == "Backward":
        message = f"Now Playing: <{queue[0]}>\n"
        message = message + f"Shuffle is currently " + ("off\n" if not shuffle else "on!\n")
        offset = int((int(page[0]) - 2) * 10)
        for i in range(11):
            if i == 0:
                continue
            try:
                message = message + f"{int(i+offset)}. <{queue[int(i+offset)]}>\n"
            except IndexError:
                break
        message = message + f"{int(page[0])-1} of {math.ceil((len(queue)-1)/10) if (len(queue)-1)/10 > 1 else 1}"
        if (int(page[0])-1) <= 1 and int(int(page[1]) == 1):
            await inter.response.edit_message(message)
        elif (int(page[0])-1) <= 1 and int(int(page[1]) > 1):
            await inter.response.edit_message(message, components=[disnake.ui.Button(label=">>", style=disnake.ButtonStyle.primary, custom_id="Forward"),])
        else:
            await inter.response.edit_message(message, components=[disnake.ui.Button(label="<<", style=disnake.ButtonStyle.primary, custom_id="Backward"),
            disnake.ui.Button(label=">>", style=disnake.ButtonStyle.primary, custom_id="Forward"),])
        await inter.response.edit_message(message, components=[])
        return

@bot.slash_command(
    name="toggleplayback",
    description="pauses or unpauses the video, does not bypass the video timelimit",
)
async def toggleplayback(inter: disnake.AppCmdInter):
    await inter.response.defer(ephemeral=True)
    try:
        elem = driver.find_element(By.XPATH, '//button[@aria-keyshortcuts="k"]')
        elem.send_keys(Keys.RETURN)
    except Exception:
        pass
    await inter.edit_original_response("toggled!")

@bot.slash_command(
    name="skip",
    description="skips the current video",
    default_member_permissions=disnake.Permissions(8192),
)
async def skip(inter: disnake.AppCmdInter):
    await inter.response.defer(ephemeral=False)
    global queuetask
    queuetask.cancel()
    try:
        await queuetask
    except asyncio.CancelledError:
        queue.pop(0)
        user_queue.pop(0)
        skip_list.clear()
        if len(queue) < 1:
            driver.get(f"file://{os.getcwd()}/waitingforvideo.png")
            driver.fullscreen_window()
        else:
            queuetask = asyncio.create_task(queuehandler())
        await inter.edit_original_response("skipped")

@bot.slash_command(
    name="voteskip",
    description="vote to skip the current video",
)
async def voteskip(inter: disnake.AppCmdInter):
    await inter.response.defer(ephemeral=False)
    if not inter.user.voice:
        await inter.edit_original_response("You are not in the voice channel")
        return
    if inter.user.id in skip_list:
        await inter.edit_original_response("You have already voted to skip this video")
        return
    vc = inter.user.voice.channel.members #this could be better due to potential for abuse, but it's fine for now
    print(inter.user.voice.channel.name)
    broadcaster = False
    for m in vc:
        if m.voice.self_video or m.voice.self_stream:
            broadcaster = True
            break
    if not broadcaster:
        await inter.edit_original_response("No one is playing video so how can this be the correct vc?")
        return
    skip_list.append(inter.user.id)
    print(len(skip_list))
    print(math.floor(len(vc)/2))
    print(vc)
    if len(skip_list) >= (math.floor(len(vc)/2)):
        global queuetask
        queuetask.cancel()
        try:
            await queuetask
        except asyncio.CancelledError:
            queue.pop(0)
            user_queue.pop(0)
            skip_list.clear()
            if len(queue) < 1:
                driver.get(f"file://{os.getcwd()}/waitingforvideo.png")
                driver.fullscreen_window()
            else:
                queuetask = asyncio.create_task(queuehandler())
            await inter.edit_original_response("skipped as sufficient votes were reached")
    else:
        await inter.edit_original_response(f"**{inter.user.display_name}** has voted to skip the video, {len(skip_list)}/{math.floor(len(vc)/2)}")


@bot.slash_command(
    name="remove",
    description="removes a video from the queue",
)
async def remove(inter: disnake.AppCmdInter, toremove: int):
    await inter.response.defer(ephemeral=True)
    if toremove == 0:
        await inter.edit_original_response("that is the currently playing video!", ephemeral=True)
        return
    else:
        queue.pop(toremove)
        user_queue.pop(toremove)
        await inter.edit_original_response("removed!")    

async def queuehandler():
    loop = asyncio.get_running_loop()
    global queue
    while queue:
        if shuffle:
            seed = time.time()
            random.seed(seed)
            random.shuffle(queue)
            random.seed(seed)
            random.shuffle(user_queue)
        print(queue)
        print(user_queue)
        await loop.run_in_executor(None, play_video, queue[0])
        queue.pop(0)
        user_queue.pop(0)
        skip_list.clear()
    driver.get(f"file://{os.getcwd()}/waitingforvideo.png")
    driver.fullscreen_window()

bot.run(os.getenv("TOKEN"))