140 lines
4.6 KiB
Python
140 lines
4.6 KiB
Python
"""
|
|
TeamSpeak 6 WebQuery HTTP API Client.
|
|
|
|
Communicates with the TS6 server via the WebQuery HTTP interface
|
|
to retrieve server information, client lists, channels, bans, etc.
|
|
"""
|
|
|
|
import urllib.parse
|
|
import logging
|
|
import requests
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TS6Client:
|
|
"""Client for the TeamSpeak 6 WebQuery HTTP API."""
|
|
|
|
def __init__(self, host: str, port: int, api_key: str, server_id: int = 1, timeout: int = 10):
|
|
self.base_url = f"http://{host}:{port}"
|
|
self.server_id = server_id
|
|
self.api_key = api_key
|
|
self.timeout = timeout
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
"x-api-key": self.api_key,
|
|
})
|
|
|
|
def _request(self, command: str, params: dict = None, use_sid: bool = True) -> list[dict]:
|
|
"""Execute a WebQuery command and return parsed response."""
|
|
if use_sid:
|
|
url = f"{self.base_url}/{self.server_id}/{command}"
|
|
else:
|
|
url = f"{self.base_url}/{command}"
|
|
|
|
if params:
|
|
query_string = "&".join(f"{k}={urllib.parse.quote(str(v))}" for k, v in params.items())
|
|
url = f"{url}?{query_string}"
|
|
|
|
try:
|
|
response = self.session.get(url, timeout=self.timeout)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
if "body" in data:
|
|
body = data["body"]
|
|
if isinstance(body, list):
|
|
return body
|
|
return [body] if body else []
|
|
|
|
if "status" in data and data["status"].get("code", 0) != 0:
|
|
error_msg = data["status"].get("message", "Unknown error")
|
|
logger.error("TS6 API error for '%s': %s", command, error_msg)
|
|
return []
|
|
|
|
return []
|
|
|
|
except requests.exceptions.ConnectionError:
|
|
logger.error("Cannot connect to TS6 WebQuery at %s", self.base_url)
|
|
raise
|
|
except requests.exceptions.Timeout:
|
|
logger.error("Timeout connecting to TS6 WebQuery at %s", self.base_url)
|
|
raise
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error("Request error for '%s': %s", command, e)
|
|
raise
|
|
except ValueError:
|
|
# JSON decode error — try raw text parsing
|
|
return self._parse_raw_response(response.text)
|
|
|
|
@staticmethod
|
|
def _parse_raw_response(text: str) -> list[dict]:
|
|
"""Parse legacy-style key=value response format."""
|
|
results = []
|
|
for entry in text.split("|"):
|
|
item = {}
|
|
for pair in entry.strip().split(" "):
|
|
if "=" in pair:
|
|
key, value = pair.split("=", 1)
|
|
# Unescape TS6 special chars
|
|
value = value.replace("\\s", " ").replace("\\p", "|").replace("\\/", "/")
|
|
item[key] = value
|
|
else:
|
|
item[pair] = ""
|
|
if item:
|
|
results.append(item)
|
|
return results
|
|
|
|
def is_alive(self) -> bool:
|
|
"""Check if the TS6 server is reachable."""
|
|
try:
|
|
self._request("version", use_sid=False)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def version(self) -> dict:
|
|
"""Get server version info."""
|
|
result = self._request("version", use_sid=False)
|
|
return result[0] if result else {}
|
|
|
|
def server_info(self) -> dict:
|
|
"""Get detailed server information."""
|
|
result = self._request("serverinfo")
|
|
return result[0] if result else {}
|
|
|
|
def client_list(self) -> list[dict]:
|
|
"""Get list of connected clients with extended info."""
|
|
return self._request("clientlist", params={
|
|
"-uid": "",
|
|
"-away": "",
|
|
"-voice": "",
|
|
"-groups": "",
|
|
"-info": "",
|
|
"-country": "",
|
|
})
|
|
|
|
def channel_list(self) -> list[dict]:
|
|
"""Get list of channels."""
|
|
return self._request("channellist", params={
|
|
"-topic": "",
|
|
"-flags": "",
|
|
"-voice": "",
|
|
})
|
|
|
|
def ban_list(self) -> list[dict]:
|
|
"""Get list of active bans."""
|
|
try:
|
|
return self._request("banlist")
|
|
except Exception:
|
|
return []
|
|
|
|
def server_group_list(self) -> list[dict]:
|
|
"""Get list of server groups."""
|
|
return self._request("servergrouplist")
|
|
|
|
def whoami(self) -> dict:
|
|
"""Get info about the current query connection."""
|
|
result = self._request("whoami")
|
|
return result[0] if result else {}
|