nikdoof/python-ts3
1
0
Fork 0
mirror of https://github.com/nikdoof/python-ts3.git synced 2025-12-17 11:59:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

better error reporting

Browse Source
This commit is contained in:
Krzysztof Jagiello
2011-06-04 20:13:13 +02:00
parent 15c8e3d87e
commit 4911948cc3
1 changed files with 23 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
ts3.py
View File

@@ -29,15 +29,19 @@ import time
import telnetlib import telnetlib
import logging import logging
class ConnectionError(): class ConnectionError(Exception):
def __init__(self, ip, port): def __init__(self, ip, port):
self.ip = ip self.ip = ip
self.port = port self.port = port
def __str__(): def __str__():
return 'Error connecting to host %s port %s' % (self.ip, self.port) return 'Error connecting to host %s port %s.' % (self.ip, self.port,)
ts3_escape = { "\\", r'\\', class NoConnection(Exception):
def __str__():
return 'No connection established.' % (self.ip, self.port,)
ts3_escape = { "\\": r'\\',
'/': r"\/", '/': r"\/",
' ': r'\s', ' ': r'\s',
'|': r'\p', '|': r'\p',
@@ -64,16 +68,16 @@ class TS3Response():
return self.response return self.response
class TS3Proto(): class TS3Proto():
def __init__(self):
self._log = logging.getLogger('%s.%s' % (__name__, self.__class__.__name__))
pass
def connect(self, ip, port, timeout=5): def connect(self, ip, port, timeout=5):
try:
self._telnet = telnetlib.Telnet(ip, port) self._telnet = telnetlib.Telnet(ip, port)
except telnetlib.socket.error:
raise ConnectionError(ip, port)
self._timeout = timeout self._timeout = timeout
self._connected = False self._connected = False
data = self._telnet.read_until("TS3\n", self._timeout) data = self._telnet.read_until("\n", self._timeout)
if data.endswith("TS3\n"): if data.endswith("TS3\n"):
self._connected = True self._connected = True
@@ -81,18 +85,23 @@ class TS3Proto():
return self._connected return self._connected
def disconnect(self): def disconnect(self):
self.check_connection()
self.send_command("quit") self.send_command("quit")
self._telnet.close()
self._connected = False self._connected = False
self._log.info('Disconnected')
def send_command(self, command, keys=None, opts=None): def send_command(self, command, keys=None, opts=None):
cmd = self.construct_command(command, keys=keys, opts=opts) self.check_connection()
self._telnet.write("%s\n" % self.construct_command(command, keys=keys, opts=opts))
self._telnet.write("%s\n" % cmd)
return TS3Response(self._telnet.read_until("\n", self._timeout)) return TS3Response(self._telnet.read_until("\n", self._timeout))
def check_connection(self):
if not self.is_connected:
raise NoConnectionError
def is_connected(self): def is_connected(self):
return self._connected return self._connected
@@ -218,8 +227,6 @@ class TS3Server(TS3Proto):
@type port: int @type port: int
""" """
TS3Proto.__init__(self)
if self.connect(ip, port) and id > 0: if self.connect(ip, port) and id > 0:
self.use(id) self.use(id)
@@ -234,13 +241,7 @@ class TS3Server(TS3Proto):
""" """
response = self.send_command('login', keys={'client_login_name': username, 'client_login_password': password }) response = self.send_command('login', keys={'client_login_name': username, 'client_login_password': password })
return response.is_successful()
if response.is_successful():
self._log.info('Login error: %s.')
return False
else:
self._log.info('Login successful.')
return True
def serverlist(self): def serverlist(self):
""" """