{str(result.id)[-8:]}
{result.id!s}
Version: {version}
PWD: {pwd_text}
{full_output}
{}
', obj.end_ts.strftime("%Y-%m-%d %H:%M:%S")) def result_id(self, obj): return format_html( '[{}]',
reverse("admin:core_archiveresult_change", args=(obj.id,)),
str(obj.id)[:8],
)
def command(self, obj):
return format_html("{}", " ".join(obj.cmd or []))
def version(self, obj):
return format_html("{}", obj.cmd_version or "-")
def get_formset(self, request, obj=None, **kwargs):
formset = super().get_formset(request, obj, **kwargs)
snapshot = self.get_parent_object_from_request(request)
base_fields = formset.form.base_fields
snapshot_output_dir = str(snapshot.output_dir) if snapshot else ""
# formset.form.base_fields['id'].widget = formset.form.base_fields['id'].hidden_widget()
# default values for new entries
base_fields["status"].initial = "succeeded"
base_fields["start_ts"].initial = timezone.now()
base_fields["end_ts"].initial = timezone.now()
base_fields["cmd_version"].initial = "-"
base_fields["pwd"].initial = snapshot_output_dir
base_fields["cmd"].initial = '["-"]'
base_fields["output_str"].initial = "Manually recorded cmd output..."
if obj is not None:
# hidden values for existing entries and new entries
base_fields["start_ts"].widget = base_fields["start_ts"].hidden_widget()
base_fields["end_ts"].widget = base_fields["end_ts"].hidden_widget()
base_fields["cmd"].widget = base_fields["cmd"].hidden_widget()
base_fields["pwd"].widget = base_fields["pwd"].hidden_widget()
base_fields["cmd_version"].widget = base_fields["cmd_version"].hidden_widget()
return formset
def get_readonly_fields(self, request, obj=None):
if obj is not None:
return self.readonly_fields
else:
return []
class ArchiveResultAdmin(BaseModelAdmin):
list_select_related = ()
list_display = (
"details_link",
"zip_link",
"created_at",
"snapshot_info",
"tags_inline",
"status_badge",
"plugin_with_icon",
"process_link",
"machine_link",
"cmd_str",
"output_str_display",
)
list_display_links = None
sort_fields = ("id", "created_at", "plugin", "status")
readonly_fields = (
"cmd",
"cmd_version",
"pwd",
"cmd_str",
"admin_actions",
"snapshot_info",
"tags_str",
"created_at",
"modified_at",
"output_summary",
"plugin_with_icon",
"process_link",
)
search_fields = (
"snapshot__id",
"snapshot__url",
"snapshot__tags__name",
"snapshot__crawl_id",
"plugin",
"hook_name",
"output_str",
"output_json",
"process__cmd",
)
autocomplete_fields = ("snapshot",)
fieldsets = (
(
"Snapshot",
{
"fields": ("snapshot", "snapshot_info", "tags_str", "admin_actions"),
"classes": ("card", "wide"),
},
),
(
"Plugin",
{
"fields": ("plugin_with_icon", "process_link", "status"),
"classes": ("card",),
},
),
(
"Timing",
{
"fields": ("start_ts", "end_ts", "created_at", "modified_at"),
"classes": ("card",),
},
),
(
"Command",
{
"fields": ("cmd", "cmd_str", "cmd_version", "pwd"),
"classes": ("card",),
},
),
(
"Output",
{
"fields": ("output_str", "output_json", "output_files", "output_size", "output_mimetypes", "output_summary"),
"classes": ("card", "wide"),
},
),
)
list_filter = ("status", "plugin", "start_ts")
ordering = ("-start_ts",)
list_per_page = 50
paginator = AcceleratedPaginator
save_on_top = True
show_full_result_count = False
actions = ("delete_selected",)
class Meta:
verbose_name = "Archive Result"
verbose_name_plural = "Archive Results"
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 get_admin_toolbar_actions(self, request, obj):
if obj is None:
return []
self.request = request
return [
{
"label": "View Output",
"icon": "📄",
"url": self.get_output_view_url(obj),
"title": "Open the archived output for this result",
},
{
"label": "Output files",
"icon": "📁",
"url": self.get_output_files_url(obj),
"title": "Browse the output files for this result",
},
{
"label": "Download Zip",
"icon": "⬇",
"url": self.get_output_zip_url(obj),
"kind": "accent",
"css_classes": "archivebox-zip-button",
"onclick": "return window.archiveboxHandleZipClick(this, event);",
"extra_attrs": [("data-loading-label", "Preparing...")],
"title": "Download all output files as a zip",
},
{"label": "Snapshot", "icon": "🗂", "url": self.get_snapshot_view_url(obj), "title": "Open the parent snapshot view"},
]
def changelist_view(self, request, extra_context=None):
self.request = request
selected = request.GET.getlist(ACTION_CHECKBOX_NAME)
if request.method == "GET" and request.GET.get("action") == "delete_selected" and selected:
if not request.user.is_superuser:
raise PermissionDenied
if len(selected) > 100:
raise SuspiciousOperation("Too many ArchiveResults selected for deletion")
try:
queryset = self.get_queryset(request).filter(pk__in=selected)
if not queryset.exists():
snapshot = Snapshot.objects.only("id").filter(pk=request.GET.get("snapshot")).first()
return redirect(build_snapshot_url(str(snapshot.id), "index.html", request=request) if snapshot else request.path)
except (ValidationError, ValueError):
return redirect(request.path)
return delete_selected(self, request, queryset)
handoff_snapshot = (
request.GET.get("snapshot") if request.method == "POST" and request.GET.get("action") == "delete_selected" else None
)
if handoff_snapshot:
request.GET = request.GET.copy()
request.GET.clear()
request.META["QUERY_STRING"] = ""
try:
handoff_snapshot = Snapshot.objects.only("id").filter(pk=handoff_snapshot).first()
except (ValidationError, ValueError):
handoff_snapshot = None
saved_list_per_page = self.list_per_page
self.list_per_page = request.archivebox_config.SNAPSHOTS_PER_PAGE
try:
response = super().changelist_view(request, extra_context)
if (
handoff_snapshot
and response.status_code in (301, 302)
and not ArchiveResult.objects.filter(
pk__in=request.POST.getlist(ACTION_CHECKBOX_NAME),
).exists()
):
return redirect(build_snapshot_url(str(handoff_snapshot.id), "index.html", request=request))
return response
finally:
self.list_per_page = saved_list_per_page
def get_queryset(self, request):
ordering = request.GET.get("o")
ordering_fields = set()
if ordering:
for part in ordering.split("."):
if not part:
continue
try:
idx = abs(int(part)) - 1
except ValueError:
continue
if 0 <= idx < len(self.list_display):
ordering_fields.add(self.list_display[idx])
qs = (
super()
.get_queryset(request)
.prefetch_related(
Prefetch(
"snapshot",
queryset=Snapshot.objects.defer("config", "notes").prefetch_related("crawl__created_by", "tags"),
),
"process__machine",
)
)
if request.resolver_match.url_name == "core_archiveresult_change":
qs = qs.defer("notes")
else:
qs = qs.defer("notes", "output_json")
if "tags_inline" in ordering_fields:
qs = qs.annotate(snapshot_first_tag=Min("snapshot__tags__name"))
return qs
def get_search_results(self, request, queryset, search_term):
if not search_term:
return queryset, False
queryset = queryset.annotate(
snapshot_id_text=Cast("snapshot__id", output_field=TextField()),
snapshot_crawl_id_text=Cast("snapshot__crawl_id", output_field=TextField()),
output_json_text=Cast("output_json", output_field=TextField()),
cmd_text=Cast("process__cmd", output_field=TextField()),
)
search_bits = [
bit[1:-1] if len(bit) >= 2 and bit[0] == bit[-1] and bit[0] in {'"', "'"} else bit for bit in smart_split(search_term)
]
search_bits = [bit.strip() for bit in search_bits if bit.strip()]
if not search_bits:
return queryset, False
filters = []
for bit in search_bits:
filters.append(
Q(snapshot_id_text__icontains=bit)
| Q(snapshot__url__icontains=bit)
| Q(snapshot__tags__name__icontains=bit)
| Q(snapshot_crawl_id_text__icontains=bit)
| Q(plugin__icontains=bit)
| Q(hook_name__icontains=bit)
| Q(output_str__icontains=bit)
| Q(output_json_text__icontains=bit)
| Q(cmd_text__icontains=bit),
)
return queryset.filter(reduce(and_, filters)).distinct(), True
def get_snapshot_view_url(self, result: ArchiveResult) -> str:
request = self.request
return build_snapshot_url(str(result.snapshot_id), request=request, config=request.archivebox_config)
def get_output_view_url(self, result: ArchiveResult) -> str:
request = self.request
config = request.archivebox_config
output_path = result.embed_path()
if not output_path:
output_path = result.plugin or ""
return build_snapshot_url(str(result.snapshot_id), output_path, request=request, config=config)
def get_output_files_url(self, result: ArchiveResult) -> str:
request = self.request
return f"{build_snapshot_url(str(result.snapshot_id), result.plugin, request=request, config=request.archivebox_config)}/?files=1"
def get_output_zip_url(self, result: ArchiveResult) -> str:
return f"{self.get_output_files_url(result)}&download=zip"
@admin.display(description="Details", ordering="id")
def details_link(self, result):
return format_html(
'{}',
reverse("admin:core_archiveresult_change", args=[result.id]),
str(result.id)[-8:],
)
@admin.display(description="Zip")
def zip_link(self, result):
return format_html(
'⬇ ZIP',
self.get_output_zip_url(result),
)
@admin.display(description="")
def admin_actions(self, result):
return format_html(
"""
""",
self.get_output_view_url(result),
self.get_output_files_url(result),
self.get_output_zip_url(result),
self.get_snapshot_view_url(result),
)
@admin.display(
description="Snapshot",
ordering="snapshot__url",
)
def snapshot_info(self, result):
snapshot_id = str(result.snapshot_id)
request = self.request
return format_html(
'[{}] {} {}{}',
get_plugin_admin_url(result.plugin),
result.plugin,
icon,
get_plugin_admin_url(result.plugin),
result.plugin,
)
@admin.display(description="Process", ordering="process__pid")
def process_link(self, result):
process = result.process_record
if not process:
return "-"
process_label = get_process_link_label(process)
return format_html(
'{}',
reverse("admin:machine_process_change", args=[process.id]),
process_label,
)
@admin.display(description="Machine", ordering="process__machine__hostname")
def machine_link(self, result):
process = result.process_record
if not process or not process.machine_id:
return "-"
machine = process.machine
return format_html(
'{}',
reverse("admin:machine_machine_change", args=[machine.id]),
machine.hostname,
)
@admin.display(description="Command")
def cmd_str(self, result):
request = self.request
display_cmd = build_abx_dl_display_command(result)
replay_cmd = build_abx_dl_replay_command(result, config=request.archivebox_config)
return format_html(
"""
""",
replay_cmd,
replay_cmd,
display_cmd,
)
def output_display(self, result):
request = self.request
config = request.archivebox_config
# Determine output link path - use embed_path() which checks output_files
embed_path = result.embed_path()
output_path = embed_path if (result.status == "succeeded" and embed_path) else "index.html"
snapshot_id = str(result.snapshot_id)
return format_html(
'↗️{}',
build_snapshot_url(snapshot_id, output_path, request=request, config=config),
result.output_str_for_display(),
)
@admin.display(description="Output", ordering="output_str")
def output_str_display(self, result):
output_text = str(result.output_str_for_display() or "").strip()
if not output_text:
return "-"
request = self.request
live_path = result.embed_path()
if live_path:
return format_html(
'{}',
build_snapshot_url(str(result.snapshot_id), live_path, request=request, config=request.archivebox_config),
output_text,
output_text,
)
return format_html(
'{}',
output_text,
output_text,
)
def output_summary(self, result):
snapshot_dir = Path(result.snapshot.output_dir)
output_html = format_html(
'{}',
build_snapshot_url(snapshot_id, "index.html", request=request, config=request.archivebox_config),
)
embed_path = result.embed_path() or ""
path_from_embed = snapshot_dir / (embed_path or "")
output_html += format_html(
'{}/{}
',
str(snapshot_dir),
str(embed_path),
)
if os.access(path_from_embed, os.R_OK):
root_dir = str(path_from_embed)
else:
root_dir = str(snapshot_dir)
for root, dirs, files in os.walk(root_dir):
depth = root.replace(root_dir, "").count(os.sep) + 1
if depth > 2:
continue
indent = " " * 4 * (depth)
output_html += format_html('{}{}/
', indent, os.path.basename(root))
indentation_str = " " * 4 * (depth + 1)
for filename in sorted(files):
is_hidden = filename.startswith(".")
output_html += format_html(
'{}{}
',
int(not is_hidden),
indentation_str,
filename.strip(),
)
return output_html + mark_safe("