""" Plugin for providing a command to fetch YouTube video information from links. """ # Importing necessary libraries import re import logging import asyncio import aiohttp import yt_dlp 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: url = LYRICIST_API_URL.format(artist, song) logging.info(f"Fetching lyrics from: {url}") async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response: if response.status == 200: data = await response.json() return data.get("lyrics") else: logging.warning(f"Lyrics API returned status {response.status}") return None except asyncio.TimeoutError: logging.error("Timeout fetching lyrics") return None 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 using yt-dlp. 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: logging.info(f"Fetching YouTube info for: {youtube_url}") # Configure yt-dlp options ydl_opts = { 'quiet': True, 'no_warnings': True, 'extract_flat': False, 'skip_download': True, } # Run yt-dlp in thread pool to avoid blocking loop = asyncio.get_event_loop() def extract_info(): with yt_dlp.YoutubeDL(ydl_opts) as ydl: return ydl.extract_info(youtube_url, download=False) info = await loop.run_in_executor(None, extract_info) if not info: logging.error("No info returned from yt-dlp") return None # Extract video information title = info.get('title', 'Unknown Title') description = info.get('description', 'No description available') duration = info.get('duration', 0) view_count = info.get('view_count', 0) uploader = info.get('uploader', 'Unknown') logging.info(f"Video title: {title}") length = seconds_to_minutes_seconds(duration) # Parse artist and song from title artist, song = get_artist_title(title) logging.info(f"Parsed artist: {artist}, song: {song}") # Limit description length to avoid huge messages if len(description) > 500: description = description[:500] + "..." description_with_breaks = description.replace('\n', '
') # Build basic info message info_message = f"""🎬🎝 Title: {title}
Length: {length} | Views: {view_count:,} | Uploader: {uploader}
⤵︎Description⤵︎{description_with_breaks}
""" # Try to fetch lyrics if artist and song were parsed if artist and song: logging.info("Attempting to fetch lyrics...") lyrics = await fetch_lyrics(song, artist) if lyrics: lyrics = lyrics.replace('\n', "
") # Limit lyrics length if len(lyrics) > 3000: lyrics = lyrics[:3000] + "
...(truncated)" info_message += f"
🎵 Lyrics:
{lyrics}
" else: logging.info("No lyrics found") else: logging.info("Could not parse artist/song from title, skipping lyrics") return info_message except Exception as e: logging.error(f"Error fetching YouTube video information: {str(e)}", exc_info=True) 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) # Check if message contains a YouTube link if match.is_not_from_this_bot() and re.search(r'(youtube\.com/watch\?v=|youtu\.be/)', message.body): logging.info(f"YouTube link detected in message: {message.body}") # Match both youtube.com and youtu.be formats video_id_match = re.search(r'(?:youtube\.com/watch\?v=|youtu\.be/)([a-zA-Z0-9_-]{11})', 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 ID: {video_id}") retry_count = 2 # Reduced retries since yt-dlp is more reliable 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.warning(f"Failed to fetch info, retrying... ({retry_count-1} attempts left)") retry_count -= 1 if retry_count > 0: await asyncio.sleep(2) # wait for 2 seconds before retrying else: logging.error("Failed to fetch YouTube video information after all retries") await bot.api.send_text_message(room.room_id, "Failed to fetch YouTube video information. The video may be unavailable or age-restricted.") else: logging.warning("Could not extract video ID from YouTube URL")