57 lines
2.3 KiB
Python
57 lines
2.3 KiB
Python
"""
|
||
This plugin provides a command to fetch YouTube video information from links.
|
||
"""
|
||
|
||
# plugins/youtube.py
|
||
|
||
import re
|
||
import logging
|
||
from pytubefix import YouTube
|
||
import simplematrixbotlib as botlib
|
||
import asyncio
|
||
|
||
def seconds_to_minutes_seconds(seconds):
|
||
minutes = seconds // 60
|
||
seconds %= 60
|
||
return f"{minutes:02d}:{seconds:02d}"
|
||
|
||
async def fetch_youtube_info(youtube_url):
|
||
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', '<br>')
|
||
info_message = f"""<strong>🎬🎝 Title:</strong> {title} | <strong>Length</strong>: {length} minutes | <strong>Views</strong>: {views}\n<details><summary><strong>⤵︎Click Here For Description⤵︎</strong></summary>{description_with_breaks}</details>"""
|
||
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):
|
||
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")
|
||
|
||
|