""" Plugin for providing a command to fetch YouTube video information from links. """ # Importing necessary libraries import re import logging import asyncio from pytubefix import YouTube import simplematrixbotlib as botlib 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_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 description = video.description length = seconds_to_minutes_seconds(video.length) views = video.views author = video.author description_with_breaks = description.replace('\n', '
') info_message = f"""🎬🎝 Title: {title} | Length: {length} minutes | Views: {views}\n
⤵︎Click Here For Description⤵︎{description_with_breaks}
""" 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")