mirror of
https://github.com/house-of-vanity/OutFleet.git
synced 2025-07-06 17:14:07 +00:00
100 lines
3.5 KiB
Python
100 lines
3.5 KiB
Python
import json
|
|
from polymorphic.admin import (
|
|
PolymorphicParentModelAdmin,
|
|
)
|
|
from django.contrib import admin
|
|
from django.utils.safestring import mark_safe
|
|
from django.db.models import Count
|
|
|
|
from vpn.models import User, ACL
|
|
from vpn.forms import UserForm
|
|
from .server_plugins import (
|
|
Server,
|
|
WireguardServer,
|
|
WireguardServerAdmin,
|
|
OutlineServer,
|
|
OutlineServerAdmin)
|
|
|
|
|
|
@admin.register(Server)
|
|
class ServerAdmin(PolymorphicParentModelAdmin):
|
|
base_model = Server
|
|
child_models = (OutlineServer, WireguardServer)
|
|
list_display = ('name', 'server_type', 'comment', 'registration_date', 'user_count', 'server_status_inline')
|
|
search_fields = ('name', 'comment')
|
|
list_filter = ('server_type', )
|
|
|
|
@admin.display(description='User Count', ordering='user_count')
|
|
def user_count(self, obj):
|
|
return obj.user_count
|
|
|
|
@admin.display(description='Status')
|
|
def server_status_inline(self, obj):
|
|
status = obj.get_server_status()
|
|
if 'error' in status:
|
|
return mark_safe(f"<span style='color: red;'>Error: {status['error']}</span>")
|
|
# Преобразуем JSON в красивый формат
|
|
import json
|
|
pretty_status = ", ".join(f"{key}: {value}" for key, value in status.items())
|
|
return mark_safe(f"<pre>{pretty_status}</pre>")
|
|
server_status_inline.short_description = "Status"
|
|
|
|
def get_queryset(self, request):
|
|
qs = super().get_queryset(request)
|
|
qs = qs.annotate(user_count=Count('acl'))
|
|
return qs
|
|
|
|
@admin.register(User)
|
|
class UserAdmin(admin.ModelAdmin):
|
|
form = UserForm
|
|
list_display = ('name', 'comment', 'registration_date', 'hash', 'server_count')
|
|
search_fields = ('name', 'hash')
|
|
readonly_fields = ('hash',)
|
|
|
|
|
|
@admin.display(description='Allowed servers', ordering='server_count')
|
|
def server_count(self, obj):
|
|
return obj.server_count
|
|
|
|
def get_queryset(self, request):
|
|
qs = super().get_queryset(request)
|
|
qs = qs.annotate(server_count=Count('acl'))
|
|
return qs
|
|
|
|
def save_model(self, request, obj, form, change):
|
|
super().save_model(request, obj, form, change)
|
|
selected_servers = form.cleaned_data.get('servers', [])
|
|
|
|
ACL.objects.filter(user=obj).exclude(server__in=selected_servers).delete()
|
|
|
|
for server in selected_servers:
|
|
ACL.objects.get_or_create(user=obj, server=server)
|
|
|
|
|
|
@admin.register(ACL)
|
|
class ACLAdmin(admin.ModelAdmin):
|
|
list_display = ('user', 'server', 'server_type', 'link', 'created_at')
|
|
list_filter = ('user', 'server__server_type')
|
|
search_fields = ('user__name', 'server__name', 'server__comment', 'user__comment', 'link')
|
|
readonly_fields = ('user_info', )
|
|
|
|
@admin.display(description='Server Type', ordering='server__server_type')
|
|
def server_type(self, obj):
|
|
return obj.server.get_server_type_display()
|
|
|
|
@admin.display(description='Client info')
|
|
def user_info(self, obj):
|
|
server = obj.server
|
|
user = obj.user
|
|
try:
|
|
data = server.get_user(user)
|
|
|
|
if isinstance(data, dict):
|
|
formatted_data = json.dumps(data, indent=2)
|
|
return mark_safe(f"<pre>{formatted_data}</pre>")
|
|
elif isinstance(data, str):
|
|
return mark_safe(f"<pre>{data}</pre>")
|
|
else:
|
|
return mark_safe(f"<pre>{str(data)}</pre>")
|
|
except Exception as e:
|
|
return mark_safe(f"<span style='color: red;'>Error: {e}</span>") |