""" youtube - A maubot plugin to display YouTube video information Copyright (C) 2025 L. Bradley LaBoon This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License version 3 as published by the Free Software Foundation. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. You should have received a copy of the GNU Affero General Public License along with this program. If not, see . """ from maubot import Plugin, MessageEvent from maubot.handlers import command from mautrix.types import TextMessageEventContent, MessageType, Format from mautrix.util.config import BaseProxyConfig, ConfigUpdateHelper from typing import Type from yarl import URL class Config(BaseProxyConfig): def do_update(self, helper: ConfigUpdateHelper) -> None: helper.copy("api_key") class YouTube(Plugin): async def start(self) -> None: self.config.load_and_update() @classmethod def get_config_class(cls) -> Type[BaseProxyConfig]: return Config # Fetch an image URL, upload it to the matrix media store, and return the mcx:// address async def upload_img(self, link: str) -> str: image_req = await self.http.get(link) if (image_req.status < 400): image = await image_req.read() mxc = await self.client.upload_media(image, filename=image_req.url.name) return mxc return "" # Main !youtube command handler @command.new(name="youtube", aliases=["yt"], help="Display the details of a YouTube video") @command.argument("video", required=True, pass_raw=True) async def youtube(self, evt: MessageEvent, video: str) -> None: if (len(video) < 1): await evt.respond("You forgot to give me a video ID or URL") return # Handle URLs offset = "" if "https://" in video: video_url = URL(video) video_params = video_url.query if (video_url.host == "youtube.com" or video_url.host == "www.youtube.com"): if "shorts" in video_url.parts: video_id = video_url.name else: if "v" in video_params: video_id = video_params["v"] else: await evt.respond("Sorry, I couldn't find the video ID in the URL you gave me.") return if "t" in video_params: offset = f"&t={video_params['t']}" elif (video_url.host == "youtu.be"): video_id = video_url.name if "t" in video_params: offset = f"&t={video_params['t']}" else: await evt.respond("Sorry, I don't know how to parse that URL.") return else: video_id = video # Fetch video info params = { "key": self.config["api_key"], "part": "snippet", "id": video_id } async with self.http.get("https://www.googleapis.com/youtube/v3/videos", params=params) as response: videos = await response.json() if "error" in videos: self.log.error(f"GET {str(response.url)}: {response.status} {videos['error']['message']}") await evt.respond(f"Error: {videos['error']['message']}") return elif response.status >= 400: self.log.error(f"GET {str(response.url)}: {response.status}") await evt.respond("I ran into an issue fetching that video. Check the logs for more details.") return if (len(videos["items"]) < 1): await evt.respond("Sorry, I couldn't find a video with that ID.") return video = videos["items"][0] # Upload thumbnail mxc = await self.upload_img(video["snippet"]["thumbnails"]["maxres"]["url"]) if (len(mxc) > 0): img_html = f"" else: img_html = "" # Construct and send response content = TextMessageEventContent( msgtype=MessageType.NOTICE, format=Format.HTML, body=f"> YouTube\n> [{video['snippet']['channelTitle']}](https://www.youtube.com/channel/{video['snippet']['channelId']})\n> **[{video['snippet']['title']}](https://www.youtube.com/watch?v={video['id']}{offset})**", formatted_body=f"
YouTube

{video['snippet']['channelTitle']}

{video['snippet']['title']}

{img_html}
" ) await evt.respond(content)