Simplified, asyncio bot revamped for vortex with commenting

LICENSE

@@ -1,6 +1,6 @@
 ISC License
 
-Copyright (c) 2021, acidvegas <acid.vegas@acid.vegas>
+Copyright (c) 2023, acidvegas <acid.vegas@acid.vegas>
 
 Permission to use, copy, modify, and/or distribute this software for any
 purpose with or without fee is hereby granted, provided that the above

README.md

@@ -3,17 +3,14 @@
 
 ## Requirements
 * [Python](https://www.python.org/downloads/) *(**Note:** This script was developed to be used with the latest version of Python)*
-* [PySocks](https://pypi.python.org/pypi/PySocks) *(**Optional:** For using the `proxy` setting)*
 
 ## Information
-The repository comes with 2 skeletons. A simple, single-file skeleton for basic bots & an advanced structured skeleton for more complex bots.
-
 This is just a basic structure to help setup a bot. The bots have no use by default. It is asyncronous, can log to file, handle basic I/O, flood control, etc.
 
 ## IRC RCF Reference
 - http://www.irchelp.org/protocol/rfc/
 
-## Mirrors
+___
-- [acid.vegas](https://acid.vegas/skeleton) *(main)*
+
-- [GitHub](https://github.com/acidvegas/skeleton)
+###### Mirrors
-- [GitLab](https://gitlab.com/acidvegas/skeleton)
+[acid.vegas](https://git.acid.vegas/skeleton) • [GitHub](https://github.com/acidvegas/skeleton) • [GitLab](https://gitlab.com/acidvegas/skeleton) • [SuperNETs](https://git.supernets.org/acidvegas/skeleton)

advanced/core/bot.py

@@ -1,54 +0,0 @@
-#!/usr/bin/env python
-# Asyncronous IRC Bot Skeleton - Developed by acidvegas in Python (https://acid.vegas/skeleton)
-# bot.py
-
-import asyncio
-import logging
-
-import config
-
-from commands import Command
-from events import Event
-
-def ssl_ctx():
-	import ssl
-	ctx = ssl.create_default_context()
-	if not config.connection.ssl_verify:
-		ctx.check_hostname = False
-		ctx.verify_mode = ssl.CERT_NONE
-	if config.cert.file:
-	    ctx.load_cert_chain(config.cert.file, password=config.cert.password)
-	return ctx
-
-class IrcBot:
-	def __init__(self):
-		self.options = {
-			'host'       : config.connection.server,
-			'port'       : config.connection.port,
-			'limit'      : 1024,
-			'ssl'        : ssl_ctx() if config.connection.ssl else None,
-			'family'     : 10 if config.connection.ipv6 else 2,
-			'local_addr' : (config.connection.vhost, 0) if config.connection.vhost else None
-		}
-		self.reader = None
-		self.writer = None
-
-	async def run(self):
-		try:
-			self.reader, self.writer = await asyncio.open_connection(**self.options, timeout=config.throttle.timeout)
-		except Exception as ex:
-			logging.exception('Failed to connect to IRC server!')
-		else:
-			try:
-				await Command(Bot).register(config.ident.nickname, config.ident.username, config.ident.realname, config.login.network)
-				while not self.reader.at_eof():
-					data = await self.reader.readline()
-					Event(Bot).handle(data.decode('utf-8').strip())
-			except (UnicodeDecodeError, UnicodeEncodeError):
-				pass
-			except Exception as ex:
-				logging.exception('Unknown error has occured!')
-		finally:
-			Event.disconnect()
-
-Bot = IrcBot()

advanced/core/commands.py

@@ -1,43 +0,0 @@
-#!/usr/bin/env python
-# Asyncronous IRC Bot Skeleton - Developed by acidvegas in Python (https://acid.vegas/skeleton)
-# commands.py
-
-class Command:
-	def __init__(self, bot):
-		self.Bot = bot
-
-	def action(self, target, msg):
-		self.sendmsg(target, f'\x01ACTION {msg}\x01')
-
-	def join_channel(self, chan, key=None):
-		self.raw(f'JOIN {chan} {key}') if key else raw('JOIN ' + chan)
-
-	def mode(self, target, mode):
-		self.raw(f'MODE {target} {mode}')
-
-	def nick(self, new_nick):
-		self.raw('NICK ' + new_nick)
-
-	def notice(self, target, msg):
-		self.raw(f'NOTICE {target} :{msg}')
-
-	def part_channel(self, chan, msg=None):
-		self.raw(f'PART {chan} {msg}') if msg else raw('PART ' + chan)
-
-	def quit(self, msg=None):
-		self.raw('QUIT :' + msg) if msg else raw('QUIT')
-
-	def raw(self, data):
-		self.Bot.writer.write(data[:510].encode('utf-8') + b'\r\n')
-
-	def register(self, nickname, username, realname, password=None):
-		if password:
-			self.raw('PASS ' + password)
-		self.raw('NICK ' + nickname)
-		self.raw(f'USER {username} 0 * :{realname}')
-
-	def sendmsg(self, target, msg):
-		self.raw(f'PRIVMSG {target} :{msg}')
-
-	def topic(self, chan, data):
-		self.raw(f'TOPIC {chan} :{text}')

advanced/core/config.py

@@ -1,39 +0,0 @@
-#!/usr/bin/env python
-# Asyncronous IRC Bot Skeleton - Developed by acidvegas in Python (https://acid.vegas/skeleton)
-# config.py
-
-class connection:
-	server        = 'irc.server.com'
-	port          = 6667
-	ipv6          = False
-	ssl           = False
-	ssl_verify    = False
-	vhost         = None
-	channel       = '#dev'
-	key           = None
-	modes         = None
-
-class cert:
-	file     = None
-	password = None
-
-class ident:
-	nickname = 'skeleton'
-	username = 'skeleton'
-	realname = 'acid.vegas/skeleton'
-
-class login:
-	network  = None
-	nickserv = None
-	operator = None
-
-class settings:
-	admin = 'nick!user@host' # Must be in nick!user@host format (Wildcards accepted)
-	log   = False
-
-class throttle:
-	command   = 3
-	message   = 0.5
-	reconnect = 15
-	rejoin    = 5
-	timeout   = 15

advanced/core/events.py

@@ -1,61 +0,0 @@
-#!/usr/bin/env python
-# Asyncronous IRC Bot Skeleton - Developed by acidvegas in Python (https://acid.vegas/asyncirc)
-# events.py
-
-import asyncio
-import logging
-
-import config
-
-class Event:
-	def __init__(self, bot):
-		self.Bot = bot
-
-	def connect(self):
-		if config.settings.modes:
-			Commands.raw(f'MODE {config.ident.nickname} +{config.settings.modes}')
-		if config.login.nickserv:
-			Commands.sendmsg('NickServ', f'IDENTIFY {config.ident.nickname} {config.login.nickserv}')
-		if config.login.operator:
-			Commands.raw(f'OPER {config.ident.username} {config.login.operator}')
-		Commands.join_channel(config.connection.channel, config.connection.key)
-
-	async def disconnect(self):
-		self.writer.close()
-		await self.writer.wait_closed()
-		asyncio.sleep(config.throttle.reconnect)
-
-	def join_channel(self):
-		pass
-
-	def kick(self):
-		pass
-
-	def invite(self):
-		pass
-
-	def message(self):
-		pass
-
-	def nick_in_use(self):
-		new_nick = 'a' + str(random.randint(1000,9999))
-		Command.nick(new_nick)
-
-	def part_channel(self):
-		pass
-
-	def private_message(self):
-		pass
-
-	def quit(self):
-		pass
-
-	async def handler(self, data):
-		logging.info(data)
-		args = data.split()
-		if args[0] == 'PING':
-			self.raw('PONG ' + args[1][1:])
-		elif args[1] == '001': #RPL_WELCOME
-			self.connect()
-		elif args[1] == '433': #ERR_NICKNAMEINUSE
-			self.nick_in_use()

advanced/skeleton.py

@@ -1,40 +0,0 @@
-#!/usr/bin/env python
-# Asyncronous IRC Bot Skeleton - Developed by acidvegas in Python (https://acid.vegas/skeleton)
-# skeleton.py
-
-import asyncio
-import logging
-import logging.handlers
-import os
-import sys
-
-sys.dont_write_bytecode = True
-os.chdir(os.path.dirname(__file__) or '.')
-sys.path += ('core','modules')
-
-import config
-
-if not os.path.exists('logs'):
-	os.makedirs('logs')
-sh = logging.StreamHandler()
-sh.setFormatter(logging.Formatter('%(asctime)s | %(levelname)9s | %(message)s', '%I:%M %p'))
-if config.settings.log:
-	fh = logging.handlers.RotatingFileHandler('logs/debug.log', maxBytes=250000, backupCount=7, encoding='utf-8')
-	fh.setFormatter(logging.Formatter('%(asctime)s | %(levelname)9s | %(filename)s.%(funcName)s.%(lineno)d | %(message)s', '%Y-%m-%d %I:%M %p'))
-	logging.basicConfig(level=logging.NOTSET, handlers=(sh,fh))
-	del fh
-else:
-	logging.basicConfig(level=logging.NOTSET, handlers=(sh,))
-del sh
-
-print('#'*56)
-print('#{:^54}#'.format(''))
-print('#{:^54}#'.format('Asyncronous IRC Bot Skeleton'))
-print('#{:^54}#'.format('Developed by acidvegas in Python'))
-print('#{:^54}#'.format('https://acid.vegas/skeleton'))
-print('#{:^54}#'.format(''))
-print('#'*56)
-
-from bot import Bot
-
-asyncio.run(Bot.run())

skeleton.py

@@ -1,177 +1,186 @@
 #!/usr/bin/env python
-# Asyncronoua IRC Bot Skeleton - Developed by acidvegas in Python (https://acid.vegas/skeleton)
+# Skeleton IRC bot - developed by acidvegas in python (https://git.acid.vegas/skeleton)
-# skeleton.py
+import argparse
-
 import asyncio
 import logging
 import logging.handlers
-import os
+import ssl
-import random
-import time
 
-##################################################
+# Formatting Control Characters / Color Codes
+bold        = '\x02'
+italic      = '\x1D'
+underline   = '\x1F'
+reverse     = '\x16'
+reset       = '\x0f'
+white       = '00'
+black       = '01'
+blue        = '02'
+green       = '03'
+red         = '04'
+brown       = '05'
+purple      = '06'
+orange      = '07'
+yellow      = '08'
+light_green = '09'
+cyan        = '10'
+light_cyan  = '11'
+light_blue  = '12'
+pink        = '13'
+grey        = '14'
+light_grey  = '15'
 
-class config:
+def color(msg: str, foreground: str, background: str='') -> str:
-	class connection:
+    '''
-		server     = 'irc.supernets.org'
+    Color a string with the specified foreground and background colors.
-		port       = 6697
+    
-		ipv6       = False
+    :param msg: The string to color.
-		ssl        = True
+    :param foreground: The foreground color to use.
-		ssl_verify = False
+    :param background: The background color to use.
-		vhost      = None
+    '''
-		channel    = '#dev'
+    return f'\x03{foreground},{background}{msg}{reset}' if background else f'\x03{foreground}{msg}{reset}'
-		key        = None
-		modes      = None
 
-	class cert:
+def ssl_ctx() -> ssl.SSLContext:
-		file     = None
+    '''Create a SSL context for the connection.'''
-		password = None
+    ctx = ssl.create_default_context()
+    ctx.verify_mode = ssl.CERT_NONE # Comment out this line to verify hosts
+    #ctx.load_cert_chain('/path/to/cert', password='loldongs')
+    return ctx
 
-	class ident:
+class Bot():
-		nickname = 'skeleton'
+    def __init__(self):
-		username = 'skeleton'
+        self.nickname = 'skeleton'
-		realname = 'acid.vegas/skeleton'
+        self.username = 'skelly'
+        self.realname = 'Developement Bot'
+        self.reader   = None
+        self.writer   = None
 
-	class login:
+    async def action(self, chan: str, msg: str):
-		network  = None
+        '''
-		nickserv = None
+        Send an ACTION to the IRC server.
-		operator = None
 
-	class settings:
+        :param chan: The channel to send the ACTION to.
-		admin = 'nick!user@host' # Must be in nick!user@host format (Wildcards accepted)
+        :param msg: The message to send to the channel.
-		log   = False
+        '''
+        await self.sendmsg(chan, f'\x01ACTION {msg}\x01')
 
-	class throttle:
+    def raw(self, data: str):
-		command   = 3
+        '''
-		message   = 0.5
+        Send raw data to the IRC server.
-		reconnect = 15
+		
-		rejoin    = 5
+        :param data: The raw data to send to the IRC server. (512 bytes max including crlf)
-		timeout   = 15
+        '''
+        self.writer.write(data[:510].encode('utf-8') + b'\r\n')
 
-##################################################
+    async def sendmsg(self, target: str, msg: str):
+        '''
+        Send a PRIVMSG to the IRC server.
+        
+        :param target: The target to send the PRIVMSG to. (channel or user)
+        :param msg: The message to send to the target.
+        '''
+        await self.raw(f'PRIVMSG {target} :{msg}')
 
-def ssl_ctx():
+    async def connect(self):
-	import ssl
+        '''Connect to the IRC server.'''
-	ctx = ssl.create_default_context()
+        while True:
-	if not config.connection.ssl_verify:
+            try:
-		ctx.check_hostname = False
+                options = {
-		ctx.verify_mode = ssl.CERT_NONE
+                    'host'       : args.server,
-	if config.cert.file:
+                    'port'       : args.port if args.port else 6697 if args.ssl else 6667,
-	    ctx.load_cert_chain(config.cert.file, password=config.cert.password)
+                    'limit'      : 1024, # Buffer size in bytes (don't change this unless you know what you're doing)
-	return ctx
+                    'ssl'        : ssl_ctx() if args.ssl else None,
+                    'family'     : 10 if args.ipv6 else 2, # 10 = AF_INET6 (IPv6), 2 = AF_INET (IPv4)
+                    'local_addr' : args.vhost if args.vhost else None # Can we just leave this as args.vhost?
+                }
+                self.reader, self.writer = await asyncio.wait_for(asyncio.open_connection(**options), 15) # 15 second timeout
+                if args.password:
+                    await self.raw('PASS ' + args.password) # Rarely used, but IRCds may require this
+                    await self.raw(f'USER {self.username} 0 * :{self.realname}') # These lines must be sent upon connection
+                    await self.raw('NICK ' + self.nickname)                      # They are to identify the bot to the server
+                    while not self.reader.at_eof():
+                        data = await asyncio.wait_for(self.reader.readuntil(b'\r\n'), 300) # 5 minute ping timeout
+                        await self.handle(data.decode('utf-8').strip()) # Handle the data received from the IRC server
+            except Exception as ex:
+                logging.error(f'failed to connect to {self.server} ({ex})')
+            finally:
+                await asyncio.sleep(30) # Wait 30 seconds before reconnecting
 
-##################################################
+    async def handle(self, data: str):
+        '''
+        Handle the data received from the IRC server.
+        
+        :param data: The data received from the IRC server.
+        '''
+        try:
+            args = data.split()
+            if data.startswith('ERROR :Closing Link:'):
+                raise Exception('BANNED')
+            if args[0] == 'PING':
+                await self.raw('PONG ' + args[1]) # Respond to the server's PING request with a PONG to prevent ping timeout
+            elif args[1] == '001': # RPL_WELCOME
+                await self.raw(f'MODE {self.nickname} +B') # Set user mode +B (Bot)
+                await self.sendmsg('NickServ', 'IDENTIFY {self.nickname} simps0nsfan420') # Identify to NickServ
+                await self.raw('OPER MrSysadmin fartsimps0n1337') # Oper up
+                await asyncio.sleep(10) # Wait 10 seconds before joining the channel (required by some IRCds to wait before JOIN)
+                await self.raw(f'JOIN {args.channel} {args.key}') # Join the channel (if no key was provided, this will still work as the key will default to an empty string)
+            elif args[1] == '433': # ERR_NICKNAMEINUSE
+                self.nickname += '_' # If the nickname is already in use, append an underscore to the end of it
+                await self.raw('NICK ' + self.nickname) # Send the new nickname to the server
+            elif args[1] == 'KICK':
+                chan   = args[2]
+                kicked = args[3]
+                if kicked == self.nickname:
+                    await asyncio.sleep(3)
+                    await self.raw(f'JOIN {chan}')
+            elif args[1] == 'PRIVMSG':
+                ident  = args[0][1:]
+                nick   = args[0].split('!')[0][1:]
+                target = args[2]
+                msg    = ' '.join(args[3:])[1:]
+                if target == self.nickname:
+                    pass # Handle private messages here
+                if target.startswith('#'): # Channel message
+                    if msg.startswith('!'):
+                        if msg == '!hello':
+                            self.sendmsg(chan, f'Hello {nick}!')
+        except (UnicodeDecodeError, UnicodeEncodeError):
+            pass # Some IRCds allow invalid UTF-8 characters, this is a very important exception to catch
+        except Exception as ex:
+            logging.exception(f'Unknown error has occured! ({ex})')
 
-class Command:
-	def join_channel(chan, key=None):
-		Command.raw(f'JOIN {chan} {key}') if key else Command.raw('JOIN ' + chan)
 
-	def mode(target, mode):
+def setup_logger(log_filename: str, to_file: bool = False):
-		Command.raw(f'MODE {target} {mode}')
+    '''
+    Set up logging to console & optionally to file.
 
-	def nick(new_nick):
+    :param log_filename: The filename of the log file
-		Command.raw('NICK ' + new_nick)
+    '''
-
+    sh = logging.StreamHandler()
-	def raw(data):
+    sh.setFormatter(logging.Formatter('%(asctime)s | %(levelname)9s | %(message)s', '%I:%M %p'))
-		Bot.writer.write(data[:510].encode('utf-8') + b'\r\n')
+    if to_file:
-
+        fh = logging.handlers.RotatingFileHandler(log_filename+'.log', maxBytes=250000, backupCount=3, encoding='utf-8') # Max size of 250KB, 3 backups
-	def sendmsg(target, msg):
+        fh.setFormatter(logging.Formatter('%(asctime)s | %(levelname)9s | %(filename)s.%(funcName)s.%(lineno)d | %(message)s', '%Y-%m-%d %I:%M %p')) # We can be more verbose in the log file
-		Command.raw(f'PRIVMSG {target} :{msg}')
+        logging.basicConfig(level=logging.NOTSET, handlers=(sh,fh))
-
+    else:
-##################################################
+        logging.basicConfig(level=logging.NOTSET, handlers=(sh,))
-
-class Event:
-	def connect():
-		if config.connection.modes:
-			Command.raw(f'MODE {config.ident.nickname} +{config.connection.modes}')
-		if config.login.nickserv:
-			Command.sendmsg('NickServ', f'IDENTIFY {config.ident.nickname} {config.login.nickserv}')
-		if config.login.operator:
-			Command.raw(f'OPER {config.ident.username} {config.login.operator}')
-		Command.join_channel(config.connection.channel, config.connection.key)
-
-	async def disconnect():
-		Bot.writer.close()
-		await bot.writer.wait_closed()
-		asyncio.sleep(config.throttle.reconnect)
-
-	def nick_in_use():
-		new_nick = 'a' + str(random.randint(1000,9999))
-		Command.nick(new_nick)
-
-	async def handler():
-		while not Bot.reader.at_eof():
-			try:
-				data = await Bot.reader.readline()
-				data = data.decode('utf-8').strip()
-				logging.info(data)
-				args = data.split()
-				if data.startswith('ERROR :Closing Link:'):
-					raise Exception('Connection has closed.')
-				elif data.startswith('ERROR :Reconnecting too fast, throttled.'):
-					raise Exception('Connection has closed. (throttled)')
-				elif args[0] == 'PING':
-					Command.raw('PONG ' + args[1][1:])
-				elif args[1] == '001': #RPL_WELCOME
-					Event.connect()
-				elif args[1] == '433': #ERR_NICKNAMEINUSE
-					Event.nick_in_use()
-				elif args[1] == 'KICK':
-					pass # handle kick
-			except (UnicodeDecodeError, UnicodeEncodeError):
-				pass
-			except:
-				logging.exception('Unknown error has occured!')
-
-##################################################
-
-class IrcBot:
-	def __init__(self):
-		self.options = {
-			'host'       : config.connection.server,
-			'port'       : config.connection.port,
-			'limit'      : 1024,
-			'ssl'        : ssl_ctx() if config.connection.ssl else None,
-			'family'     : socket.AF_INET6 if config.connection.ipv6 else socket.AF_INET,
-			'local_addr' : (config.connection.vhost, 0) if config.connection.vhost else None
-		}
-		self.reader, self.writer = (None, None)
-
-	async def connect(self):
-		try:
-			self.reader, self.writer = await asyncio.wait_for(asyncio.open_connection(**self.options), timeout=config.throttle.timeout)
-			if config.login.network:
-				Command.raw('PASS ' + config.login.network)
-			Command.raw(f'USER {config.ident.username} 0 * :{config.ident.realname}')
-			Command.raw('NICK ' + config.ident.nickname)
-		except:
-			logging.exception('Failed to connect to IRC server!')
-		else:
-			await Event.handler()
-
-##################################################
 
 if __name__ == '__main__':
-	if not os.path.exists('logs'):
+    parser = argparse.ArgumentParser(description="Connect to an IRC server.") # The arguments without -- are required arguments.
-		os.makedirs('logs')
+    parser.add_argument("server", help="The IRC server address.")
-	sh = logging.StreamHandler()
+    parser.add_argument("channel", help="The IRC channel to join.")
-	sh.setFormatter(logging.Formatter('%(asctime)s | %(levelname)9s | %(message)s', '%I:%M %p'))
+    parser.add_argument("--password", help="The password for the IRC server.")
-	if config.settings.log:
+    parser.add_argument("--port", type=int, help="The port number for the IRC server.") # Port is optional, will default to 6667/6697 depending on SSL.
-		fh = logging.handlers.RotatingFileHandler('logs/debug.log', maxBytes=250000, backupCount=7, encoding='utf-8')
+    parser.add_argument("--ssl", action="store_true", help="Use SSL for the connection.")
-		fh.setFormatter(logging.Formatter('%(asctime)s | %(levelname)9s | %(filename)s.%(funcName)s.%(lineno)d | %(message)s', '%Y-%m-%d %I:%M %p'))
+    parser.add_argument("--v4", action="store_true", help="Use IPv4 for the connection.")
-		logging.basicConfig(level=logging.NOTSET, handlers=(sh,fh))
+    parser.add_argument("--v6", action="store_true", help="Use IPv6 for the connection.")
-		del fh,sh
+    parser.add_argument("--key",  default="", help="The key (password) for the IRC channel, if required.")
-	else:
+    parser.add_argument("--vhost", help="The VHOST to use for connection.")
-		logging.basicConfig(level=logging.NOTSET, handlers=(sh,))
+    args = parser.parse_args()
-		del sh
 
-	print('#'*56)
+    print(f"Connecting to {args.server}:{args.port} (SSL: {args.ssl}) and joining {args.channel} (Key: {args.key or 'None'})")
-	print('#{:^54}#'.format(''))
-	print('#{:^54}#'.format('Asyncronous IRC Bot Skeleton'))
-	print('#{:^54}#'.format('Developed by acidvegas in Python'))
-	print('#{:^54}#'.format('https://acid.vegas/skeleton'))
-	print('#{:^54}#'.format(''))
-	print('#'*56)
 
-	Bot = IrcBot()
+    setup_logger('skeleton', to_file=True) # Optionally, you can log to a file, change to_file to False to disable this.
-	asyncio.run(Bot.connect())
+
+    bot = Bot() # We define this here as an object so we can call it from an outside function if we need to.
+
+    asyncio.run(bot.connect())