__package__ = "archivebox.crawls"
import json
from copy import copy
from typing import ClassVar
from urllib.parse import urlencode, urlparse
from django import forms
from django.contrib import admin, messages
from django.core.paginator import Paginator
from django.db.models import Count, F, Q
from django.http import HttpRequest, HttpResponseBadRequest, HttpResponseNotAllowed, JsonResponse
from django.shortcuts import get_object_or_404, redirect
from django.template.loader import render_to_string
from django.template.response import TemplateResponse
from django.urls import path, reverse
from django.utils import timezone
from django.utils.html import escape, format_html, format_html_join
from django.utils.safestring import mark_safe
from django_object_actions import action
from archivebox.base_models.admin import BaseModelAdmin, ConfigEditorMixin
from archivebox.core.models import ArchiveResult, Snapshot
from archivebox.core.permissions import (
PERMISSIONS_CHOICES,
PERMISSIONS_META,
PERMISSIONS_PRIVATE,
PERMISSIONS_PUBLIC,
PERMISSIONS_UNLISTED,
PERMISSIONS_VALUES,
normalize_permissions,
)
from archivebox.core.widgets import TagEditorWidget, URLFiltersWidget
from archivebox.crawls.models import Crawl, CrawlSchedule
from archivebox.misc.paginators import AcceleratedPaginator
from archivebox.progressmonitor.views import progress_endpoint
from archivebox.workers.models import RETRY_AT_MAX
class MaxDepthListFilter(admin.SimpleListFilter):
title = "max depth"
parameter_name = "max_depth"
def lookups(self, request, model_admin):
return [(str(depth), str(depth)) for depth in range(5)]
def queryset(self, request, queryset):
value = self.value()
if value is not None and value.isdigit():
return queryset.filter(max_depth=int(value))
return queryset
def render_snapshots_list(snapshots_qs, request=None, crawl=None, page_size=50, prefix="snapshots"):
"""Render a nice inline list view of snapshots with status, title, URL, and progress."""
query_param = f"{prefix}_q"
status_param = f"{prefix}_status"
page_param = f"{prefix}_page"
query = (request.GET.get(query_param, "") if request is not None else "").strip()
status_filter = (request.GET.get(status_param, "") if request is not None else "").strip()
valid_statuses = {choice[0] for choice in Snapshot.StatusChoices.choices}
filtered_qs = snapshots_qs
if query:
from archivebox.misc.util import filter_queryset_by_uuid_substring
id_match_pks = list(filter_queryset_by_uuid_substring(Snapshot.objects.all(), query).values_list("pk", flat=True)[:100])
filtered_qs = filtered_qs.filter(Q(pk__in=id_match_pks) | Q(url__icontains=query) | Q(title__icontains=query))
if status_filter in valid_statuses:
filtered_qs = filtered_qs.filter(status=status_filter)
# Keep ArchiveResult counters as scalar subqueries so the paginated
# Snapshot queryset does not become a join+GROUP BY over every result row.
snapshots_qs = filtered_qs.order_by("-created_at").annotate(
total_results=ArchiveResult.snapshot_count_expr(),
succeeded_results=ArchiveResult.snapshot_count_expr(status=ArchiveResult.StatusChoices.SUCCEEDED),
failed_results=ArchiveResult.snapshot_count_expr(status=ArchiveResult.StatusChoices.FAILED),
started_results=ArchiveResult.snapshot_count_expr(status=ArchiveResult.StatusChoices.STARTED),
skipped_results=ArchiveResult.snapshot_count_expr(status=ArchiveResult.StatusChoices.SKIPPED),
snapshot_permissions=F("permissions"),
)
page_number = request.GET.get(page_param, 1) if request is not None else 1
paginator = Paginator(snapshots_qs, page_size)
page_obj = paginator.get_page(page_number)
snapshots = page_obj.object_list
total_count = paginator.count
def querystring(**updates):
if request is None:
return "#"
params = request.GET.copy()
for key, value in updates.items():
if value in (None, ""):
params.pop(key, None)
else:
params[key] = str(value)
return f"?{params.urlencode()}" if params else "?"
preserved_inputs = ""
if request is not None:
managed_params = {query_param, status_param, page_param}
preserved_inputs = "".join(
f'{escape(label)} '
for value, label in Snapshot.StatusChoices.choices
)
controls = f"""
"""
if not snapshots:
return mark_safe(f"""
{controls}
No Snapshots found.
๐
'''
exclude_button = f'''
โ
'''
# Format date
date_str = snapshot.created_at.strftime("%Y-%m-%d %H:%M") if snapshot.created_at else "-"
rows.append(f'''
{status}
{permission_icon}
{escape(title)}
{escape(url_display)}
{date_str}
{f'{exclude_button}{delete_button}
''')
pagination = ""
if paginator.num_pages > 1:
pagination = f"""
{"
Previous " if page_obj.has_previous() else "
Previous "}
Page {page_obj.number} of {paginator.num_pages}
{"
Next " if page_obj.has_next() else "
Next "}
{controls}
Status
๐
Title
URL
Progress
Created
{
'Actions ' if crawl is not None else ""
}
{"".join(rows)}
{pagination}
/snapshot//delete/",
self.admin_site.admin_view(self.delete_snapshot_view),
name="crawls_crawl_snapshot_delete",
),
path(
"/snapshot//exclude-domain/",
self.admin_site.admin_view(self.exclude_domain_view),
name="crawls_crawl_snapshot_exclude_domain",
),
path(
"/set-permissions/",
self.admin_site.admin_view(self.set_permissions_view),
name="crawls_crawl_set_permissions",
),
]
return custom_urls + urls
def get_actions(self, request):
actions = super().get_actions(request)
actions.pop("delete_selected", None)
return actions
@admin.action(description="Delete")
def delete_selected_batched(self, request, queryset):
"""Delete crawls in a single transaction to avoid SQLite concurrency issues."""
from django.db import transaction
total = queryset.count()
# Get list of IDs to delete first (outside transaction)
ids_to_delete = list(queryset.values_list("pk", flat=True))
# Delete everything in a single atomic transaction
with transaction.atomic():
deleted_count, _ = Crawl.objects.filter(pk__in=ids_to_delete).delete()
messages.success(request, f"Successfully deleted {total} crawls ({deleted_count} total objects including related records).")
@admin.action(description="Pause")
def pause_selected_crawls(self, request, queryset):
# Admin changelist actions must stay set-based. Calling crawl.pause()
# here fans out into per-crawl Snapshot/ArchiveResult writes and can
# hold SQLite behind the request for minutes on large archives. The
# Crawl row is the scheduler signal; the runner observes PAUSED and
# owns child-row lifecycle work.
paused = queryset.exclude(status__in=Crawl.INACTIVE_STATES).update(
status=Crawl.StatusChoices.PAUSED,
retry_at=RETRY_AT_MAX,
modified_at=timezone.now(),
)
if paused:
messages.success(request, f"Paused {paused} crawl(s). The runner will stop scheduling new work on the next sweep.")
else:
messages.warning(request, "No active crawls were selected to pause.")
@admin.action(description="Resume")
def resume_selected_crawls(self, request, queryset):
# Keep resume symmetrical with pause: one tight scheduler UPDATE, no
# save() hooks and no child fanout in the request path. Paused child
# rows become runnable through their own resume/maintenance paths.
resumed = queryset.filter(status__in=Crawl.INACTIVE_STATES).update(
status=Crawl.StatusChoices.QUEUED,
retry_at=timezone.now(),
modified_at=timezone.now(),
)
if resumed:
messages.success(request, f"Resumed {resumed} crawl(s). The runner will pick them up on the next sweep.")
else:
messages.warning(request, "No paused or sealed crawls were selected to resume.")
@admin.action(description="Seal")
def seal_selected_crawls(self, request, queryset):
now = timezone.now()
crawl_ids = list(queryset.exclude(status=Crawl.StatusChoices.SEALED).values_list("pk", flat=True))
if not crawl_ids:
messages.warning(request, "No unsealed crawls were selected to seal.")
return
Snapshot.objects.filter(
crawl_id__in=crawl_ids,
status__in=Snapshot.OPEN_STATES,
).filter(
Q(retry_at__isnull=True) | Q(retry_at__gt=now),
).update(
retry_at=now,
modified_at=now,
)
sealed = (
Crawl.objects.filter(pk__in=crawl_ids)
.exclude(status=Crawl.StatusChoices.SEALED)
.update(
status=Crawl.StatusChoices.SEALED,
retry_at=now,
modified_at=now,
)
)
messages.success(request, f"Sealed {sealed} crawl(s). The runner will finish cleanup on the next sweep.")
@admin.action(description="Permissions โพ")
def set_crawl_permissions(self, request, queryset):
permissions = (request.POST.get("permissions") or "").strip().lower()
if permissions not in PERMISSIONS_VALUES:
messages.error(request, "Choose a valid permissions value.")
return
updated = self.update_crawl_permissions(queryset, permissions)
messages.success(request, f"Set permissions to {permissions} on {updated} crawl(s).")
def update_crawl_permissions(self, queryset, permissions):
now = timezone.now()
updated = 0
batch = []
crawls_to_update = []
for crawl in queryset.only("id", "config", "permissions").iterator(chunk_size=500):
old_permissions = crawl.permissions
config = dict(crawl.config or {})
config["PERMISSIONS"] = permissions
crawl.config = config
crawl.modified_at = now
crawls_to_update.append((crawl, old_permissions))
batch.append(crawl)
if len(batch) >= 500:
Crawl.objects.bulk_update(batch, ["config", "modified_at"], batch_size=500)
updated += len(batch)
batch.clear()
if batch:
Crawl.objects.bulk_update(batch, ["config", "modified_at"], batch_size=500)
updated += len(batch)
for crawl, old_permissions in crawls_to_update:
crawl.update_child_snapshot_permissions(old_permissions, permissions)
return updated
@action(label="Recrawl", description="Create a new crawl with the same settings", methods=("POST",))
def recrawl(self, request, obj):
"""Duplicate this crawl as a new crawl with the same URLs and settings."""
# Validate URLs (required for crawl to start)
if not obj.urls:
messages.error(request, "Cannot recrawl: original crawl has no URLs.")
return redirect("admin:crawls_crawl_change", obj.id)
new_crawl = Crawl.create_scheduler_row(
urls=obj.urls,
max_depth=obj.max_depth,
tags_str=obj.tags_str,
config=obj.config,
schedule=obj.schedule,
label=f"{obj.label} (recrawl)" if obj.label else "",
notes=obj.notes,
created_by=request.user,
status=Crawl.StatusChoices.QUEUED,
retry_at=timezone.now(),
)
messages.success(request, f"Created new crawl {new_crawl.id} with the same settings. It will start processing shortly.")
return redirect("admin:crawls_crawl_change", new_crawl.id)
@admin.display(description="Stop Reason")
def stop_reason_display(self, obj):
reason = self.stop_reason_for_crawl(obj) if obj else ""
if not reason:
return mark_safe('None ')
return format_html('{} ', reason)
def stop_reason_for_crawl(self, obj):
if obj.pk in self.stop_reason_cache:
return self.stop_reason_cache[obj.pk]
output_dir = obj.output_dir
config = self.limit_config_for_crawl(obj, output_dir)
reason = obj.stop_reason(
config=config,
output_dir=output_dir,
num_snapshots=obj.num_snapshots_cached,
num_sealed_snapshots=obj.num_archived_snapshots_cached,
)
self.stop_reason_cache[obj.pk] = reason
return reason
def limit_config_for_crawl(self, obj, output_dir):
from archivebox.config.common import get_config
return get_config(crawl=obj).for_crawl_runtime(
crawl=obj,
persona=obj.resolve_persona(),
crawl_output_dir=output_dir,
)
@admin.display(description="Status", ordering="status")
def status_with_stop_reason(self, obj):
status = "PAUSED" if obj.is_paused else str(obj.status or "").upper()
reason = self.stop_reason_for_crawl(obj) if obj.is_paused or obj.status == Crawl.StatusChoices.SEALED else ""
if reason:
reason_label = reason.removeprefix("crawl_").replace("_", " ")
return format_html(
'{} {} {} ', obj.status, status)
@admin.display(description="ID", ordering="id")
def short_id(self, obj):
short_id = str(obj.pk)[-8:]
return format_html('{} ', obj.admin_change_url, short_id)
@admin.display(description="Owner", ordering="created_by")
def owner(self, obj):
return obj.created_by
@admin.display(description="Depth", ordering="max_depth")
def depth(self, obj):
return obj.max_depth
@admin.display(description="๐", ordering="permissions")
def permissions_badge(self, obj):
permissions = normalize_permissions(obj.permissions)
icon, label, fg, bg = PERMISSIONS_META[permissions]
menu_items = format_html_join(
"",
(
'"
),
(
(
" is-active" if choice_value == permissions else "",
choice_value,
choice_fg,
choice_bg,
choice_icon,
choice_label,
)
for choice_value, choice_label in PERMISSIONS_CHOICES
for choice_icon, _choice_title, choice_fg, choice_bg in [PERMISSIONS_META[choice_value]]
),
)
return format_html(
''
''
'{} '
" "
''
" ",
permissions,
reverse(f"{self.admin_site.name}:crawls_crawl_set_permissions", args=[obj.pk]),
permissions,
label,
label,
fg,
bg,
icon,
menu_items,
)
@admin.display(description="Pause")
def pause_resume_control(self, obj):
if obj.is_paused or obj.status == Crawl.StatusChoices.SEALED:
reason = "paused" if obj.is_paused else (self.stop_reason_for_crawl(obj) or "sealed")
return format_html(
'Resume ',
obj.pk,
reason,
)
return format_html(
'Pause ',
obj.pk,
)
@admin.display(description="Archived", ordering="num_archived_snapshots_cached")
def num_archived_snapshots(self, obj):
return obj.num_archived_snapshots_cached
@admin.display(description="Snapshots", ordering="num_snapshots_cached")
def num_total_snapshots(self, obj):
return obj.num_snapshots_cached
@admin.display(description="Snapshots")
def snapshots_changelist(self, obj):
request = self.request
snapshot_changelist = reverse("admin:core_snapshot_changelist")
scoped_params = {"crawl_id": str(obj.pk)}
full_url = f"{snapshot_changelist}?{urlencode(scoped_params)}"
snapshot_admin = self.admin_site._registry[Snapshot]
changelist_request = copy(request)
changelist_request.method = "GET"
changelist_request.path = snapshot_changelist
changelist_request.GET = request.GET.copy()
changelist_request.GET.update(
{
**scoped_params,
"_embedded": "crawl",
"per_page": "200",
},
)
changelist_request.POST = request.POST.copy()
changelist_request.POST.clear()
response = snapshot_admin.changelist_view(
changelist_request,
extra_context={"embedded_changelist": True},
)
context = {
**response.context_data,
"snapshot_changelist_url": full_url,
"crawl": obj,
}
return mark_safe(render_to_string("admin/crawls/crawl/snapshots_changelist.html", context, request=request))
def delete_snapshot_view(self, request: HttpRequest, object_id: str, snapshot_id: str):
if request.method != "POST":
return HttpResponseNotAllowed(["POST"])
crawl = get_object_or_404(Crawl, pk=object_id)
snapshot = get_object_or_404(Snapshot, pk=snapshot_id, crawl=crawl)
if snapshot.status == Snapshot.StatusChoices.STARTED:
snapshot.cancel_running_hooks()
removed_urls = crawl.prune_url(snapshot.url)
snapshot.delete()
return JsonResponse(
{
"ok": True,
"snapshot_id": str(snapshot.id),
"removed_urls": removed_urls,
},
)
def exclude_domain_view(self, request: HttpRequest, object_id: str, snapshot_id: str):
if request.method != "POST":
return HttpResponseNotAllowed(["POST"])
crawl = get_object_or_404(Crawl, pk=object_id)
snapshot = get_object_or_404(Snapshot, pk=snapshot_id, crawl=crawl)
result = crawl.exclude_domain(snapshot.url)
return JsonResponse(
{
"ok": True,
**result,
},
)
def set_permissions_view(self, request: HttpRequest, object_id: str):
if request.method != "POST":
return HttpResponseNotAllowed(["POST"])
permissions = (request.POST.get("permissions") or "").strip().lower()
if permissions not in PERMISSIONS_VALUES:
return HttpResponseBadRequest("Invalid permissions value")
crawl = get_object_or_404(Crawl, pk=object_id)
self.update_crawl_permissions(Crawl.objects.filter(pk=crawl.pk), permissions)
icon, label, fg, bg = PERMISSIONS_META[permissions]
return JsonResponse({"permissions": permissions, "icon": icon, "label": label, "fg": fg, "bg": bg})
@admin.display(description="Schedule", ordering="schedule")
def schedule_str(self, obj):
if not obj.schedule:
return mark_safe("None ")
return format_html('{} ', obj.schedule.admin_change_url, obj.schedule)
@admin.display(description="URLs", ordering="urls")
def urls_preview(self, obj):
first_url = next((line.strip() for line in (obj.urls or "").splitlines() if line.strip() and not line.strip().startswith("#")), "")
return first_url[:80] + "..." if len(first_url) > 80 else first_url
@admin.display(description="URLs")
def urls_editor(self, obj):
"""Editor for crawl URLs."""
widget_id = f"crawl_urls_{obj.pk}"
# Escape for safe HTML embedding
escaped_urls = (obj.urls or "").replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """)
# Count lines for auto-expand logic
line_count = len((obj.urls or "").split("\n"))
uri_rows = min(max(3, line_count), 10)
html = f'''
'''
return mark_safe(html)
class CrawlScheduleAdmin(BaseModelAdmin):
list_display = ("id", "created_at", "created_by", "label", "notes", "template_str", "crawls", "num_crawls", "num_snapshots")
sort_fields = ("id", "created_at", "created_by", "label", "notes", "template_str")
search_fields = ("id", "created_by__username", "label", "notes", "schedule_id", "template_id", "template__urls")
readonly_fields = ("created_at", "modified_at", "crawls", "snapshots")
autocomplete_fields = ("template", "created_by")
fieldsets = (
(
"Schedule Info",
{
"fields": ("label", "notes"),
"classes": ("card",),
},
),
(
"Configuration",
{
"fields": ("schedule", "template"),
"classes": ("card",),
},
),
(
"Metadata",
{
"fields": ("created_by", "created_at", "modified_at"),
"classes": ("card",),
},
),
(
"Crawls",
{
"fields": ("crawls",),
"classes": ("card", "wide"),
},
),
(
"Snapshots",
{
"fields": ("snapshots",),
"classes": ("card", "wide"),
},
),
)
list_filter = ("created_by",)
ordering = ("-created_at",)
list_per_page = 100
actions = ("delete_selected",)
def get_queryset(self, request):
self.request = request
return (
super()
.get_queryset(request)
.select_related("created_by", "template")
.annotate(
crawl_count=Count("crawl", distinct=True),
snapshot_count=Count("crawl__snapshot_set", distinct=True),
)
)
def change_view(self, request, object_id, form_url="", extra_context=None):
self.request = request
return super().change_view(request, object_id, form_url, extra_context)
def add_view(self, request, form_url="", extra_context=None):
return redirect("/add/#schedule")
def get_fieldsets(self, request, obj=None):
if obj is None:
return tuple(fieldset for fieldset in self.fieldsets if fieldset[0] not in {"Crawls", "Snapshots"})
return self.fieldsets
def save_model(self, request, obj, form, change):
if not obj.created_by_id and request.user.is_authenticated:
obj.created_by = request.user
super().save_model(request, obj, form, change)
@admin.display(description="Template", ordering="template")
def template_str(self, obj):
return format_html('{} ', obj.template.admin_change_url, obj.template)
@admin.display(description="# Crawls", ordering="crawl_count")
def num_crawls(self, obj):
count = obj.__dict__.get("crawl_count")
if count is None:
count = obj.crawl_set.count()
return count
@admin.display(description="# Snapshots", ordering="snapshot_count")
def num_snapshots(self, obj):
count = obj.__dict__.get("snapshot_count")
if count is None:
count = Snapshot.objects.filter(crawl__schedule=obj).count()
return count
def crawls(self, obj):
return format_html_join(
"{} ',
((crawl.admin_change_url, crawl) for crawl in obj.crawl_set.all().order_by("-created_at")[:20]),
) or mark_safe("No Crawls yet... ")
def snapshots(self, obj):
crawl_ids = obj.crawl_set.values_list("pk", flat=True)
return render_snapshots_list(
Snapshot.objects.filter(crawl_id__in=crawl_ids),
request=self.request,
prefix="schedule_snapshots",
)
def register_admin(admin_site):
admin_site.register(Crawl, CrawlAdmin)
admin_site.register(CrawlSchedule, CrawlScheduleAdmin)