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 time import sleep
import disnake
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

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 = []
load_dotenv()
options = Options()
options.profile = webdriver.FirefoxProfile(os.getenv("PROFILE_PATH"))
driver = webdriver.Firefox(options=options)
intents = disnake.Intents.default()
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 just fail but it establishes checks

def play_video(videourl):
    driver.get(videourl)
    sleep(4)
    try:
        elem = driver.find_element(By.CLASS_NAME, "ytp-fullscreen-button")
        elem.send_keys(Keys.RETURN)
    except Exception:
        return #if this errors there is no fullscreen options, such as playlists, so skip the link
    #guess I don't need this
    #sleep(1.5)
    #elem = driver.find_element(By.XPATH, '//button[@aria-keyshortcuts="k"]')
    #elem.send_keys(Keys.RETURN)
    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 Exception:
        pass
    sleep(10)
    try:
        elem = driver.find_element(By.XPATH, '//button[@aria-label="Dismiss"]')
        elem.send_keys(Keys.RETURN)
    except Exception:
        pass
    try:
        endscreen = driver.find_element(By.CLASS_NAME, "html5-endscreen")
    except Exception:
        return #same as above
    while str(endscreen.get_attribute('style')) == "display: none;":
        pass
    sleep(2)
    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 urlparse(link).netloc == 'youtube.com' or urlparse(link).netloc == 'www.youtube.com' or urlparse(link).netloc == 'youtu.be':
        queue.append(link)
        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="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")
    message = ""
    for idx, item in enumerate(queue):
        if idx == 0:
            message = message + f"Now playing: <{item}>\n"
            continue
        message = message + f"{idx}. <{item}>\n"
    await inter.edit_original_response(message)

@bot.slash_command(
    name="toggleplayback",
    description="pauses or unpauses the video",
)
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 DON'T SKIP LESS THAN 18 SECONDS INTO A VIDEO",
)
async def skip(inter: disnake.AppCmdInter):
    await inter.response.defer(ephemeral=True)
    global queuetask
    queuetask.cancel()
    try:
        await queuetask
    except asyncio.CancelledError:
        queue.pop(0)
        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("skipepd")

@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)
        await inter.edit_original_response("removed!")    

async def queuehandler():
    loop = asyncio.get_running_loop()
    while queue:
        driver.maximize_window()
        await loop.run_in_executor(None, play_video, queue[0])
        queue.pop(0)
    driver.get(f"file://{os.getcwd()}/waitingforvideo.png")
    driver.fullscreen_window()

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