''' Created on 17.04.2010 Python Warsow Module @author: Orbital ''' import socket import threading import time def irc_color( text #note: does not have to be string (is converted anyways) ): ''' Converts the colour codes from wars ^# to irc color codes (^^ is handled correctly) ''' text = str( text ) #warsow color code : irc color code, ^^ is a special case colors = { "^0" : "\x0301,00", "^7" : "\x0300,01", "^4" : "\x0302,00", "^2" : "\x0303,00", "^1" : "\x0304,00", "^6" : "\x0306,00", "^8" : "\x0307,00", "^3" : "\x0308,01", "^5" : "\x0310,00", "^9" : "\x0314,00", "^^" : "^" } #---------------------------------- for color in colors: text = text.replace( color, colors[color] ) return text + "\x03" #We don't want it to take any effect to the text that follows up def decolor( text #note: does not have to be string (is converted anyways) ): ''' Removes the color codes from a string, ^^ case is handled correctly! ''' text = str( text ) #warsow color code : None, ^^ is a special case colors = { "^0" : "", "^7" : "", "^4" : "", "^2" : "", "^1" : "", "^6" : "", "^8" : "", "^3" : "", "^5" : "", "^9" : "", "^^" : "^" } #---------------------------------- for color in colors: text = text.replace( color, colors[color] ) return text class Query( threading.Thread ): ''' Query class, when started: creates an pretty up_to_date serverlist which you can get information from ''' def run( self ): self.ips = list() self.infos = list() self.reload() def info_get( self, address, #should be a tuple like (ip, port) * modes #should be strings like "sv_hostname", "gametype" ): ''' main query function: this gets information from a server ''' server = socket.socket( socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP ) #connection to the warsow server (udp) try: server.connect( address ) server.settimeout( 2 ) server.send( "\xFF\xFF\xFF\xFFgetstatus" ) data = server.recv( 4096 ).split( "\\" )[1:] #we will receive a string with pairs separated by \\ like gametype\\ca except: #in case of timeout: we don't want it to take for ever so we ignore servers that are bugged or don't reply return None vars = dict() for i in xrange( 0, len( data ), 2 ): #this handles the output string that we got if data[i] == "clients": #clients are separated by \n vars[data[i]] = data[i + 1].split( "\n" )[0], data[i + 1].split( "\n" )[1:] else: #whole rest vars[data[i]] = data[i + 1] if modes: out = list() for mode in modes: out.append( vars[mode] ) return out else: return vars def reload( self ): ''' this reloads the list of servers it loads the list and then switches it to prevent having an almost empty list at some time this is done every 30 seconds ( can be changed ) ''' master = socket.socket( socket.AF_INET, socket.SOCK_DGRAM ) master.connect( ( "dpmaster.deathmask.net", 27950 ) ) #most common warsow master server master.send( "\xFF\xFF\xFF\xFFgetservers Warsow 11 full empty" ) data = master.recv( 65535 ).lstrip( "\xFF\xFF\xFF\xFFgetserversResponse\\" ).split( "\\" ) #we'll get addresses separated by \\ master.close() #we don't need you anymore, buddy! servers, infs = list(), list() #temporary lists for address in data: #the strings will usually contain 6 bytes, the first 4 are the ip and the last 2 the port try: #IP: ip = str( ord( address[0] ) ) + "." + str( ord( address[1] ) ) + "." + str( ord( address[2] ) ) + "." + str( ord( address[3] ) ) #PORT is build by calculating the 5th byte * 2^8 and adding the 6th one port = str( ( ord( address[4] ) << 8 ) + ord( address[5] ) ) servers.append( ip + ":" + port ) except: pass #in case the address didn't contain 6 bytes, which strangely happens from time to time: ignore it for address in servers: #collect all the informations using the info_get function infs.append( self.info_get( ( address.split( ":" )[0], int( address.split( ":" )[1] ) ) ) ) self.ips, self.infos = servers, infs #switching the lists time.sleep( 30 ) #you can change the reload time here self.reload() def getplayers( self, ip, #should be like "ip:port" colors = None #None: return without colors, False: return with color codes, True: return with irc_colors ): ''' function to return the clients that are currently online on a specific server ''' clients = list() for client in self.infos[ self.ips.index( ip ) ]["clients"][1]: try: clients.append( client.split( '"' )[1] ) except IndexError: pass if colors == True: for i in range( len( clients ) ): clients[i] = irc_color( clients[i] ) if colors == None: for i in range( len( clients ) ): clients[i] = decolor( clients[i] ) return clients def top3( self, gametype, #should be a string like "ca" instagib ): clients = list() ips = list() final = list() for info in self.infos: try: if info["gametype"] == gametype: if ( instagib and info["g_instagib"] == "1" ) or ( not instagib and info["g_instagib"] == "0" ) : clients.append( info["clients"][0] ) ips.append( self.ips[self.infos.index( info )] ) except TypeError: pass except KeyError: pass for i in range( 3 ): final.append( ( ips[clients.index( max( clients ) )], max( clients ) ) ) del clients[clients.index( max( clients ) )] del ips[clients.index( max( clients ) )] return final def getname( self, ip #should be like "ip:port" ): ''' function to return the name of a server (ip) ''' try: if ip in self.ips and self.infos[ self.ips.index( ip ) ]: return self.infos[ self.ips.index( ip ) ]["sv_hostname"] except IndexError: pass return None def getinfo( self, ip, #should be like "ip:port" info #any information in form of a string like: "sv_hostname" ): ''' this returns the information for info ''' if ip in self.ips and self.infos[ self.ips.index( ip ) ]: return self.infos[ self.ips.index( ip ) ][info] return None def getplayer( self, player, #a playername (without colors): doesn't have to be the full name colors = None #None: return without colors, False: return with color codes, True: return with irc_colors ): ''' Searches for a player name in the current server list ''' for info in self.infos: if info != None: try: for cplayer in info["clients"][1]: if player.lower() in decolor( cplayer ).lower(): if colors == True: return ( irc_color( cplayer.split( '"' )[1] ), irc_color( info["sv_hostname"] ), self.ips[ self.infos.index( info ) ] ) if colors == None: return ( decolor( cplayer.split( '"' )[1] ), decolor( info["sv_hostname"] ), self.ips[ self.infos.index( info ) ] ) if colors == False: return ( cplayer.split( '"' )[1] , info["sv_hostname"] , self.ips[ self.infos.index( info ) ] ) except KeyError: pass return None def getip( self, name #doesn't have to be the full name, ignores colors! ): ''' returns the ip to a given port of the server name, ignoring any color codes ''' for info in self.infos: if info != None: try: if name.lower() in decolor( info["sv_hostname"].lower() ): print self.ips return self.ips[ self.infos.index( info )] except KeyError: pass return None if __name__ == "__main__": quer = Query() quer.start() while True: eval( raw_input( ">>> " ) )