220 lines
7.5 KiB
Python
220 lines
7.5 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
|
||
|
|
from irc3.plugins.command import command
|
||
|
|
import irc3
|
||
|
|
import html
|
||
|
|
import googleapiclient.discovery
|
||
|
|
import re
|
||
|
|
import datetime
|
||
|
|
import shlex
|
||
|
|
|
||
|
|
# Constants for YouTube API
|
||
|
|
API_SERVICE_NAME = "youtube"
|
||
|
|
API_VERSION = "v3"
|
||
|
|
DEVELOPER_KEY = "AIzaSyBNrqOA0ZIziUVLYm0K5W76n9ndqz6zTxI"
|
||
|
|
|
||
|
|
# Initialize YouTube API client
|
||
|
|
youtube = googleapiclient.discovery.build(
|
||
|
|
API_SERVICE_NAME, API_VERSION, developerKey=DEVELOPER_KEY
|
||
|
|
)
|
||
|
|
|
||
|
|
# Regular expression for matching YouTube video URLs
|
||
|
|
YT_URL_REGEX = (
|
||
|
|
r"((?:https?:)?\/\/)?((?:www|m)\.)?((?:youtube(-nocookie)?\.com|youtu\.be))"
|
||
|
|
r"(\/(?:[\w\-]+\?v=|embed\/|live\/|v\/)?)([\w\-]+)(\S+)?"
|
||
|
|
)
|
||
|
|
|
||
|
|
@irc3.plugin
|
||
|
|
class YouTubePlugin:
|
||
|
|
"""
|
||
|
|
An IRC plugin to fetch and display YouTube video information.
|
||
|
|
|
||
|
|
This plugin responds to both command inputs and messages containing YouTube links.
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self, bot):
|
||
|
|
"""
|
||
|
|
Initialize the YouTubePlugin with an IRC bot instance.
|
||
|
|
|
||
|
|
:param bot: IRC bot instance
|
||
|
|
"""
|
||
|
|
self.bot = bot
|
||
|
|
self.yt_reg = re.compile(YT_URL_REGEX)
|
||
|
|
|
||
|
|
def parse_duration(self, duration):
|
||
|
|
"""
|
||
|
|
Parse ISO 8601 duration format into a human-readable string.
|
||
|
|
|
||
|
|
:param duration: ISO 8601 duration string
|
||
|
|
:return: Formatted duration string
|
||
|
|
"""
|
||
|
|
duration = duration[2:] # Remove 'PT' prefix
|
||
|
|
hours = minutes = seconds = 0
|
||
|
|
|
||
|
|
if 'H' in duration:
|
||
|
|
hours, duration = duration.split('H', 1)
|
||
|
|
hours = int(hours)
|
||
|
|
if 'M' in duration:
|
||
|
|
minutes, duration = duration.split('M', 1)
|
||
|
|
minutes = int(minutes)
|
||
|
|
if 'S' in duration:
|
||
|
|
seconds = int(duration.split('S', 1)[0])
|
||
|
|
|
||
|
|
parts = []
|
||
|
|
if hours > 0:
|
||
|
|
parts.extend([str(hours), f"{minutes:02}", f"{seconds:02}"])
|
||
|
|
elif minutes > 0:
|
||
|
|
parts.extend([str(minutes), f"{seconds:02}"])
|
||
|
|
else:
|
||
|
|
parts.append(str(seconds))
|
||
|
|
|
||
|
|
return ":".join(parts)
|
||
|
|
|
||
|
|
def format_number(self, num):
|
||
|
|
"""
|
||
|
|
Format numbers to use K for thousands and M for millions.
|
||
|
|
|
||
|
|
:param num: Integer number to format
|
||
|
|
:return: Formatted string
|
||
|
|
"""
|
||
|
|
if num >= 1_000_000:
|
||
|
|
return f"{num/1_000_000:.1f}M"
|
||
|
|
elif num >= 1_000:
|
||
|
|
return f"{num/1_000:.1f}k"
|
||
|
|
return str(num)
|
||
|
|
|
||
|
|
async def get_video_info(self, video_id):
|
||
|
|
"""
|
||
|
|
Retrieve video information from YouTube API.
|
||
|
|
|
||
|
|
:param video_id: ID of the video
|
||
|
|
:return: Dictionary with video info or None if data couldn't be fetched
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
request = youtube.videos().list(
|
||
|
|
part="contentDetails,statistics,snippet",
|
||
|
|
id=video_id
|
||
|
|
)
|
||
|
|
response = await self.bot.loop.run_in_executor(None, request.execute)
|
||
|
|
if not response.get("items"):
|
||
|
|
return None
|
||
|
|
|
||
|
|
item = response["items"][0]
|
||
|
|
stats = item["statistics"]
|
||
|
|
snippet = item["snippet"]
|
||
|
|
details = item["contentDetails"]
|
||
|
|
|
||
|
|
info = {
|
||
|
|
"title": html.unescape(snippet["title"]).title(),
|
||
|
|
"duration": self.parse_duration(details["duration"]),
|
||
|
|
"views": int(stats.get("viewCount", 0)),
|
||
|
|
"likes": int(stats.get("likeCount", 0)),
|
||
|
|
"comments": int(stats.get("commentCount", 0)),
|
||
|
|
"channel": snippet.get("channelTitle", "Unknown Channel"),
|
||
|
|
"videoId": video_id,
|
||
|
|
}
|
||
|
|
|
||
|
|
if published := snippet.get("publishedAt"):
|
||
|
|
dt = datetime.datetime.fromisoformat(published.replace("Z", "+00:00"))
|
||
|
|
info["date"] = dt.strftime("%b %d, %Y")
|
||
|
|
else:
|
||
|
|
info["date"] = "Unknown Date"
|
||
|
|
|
||
|
|
return info
|
||
|
|
except Exception as e:
|
||
|
|
self.bot.log.error(f"YouTube API error: {e}")
|
||
|
|
return None
|
||
|
|
|
||
|
|
def format_message(self, info):
|
||
|
|
"""
|
||
|
|
Format video information into an attractive mIRC color-coded message.
|
||
|
|
|
||
|
|
:param info: Dictionary containing video information
|
||
|
|
:return: Formatted message string
|
||
|
|
"""
|
||
|
|
return (
|
||
|
|
f"\x0300,04\x0315Youtube\x03 ► \x03\x1F{info['title']}\x0F\x03 | "
|
||
|
|
f"\x0308Duration:\x03 \x0307\x02{info['duration']}\x0F\x03 | "
|
||
|
|
f"\x02Views:\x02 \x0309{self.format_number(info['views'])}\x0F\x03 | "
|
||
|
|
f"\x0306Published:\x03 \x0314{info['date']}\x0F\x03 | "
|
||
|
|
f"\x0312https://youtu.be/{info['videoId']}\x0F\x03"
|
||
|
|
)
|
||
|
|
|
||
|
|
@command(options_first=True, use_shlex=True)
|
||
|
|
async def yt(self, mask, target, args):
|
||
|
|
"""
|
||
|
|
%%yt [--<amount>] <search_query>...
|
||
|
|
|
||
|
|
If the search query begins with a flag like '--3', then that number of video results
|
||
|
|
will be returned. By default only one result is returned.
|
||
|
|
"""
|
||
|
|
tokens = args["<search_query>"]
|
||
|
|
# Default to one result if no flag is provided.
|
||
|
|
amount = 1
|
||
|
|
|
||
|
|
# Separate amount flag from the rest of the tokens
|
||
|
|
new_tokens = []
|
||
|
|
for token in tokens:
|
||
|
|
if token.startswith("--"):
|
||
|
|
try:
|
||
|
|
amount = int(token[2:])
|
||
|
|
except ValueError:
|
||
|
|
self.bot.privmsg(target, "Invalid amount specified. Using default of 1 result.")
|
||
|
|
else:
|
||
|
|
new_tokens.append(token)
|
||
|
|
|
||
|
|
query = " ".join(new_tokens)
|
||
|
|
if not query:
|
||
|
|
self.bot.privmsg(target, "You must specify a search query.")
|
||
|
|
return
|
||
|
|
|
||
|
|
try:
|
||
|
|
request = youtube.search().list(
|
||
|
|
part="id,snippet",
|
||
|
|
q=query,
|
||
|
|
type="video",
|
||
|
|
maxResults=amount
|
||
|
|
)
|
||
|
|
result = await self.bot.loop.run_in_executor(None, request.execute)
|
||
|
|
if not result.get("items"):
|
||
|
|
self.bot.privmsg(target, "No results found.")
|
||
|
|
return
|
||
|
|
|
||
|
|
# If more than one result is requested, show a header.
|
||
|
|
if amount > 1:
|
||
|
|
self.bot.privmsg(target, f"\x02Search Results For:\x03 \x0311{query}\x03")
|
||
|
|
self.bot.privmsg(target, "━" * 99)
|
||
|
|
|
||
|
|
count = 0
|
||
|
|
for item in result["items"]:
|
||
|
|
count += 1
|
||
|
|
video_id = item["id"]["videoId"]
|
||
|
|
if info := await self.get_video_info(video_id):
|
||
|
|
# Only number the results if more than one is returned.
|
||
|
|
if amount == 1:
|
||
|
|
msg = self.format_message(info)
|
||
|
|
else:
|
||
|
|
msg = f" {count}. {self.format_message(info)}"
|
||
|
|
self.bot.privmsg(target, msg)
|
||
|
|
if amount > 1:
|
||
|
|
self.bot.privmsg(target, "━" * 99)
|
||
|
|
except Exception as e:
|
||
|
|
self.bot.privmsg(target, f"Search error: {e}")
|
||
|
|
|
||
|
|
@irc3.event(irc3.rfc.PRIVMSG)
|
||
|
|
async def on_message(self, mask, event, target, data):
|
||
|
|
"""
|
||
|
|
Event handler for messages in the channel to detect and respond to YouTube links.
|
||
|
|
|
||
|
|
:param mask: Mask of the user who sent the message
|
||
|
|
:param event: Event object from IRC
|
||
|
|
:param target: Target channel or user where the message was sent
|
||
|
|
:param data: Content of the message
|
||
|
|
"""
|
||
|
|
if mask.nick == self.bot.nick:
|
||
|
|
return
|
||
|
|
|
||
|
|
for match in self.yt_reg.finditer(data):
|
||
|
|
video_id = match.group(6)
|
||
|
|
if info := await self.get_video_info(video_id):
|
||
|
|
self.bot.privmsg(target, self.format_message(info))
|