typehint teamspeak3

This commit is contained in:
Joel Falknau committed 2026-10-08 22:20:48 +10:00
1 parent 9241daf9f1
commit 7d5ab291ca
9 files changed
+118 -68

No files matched your search

@@ -1,3 +1,7 @@
__lazy_modules__ = ["django.db.models", "django.http"]
from typing import TYPE_CHECKING
from django.contrib import admin
from allianceauth.groupmanagement.models import Group, ReservedGroupName
@@ -5,6 +9,10 @@ from allianceauth.groupmanagement.models import Group, ReservedGroupName
from ...admin import ServicesUserAdmin
from .models import AuthTS, StateGroup, Teamspeak3User, TSgroup
if TYPE_CHECKING:
from django.db.models import Field
from django.http import HttpRequest
@admin.register(Teamspeak3User)
class Teamspeak3UserAdmin(ServicesUserAdmin):
@@ -27,12 +35,12 @@ class AuthTSgroupAdmin(admin.ModelAdmin):
fields = ('auth_group', 'ts_group')
filter_horizontal = ('ts_group',)
def formfield_for_foreignkey(self, db_field, request, **kwargs):
def formfield_for_foreignkey(self, db_field: "Field", request: "HttpRequest", **kwargs):
if db_field.name == 'auth_group':
kwargs['queryset'] = Group.objects.exclude(name__in=ReservedGroupName.objects.values_list('name', flat=True))
return super().formfield_for_foreignkey(db_field, request, **kwargs)
def formfield_for_manytomany(self, db_field, request, **kwargs):
def formfield_for_manytomany(self, db_field: "Field", request: "HttpRequest", **kwargs):
if db_field.name == 'ts_group':
kwargs['queryset'] = TSgroup.objects.exclude(ts_group_name__in=ReservedGroupName.objects.values_list('name', flat=True))
return super().formfield_for_manytomany(db_field, request, **kwargs)
@@ -40,7 +48,7 @@ class AuthTSgroupAdmin(admin.ModelAdmin):
@admin.display(
description='ts groups'
)
def _ts_group(self, obj):
def _ts_group(self, obj: AuthTS) -> list[TSgroup]:
return list(obj.ts_group.all().order_by('ts_group_id'))
# _ts_group.admin_order_field = 'profile__state'
@@ -7,6 +7,6 @@ class Teamspeak3ServiceConfig(AppConfig):
label = 'teamspeak3'
verbose_name = _('TeamSpeak 3 Service')
def ready(self):
def ready(self) -> None:
from allianceauth.services.modules.teamspeak3 import \
signals # noqa: F401
@@ -1,4 +1,7 @@
__lazy_modules__ = ["django.contrib.auth.models", "django.http"]
import logging
from typing import TYPE_CHECKING
from django.conf import settings
from django.template.loader import render_to_string
@@ -9,11 +12,15 @@ from allianceauth.services.hooks import ServicesHook
from .tasks import Teamspeak3Tasks
from .urls import urlpatterns
if TYPE_CHECKING:
from django.contrib.auth.models import User
from django.http import HttpRequest
logger = logging.getLogger(__name__)
class Teamspeak3Service(ServicesHook):
def __init__(self):
def __init__(self) -> None:
ServicesHook.__init__(self)
self.name = 'teamspeak3'
self.urlpatterns = urlpatterns
@@ -21,27 +28,27 @@ class Teamspeak3Service(ServicesHook):
self.access_perm = 'teamspeak3.access_teamspeak3'
self.name_format = '[{corp_ticker}]{character_name}'
def delete_user(self, user, notify_user=False):
def delete_user(self, user: "User", notify_user: bool = False) -> bool:
logger.debug(f'Deleting user {user} {self.name} account')
return Teamspeak3Tasks.delete_user(user, notify_user=notify_user)
def update_groups(self, user):
def update_groups(self, user: "User") -> None:
logger.debug(f'Updating {self.name} groups for {user}')
Teamspeak3Tasks.update_groups.delay(user.pk)
def validate_user(self, user):
def validate_user(self, user: "User") -> None:
logger.debug(f'Validating user {user} {self.name} account')
if Teamspeak3Tasks.has_account(user) and not self.service_active_for_user(user):
self.delete_user(user, notify_user=True)
def update_all_groups(self):
def update_all_groups(self) -> None:
logger.debug(f'Update all {self.name} groups called')
Teamspeak3Tasks.update_all_groups.delay()
def service_active_for_user(self, user):
def service_active_for_user(self, user: "User") -> bool:
return user.has_perm(self.access_perm)
def render_services_ctrl(self, request):
def render_services_ctrl(self, request: "HttpRequest") -> str:
authinfo = {'teamspeak3_uid': '',
'teamspeak3_perm_key': '',}
if Teamspeak3Tasks.has_account(request.user):
@@ -54,5 +61,5 @@ class Teamspeak3Service(ServicesHook):
@hooks.register('services_hook')
def register_service():
def register_service() -> Teamspeak3Service:
return Teamspeak3Service()
@@ -1,3 +1,5 @@
from typing import Any
from django import forms
from django.utils.translation import gettext_lazy as _
@@ -7,7 +9,7 @@ from .manager import Teamspeak3Manager
class TeamspeakJoinForm(forms.Form):
username = forms.CharField(widget=forms.HiddenInput())
def clean(self):
def clean(self) -> dict[str, Any]:
with Teamspeak3Manager() as ts3man:
if ts3man._get_userid(self.cleaned_data['username']):
return self.cleaned_data
@@ -1,4 +1,7 @@
__lazy_modules__ = ["django.contrib.auth.models", "types"]
import logging
from typing import TYPE_CHECKING, Any
from django.conf import settings
@@ -7,39 +10,46 @@ from allianceauth.groupmanagement.models import ReservedGroupName
from .models import TSgroup
from .util.ts3 import TeamspeakError, TS3Server
if TYPE_CHECKING:
from types import TracebackType
from django.contrib.auth.models import User
logger = logging.getLogger(__name__)
class Teamspeak3Manager:
def __init__(self):
def __init__(self) -> None:
self._server = None
@property
def server(self):
def server(self) -> TS3Server:
if self._server is not None and self._server._connected:
return self._server
else:
raise ValueError("Teamspeak not connected")
def connect(self):
def connect(self) -> "Teamspeak3Manager":
self._server = self.__get_created_server()
return self
def disconnect(self):
def disconnect(self) -> None:
self._server.disconnect()
self._server = None
def __enter__(self):
def __enter__(self) -> "Teamspeak3Manager":
logger.debug("Entering with statement, connecting")
self.connect()
return self
def __exit__(self, _type, value, traceback):
def __exit__(
self, _type: type[BaseException] | None, value: BaseException | None, traceback: "TracebackType | None"
) -> None:
logger.debug("Exiting with statement, cleaning up")
self.disconnect()
@staticmethod
def __get_created_server():
def __get_created_server() -> TS3Server:
server = TS3Server(settings.TEAMSPEAK3_SERVER_IP, settings.TEAMSPEAK3_SERVER_PORT)
server.login(settings.TEAMSPEAK3_SERVERQUERY_USER, settings.TEAMSPEAK3_SERVERQUERY_PASSWORD)
server.use(settings.TEAMSPEAK3_VIRTUAL_SERVER)
@@ -47,12 +57,12 @@ class Teamspeak3Manager:
return server
@staticmethod
def __santatize_username(username):
def __santatize_username(username: str) -> str:
sanatized = username.replace(" ", "_")
sanatized = sanatized.replace("'", "-")
return sanatized
def _get_userid(self, uid):
def _get_userid(self, uid: str):
logger.debug(f"Looking for uid {uid} on TS3 server.")
try:
ret = self.server.send_command('customsearch', {'ident': 'sso_uid', 'pattern': uid})
@@ -64,7 +74,7 @@ class Teamspeak3Manager:
raise e
return None
def _group_id_by_name(self, groupname):
def _group_id_by_name(self, groupname: str):
logger.debug(f"Looking for group {groupname} on TS3 server.")
group_cache = self.server.send_command('servergrouplist')
logger.debug(f"Received group cache from server: {group_cache}")
@@ -78,7 +88,7 @@ class Teamspeak3Manager:
logger.debug(f"Group {groupname} not found on server.")
return None
def _create_group(self, groupname):
def _create_group(self, groupname: str):
logger.debug(f"Creating group {groupname} on TS3 server.")
sgid = self._group_id_by_name(groupname)
if not sgid:
@@ -98,7 +108,7 @@ class Teamspeak3Manager:
logger.info(f"Created group on TS3 server with name {groupname} and id {sgid}")
return sgid
def _user_group_list(self, cldbid):
def _user_group_list(self, cldbid) -> dict[str, Any]:
logger.debug(f"Retrieving group list for user with id {cldbid}")
Teamspeak3Manager.__get_created_server()
try:
@@ -121,7 +131,7 @@ class Teamspeak3Manager:
logger.debug(f"Returning name/id pairing: {outlist}")
return outlist
def _group_list(self):
def _group_list(self) -> dict[str, Any]:
logger.debug("Retrieving group list on TS3 server.")
group_cache = self.server.send_command('servergrouplist')
logger.debug(f"Received group cache from server: {group_cache}")
@@ -137,7 +147,7 @@ class Teamspeak3Manager:
logger.debug(f"Returning name/id pairing: {outlist}")
return outlist
def _add_user_to_group(self, uid, groupid):
def _add_user_to_group(self, uid, groupid) -> None:
logger.debug(f"Adding group id {groupid} to TS3 user id {uid}")
user_groups = self._user_group_list(uid)
@@ -147,7 +157,7 @@ class Teamspeak3Manager:
{'sgid': str(groupid), 'cldbid': uid})
logger.info(f"Added user id {uid} to group id {groupid} on TS3 server.")
def _remove_user_from_group(self, uid, groupid):
def _remove_user_from_group(self, uid, groupid) -> None:
logger.debug(f"Removing group id {groupid} from TS3 user id {uid}")
user_groups = self._user_group_list(uid)
@@ -157,7 +167,7 @@ class Teamspeak3Manager:
{'sgid': str(groupid), 'cldbid': uid})
logger.info(f"Removed user id {uid} from group id {groupid} on TS3 server.")
def _sync_ts_group_db(self):
def _sync_ts_group_db(self) -> None:
try:
remote_groups = self._group_list()
managed_groups = {g:int(remote_groups[g]) for g in remote_groups if g in set(remote_groups.keys()) - set(ReservedGroupName.objects.values_list('name', flat=True))}
@@ -178,7 +188,7 @@ class Teamspeak3Manager:
except Exception:
logger.exception("An unhandled exception has occurred while syncing TS groups.")
def add_user(self, user, fmt_name):
def add_user(self, user: "User", fmt_name: str) -> tuple[str, str]:
username_clean = self.__santatize_username(fmt_name[:30])
logger.debug(f"Adding user to TS3 server with cleaned username {username_clean}")
server_groups = self._group_list()
@@ -205,7 +215,7 @@ class Teamspeak3Manager:
logger.exception(f"Failed to add teamspeak user {username_clean} - received response: {ret}")
return "", ""
def delete_user(self, uid):
def delete_user(self, uid: str) -> bool:
user = self._get_userid(uid)
logger.debug(f"Deleting user {user} with id {uid} from TS3 server.")
if user:
@@ -238,18 +248,18 @@ class Teamspeak3Manager:
logger.warning(f"User with id {uid} not found on TS3 server. Assuming succesful deletion.")
return True
def check_user_exists(self, uid):
def check_user_exists(self, uid: str) -> bool:
if self._get_userid(uid):
return True
return False
def generate_new_permissionkey(self, uid, user, username):
def generate_new_permissionkey(self, uid: str, user: "User", username: str) -> tuple[str, str]:
logger.debug(f"Re-issuing permission key for user id {uid}")
self.delete_user(uid)
return self.add_user(user, username)
def update_groups(self, uid, ts_groups):
def update_groups(self, uid: str, ts_groups: dict[str, Any]) -> None:
logger.debug(f"Updating uid {uid} TS3 groups {ts_groups}")
userid = self._get_userid(uid)
addgroups = []
@@ -1,4 +1,7 @@
__lazy_modules__ = ["allianceauth.authentication.models", "django.contrib.auth.models"]
import logging
from typing import TYPE_CHECKING
from django.db import transaction
from django.db.models.signals import m2m_changed, post_delete, post_save
@@ -9,29 +12,34 @@ from allianceauth.authentication.signals import state_changed
from .models import AuthTS, StateGroup
from .tasks import Teamspeak3Tasks
if TYPE_CHECKING:
from django.contrib.auth.models import User
from allianceauth.authentication.models import State
logger = logging.getLogger(__name__)
def trigger_all_ts_update():
def trigger_all_ts_update() -> None:
logger.debug("Triggering update_all_groups")
Teamspeak3Tasks.update_all_groups()
@receiver(m2m_changed, sender=AuthTS.ts_group.through)
def m2m_changed_authts_group(sender, instance, action, *args, **kwargs):
def m2m_changed_authts_group(sender: type, instance: AuthTS, action: str, *args, **kwargs) -> None:
logger.debug(f"Received m2m_changed from {instance} ts_group with action {action}")
if action == "post_add" or action == "post_remove":
transaction.on_commit(trigger_all_ts_update)
@receiver(post_save, sender=AuthTS)
def post_save_authts(sender, instance, *args, **kwargs):
def post_save_authts(sender: type[AuthTS], instance: AuthTS, *args, **kwargs) -> None:
logger.debug(f"Received post_save from {instance}")
transaction.on_commit(trigger_all_ts_update)
@receiver(post_delete, sender=AuthTS)
def post_delete_authts(sender, instance, *args, **kwargs):
def post_delete_authts(sender: type[AuthTS], instance: AuthTS, *args, **kwargs) -> None:
logger.debug(f"Received post_delete signal from {instance}")
transaction.on_commit(trigger_all_ts_update)
@@ -42,8 +50,8 @@ post_delete.connect(post_delete_authts, sender=StateGroup)
@receiver(state_changed)
def check_groups_on_state_change(sender, user, state, **kwargs):
def trigger_update():
def check_groups_on_state_change(sender, user: "User", state: "State", **kwargs) -> None:
def trigger_update() -> None:
Teamspeak3Tasks.update_groups.delay(user.pk)
logger.debug(f"Received state_changed signal from {user}")
transaction.on_commit(trigger_update)
@@ -1,4 +1,7 @@
__lazy_modules__ = ["celery"]
import logging
from typing import TYPE_CHECKING
from celery import shared_task
@@ -13,15 +16,18 @@ from .manager import Teamspeak3Manager
from .models import AuthTS, Teamspeak3User, TSgroup, UserTSgroup
from .util.ts3 import TeamspeakError
if TYPE_CHECKING:
from celery import Task
logger = logging.getLogger(__name__)
class Teamspeak3Tasks:
def __init__(self):
def __init__(self) -> None:
pass
@classmethod
def delete_user(cls, user, notify_user=False):
def delete_user(cls, user: User, notify_user: bool = False) -> bool:
if cls.has_account(user):
logger.debug(f"User {user} has TS3 account {user.teamspeak3.uid}. Deleting.")
with Teamspeak3Manager() as ts3man:
@@ -33,7 +39,7 @@ class Teamspeak3Tasks:
return False
@staticmethod
def has_account(user):
def has_account(user: User) -> bool:
try:
return user.teamspeak3.uid != ''
except ObjectDoesNotExist:
@@ -41,13 +47,13 @@ class Teamspeak3Tasks:
@staticmethod
@shared_task
def run_ts3_group_update():
def run_ts3_group_update() -> None:
logger.debug("TS3 installed. Syncing local group objects.")
with Teamspeak3Manager() as ts3man:
ts3man._sync_ts_group_db()
@staticmethod
def disable():
def disable() -> None:
logger.info("Deleting all Teamspeak3Users")
Teamspeak3User.objects.all().delete()
logger.info("Deleting all UserTSgroup models")
@@ -60,7 +66,7 @@ class Teamspeak3Tasks:
@staticmethod
@shared_task(bind=True, name="teamspeak3.update_groups", base=QueueOnce)
def update_groups(self, pk):
def update_groups(self: "Task", pk: int) -> None:
user = User.objects.get(pk=pk)
logger.debug(f"Updating user {user} teamspeak3 groups")
if Teamspeak3Tasks.has_account(user):
@@ -87,12 +93,12 @@ class Teamspeak3Tasks:
@staticmethod
@shared_task(name="teamspeak3.update_all_groups")
def update_all_groups():
def update_all_groups() -> None:
logger.debug("Updating ALL teamspeak3 groups")
for user in Teamspeak3User.objects.exclude(uid__exact=''):
Teamspeak3Tasks.update_groups.delay(user.user_id)
@staticmethod
def get_username(user):
def get_username(user: User) -> str:
from .auth_hooks import Teamspeak3Service
return NameFormatter(Teamspeak3Service(), user).format_name()
@@ -1,10 +1,11 @@
import logging
from typing import Any
from telnetlib3 import telnetlib
class ConnectionError:
def __init__(self, ip, port):
def __init__(self, ip: str, port: int) -> None:
self.ip = ip
self.port = port
@@ -32,12 +33,12 @@ class TS3Proto:
EOL = b'\n\r'
def __init__(self):
def __init__(self) -> None:
self._log = logging.getLogger(f'{__name__}.{self.__class__.__name__}')
self._conn = None
self._connected = False
def connect(self, ip, port):
def connect(self, ip: str, port: int) -> bool | None:
try:
self._conn = telnetlib.Telnet(host=ip, port=port, timeout=5)
self._connected = True
@@ -51,7 +52,7 @@ class TS3Proto:
self._connected = True
return True
def disconnect(self):
def disconnect(self) -> None:
if self._connected:
try:
self.send("quit")
@@ -63,7 +64,7 @@ class TS3Proto:
else:
self._log.info("Not connected")
def send_command(self, command, keys=None, opts=None):
def send_command(self, command: str, keys: dict[str, Any] | None = None, opts: list[str] | None = None):
cmd = self.construct_command(command, keys=keys, opts=opts)
# Clear read buffer of any stray bytes
@@ -103,7 +104,9 @@ class TS3Proto:
else:
raise TeamspeakError(resp['keys']['id'])
def construct_command(self, command, keys=None, opts=None):
def construct_command(
self, command: str, keys: dict[str, Any] | None = None, opts: list[str] | None = None
) -> str:
"""
Constructs a TS3 formatted command string
Keys can have a single nested list to construct a nested parameter
@@ -135,7 +138,7 @@ class TS3Proto:
return " ".join(cstr)
def parse_command(self, commandstr):
def parse_command(self, commandstr: str):
"""
Parses a TS3 command string into command/keys/opts tuple
@param commandstr: Command string
@@ -175,7 +178,7 @@ class TS3Proto:
return d
@staticmethod
def _escape_str(value):
def _escape_str(value: str | int) -> str:
"""
Escape a value into a TS3 compatible string
@param value: Value
@@ -190,7 +193,7 @@ class TS3Proto:
return value
@staticmethod
def _unescape_str(value):
def _unescape_str(value: str | int) -> str:
"""
Unescape a TS3 compatible string into a normal string
@param value: Value
@@ -204,14 +207,14 @@ class TS3Proto:
value = value.replace(j, i)
return value
def send(self, payload):
def send(self, payload: str) -> None:
if self._connected:
self._log.debug(f'Sent: {payload}')
self._conn.write(payload.encode('utf-8'))
class TS3Server(TS3Proto):
def __init__(self, ip, port, id=0):
def __init__(self, ip: str, port: int, id: int = 0) -> None:
"""
Abstraction class for TS3 Servers
@param ip: IP Address
@@ -223,7 +226,7 @@ class TS3Server(TS3Proto):
self.connect(ip, port)
def login(self, username, password):
def login(self, username: str, password: str) -> bool:
"""
Login to the TS3 Server
@param username: Username
@@ -244,7 +247,7 @@ class TS3Server(TS3Proto):
if self._connected:
return self.send_command('serverlist')
def gm(self, msg):
def gm(self, msg: str):
"""
Send a global message to the current Virtual Server
@param msg: Message
@@ -253,7 +256,7 @@ class TS3Server(TS3Proto):
if self._connected:
return self.send_command('gm', keys={'msg': msg})
def use(self, id):
def use(self, id: int) -> None:
"""
Use a particular Virtual Server instance
@param id: Virtual Server ID
@@ -264,7 +267,7 @@ class TS3Server(TS3Proto):
class TeamspeakError(Exception):
def __init__(self, code, msg=None):
def __init__(self, code: int | str, msg: str | None = None) -> None:
self.code = str(code)
if not msg:
msg = ts3_errors[self.code]
@@ -1,4 +1,7 @@
__lazy_modules__ = ["django.http"]
import logging
from typing import TYPE_CHECKING
from django.conf import settings
from django.contrib import messages
@@ -12,6 +15,9 @@ from .manager import Teamspeak3Manager
from .models import Teamspeak3User
from .tasks import Teamspeak3Tasks
if TYPE_CHECKING:
from django.http import HttpRequest, HttpResponse
logger = logging.getLogger(__name__)
ACCESS_PERM = 'teamspeak3.access_teamspeak3'
@@ -19,7 +25,7 @@ ACCESS_PERM = 'teamspeak3.access_teamspeak3'
@login_required
@permission_required(ACCESS_PERM)
def activate_teamspeak3(request):
def activate_teamspeak3(request: "HttpRequest") -> "HttpResponse":
logger.debug(f"activate_teamspeak3 called by user {request.user}")
character = request.user.profile.main_character
@@ -41,7 +47,7 @@ def activate_teamspeak3(request):
@login_required
@permission_required(ACCESS_PERM)
def verify_teamspeak3(request):
def verify_teamspeak3(request: "HttpRequest") -> "HttpResponse":
logger.debug(f"verify_teamspeak3 called by user {request.user}")
if not Teamspeak3Tasks.has_account(request.user):
logger.warning(f"Unable to validate user {request.user} teamspeak: no teamspeak data")
@@ -64,7 +70,7 @@ def verify_teamspeak3(request):
@login_required
@permission_required(ACCESS_PERM)
def deactivate_teamspeak3(request):
def deactivate_teamspeak3(request: "HttpRequest") -> "HttpResponse":
logger.debug(f"deactivate_teamspeak3 called by user {request.user}")
if Teamspeak3Tasks.has_account(request.user) and Teamspeak3Tasks.delete_user(request.user):
logger.info(f"Successfully deactivated TeamSpeak3 for user {request.user}")
@@ -77,7 +83,7 @@ def deactivate_teamspeak3(request):
@login_required
@permission_required(ACCESS_PERM)
def reset_teamspeak3_perm(request):
def reset_teamspeak3_perm(request: "HttpRequest") -> "HttpResponse":
logger.debug(f"reset_teamspeak3_perm called by user {request.user}")
if not Teamspeak3Tasks.has_account(request.user):
return redirect("services:services")
@@ -103,7 +109,7 @@ def reset_teamspeak3_perm(request):
@login_required
@staff_member_required
def admin_update_ts3_groups(request):
def admin_update_ts3_groups(request: "HttpRequest") -> "HttpResponse":
Teamspeak3Tasks.run_ts3_group_update.delay()
messages.info(request, "Started updating TeamSpeak3 server groups...")
return redirect("admin:teamspeak3_authts_changelist")