mirror of
https://github.com/ArchiveBox/ArchiveBox.git
synced 2026-09-12 19:50:57 +05:00
233 lines
9.4 KiB
Python
233 lines
9.4 KiB
Python
__package__ = "archivebox.core"
|
|
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from admin_data_views.admin import (
|
|
admin_data_index_view as adv_admin_data_index_view,
|
|
)
|
|
from admin_data_views.admin import (
|
|
get_admin_data_urls as adv_get_admin_data_urls,
|
|
)
|
|
from admin_data_views.admin import (
|
|
get_app_list as adv_get_app_list,
|
|
)
|
|
from django.contrib import admin
|
|
from django.contrib.auth import REDIRECT_FIELD_NAME, get_user_model, login as auth_login
|
|
from django.contrib.auth.decorators import login_not_required
|
|
from django.contrib.auth.forms import UserCreationForm
|
|
from django.contrib.auth.views import LoginView
|
|
from django.db import DatabaseError, connection, transaction
|
|
from django.http import HttpResponseRedirect
|
|
from django.template.response import TemplateResponse
|
|
from django.urls import reverse
|
|
from django.utils.decorators import method_decorator
|
|
from django.utils.translation import gettext as _
|
|
from django.views.decorators.cache import never_cache
|
|
|
|
from archivebox.config import VERSION
|
|
from archivebox.core.context_processors import get_static_cache_key
|
|
from archivebox.core.routes_util import is_allowed_archivebox_redirect_url
|
|
|
|
if TYPE_CHECKING:
|
|
from admin_data_views.typing import AppDict
|
|
from django.http import HttpRequest
|
|
from django.template.response import TemplateResponse
|
|
from django.urls import URLPattern, URLResolver
|
|
|
|
|
|
class ArchiveBoxLoginView(LoginView):
|
|
def get_redirect_url(self) -> str:
|
|
redirect_to = self.request.POST.get(
|
|
self.redirect_field_name,
|
|
self.request.GET.get(self.redirect_field_name),
|
|
)
|
|
if is_allowed_archivebox_redirect_url(redirect_to, request=self.request):
|
|
return redirect_to
|
|
return ""
|
|
|
|
|
|
class ArchiveBoxAdmin(admin.AdminSite):
|
|
site_header = "ArchiveBox"
|
|
index_title = "Admin Views"
|
|
site_title = "Admin"
|
|
namespace = "admin"
|
|
|
|
def each_context(self, request: "HttpRequest") -> dict[str, Any]:
|
|
context = super().each_context(request)
|
|
context["VERSION"] = VERSION
|
|
context["STATIC_CACHE_KEY"] = get_static_cache_key()
|
|
return context
|
|
|
|
@staticmethod
|
|
def _format_object_count(count: int) -> tuple[int, str, str]:
|
|
if count >= 1_000_000_000:
|
|
count_label = f"{count / 1_000_000_000:.1f}B"
|
|
elif count >= 1_000_000:
|
|
count_label = f"{count / 1_000_000:.1f}M"
|
|
elif count >= 1_000:
|
|
count_label = f"{count / 1_000:.1f}K"
|
|
else:
|
|
count_label = f"{count:,}"
|
|
count_label = count_label.replace(".0", "")
|
|
return count, count_label, f"Object count: {count:,}"
|
|
|
|
def _set_model_object_count(
|
|
self,
|
|
models_by_table: dict[str, list[dict[str, Any]]],
|
|
table: str,
|
|
count: int,
|
|
title: str | None = None,
|
|
) -> None:
|
|
models = models_by_table.get(table)
|
|
if not models:
|
|
return
|
|
count, count_label, count_title = self._format_object_count(count)
|
|
if title:
|
|
count_title = title
|
|
for model in models:
|
|
model["object_count"] = count
|
|
model["object_count_label"] = count_label
|
|
model["object_count_title"] = count_title
|
|
|
|
def get_app_list(self, request: "HttpRequest", app_label: str | None = None) -> list["AppDict"]:
|
|
if app_label is None:
|
|
return adv_get_app_list(self, request)
|
|
return adv_get_app_list(self, request, app_label)
|
|
|
|
def admin_data_index_view(self, request: "HttpRequest", **kwargs: Any) -> "TemplateResponse":
|
|
return adv_admin_data_index_view(self, request, **kwargs)
|
|
|
|
@method_decorator(never_cache)
|
|
@login_not_required
|
|
def login(self, request: "HttpRequest", extra_context: dict[str, Any] | None = None) -> "TemplateResponse":
|
|
if request.method == "GET" and self.has_permission(request):
|
|
return HttpResponseRedirect(reverse("admin:index", current_app=self.name))
|
|
|
|
from django.contrib.admin.forms import AdminAuthenticationForm
|
|
|
|
User = get_user_model()
|
|
first_admin_setup = not User.objects.filter(is_superuser=True).exclude(username="system").exists()
|
|
context = {
|
|
**self.each_context(request),
|
|
"title": _("Set up ArchiveBox") if first_admin_setup else _("Log in"),
|
|
"subtitle": None,
|
|
"app_path": request.get_full_path(),
|
|
"username": request.user.get_username(),
|
|
"first_admin_setup": first_admin_setup,
|
|
}
|
|
if REDIRECT_FIELD_NAME not in request.GET and REDIRECT_FIELD_NAME not in request.POST:
|
|
context[REDIRECT_FIELD_NAME] = reverse("admin:index", current_app=self.name)
|
|
context.update(extra_context or {})
|
|
|
|
index_path = reverse("admin:index", current_app=self.name)
|
|
request.current_app = self.name
|
|
if first_admin_setup:
|
|
form = UserCreationForm(request.POST or None)
|
|
if request.method == "POST" and form.is_valid():
|
|
with transaction.atomic():
|
|
# Fresh collections contain the system superuser, which also
|
|
# provides a row lock while the first real admin is created.
|
|
list(User.objects.select_for_update().filter(is_superuser=True).values_list("pk", flat=True))
|
|
if User.objects.filter(is_superuser=True).exclude(username="system").exists():
|
|
form.add_error(None, _("An admin account was already created. Log in instead."))
|
|
else:
|
|
user = form.save(commit=False)
|
|
user.is_staff = True
|
|
user.is_superuser = True
|
|
user.save()
|
|
auth_login(request, user, backend="django.contrib.auth.backends.ModelBackend")
|
|
return HttpResponseRedirect(index_path)
|
|
|
|
context["form"] = form
|
|
return TemplateResponse(
|
|
request,
|
|
self.login_template or "admin/login.html",
|
|
context,
|
|
)
|
|
|
|
return ArchiveBoxLoginView.as_view(
|
|
extra_context=context,
|
|
authentication_form=self.login_form or AdminAuthenticationForm,
|
|
template_name=self.login_template or "admin/login.html",
|
|
next_page=index_path,
|
|
)(request)
|
|
|
|
def index(self, request: "HttpRequest", extra_context: dict[str, Any] | None = None) -> "TemplateResponse":
|
|
response = super().index(request, extra_context)
|
|
|
|
models_by_table: dict[str, list[dict[str, Any]]] = {}
|
|
for app in response.context_data.get("app_list", []):
|
|
for model in app.get("models", []):
|
|
model_class = model.get("model")
|
|
if not model_class or not model.get("perms", {}).get("view"):
|
|
continue
|
|
models_by_table.setdefault(model_class._meta.db_table, []).append(model)
|
|
|
|
if not models_by_table:
|
|
return response
|
|
|
|
from archivebox.misc.db import approximate_row_counts
|
|
|
|
for table, count in approximate_row_counts(connection).items():
|
|
if table not in models_by_table:
|
|
continue
|
|
self._set_model_object_count(
|
|
models_by_table,
|
|
table,
|
|
count,
|
|
title=f"Approximate count from database stats: {count:,}",
|
|
)
|
|
models_by_table.pop(table, None)
|
|
|
|
for table in list(models_by_table):
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(f"SELECT COUNT(*) FROM {connection.ops.quote_name(table)}")
|
|
count = int(cursor.fetchone()[0])
|
|
except DatabaseError:
|
|
continue
|
|
self._set_model_object_count(models_by_table, table, count)
|
|
models_by_table.pop(table, None)
|
|
return response
|
|
|
|
def get_admin_data_urls(self) -> list["URLResolver | URLPattern"]:
|
|
return adv_get_admin_data_urls(self)
|
|
|
|
def get_urls(self) -> list["URLResolver | URLPattern"]:
|
|
return self.get_admin_data_urls() + super().get_urls()
|
|
|
|
|
|
archivebox_admin = ArchiveBoxAdmin()
|
|
# Note: delete_selected is enabled per-model via actions = ['delete_selected'] in each ModelAdmin
|
|
# TODO: https://stackoverflow.com/questions/40760880/add-custom-button-to-django-admin-panel
|
|
|
|
|
|
############### Admin Data View sections are defined in settings.ADMIN_DATA_VIEWS #########
|
|
|
|
|
|
def register_admin_site():
|
|
"""Replace the default admin site with our custom ArchiveBox admin site."""
|
|
from django.contrib import admin
|
|
from django.contrib.admin import sites
|
|
|
|
admin.site = archivebox_admin
|
|
sites.site = archivebox_admin
|
|
|
|
# Register admin views for each app
|
|
# (Previously handled by ABX plugin system, now called directly)
|
|
from archivebox.api.admin import register_admin as register_api_admin
|
|
from archivebox.core.admin import register_admin as register_core_admin
|
|
from archivebox.crawls.admin import register_admin as register_crawls_admin
|
|
from archivebox.machine.admin import register_admin as register_machine_admin
|
|
from archivebox.personas.admin import register_admin as register_personas_admin
|
|
from archivebox.workers.admin import register_admin as register_workers_admin
|
|
|
|
register_core_admin(archivebox_admin)
|
|
register_crawls_admin(archivebox_admin)
|
|
register_api_admin(archivebox_admin)
|
|
register_machine_admin(archivebox_admin)
|
|
register_personas_admin(archivebox_admin)
|
|
register_workers_admin(archivebox_admin)
|
|
|
|
return archivebox_admin
|