138 lines
6.0 KiB
Python
138 lines
6.0 KiB
Python
from django.contrib import admin
|
|
from .models import UserActivityLog, ExternalDBSettings, RoomDiscrepancy
|
|
from django.urls import path
|
|
from django.http import JsonResponse
|
|
from django.shortcuts import render
|
|
from .models import ExternalDBSettings
|
|
import pymysql
|
|
from django.shortcuts import redirect
|
|
from django.urls import reverse
|
|
|
|
@admin.register(ExternalDBSettings)
|
|
class ExternalDBSettingsAdmin(admin.ModelAdmin):
|
|
change_form_template = "antifroud/admin/external_db_settings_change_form.html"
|
|
def add_view(self, request, form_url='', extra_context=None):
|
|
# Создаем новую запись
|
|
new_instance = ExternalDBSettings.objects.create(
|
|
name="Новая настройка", # Задайте значение по умолчанию
|
|
host="",
|
|
port=3306,
|
|
user="",
|
|
password="",
|
|
is_active=False
|
|
)
|
|
# Перенаправляем пользователя на страницу редактирования новой записи
|
|
return redirect(reverse('admin:antifroud_externaldbsettings_change', args=(new_instance.id,)))
|
|
|
|
def get_urls(self):
|
|
urls = super().get_urls()
|
|
custom_urls = [
|
|
path(
|
|
'test-connection/',
|
|
self.admin_site.admin_view(self.test_connection),
|
|
name='test_connection',
|
|
),
|
|
path(
|
|
'fetch-tables/',
|
|
self.admin_site.admin_view(self.fetch_tables),
|
|
name='fetch_tables',
|
|
),
|
|
path(
|
|
'fetch-table-data/',
|
|
self.admin_site.admin_view(self.fetch_table_data),
|
|
name='fetch_table_data',
|
|
),
|
|
]
|
|
return custom_urls + urls
|
|
|
|
def test_connection(self, request):
|
|
db_id = request.GET.get('db_id')
|
|
if not db_id:
|
|
return JsonResponse({"status": "error", "message": "ID подключения отсутствует."}, status=400)
|
|
try:
|
|
# Получаем объект настроек подключения
|
|
db_settings = ExternalDBSettings.objects.get(id=db_id)
|
|
|
|
# Проверяем, что все необходимые поля заполнены
|
|
if not db_settings.user or not db_settings.password:
|
|
return JsonResponse({"status": "error", "message": "Имя пользователя или пароль не указаны."}, status=400)
|
|
|
|
# Проверяем подключение к базе данных
|
|
import pymysql
|
|
connection = pymysql.connect(
|
|
host=db_settings.host,
|
|
port=db_settings.port,
|
|
user=db_settings.user,
|
|
password=db_settings.password,
|
|
database=db_settings.database
|
|
)
|
|
connection.close()
|
|
return JsonResponse({"status": "success", "message": "Подключение успешно установлено."})
|
|
except ExternalDBSettings.DoesNotExist:
|
|
return JsonResponse({"status": "error", "message": "Настройки подключения не найдены."}, status=404)
|
|
except pymysql.MySQLError as e:
|
|
return JsonResponse({"status": "error", "message": f"Ошибка MySQL: {str(e)}"}, status=500)
|
|
except Exception as e:
|
|
return JsonResponse({"status": "error", "message": f"Неизвестная ошибка: {str(e)}"}, status=500)
|
|
|
|
|
|
def fetch_tables(self, request):
|
|
"""Возвращает список таблиц в базе данных."""
|
|
try:
|
|
db_id = request.GET.get('db_id')
|
|
db_settings = ExternalDBSettings.objects.get(id=db_id)
|
|
connection = pymysql.connect(
|
|
host=db_settings.host,
|
|
port=db_settings.port,
|
|
user=db_settings.user,
|
|
password=db_settings.password,
|
|
database=db_settings.database
|
|
)
|
|
cursor = connection.cursor()
|
|
cursor.execute("SHOW TABLES;")
|
|
tables = [row[0] for row in cursor.fetchall()]
|
|
connection.close()
|
|
return JsonResponse({"status": "success", "tables": tables})
|
|
except Exception as e:
|
|
return JsonResponse({"status": "error", "message": str(e)})
|
|
|
|
def fetch_table_data(self, request):
|
|
"""Возвращает первые 10 записей из выбранной таблицы."""
|
|
try:
|
|
db_id = request.GET.get('db_id')
|
|
table_name = request.GET.get('table_name')
|
|
db_settings = ExternalDBSettings.objects.get(id=db_id)
|
|
connection = pymysql.connect(
|
|
host=db_settings.host,
|
|
port=db_settings.port,
|
|
user=db_settings.user,
|
|
password=db_settings.password,
|
|
database=db_settings.database
|
|
)
|
|
cursor = connection.cursor()
|
|
cursor.execute(f"SELECT * FROM `{table_name}` LIMIT 10;")
|
|
columns = [desc[0] for desc in cursor.description]
|
|
rows = cursor.fetchall()
|
|
connection.close()
|
|
return JsonResponse({"status": "success", "columns": columns, "rows": rows})
|
|
except Exception as e:
|
|
return JsonResponse({"status": "error", "message": str(e)})
|
|
|
|
@admin.register(UserActivityLog)
|
|
class UserActivityLogAdmin(admin.ModelAdmin):
|
|
list_display = ("id", "user_id", "ip", "created", "page_title", "type", "hits")
|
|
search_fields = ("user_id", "ip", "page_title")
|
|
list_filter = ("type", "created")
|
|
readonly_fields = ("created", "timestamp")
|
|
|
|
|
|
|
|
|
|
|
|
@admin.register(RoomDiscrepancy)
|
|
class RoomDiscrepancyAdmin(admin.ModelAdmin):
|
|
list_display = ("hotel", "room_number", "booking_id", "check_in_date_expected", "check_in_date_actual", "discrepancy_type", "created_at")
|
|
search_fields = ("hotel__name", "room_number", "booking_id")
|
|
list_filter = ("discrepancy_type", "created_at")
|
|
readonly_fields = ("created_at",)
|