-
Notifications
You must be signed in to change notification settings - Fork 2
/
interpreter.py
209 lines (177 loc) · 6.48 KB
/
interpreter.py
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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
import os
from cmd import Cmd
from client import FtpClient
class FtpInterpreter(Cmd):
"""
FTP client command line utility.
"""
def __init__(self, debug=False):
Cmd.__init__(self)
self.intro = ('FTP Client. Start typing help or ? to see available '
'commands.')
self.prompt = 'FTP > '
self._ftp_client = FtpClient(debug=debug)
def _update_prompt(self):
prompt = 'FTP'
if self._ftp_client.host is not None:
prompt = '{} {}'.format(prompt, self._ftp_client.host)
if self._ftp_client.user is not None:
prompt = '{} ({})'.format(prompt, self._ftp_client.user)
self.prompt = '{} > '.format(prompt)
def _perform_ftp_command(self, command, *args):
method = getattr(self._ftp_client, command)
try:
response = method(*args)
except (FtpClient.TimeoutException,
FtpClient.UnknownHostException,
FtpClient.ConnectionRefusedException) as e:
response = e.msg
except FtpClient.NotConnectedException as e:
response = e.msg
response = ('{}\nPlease connect to an FTP server using'
' the `connect` command').format(response)
except FtpClient.NotAuthenticatedException as e:
response = e.msg
response = ('{}\nPlease authenticate using the `login` command.')\
.format(response)
except FtpClient.LocalIOException as e:
response = e.msg
response = ('{}\nSomething went wrong trying to {} the file,'
' please try again.').format(response, command)
return response
def emptyline(self):
pass
def do_connect(self, host):
"""
Command to connect to an FTP server in the specified host.
Args:
host (str): The host to connect to.
"""
response = self._perform_ftp_command('connect', host)
print response
self._update_prompt()
def do_login(self, *args):
"""
Command to login with user and password in the connected FTP host.
"""
user = ''
while not user:
user = raw_input('User: ')
password = ''
while not password:
password = raw_input('Password: ')
response = self._perform_ftp_command('login', user, password)
print response
self._update_prompt()
def do_logout(self, *args):
"""
Command to logout the current user from the connected FTP host.
"""
self._perform_ftp_command('logout')
self._update_prompt()
def do_list(self, filename):
"""
Command to perform LIST command on the connected FTP host.
Args:
filename (str): Name of file or directory to retrieve info for.
"""
response = self._perform_ftp_command('list', filename)
print response
def do_disconnect(self, *args):
"""
Command to disconnect from connected FTP host.
"""
response = self._perform_ftp_command('disconnect')
print response
self._update_prompt()
def do_retrieve(self, *args):
"""
Command to retrieve a file from the connected FTP host and store
it locally.
"""
filename = ''
while not filename:
filename = raw_input('Remote file: ')
local_filename = ''
while not local_filename:
local_filename = raw_input('Local file: ')
response = self._perform_ftp_command('retrieve', filename,
local_filename)
local_file = None
if isinstance(response, tuple):
response, local_file = response
print response
if local_file is not None:
local_path = os.path.realpath(local_file.name)
print 'Local file created: {}'.format(local_path)
def do_store(self, *args):
"""
Command to send a local file to the connected FTP host.
"""
local_filename = ''
while not local_filename:
local_filename = raw_input('Local file: ')
filename = ''
while not filename:
filename = raw_input('Remote file: ')
response = self._perform_ftp_command('store', local_filename,
filename)
print response
def do_pwd(self, *args):
"""
Command to retrieve the current directory on the connected FTP host.
"""
response = self._perform_ftp_command('pwd')
print response
def do_cwd(self, directory):
"""
Command to change current directory on the connected FTP host.
Args:
directory (str): Name of directory to work on.
"""
response = self._perform_ftp_command('cwd', directory)
print response
def do_cdup(self, *args):
"""
Command to set parent directory as current working directory
on the connected FTP host.
"""
response = self._perform_ftp_command('cdup')
print response
def do_mkdir(self, directory):
"""
Command to create directory on the connected FTP host.
Args:
directory (str): Name of directory to create.
"""
response = self._perform_ftp_command('mkdir', directory)
print response
def do_rm(self, filename):
"""
Command to remove file on the connected FTP host.
Args:
filename (str): Name of file to delete.
"""
response = self._perform_ftp_command('rm', filename)
print response
def do_rmdir(self, directory):
"""
Command to remove directory on the connected FTP host.
Args:
directory (str): Name of directory to delete.
"""
response = self._perform_ftp_command('rmdir', directory)
print response
def do_rename(self, *args):
"""
Command to rename a file or directory on the connected FTP host.
"""
original_filename = ''
while not original_filename:
original_filename = raw_input('Name of original remote file: ')
new_filename = ''
while not new_filename:
new_filename = raw_input('New name for remote file: ')
response = self._perform_ftp_command('rename', original_filename,
new_filename)
print response