CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-irc

IRC (Internet Relay Chat) protocol library for Python

Pending
Overview
Eval results
Files

event-system.mddocs/

Event System

Comprehensive event handling system supporting all IRC protocol events, custom handlers with priorities, and extensible event processing for advanced IRC applications. The event system is the core of the IRC library's architecture.

Capabilities

Event Class

Represents IRC events with structured data including source, target, arguments, and IRCv3 message tags.

class Event:
    def __init__(self, type: str, source, target, arguments: list = None, tags: list = None):
        """
        Initialize IRC event.
        
        Parameters:
        - type: str, event type (e.g., "pubmsg", "join", "privmsg")
        - source: NickMask or str, event source (user who triggered event)
        - target: str, event target (channel or user)
        - arguments: list, event-specific arguments
        - tags: list, IRCv3 message tags
        """
    
    @property
    def type(self) -> str:
        """Event type identifier."""
    
    @property
    def source(self):
        """Event source (NickMask or server name)."""
    
    @property
    def target(self) -> str:
        """Event target (channel, user, or server)."""
    
    @property
    def arguments(self) -> list:
        """List of event-specific arguments."""
    
    @property
    def tags(self) -> list:
        """IRCv3 message tags list."""

Event Handler Management

Functions for registering and managing event handlers with priority support.

def add_global_handler(event: str, handler, priority: int = 0):
    """
    Add global event handler for all connections.
    
    Parameters:
    - event: str, event type to handle
    - handler: callable, handler function with signature (connection, event)
    - priority: int, handler priority (lower numbers = higher priority)
    """

def remove_global_handler(event: str, handler):
    """
    Remove global event handler.
    
    Parameters:
    - event: str, event type
    - handler: callable, handler function to remove
    """

class PrioritizedHandler:
    """Handler with priority for event processing ordering."""
    priority: int
    callback: callable

IRC Protocol Events

Standard IRC protocol events that can be handled by the event system.

# Protocol events (generated from IRC messages)
protocol_events = [
    "error",        # Server error message
    "join",         # User joined channel
    "kick",         # User kicked from channel
    "mode",         # Mode change (user or channel)
    "part",         # User left channel
    "ping",         # Server ping
    "privmsg",      # Private message to user
    "privnotice",   # Private notice to user
    "pubmsg",       # Public message in channel
    "pubnotice",    # Public notice in channel
    "quit",         # User quit IRC
    "invite",       # Channel invitation
    "pong",         # Server pong response
    "action",       # CTCP ACTION message (/me)
    "topic",        # Channel topic change
    "nick"          # Nickname change
]

# Generated events (library-specific)
generated_events = [
    "dcc_connect",      # DCC connection established
    "dcc_disconnect",   # DCC connection closed
    "dccmsg",          # DCC message received
    "disconnect",       # Server disconnection
    "ctcp",            # CTCP query received
    "ctcpreply",       # CTCP reply received
    "login_failed"     # Authentication failed
]

# Numeric events (IRC numeric codes)
numeric_events = [
    "welcome",          # 001 - Welcome message
    "yourhost",         # 002 - Host information
    "created",          # 003 - Server creation date
    "myinfo",           # 004 - Server information
    "isupport",         # 005 - Server features
    "statsconn",        # 250 - Connection statistics
    "luserclient",      # 251 - User statistics
    "luserop",          # 252 - Operator count
    "luserunknown",     # 253 - Unknown connections
    "luserchannels",    # 254 - Channel count
    "luserme",          # 255 - Local user count
    "namreply",         # 353 - Channel user list
    "endofnames",       # 366 - End of names list
    "motdstart",        # 375 - MOTD start
    "motd",             # 372 - MOTD line
    "endofmotd",        # 376 - MOTD end
    "whoisuser",        # 311 - WHOIS user info
    "whoisserver",      # 312 - WHOIS server info
    "whoisoperator",    # 313 - WHOIS operator
    "whoisidle",        # 317 - WHOIS idle time
    "endofwhois",       # 318 - End of WHOIS
    "whoischannels",    # 319 - WHOIS channels
    "liststart",        # 321 - Channel list start
    "list",             # 322 - Channel list entry
    "listend",          # 323 - Channel list end
    "topic",            # 332 - Channel topic
    "topicinfo",        # 333 - Topic set info
    "inviting",         # 341 - Invitation sent
    "version",          # 351 - Server version
    "whoreply",         # 352 - WHO reply
    "endofwho",         # 315 - End of WHO
    "banlist",          # 367 - Ban list entry
    "endofbanlist",     # 368 - End of ban list
    "youreoper",        # 381 - You are operator
    "rehashing",        # 382 - Rehashing config
    "time",             # 391 - Server time
    "nomotd",           # 422 - No MOTD
    "nicknameinuse",    # 433 - Nickname in use
    "nickcollision",    # 436 - Nickname collision
    "unavailresource",  # 437 - Resource unavailable
    "usernotinchannel", # 441 - User not in channel
    "notonchannel",     # 442 - Not on channel
    "useronchannel",    # 443 - User already on channel
    "nologin",          # 444 - No login
    "summondisabled",   # 445 - SUMMON disabled
    "usersdisabled",    # 446 - USERS disabled
    "notregistered",    # 451 - Not registered
    "needmoreparams",   # 461 - Need more parameters
    "alreadyregistred", # 462 - Already registered
    "nopermforhost",    # 463 - No permission for host
    "passwdmismatch",   # 464 - Password mismatch
    "yourebannedcreep", # 465 - You are banned
    "youwillbebanned",  # 466 - You will be banned
    "keyset",           # 467 - Channel key set
    "channelisfull",    # 471 - Channel is full
    "unknownmode",      # 472 - Unknown mode
    "inviteonlychan",   # 473 - Invite only channel
    "bannedfromchan",   # 474 - Banned from channel
    "badchannelkey",    # 475 - Bad channel key
    "badchanmask",      # 476 - Bad channel mask
    "nochanmodes",      # 477 - No channel modes
    "banlistfull",      # 478 - Ban list full
    "noprivileges",     # 481 - No privileges
    "chanoprivsneeded", # 482 - Channel operator privileges needed
    "cantkillserver",   # 483 - Can't kill server
    "restricted",       # 484 - Connection restricted
    "uniqopprivsneeded",# 485 - Unique operator privileges needed
    "nooperhost",       # 491 - No operator host
    "umodeunknownflag", # 501 - Unknown user mode flag
    "usersdontmatch"    # 502 - Users don't match
]

# All available events
all_events = protocol_events + generated_events + numeric_events

Command Processing

IRC command representation and numeric code mapping for event processing.

class Command:
    """IRC command with numeric code mapping."""
    
    def __init__(self, code: int, name: str):
        """
        Initialize IRC command.
        
        Parameters:
        - code: int, numeric IRC code
        - name: str, command name
        """
    
    @property
    def code(self) -> int:
        """Numeric IRC code."""
    
    def __int__(self) -> int:
        """Return numeric code when converted to int."""
    
    @staticmethod
    def lookup(raw: str) -> str:
        """
        Look up command name from raw IRC data.
        
        Parameters:
        - raw: str, raw IRC command
        
        Returns:
        str, event type name
        """

# Mapping of numeric codes to command names
numeric_codes = {
    1: "welcome",
    2: "yourhost",
    3: "created",
    4: "myinfo",
    5: "isupport",
    # ... (full numeric mapping available)
}

# Mapping of command names to numeric codes
command_codes = {
    "welcome": 1,
    "yourhost": 2,
    "created": 3,
    # ... (reverse mapping)
}

Usage Examples

Basic Event Handling

import irc.client

def on_connect(connection, event):
    """Handle successful connection."""
    print(f"Connected to {event.source}")
    connection.join("#test")

def on_join(connection, event):
    """Handle channel join."""
    channel = event.target
    nick = event.source.nick
    
    if nick == connection.get_nickname():
        print(f"Joined {channel}")
    else:
        print(f"{nick} joined {channel}")
        connection.privmsg(channel, f"Welcome {nick}!")

def on_pubmsg(connection, event):
    """Handle public channel messages."""
    channel = event.target
    nick = event.source.nick
    message = event.arguments[0]
    
    print(f"[{channel}] <{nick}> {message}")
    
    # Respond to mentions
    bot_nick = connection.get_nickname()
    if bot_nick.lower() in message.lower():
        connection.privmsg(channel, f"{nick}: You mentioned me!")

def on_privmsg(connection, event):
    """Handle private messages."""
    nick = event.source.nick
    message = event.arguments[0]
    
    print(f"PM from {nick}: {message}")
    connection.privmsg(nick, f"Echo: {message}")

# Create client and add handlers
client = irc.client.SimpleIRCClient()
client.connection.add_global_handler("welcome", on_connect)
client.connection.add_global_handler("join", on_join)
client.connection.add_global_handler("pubmsg", on_pubmsg)
client.connection.add_global_handler("privmsg", on_privmsg)

client.connect("irc.libera.chat", 6667, "eventbot")
client.start()

Priority-Based Event Handling

import irc.client

def high_priority_handler(connection, event):
    """High priority handler - runs first."""
    message = event.arguments[0]
    if message.startswith("!stop"):
        print("Emergency stop requested!")
        connection.quit("Emergency stop")
        return "stop_processing"  # Can halt further processing

def medium_priority_handler(connection, event):
    """Medium priority handler."""
    message = event.arguments[0]
    if message.startswith("!"):
        print(f"Command detected: {message}")

def low_priority_handler(connection, event):
    """Low priority handler - runs last."""
    print(f"Logging message: {event.arguments[0]}")

client = irc.client.SimpleIRCClient()

# Add handlers with different priorities (lower = higher priority)
client.connection.add_global_handler("pubmsg", high_priority_handler, priority=0)
client.connection.add_global_handler("pubmsg", medium_priority_handler, priority=5)
client.connection.add_global_handler("pubmsg", low_priority_handler, priority=10)

client.connect("irc.libera.chat", 6667, "prioritybot")
client.start()

Comprehensive Event Logger

import irc.client
import datetime
import json

class EventLogger:
    def __init__(self, filename="irc_events.log"):
        self.filename = filename
        self.client = irc.client.SimpleIRCClient()
        self.setup_handlers()
    
    def log_event(self, event_type, connection, event):
        """Log event to file."""
        log_entry = {
            "timestamp": datetime.datetime.now().isoformat(),
            "event_type": event_type,
            "server": connection.get_server_name(),
            "source": str(event.source) if event.source else None,
            "target": event.target,
            "arguments": event.arguments,
            "tags": event.tags
        }
        
        with open(self.filename, "a") as f:
            f.write(json.dumps(log_entry) + "\n")
        
        print(f"[{event_type}] {event.source} -> {event.target}: {event.arguments}")
    
    def setup_handlers(self):
        """Set up event handlers for all event types."""
        events_to_log = [
            "welcome", "join", "part", "quit", "kick", "mode", "topic",
            "pubmsg", "privmsg", "pubnotice", "privnotice", "nick",
            "disconnect", "error"
        ]
        
        for event_type in events_to_log:
            handler = lambda conn, evt, etype=event_type: self.log_event(etype, conn, evt)
            self.client.connection.add_global_handler(event_type, handler)
    
    def connect_and_join(self, server, port, nickname, channels):
        """Connect to server and join channels."""
        def on_connect(connection, event):
            for channel in channels:
                connection.join(channel)
        
        self.client.connection.add_global_handler("welcome", on_connect)
        self.client.connect(server, port, nickname)
        self.client.start()

# Usage
logger = EventLogger("irc_events.log")
logger.connect_and_join("irc.libera.chat", 6667, "logbot", ["#test", "#logging"])

Event-Driven Bot Framework

import irc.client
from functools import wraps

class EventBot:
    def __init__(self):
        self.client = irc.client.SimpleIRCClient()
        self.commands = {}
        self.event_handlers = {}
        self.setup_base_handlers()
    
    def command(self, trigger):
        """Decorator for command handlers."""
        def decorator(func):
            self.commands[trigger] = func
            return func
        return decorator
    
    def event_handler(self, event_type, priority=5):
        """Decorator for event handlers."""
        def decorator(func):
            if event_type not in self.event_handlers:
                self.event_handlers[event_type] = []
            self.event_handlers[event_type].append((priority, func))
            return func
        return decorator
    
    def setup_base_handlers(self):
        """Set up base event handling."""
        def handle_pubmsg(connection, event):
            message = event.arguments[0]
            channel = event.target
            nick = event.source.nick
            
            # Check for commands
            if message.startswith("!"):
                command = message[1:].split()[0]
                if command in self.commands:
                    try:
                        self.commands[command](connection, event)
                    except Exception as e:
                        connection.privmsg(channel, f"Error: {e}")
            
            # Call custom event handlers
            if "pubmsg" in self.event_handlers:
                handlers = sorted(self.event_handlers["pubmsg"], key=lambda x: x[0])
                for priority, handler in handlers:
                    try:
                        handler(connection, event)
                    except Exception as e:
                        print(f"Handler error: {e}")
        
        self.client.connection.add_global_handler("pubmsg", handle_pubmsg)
    
    def connect(self, server, port, nickname):
        """Connect to IRC server."""
        def on_connect(connection, event):
            print(f"Connected as {nickname}")
        
        self.client.connection.add_global_handler("welcome", on_connect)
        self.client.connect(server, port, nickname)
    
    def start(self):
        """Start bot."""
        self.client.start()

# Example bot using the framework
bot = EventBot()

@bot.command("hello")
def hello_command(connection, event):
    """Say hello to user."""
    nick = event.source.nick
    channel = event.target
    connection.privmsg(channel, f"Hello {nick}!")

@bot.command("time")
def time_command(connection, event):
    """Show current time."""
    import datetime
    now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    connection.privmsg(event.target, f"Current time: {now}")

@bot.event_handler("join", priority=1)
def welcome_users(connection, event):
    """Welcome new users."""
    nick = event.source.nick
    channel = event.target
    
    if nick != connection.get_nickname():
        connection.privmsg(channel, f"Welcome to {channel}, {nick}!")

@bot.event_handler("pubmsg", priority=10)
def message_counter(connection, event):
    """Count messages (low priority)."""
    if not hasattr(message_counter, "count"):
        message_counter.count = 0
    message_counter.count += 1
    
    if message_counter.count % 100 == 0:
        connection.privmsg(event.target, f"Message #{message_counter.count}!")

# Connect and start
bot.connect("irc.libera.chat", 6667, "eventbot")
bot.start()

IRCv3 Message Tags Handler

import irc.client

def handle_tagged_message(connection, event):
    """Handle messages with IRCv3 tags."""
    if event.tags:
        print(f"Message tags: {event.tags}")
        
        # Check for specific tags
        if "account" in event.tags:
            print(f"User is authenticated as: {event.tags['account']}")
        
        if "time" in event.tags:
            print(f"Message timestamp: {event.tags['time']}")
        
        if "batch" in event.tags:
            print(f"Part of batch: {event.tags['batch']}")
    
    # Process message normally
    channel = event.target
    nick = event.source.nick
    message = event.arguments[0]
    print(f"[{channel}] <{nick}> {message}")

def handle_cap_ack(connection, event):
    """Handle capability acknowledgment."""
    caps = event.arguments[1].split()
    print(f"Server acknowledged capabilities: {caps}")

def on_connect(connection, event):
    """Request IRCv3 capabilities on connect."""
    connection.send_raw("CAP LS 302")  # Request capability list
    connection.send_raw("CAP REQ :message-tags server-time batch")  # Request specific caps
    connection.send_raw("CAP END")  # End capability negotiation
    connection.join("#test")

client = irc.client.SimpleIRCClient()
client.connection.add_global_handler("welcome", on_connect)
client.connection.add_global_handler("pubmsg", handle_tagged_message)
client.connection.add_global_handler("cap", handle_cap_ack)

client.connect("irc.libera.chat", 6667, "tagbot")
client.start()

Install with Tessl CLI

npx tessl i tessl/pypi-irc

docs

asynchronous-client.md

bot-framework.md

connection-management.md

event-system.md

index.md

protocol-extensions.md

synchronous-client.md

utilities.md

tile.json