IRC (Internet Relay Chat) protocol library for Python
—
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.
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."""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: callableStandard 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_eventsIRC 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)
}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()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()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"])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()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