<s> import string <EOL> import logging <EOL> from twisted . conch import recvline <EOL> IDENTCHARS = string . ascii_letters + string . digits + '<STR_LIT:_>' <EOL> def str2num ( s ) : <EOL> '<STR_LIT>' <EOL> try : <EOL> return int ( s ) <EOL> except ValueError : <EOL> pass <EOL> try : <EOL> return float ( s ) <EOL> except ValueError : <EOL> return s <EOL> class CmdProtocol ( recvline . HistoricRecvLine ) : <EOL> """<STR_LIT>""" <EOL> identchars = IDENTCHARS <EOL> motd = '<STR_LIT>' <EOL> nohelp = "<STR_LIT>" <EOL> helpHeaders = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:=>' , } <EOL> prompt = '<STR_LIT>' <EOL> sessionLineCallback = None <EOL> sessionCompletitions = None <EOL> baseCommands = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> def __init__ ( self , log_category = '<STR_LIT>' ) : <EOL> recvline . HistoricRecvLine . __init__ ( self ) <EOL> self . commands = [ ] <EOL> self . log = logging . getLogger ( log_category ) <EOL> self . log . propagate = False <EOL> def initializeScreen ( self ) : <EOL> '<STR_LIT>' <EOL> self . terminal . reset ( ) <EOL> self . setInsertMode ( ) <EOL> def terminalSize ( self , width , height ) : <EOL> self . width = width <EOL> self . height = height <EOL> self . drawInputLine ( ) <EOL> def drawInputLine ( self ) : <EOL> '<STR_LIT>' <EOL> self . terminal . write ( self . prompt + '<STR_LIT>' . join ( self . lineBuffer ) ) <EOL> def drawMotd ( self ) : <EOL> self . sendData ( self . motd , False ) <EOL> self . sendData ( '<STR_LIT>' % self . sessionRef , False ) <EOL> def connectionMade ( self , motd = True ) : <EOL> recvline . HistoricRecvLine . connectionMade ( self ) <EOL> transport = self . terminal . transport . transport <EOL> self . peer = transport . getPeer ( ) <EOL> self . factory . sessionRef += <NUM_LIT:1> <EOL> self . factory . sessionsOnline += <NUM_LIT:1> <EOL> self . sessionRef = self . factory . sessionRef <EOL> self . factory . sessions [ self . sessionRef ] = self <EOL> if motd : <EOL> self . drawMotd ( ) <EOL> self . log . info ( <EOL> '<STR_LIT>' , self . sessionRef , self . peer . host , self . peer . port ) <EOL> def connectionLost ( self , reason ) : <EOL> recvline . HistoricRecvLine . connectionLost ( self , reason ) <EOL> self . factory . sessionsOnline -= <NUM_LIT:1> <EOL> del self . factory . sessions [ self . sessionRef ] <EOL> if reason is not None : <EOL> self . log . info ( '<STR_LIT>' , self . sessionRef , reason . value ) <EOL> else : <EOL> self . log . info ( '<STR_LIT>' , self . sessionRef ) <EOL> def sendData ( self , data = None , prompt = None , append = '<STR_LIT>' ) : <EOL> self . log . debug ( '<STR_LIT>' , self . sessionRef , data ) <EOL> if data is not None : <EOL> self . terminal . write ( data ) <EOL> self . terminal . nextLine ( ) <EOL> if prompt is not False : <EOL> if prompt is None : <EOL> prompt = self . prompt <EOL> self . terminal . write ( prompt + append ) <EOL> elif append != '<STR_LIT>' : <EOL> self . terminal . write ( append ) <EOL> def parseline ( self , line ) : <EOL> """<STR_LIT>""" <EOL> if not line : <EOL> self . log . debug ( <EOL> '<STR_LIT>' , self . sessionRef , line ) <EOL> return None , None , line <EOL> elif line [ <NUM_LIT:0> ] == '<STR_LIT:?>' : <EOL> line = '<STR_LIT>' + line [ <NUM_LIT:1> : ] <EOL> i , n = <NUM_LIT:0> , len ( line ) <EOL> while i < n and line [ i ] in self . identchars : <EOL> i = i + <NUM_LIT:1> <EOL> cmd , arg = line [ : i ] , line [ i : ] . strip ( ) <EOL> self . log . debug ( <EOL> '<STR_LIT>' , self . sessionRef , cmd , arg , line ) <EOL> return cmd , arg , line <EOL> def lineReceived ( self , line ) : <EOL> line = line . strip ( ) <EOL> self . log . debug ( '<STR_LIT>' , self . sessionRef , line ) <EOL> cmd , arg , line = self . parseline ( line ) <EOL> if self . sessionLineCallback is not None : <EOL> return self . sessionLineCallback ( cmd , arg , line ) <EOL> if not line : <EOL> return self . sendData ( ) <EOL> if cmd is None or cmd not in self . findCommands ( ) : <EOL> return self . default ( line ) <EOL> funcName = '<STR_LIT>' + cmd <EOL> try : <EOL> func = getattr ( self , funcName ) <EOL> except AttributeError : <EOL> return self . default ( line ) <EOL> self . log . debug ( '<STR_LIT>' , self . sessionRef , funcName , arg ) <EOL> return func ( arg ) <EOL> def findCommands ( self , prefix = None ) : <EOL> if self . sessionCompletitions is not None : <EOL> completetions = self . sessionCompletitions <EOL> else : <EOL> completetions = self . commands + self . baseCommands <EOL> if prefix is None : <EOL> return completetions <EOL> foundCommands = [ ] <EOL> for availableCmd in completetions : <EOL> if availableCmd . find ( prefix ) == <NUM_LIT:0> : <EOL> foundCommands . append ( availableCmd ) <EOL> return foundCommands <EOL> def handle_TAB ( self ) : <EOL> line = '<STR_LIT>' . join ( self . lineBuffer ) <EOL> self . log . debug ( '<STR_LIT>' , self . sessionRef , line ) <EOL> cmd , arg , line = self . parseline ( line ) <EOL> if cmd is None : <EOL> return self . sendData ( '<STR_LIT:\n>' + '<STR_LIT:U+0020>' . join ( self . findCommands ( ) ) ) <EOL> elif cmd is not None and arg == '<STR_LIT>' : <EOL> completions = self . findCommands ( cmd ) <EOL> if len ( completions ) > <NUM_LIT:1> : <EOL> return self . sendData ( '<STR_LIT:\n>' + '<STR_LIT:U+0020>' . join ( completions ) , append = cmd ) <EOL> elif len ( completions ) == <NUM_LIT:1> and cmd != completions [ <NUM_LIT:0> ] : <EOL> completetion = completions [ <NUM_LIT:0> ] + '<STR_LIT:U+0020>' <EOL> self . lineBuffer = list ( completetion ) <EOL> self . lineBufferIndex = len ( self . lineBuffer ) <EOL> return self . sendData ( data = None , prompt = False , append = completetion [ len ( cmd ) : ] ) <EOL> def default ( self , line ) : <EOL> self . sendData ( '<STR_LIT>' % line ) <EOL> def do_quit ( self , arg ) : <EOL> '<STR_LIT>' <EOL> self . terminal . loseConnection ( ) <EOL> def do_help ( self , arg ) : <EOL> '<STR_LIT>' <EOL> if arg : <EOL> if arg not in self . commands + self . baseCommands : <EOL> return self . sendData ( "<STR_LIT:%s>" % str ( self . nohelp % ( arg , ) ) ) <EOL> DOC = '<STR_LIT>' <EOL> try : <EOL> doc = getattr ( self , '<STR_LIT>' + arg ) . __doc__ <EOL> if doc : <EOL> DOC += doc <EOL> extended_doc = getattr ( self , '<STR_LIT>' + arg ) . __extended_doc__ <EOL> if extended_doc : <EOL> DOC += '<STR_LIT:\n>' + extended_doc <EOL> except Exception : <EOL> if DOC == '<STR_LIT>' : <EOL> return self . sendData ( "<STR_LIT:%s>" % str ( self . nohelp % ( arg , ) ) ) <EOL> return self . sendData ( "<STR_LIT:%s>" % str ( DOC ) ) <EOL> else : <EOL> helpText = self . helpHeaders [ '<STR_LIT>' ] + '<STR_LIT:\n>' + self . helpHeaders [ '<STR_LIT>' ] * len ( <EOL> self . helpHeaders [ '<STR_LIT>' ] ) <EOL> for cmd in self . commands : <EOL> helpText += "<STR_LIT:\n>" <EOL> helpText += '<STR_LIT:%s>' % cmd . ljust ( <NUM_LIT:20> ) <EOL> doc = getattr ( self , '<STR_LIT>' + cmd ) . __doc__ <EOL> if doc : <EOL> helpText += str ( doc ) <EOL> else : <EOL> helpText += "<STR_LIT:%s>" % str ( self . nohelp % ( cmd , ) ) <EOL> helpText += '<STR_LIT>' + self . helpHeaders [ '<STR_LIT>' ] + '<STR_LIT:\n>' + self . helpHeaders [ '<STR_LIT>' ] * len ( <EOL> self . helpHeaders [ '<STR_LIT>' ] ) <EOL> for cmd in self . baseCommands : <EOL> helpText += "<STR_LIT:\n>" <EOL> helpText += '<STR_LIT:%s>' % cmd . ljust ( <NUM_LIT:20> ) <EOL> doc = getattr ( self , '<STR_LIT>' + cmd ) . __doc__ <EOL> if doc : <EOL> helpText += str ( doc ) <EOL> else : <EOL> helpText += "<STR_LIT:%s>" % str ( self . nohelp % ( cmd , ) ) <EOL> return self . sendData ( helpText ) </s>
