stream-discord-bot/bot.py

#!/usr/bin/env python3

import os
from urllib.parse import urlparse, parse_qs
import aiohttp
import discord


DISCORD_TOKEN = os.getenv("DISCORD_TOKEN")
DISCORD_CHANNEL = int(os.getenv("DISCORD_CHANNEL"))
DISCORD_GUILD = int(os.getenv("DISCORD_GUILD"))


class CustomClient(discord.Client):
    async def on_ready(self):
        print(f"{self.user} has connected to Discord!")

    async def on_message(self, message: discord.message.Message):
        if (
            message.channel.id == DISCORD_CHANNEL
            and message.guild.id == DISCORD_GUILD
            and message.author.id != self.user.id
        ):
            url = urlparse(message.content)
            if "youtu" in url.hostname:
                if (
                    url.path.strip("/").startswith("shorts")
                    and len(splitted := url.path.split("/")) >= 2
                ):
                    video_id = splitted[len(splitted) - 1]

                    await message.reply(
                        content=f"https://stream.purser.it/{video_id}?provider=youtube&direct=true"
                    )
                elif video := parse_qs(url.query).get("v"):
                    if isinstance(video, str):
                        await message.reply(
                            content=f"https://stream.purser.it/{video}?provider=youtube&direct=true"
                        )
                    elif isinstance(video, list):
                        await message.reply(
                            content=f"https://stream.purser.it/{video[0]}?provider=youtube&direct=true"
                        )
                else:
                    video_id = url.path.strip("/")
                    await message.reply(
                        content=f"https://stream.purser.it/{video_id}?provider=youtube&direct=true"
                    )

            elif "seafile" in url.hostname:
                stripped = url.path.strip("/")
                async with aiohttp.ClientSession(
                    timeout=aiohttp.ClientTimeout(total=5)
                ) as session:
                    resp = await session.head(
                        f"https://stream.purser.it/{stripped}/?provider=seafile&raw=true",
                        allow_redirects=True,
                    )
                    if (ctype := resp.headers.get("content-type")) and "image" in ctype:
                        await message.reply(
                            content=f"<https://stream.purser.it/{stripped}/?provider=seafile&raw=true>"
                        )
                    else:
                        await message.reply(
                            content=f"https://stream.purser.it/{stripped}/?provider=seafile&raw=false"
                        )

            elif "twitch" in url.hostname:
                stripped = url.path.strip("/")
                await message.reply(
                    content=f"https://stream.purser.it/{stripped}?provider=twitch"
                )

            elif "twitter" in url.hostname:
                stripped = url.path.strip("/")
                await message.reply(
                    content=f"https://stream.purser.it/{stripped}?provider=twitter"
                )

            elif "x.com" in url.hostname:
                stripped = url.path.strip("/")
                await message.reply(
                    content=f"https://stream.purser.it/{stripped}?provider=twitter"
                )


intents = discord.Intents.default()
intents.message_content = True
CustomClient(intents=intents).run(DISCORD_TOKEN)