from twisted.internet import stdio , reactor
from twisted.protocols import basic
from twisted.web import client
class URLGetCommandProtocol( basic.LineReceiver ):
delimiter = '\n' # unix terminal style newlines. remove this line if using with Telnet
def connectionMade( self ):
self.sendLine( "Type 'help' for help." )
def lineReceived( self , line ):
if not line: return # Ignore blank lines
commandParts = line.split( )
command = commandParts[ 0 ].lower( ) # first part is the command
args = commandParts[ 1: ] # second parts are the argument(s)
# Dispatching: to implement a new command add another do_ method.
try:
method = getattr( self , 'do_' + command )
except AttributeError , e:
self.sendLine( 'Error: no such command.' )
else:
try:
method( *args ) # the method name created by concatenating the do_ with the user provided command
except Exception , e:
self.sendLine( 'Error: ' + str( e ) )
def do_help( self , command = None ):
"""help [command]: List commands, or show help on the given command"""
if command:
self.sendLine( getattr( self , 'do_' + command ).__doc__ )
else:
commands = [ cmd[ 3: ] for cmd in dir( self ) if cmd.startswith( 'do_' ) ]
self.sendLine( "Valid commands: " + " ".join( commands ) )
def do_exit( self ):
self.sendLine( 'Exiting.' )
self.transport.loseConnection( )
def do_get( self, url ):
"""get <url>: download the url"""
client.getPage( url ).addCallback( self.__checkSuccess ).addErrback( self.__checkFailure )
def __checkSuccess( self, pageData ):
self.sendLine( "Success: got %i bytes." % len( pageData ) )
def __checkFailure( self, failure ):
self.sendLine( "Failure: " + failure.getErrorMessage( ) )
def connectionLost( self, reason ):
reactor.stop( ) # stop the reactor, only because this is meant to be run in Stdio.
if __name__ == "__main__":
stdio.StandardIO( URLGetCommandProtocol( ) )
reactor.run( )