116 lines
4.8 KiB
Python
116 lines
4.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
IRC3 Bot Plugin: URL Title Fetcher
|
|
|
|
A plugin for IRC3 bots that monitors chat messages for URLs, fetches their webpage titles, and displays them
|
|
with formatted styling in the chat. Provides visual enhancement to URL sharing in IRC channels.
|
|
|
|
Features:
|
|
- Asynchronous URL processing using aiohttp for efficient network operations
|
|
- Robust HTML parsing with lxml for accurate title extraction
|
|
- Configurable message styling with color and formatting options
|
|
- Built-in exclusion of YouTube URLs to avoid conflicts with dedicated YouTube plugins
|
|
- Error handling for network and parsing operations
|
|
- Proper resource cleanup through session management
|
|
- Queue-based processing system for better flow control
|
|
|
|
Dependencies:
|
|
- aiohttp: For asynchronous HTTP requests
|
|
- irc3: Core IRC bot functionality
|
|
- ircstyle: IRC text formatting utilities
|
|
- lxml: HTML parsing capabilities
|
|
|
|
Author: Zodiac
|
|
Date: 2025-02-14
|
|
"""
|
|
|
|
import re
|
|
import aiohttp
|
|
import ircstyle
|
|
from lxml import html
|
|
import irc3
|
|
from irc3 import event
|
|
from irc3.compat import Queue
|
|
from plugins.services.permissions import check_ignore
|
|
|
|
|
|
@irc3.plugin
|
|
class URLTitlePlugin:
|
|
"""Plugin for fetching and displaying webpage titles from URLs shared in IRC messages.
|
|
|
|
Monitors IRC messages for URLs, retrieves their webpage titles, and posts formatted responses
|
|
back to the channel. Supports styled text output with configurable formatting options.
|
|
|
|
Attributes:
|
|
bot (irc3.IrcBot): Reference to the main IRC bot instance
|
|
session (aiohttp.ClientSession): Persistent HTTP session for making web requests
|
|
url_pattern (re.Pattern): Compiled regex for URL detection in messages
|
|
queue (Queue): Processing queue for URL handling tasks
|
|
"""
|
|
|
|
def __init__(self, bot):
|
|
"""Initialize plugin with bot instance and set up components."""
|
|
self.bot = bot
|
|
self.session = aiohttp.ClientSession(loop=self.bot.loop)
|
|
self.url_pattern = re.compile(r"https?://[^\s<>\"']+|www\.[^\s<>\"']+")
|
|
self.queue = Queue()
|
|
self.bot.create_task(self.process_queue())
|
|
|
|
@event(irc3.rfc.PRIVMSG)
|
|
@check_ignore
|
|
async def on_privmsg(self, mask, event, target, data):
|
|
"""Handle incoming messages and enqueue URLs for processing."""
|
|
urls = self.url_pattern.findall(data)
|
|
|
|
for url in urls:
|
|
if "youtube.com" in url.lower() or "youtu.be" in url.lower():
|
|
continue
|
|
self.queue.put_nowait((target, url))
|
|
|
|
async def process_queue(self):
|
|
"""Process URLs from the queue asynchronously."""
|
|
while True:
|
|
target, url = await self.queue.get()
|
|
try:
|
|
title = await self.fetch_title(url)
|
|
if title:
|
|
formatted_message = self.format_message(title, url)
|
|
await self.bot.privmsg(target, formatted_message)
|
|
except Exception as e:
|
|
self.bot.log.error(f"Error processing URL {url}: {e}")
|
|
finally:
|
|
self.queue.task_done()
|
|
|
|
def format_message(self, title, url):
|
|
"""Create a styled IRC message containing the webpage title and source URL."""
|
|
prefix = ircstyle.style("►", fg="cyan", bold=True, reset=True)
|
|
title_label = ircstyle.style("Title", fg="blue", bold=True, reset=True)
|
|
title_text = ircstyle.style(title, fg="green", italics=True, underline=True, reset=True)
|
|
separator = ircstyle.style("❘", fg="grey", bold=True, reset=True)
|
|
url_label = ircstyle.style("Source", fg="blue", bold=True, underline=True, reset=True)
|
|
url_text = ircstyle.style(url, fg="cyan", italics=True, reset=True)
|
|
suffix = ircstyle.style("◄", fg="cyan", bold=True, reset=True)
|
|
|
|
return f"{prefix} {title_label}: {title_text} {separator} {url_label}: {url_text} {suffix}"
|
|
|
|
async def fetch_title(self, url):
|
|
"""Retrieve the title of a webpage using asynchronous HTTP requests."""
|
|
headers = {
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
|
|
}
|
|
|
|
async with self.session.get(url, headers=headers, timeout=10) as response:
|
|
response.raise_for_status()
|
|
content = await response.text()
|
|
tree = html.fromstring(content)
|
|
title = tree.findtext(".//title")
|
|
return title.strip() if title else "No title found"
|
|
|
|
async def close(self):
|
|
"""Clean up resources by closing the HTTP session."""
|
|
await self.session.close()
|
|
|
|
def __del__(self):
|
|
"""Ensure proper cleanup when the plugin is destroyed."""
|
|
self.bot.create_task(self.close()) |