1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
#!/usr/bin/env python
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Example using stdio, Deferreds, LineReceiver and twisted.web.client.
Note that the WebCheckerCommandProtocol protocol could easily be used in e.g.
a telnet server instead; see the comments for details.
Based on an example by Abe Fettig.
"""
from twisted.internet import reactor, stdio
from twisted.protocols import basic
from twisted.web import client
class WebCheckerCommandProtocol(basic.LineReceiver):
delimiter = b"\n" # unix terminal style newlines. remove this line
# for use with Telnet
def connectionMade(self):
self.sendLine(b"Web checker console. Type 'help' for help.")
def lineReceived(self, line):
# Ignore blank lines
if not line:
return
line = line.decode("ascii")
# Parse the command
commandParts = line.split()
command = commandParts[0].lower()
args = commandParts[1:]
# Dispatch the command to the appropriate method. Note that all you
# need to do to implement a new command is add another do_* method.
try:
method = getattr(self, "do_" + command)
except AttributeError:
self.sendLine(b"Error: no such command.")
else:
try:
method(*args)
except Exception as e:
self.sendLine(b"Error: " + str(e).encode("ascii"))
def do_help(self, command=None):
"""help [command]: List commands, or show help on the given command"""
if command:
doc = getattr(self, "do_" + command).__doc__
self.sendLine(doc.encode("ascii"))
else:
commands = [
cmd[3:].encode("ascii") for cmd in dir(self) if cmd.startswith("do_")
]
self.sendLine(b"Valid commands: " + b" ".join(commands))
def do_quit(self):
"""quit: Quit this session"""
self.sendLine(b"Goodbye.")
self.transport.loseConnection()
def do_check(self, url):
"""check <url>: Attempt to download the given web page"""
url = url.encode("ascii")
client.Agent(reactor).request(b"GET", url).addCallback(
client.readBody
).addCallback(self.__checkSuccess).addErrback(self.__checkFailure)
def __checkSuccess(self, pageData):
msg = f"Success: got {len(pageData)} bytes."
self.sendLine(msg.encode("ascii"))
def __checkFailure(self, failure):
msg = "Failure: " + failure.getErrorMessage()
self.sendLine(msg.encode("ascii"))
def connectionLost(self, reason):
# stop the reactor, only because this is meant to be run in Stdio.
reactor.stop()
if __name__ == "__main__":
stdio.StandardIO(WebCheckerCommandProtocol())
reactor.run()
|
