""" Plugin for providing a command to fetch YouTube video information from links. """ # Importing necessary libraries import re import logging import asyncio import aiohttp from pytubefix import YouTube import simplematrixbotlib as botlib from youtube_title_parse import get_artist_title LYRICIST_API_URL = "https://lyrist.vercel.app/api/{}" def seconds_to_minutes_seconds(seconds): """ Converts seconds to a string representation of minutes and seconds. Args: seconds (int): The number of seconds. Returns: str: A string representation of minutes and seconds in the format MM:SS. """ minutes = seconds // 60 seconds %= 60 return f"{minutes:02d}:{seconds:02d}" async def fetch_lyrics(song, artist): """ Asynchronously fetches lyrics for a song from the Lyricist API. Args: song (str): The name of the song. artist (str): The name of the artist. Returns: str: Lyrics of the song. None if an error occurs during fetching. """ try: async with aiohttp.ClientSession() as session: async with session.get(LYRICIST_API_URL.format(song, artist)) as response: data = await response.json() return data.get("lyrics") except Exception as e: logging.error(f"Error fetching lyrics: {str(e)}") return None async def fetch_youtube_info(youtube_url): """ Asynchronously fetches information about a YouTube video. Args: youtube_url (str): The URL of the YouTube video. Returns: str: A message containing information about the YouTube video. None if an error occurs during fetching. """ try: video = YouTube(youtube_url) title = video.title artist, song = get_artist_title(title) description = video.description length = seconds_to_minutes_seconds(video.length) views = video.views author = video.author description_with_breaks = description.replace('\n', '
') # Fetching lyrics lyrics = await fetch_lyrics(song, artist) lyrics = lyrics.replace('\n', "
") info_message = f"""🎬🎝 Title: {title} | Length: {length} minutes | Views: {views}\n
⤵︎Description⤵︎{description_with_breaks}
""" if lyrics: info_message += f"
🎵 Lyrics:
{lyrics}
" return info_message except Exception as e: logging.error(f"Error fetching YouTube video information: {str(e)}") return None async def handle_command(room, message, bot, prefix, config): """ Asynchronously handles the command to fetch YouTube video information. Args: room (Room): The Matrix room where the command was invoked. message (RoomMessage): The message object containing the command. bot (MatrixBot): The Matrix bot instance. prefix (str): The command prefix. config (dict): The bot's configuration. Returns: None """ match = botlib.MessageMatch(room, message, bot, prefix) if match.is_not_from_this_bot() and re.search(r'youtube\.com/watch\?v=', message.body): logging.info("YouTube link detected") video_id_match = re.search(r'youtube\.com/watch\?v=([^\s]+)', message.body) if video_id_match: video_id = video_id_match.group(1) youtube_url = f"https://www.youtube.com/watch?v={video_id}" logging.info(f"Fetching information for YouTube video: {youtube_url}") retry_count = 3 while retry_count > 0: info_message = await fetch_youtube_info(youtube_url) if info_message: await bot.api.send_markdown_message(room.room_id, info_message) logging.info("Sent YouTube video information to the room") break else: logging.info("Retrying...") retry_count -= 1 await asyncio.sleep(1) # wait for 1 second before retrying else: logging.error("Failed to fetch YouTube video information after retries")