2024-10-19 19:43:48 -07:00
"""
IRC Parser for the SugarCaneIRC family .
"""
2024-10-19 19:25:37 -07:00
import socket
2024-10-19 19:43:48 -07:00
import ssl as ssl_module
2024-10-19 20:39:58 -07:00
import threading
2024-10-20 20:54:08 -07:00
import traceback
2024-10-20 12:15:30 -07:00
__version__ = " _TEST_ "
2024-10-20 19:52:10 -07:00
class User : # User object
2024-10-20 20:54:08 -07:00
def __init__ ( self , name : str , system : bool = False , realname : str | None = None , username : str | None = None , host : str | None = None , mention : bool = False ) :
2024-10-20 19:52:10 -07:00
self . name , self . system , self . realname , self . username = name , system , realname , username
2024-10-20 12:13:02 -07:00
class Channel : # Channel object
2024-10-19 19:43:48 -07:00
is_init = False # If the channel's properties are initialized yet
topic = " " # Channel topic
modes = " +nt " # Channel modes
2024-10-19 20:39:58 -07:00
in_channel = True # If the user is in the channel
2024-10-19 19:25:37 -07:00
def __init__ ( self , name : str ) :
self . name = name
def info_set ( self , topic : str , modes : str ) : # Socket will automatically initialize the channel object
self . is_init = True
self . topic , self . modes = topic , modes
2024-10-20 19:52:10 -07:00
class SystemMessage : # System message object
def __init__ ( self , content : str , user : User , typ : str , mention : bool , chan : str | None = None ) :
self . content , self . user , self . type , self . mention , self . chan = content , user , typ , mention , chan
class Message : # Message object
def __init__ ( self , content : str , user : User , target : User | Channel | str ) :
self . content = content
self . target = target
self . author = user
class ParserMessage : # Parser message
def __init__ ( self , content , chan : str | None = None ) :
self . content , self . chan = content , chan
2024-10-19 19:25:37 -07:00
class IRCSession : # Actual IRC session
2024-10-19 20:39:58 -07:00
messages = [ ] # Cached messages
raw_text = " " # Cached RAW data
2024-10-20 10:13:43 -07:00
chans = [ ] # Cached channels
2024-10-19 21:29:14 -07:00
connected = False # Connection status
2024-10-19 19:43:48 -07:00
is_ssl = False # Wether the connection uses TLS/SSL
ssl_accept_invalid = False # If SSL is enabled, do not fail to connect if the certificate is invalid.
2024-10-19 21:03:02 -07:00
socket = socket . socket ( ) # Socket
2024-10-19 19:43:48 -07:00
wsocket = None # Wrapped socket (if SSL is enabled)
2024-10-19 20:39:58 -07:00
context = ssl_module . create_default_context ( ) # Context of the SSL module, not to be changed by the client.
2024-10-20 12:17:02 -07:00
def __init__ ( self , address : str = " irc.libera.chat " , port : int = 6697 , nick : str = " CaneSugar " , user : str = " ScParse " , ssl : bool = True , ssl_igninvalid : bool = False , realname : str = " SugarcaneParseIRC user " , * * kwargs ) : # Contains the configuration
2024-10-19 21:33:48 -07:00
self . socket = socket . socket ( socket . AF_INET , socket . SOCK_STREAM )
2024-10-19 21:03:02 -07:00
self . context = ssl_module . create_default_context ( )
self . wsocket = None
2024-10-20 10:27:49 -07:00
self . server , self . port , self . nick , self . user , self . ssl , self . ssl_accept_invalid , self . realname = address , port , nick , user , ssl , ssl_igninvalid , realname
2024-10-20 15:11:20 -07:00
self . msgcache_index = 0
2024-10-20 19:35:58 -07:00
self . motd = " "
2024-10-20 20:54:08 -07:00
self . yourself = User ( name = nick , username = user )
2024-10-19 19:43:48 -07:00
if ssl :
if ssl_igninvalid :
self . context = ssl_module . _create_unverified_context ( )
2024-10-19 19:45:35 -07:00
self . wsocket = self . context . wrap_socket ( self . socket , server_hostname = address )
2024-10-19 19:25:37 -07:00
def connect ( self ) : # Attempt to connect
2024-10-19 19:27:56 -07:00
print ( " Connecting to " + self . server + " : " + str ( self . port ) + " ... " )
2024-10-19 19:43:48 -07:00
if self . ssl :
self . wsocket . connect ( ( self . server , self . port ) )
else :
self . socket . connect ( ( self . server , self . port ) )
2024-10-20 10:09:47 -07:00
self . connected = True
2024-10-19 21:13:52 -07:00
self . send ( " USER " + self . user + " " + self . user + " " + self . nick + " :SugarCaneIRC user \n " )
self . send ( f " NICK { self . nick } \n " )
2024-10-20 14:18:43 -07:00
def detach_connection ( self ) :
print ( " Detaching connection to a thread... " )
threading . Thread ( target = self . getloop , daemon = True ) . start ( )
def getloop ( self ) :
while self . connected :
try :
self . get ( )
except :
2024-10-20 20:54:08 -07:00
print ( traceback . format_exc ( ) )
2024-10-19 21:13:52 -07:00
def send ( self , content : str ) : # Attempt to send raw data to the socket
2024-10-19 21:29:14 -07:00
if content [ len ( content ) - 1 ] != " \n " :
content + = " \n "
2024-10-19 21:13:52 -07:00
if self . ssl :
self . wsocket . send ( bytes ( content , " UTF-8 " ) )
else :
self . socket . send ( bytes ( content , " UTF-8 " ) )
2024-10-20 10:39:33 -07:00
def quit ( self , message : str = " ScParseIRC v " + str ( __version__ ) ) : # Send the server a signal that the client is about to quit, and rely on the server to close the connection.
2024-10-20 10:39:20 -07:00
self . send ( " QUIT : " + message + " \n " )
2024-10-20 10:09:47 -07:00
self . connected = False
def join ( self , chan ) :
2024-10-20 12:13:02 -07:00
self . chans . append ( Channel ( chan ) )
2024-10-20 10:09:47 -07:00
self . send ( f " JOIN { chan } " )
2024-10-20 12:38:43 -07:00
def part ( self , chan : str , message : str = " ScParseIRC v " + str ( __version__ ) ) :
complete = False
for i in self . chans :
if i . name == chan :
i . in_channel = False
complete = True
break
if complete :
self . send ( f " PART { chan } " )
def privmsg ( self , target : str , content : str ) :
self . send ( f " PRIVMSG { target } : { content } " )
2024-10-20 12:47:13 -07:00
self . messages . append ( Message ( content = content , chan = target , nick = self . nick ) )
2024-10-19 21:29:14 -07:00
def close ( self ) :
if self . ssl :
self . wsocket . close ( )
else :
self . socket . close ( )
self . connected = False
2024-10-19 20:39:58 -07:00
def get ( self ) : # Attempt to get the raw data and parse it.
# The code is copied from sweeBotIRC btw
2024-10-19 20:58:17 -07:00
if self . ssl :
r = self . wsocket . recv ( 2040 ) . decode ( )
else :
r = self . socket . recv ( 2040 ) . decode ( )
2024-10-19 20:39:58 -07:00
self . raw_text + = r
self . parseall ( )
2024-10-20 20:54:08 -07:00
print ( r )
2024-10-19 20:39:58 -07:00
if r . find ( " PING " ) != - 1 :
2024-10-19 21:29:14 -07:00
self . send (
" PONG " + r . split ( ) [ 1 ] + " \r \n "
2024-10-19 20:39:58 -07:00
)
2024-10-20 11:04:26 -07:00
if not r :
self . connected = False
2024-10-20 11:33:34 -07:00
return r
2024-10-19 20:39:58 -07:00
def parseall ( self ) : # Parse all of the fetched raw data, in a thread.
2024-10-20 12:13:02 -07:00
threading . Thread ( target = self . _dump_message_cache , kwargs = { " content " : self . raw_text } ) . start ( )
def _dump_message_cache ( self , content : str ) : # The thread of parsing all of the raw data, dumping all of it in the messages list.
self . messages = self . parse ( content )
def parse ( self , content : str ) : # Attempt to parse raw data into a Message or SystemMessage object
cache = [ ]
2024-10-19 20:53:06 -07:00
for i in content . replace ( " \r \n " , " \n " ) . split ( " \n " ) :
2024-10-20 11:33:34 -07:00
spaced = i . split ( " " )
2024-10-20 11:37:17 -07:00
system_ = not " @ " in spaced [ 0 ]
2024-10-20 20:54:08 -07:00
if len ( spaced ) > 1 :
2024-10-20 19:52:10 -07:00
if spaced [ 1 ] == " NOTICE " : # Notice (Can be sent from the server or a user.)
2024-10-20 12:13:02 -07:00
cache . append ( SystemMessage ( content = " " . join ( spaced [ 3 : ] ) [ 1 : ] , user = User ( name = spaced [ 0 ] [ 1 : ] if not " @ " in spaced [ 0 ] else spaced [ 0 ] [ 1 : ] . split ( " ! " ) [ 0 ] , system = system_ ) , typ = " notice " , mention = not system_ ) )
2024-10-20 19:52:10 -07:00
elif spaced [ 1 ] == " 001 " : # Type 001 (Welcome to ... Internet Relay Chat Network [nick])
2024-10-20 19:39:22 -07:00
cache . append ( ParserMessage ( content = " Server reports name \" " + spaced [ 6 ] + " \" " ) )
2024-10-20 19:52:10 -07:00
elif spaced [ 1 ] == " 003 " : # Type 003 (Server was created in...)
2024-10-20 19:39:22 -07:00
cache . append ( ParserMessage ( content = " Server reports creation time " + " " . join ( spaced [ 7 : ] ) ) )
2024-10-20 19:52:10 -07:00
elif spaced [ 1 ] == " 433 " : # Type 443 (Nick already in use)
2024-10-20 15:11:20 -07:00
cache . append ( SystemMessage ( content = " " . join ( spaced [ 4 : ] ) [ 1 : ] , user = User ( name = spaced [ 0 ] [ 1 : ] , system = True ) , typ = " error " , mention = True ) )
2024-10-20 19:52:10 -07:00
elif spaced [ 1 ] == " PRIVMSG " : # IRC messages
token1 = spaced [ 0 ] [ 1 : ]
2024-10-20 20:54:08 -07:00
channel = spaced [ 2 ]
2024-10-21 16:48:25 -07:00
cache . append ( Message ( content = " " . join ( spaced [ 3 : ] ) [ 1 : ] , user = User ( name = token1 . split ( " ! " ) [ 0 ] , username = token1 . split ( " ! " ) [ 1 ] . split ( " @ " ) [ 0 ] , host = token1 . split ( " ! " ) [ 1 ] . split ( " @ " ) [ 1 ] ) , target = Channel ( name = channel ) if channel [ 0 ] == " # " else self . yourself ) )
2024-10-20 12:13:02 -07:00
if len ( cache ) == 1 :
return cache [ 0 ]
else :
return cache
2024-10-19 19:25:37 -07:00
def alive ( self ) : # NOT FINISHED: To minimize exceptions, the client can ask the object if the socket connection is still alive.
return False
2024-10-19 19:25:46 -07:00
def whois ( self , nick : str ) : # NOT FINISHED: Try to /whois the user, will return a user() object or None.
2024-10-19 19:25:37 -07:00
return None