#!/usr/bin/env python3
from telnetlib import Telnet
from collections import defaultdict
from datetime import timedelta
import toml
from flask import Flask, jsonify, render_template_string
IDLE_TIMEOUT = timedelta(minutes=5)
app = Flask(__name__)
def parse_ts_response(response):
# entries separated by |'s
entries = response.split("|")
# entries contain key=value pairs separated by spaces
pairs = [[pr.split("=", 1) for pr in ent.split()] for ent in entries]
# rearrange these into maps for convenience
return [{k: v for k, v in pr} for pr in pairs]
def get_users():
with open("secret.toml") as infile:
cfg = toml.load(infile)
login = ("login %s %s\n" % (cfg["user"], cfg["pass"])).encode("utf-8")
with Telnet(cfg["host"], cfg["port"], 5) as tn:
print("connection")
print(tn.read_until(b"\n").decode("utf-8"))
print(tn.read_until(b"\n").decode("utf-8"))
print("----")
tn.write(login)
print("after login")
print(tn.read_until(b"\n").decode("utf-8"))
print("----")
tn.write(b"use 1 -virtual\n")
print("after use")
print(tn.read_until(b"\n").decode("utf-8"))
print("----")
tn.write(b"channellist\n")
response = tn.read_until(b"\n").decode("utf-8")
print("after channellist")
print(response)
print(tn.read_until(b"\n").decode("utf-8"))
print("----")
channel_maps = parse_ts_response(response)
# rearrange the maps into one large channel lookup map
channels = {info["cid"]: info["channel_name"].replace(r"\s", " ") for info in channel_maps}
tn.write(b"clientlist\n")
response = tn.read_until(b"\n").decode("utf-8")
print("after clientlist")
print(response)
print(tn.read_until(b"\n").decode("utf-8"))
print("----")
entry_maps = parse_ts_response(response)
# combine the maps into one large map, ignoring serveradmin query user
client_info = {info["client_nickname"]: info for info in entry_maps if "serveradmin" not in info["client_nickname"]}
for k, v in client_info.items():
tn.write(f"clientinfo clid={v['clid']}\n".encode("utf-8"))
response = tn.read_until(b"\n").decode("utf-8")
print(f"after clientinfo for {k}")
print(response)
print(tn.read_until(b"\n").decode("utf-8"))
print("----")
# info is key=value pairs separated by spaces
pairs = [ent.split("=", 1) for ent in response.split() if "=" in ent]
# rearrange into a map and put in the client_info
v["client_info"] = {k: v for k, v in pairs}
tn.write(b"quit\n")
print("after quit")
print(tn.read_until(b"\n").decode("utf-8"))
users = []
channel_users = defaultdict(list)
for name, info in client_info.items():
user_text = name
audio_status = []
if info["client_info"].get("client_input_muted", "0") == "1":
audio_status.append("Mic muted")
if info["client_info"].get("client_output_muted", "0") == "1":
audio_status.append("Sound muted")
if len(audio_status) > 0:
user_text += f" ({', '.join(audio_status)})"
idle = timedelta(milliseconds=int(info["client_info"].get("client_idle_time", "0")))
if idle >= IDLE_TIMEOUT:
# strip out the sub-second resolution
idle -= timedelta(microseconds=idle.microseconds)
user_text += f" (Idle for {idle})"
users.append(user_text)
channel_users[channels[info["cid"]]].append(user_text)
return sorted(users), {k: sorted(v) for k, v in channel_users.items()}
@app.route("/")
def get_status():
return jsonify({"users": get_users()[0]})
@app.route("/page")
def get_status_page():
# JS adapted from
# https://stackoverflow.com/questions/6152522/how-can-i-make-a-paragraph-bounce-around-in-a-div
return render_template_string("""
Teamspeak Server Status
TeamSpeak Server Status
{% if users|length == 0 %}
No one in teamspeak!
{% else %}
{% for channel, people in users.items() %}
{{ channel }}
{% for user in people %}
{{ user }}
{% endfor %}
{% endfor %}
{% endif %}
""", users=get_users()[1])
if __name__ == "__main__":
app.run("0.0.0.0", 5000, debug=True, threaded=True)