#!/usr/bin/env python
 # meshtastic irc relay - developed by acidvegas in python (https://git.acid.vegas/meshtastic)
 
 import argparse
 import asyncio
 import logging
 import ssl
 import time
 
 # 0xEF576MkXA3aEURbCfNn6p0FfZdua4I
 
 # 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'
 
 
 # Logging Configuration
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(funcName)s - %(message)s')
 
 
 def color(msg: str, foreground: str, background: str = None) -> str:
 	'''
 	Color a string with the specified foreground and background colors.
 
 	:param msg: The string to color.
 	:param foreground: The foreground color to use.
 	:param background: The background color to use.
 	'''
 
 	return f'\x03{foreground},{background}{msg}{reset}' if background else f'\x03{foreground}{msg}{reset}'
 
 
 class Bot():
 	def __init__(self):
 		self.nickname = 'MESHTASTIC'
 		self.reader   = None
 		self.writer   = None
 		self.last     = time.time()
 
 
 	async def action(self, chan: str, msg: str):
 		'''
 		Send an ACTION to the IRC server.
 
 		:param chan: The channel to send the ACTION to.
 		:param msg: The message to send to the channel.
 		'''
 
 		await self.sendmsg(chan, f'\x01ACTION {msg}\x01')
 
 
 	async def raw(self, data: str):
 		'''
 		Send raw data to the IRC server.
 
 		:param data: The raw data to send to the IRC server. (512 bytes max including crlf)
 		'''
 
 		await 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}')
 
 
 	async def connect(self):
 		'''Connect to the IRC server.'''
 
 		while True:
 			try:
 				options = {
 					'host'       : args.server,
 					'port'       : args.port,
 					'limit'      : 1024,
 					'ssl'        : ssl._create_unverified_context() if args.ssl else None, # TODO: Do not use the args variable here
 					'family'     : 2, # AF_INET = 2, AF_INET6 = 10
 					'local_addr' : None
 				}
 
 				self.reader, self.writer = await asyncio.wait_for(asyncio.open_connection(**options), 15)
 
 				await self.raw(f'USER MESHT 0 * :git.acid.vegas/meshtastic') # Static for now
 				await self.raw('NICK ' + self.nickname)
 
 				while not self.reader.at_eof():
 					data = await asyncio.wait_for(self.reader.readuntil(b'\r\n'), 300)
 					await self.handle(data.decode('utf-8').strip())
 
 			except Exception as ex:
 				logging.error(f'failed to connect to {args.server} ({str(ex)})')
 
 			finally:
 				await asyncio.sleep(15)
 
 
 	async def eventPRIVMSG(self, data: str):
 		'''
 		Handle the PRIVMSG event.
 
 		:param data: The data received from the IRC server.
 		'''
 
 		parts  = data.split()
 
 		ident  = parts[0][1:]
 		nick   = parts[0].split('!')[0][1:]
 		target = parts[2]
 		msg	   = ' '.join(parts[3:])[1:]
 
 		if target == args.channel: # TODO: Don't use the args variable here
 			if msg.startswith('!'):
 				if time.time() - self.last < 3:
 					if not self.slow:
 						self.slow = True
 						await self.sendmsg(target, color('Slow down nerd!', red))
 				else:
 					self.slow = False
 					parts = msg.split()
 					if parts[0] == '!meshage' and len(parts) > 1:
 						message = ' '.join(parts[1:])
 						if len(message) > 255:
 							await self.sendmsg(target, color('Message exceeds 255 bytes nerd!', red))
 						# TODO: Send a meshtastic message (We have to ensure our outbounds from IRC don't loop back into IRC)
 
 					self.last = time.time() # Update the last command time if it starts with ! character to prevent command flooding
 
 
 	async def handle(self, data: str):
 		'''
 		Handle the data received from the IRC server.
 
 		:param data: The data received from the IRC server.
 		'''
 
 		logging.info(data)
 
 		try:
 			parts = data.split()
 
 			if parts[0] == 'PING':
 				await self.raw('PONG ' + parts[1])
 
 			elif parts[1] == '001': # RPL_WELCOME
 				await self.raw(f'MODE {self.nickname} +B')
 				await self.sendmsg('NickServ', f'IDENTIFY {self.nickname} simps0nsfan420')
 				await asyncio.sleep(10) # Wait for NickServ to identify or any channel join delays
 				await self.raw(f'JOIN {args.channel} {args.key if args.key else ""}')
 
 			elif parts[1] == '433': # ERR_NICKNAMEINUSE
 				self.nickname += '_' # revamp this to be more unique
 				await self.raw('NICK ' + self.nickname)
 
 			elif parts[1] == 'INVITE':
 				target = parts[2]
 				chan   = parts[3][1:]
 				if target == self.nickname and chan == args.channel:
 					await self.raw(f'JOIN {chan}')
 
 			elif parts[1] == 'KICK':
 				chan   = parts[2]
 				kicked = parts[3]
 				if kicked == self.nickname and chan == args.channel:
 					await asyncio.sleep(3)
 					await self.raw(f'JOIN {args.channel} {args.key if args.key else ""}')
 
 			elif parts[1] == 'PRIVMSG':
 				await self.eventPRIVMSG(data) # We put this in a separate function since it will likely be the most used/handled event
 
 		except (UnicodeDecodeError, UnicodeEncodeError):
 			pass
 
 		except Exception as ex:
 			logging.exception(f'Unknown error has occured! ({ex})')
 
 
 
 if __name__ == '__main__':
 	parser = argparse.ArgumentParser(description='Connect to an IRC server.')
 	parser.add_argument('server', help='The IRC server address.')
 	parser.add_argument('channel', help='The IRC channel to join.')
 	parser.add_argument('--port', type=int, help='The port number for the IRC server.')
 	parser.add_argument('--ssl', action='store_true', help='Use SSL for the connection.')
 	parser.add_argument('--key',  default='', help='The key (password) for the IRC channel, if required.')
 	args = parser.parse_args()
 
 	if not args.channel.startswith('#'):
 		channel = '#' + args.channel
 
 	if not args.port:
 		args.port = 6697 if args.ssl else 6667
 	elif args.port < 1 or args.port > 65535:
 		raise ValueError('Port must be between 1 and 65535.')
 
 	print(f'Connecting to {args.server}:{args.port} (SSL: {args.ssl}) and joining {args.channel} (Key: {args.key or 'None'})')
 
 	bot = Bot()
 
 	asyncio.run(bot.connect())

© 2023 acidvegas, inc • generated with stagit