g1mp/plugins/url_title_sniffer.py

# -*- 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 with strict rate limiting

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 time
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
        last_processed (float): Timestamp of last successful URL processing
    """

    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.last_processed = 0  # Initialize to epoch start
        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 with strict 5-second cooldown between requests."""
        while True:
            target, url = await self.queue.get()
            try:
                current_time = time.time()
                elapsed = current_time - self.last_processed

                if elapsed < 5:
                    self.bot.log.info(f"Rate limited: Waiting {5 - elapsed:.1f}s to process {url}")
                    continue

                title = await self.fetch_title(url)
                if title:
                    formatted_message = self.format_message(title, url)
                    await self.bot.privmsg(target, formatted_message)
                
                self.last_processed = time.time()  # Update after successful processing

            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())