Source code for archivebox.core.admin

__package__ = 'archivebox.core'

from io import StringIO
from pathlib import Path
from contextlib import redirect_stdout
from datetime import datetime, timezone

from django.contrib import admin
from django.db.models import Count
from django.urls import path
from django.utils.html import format_html
from django.utils.safestring import mark_safe
from django.shortcuts import render, redirect
from django.contrib.auth import get_user_model
from django import forms

from ..util import htmldecode, urldecode, ansi_to_html

from core.models import Snapshot, ArchiveResult, Tag
from core.forms import AddLinkForm

from core.mixins import SearchResultsAdminMixin

from index.html import snapshot_icons
from logging_util import printable_filesize
from main import add, remove
from extractors import archive_links
from config import (


# Admin URLs
# /admin/
# /admin/login/
# /admin/core/
# /admin/core/snapshot/
# /admin/core/snapshot/:uuid/
# /admin/core/tag/
# /admin/core/tag/:uuid/


[docs] class ArchiveResultInline(admin.TabularInline): model = ArchiveResult
[docs] class TagInline(admin.TabularInline): model = Snapshot.tags.through
from django.contrib.admin.helpers import ActionForm from django.contrib.admin.widgets import AutocompleteSelectMultiple # WIP: broken by Django 3.1.2 -> 4.0 migration
[docs] class AutocompleteTags: model = Tag search_fields = ['name'] name = 'tags'
[docs] class AutocompleteTagsAdminStub: name = 'admin'
[docs] class SnapshotActionForm(ActionForm): tags = forms.ModelMultipleChoiceField( queryset=Tag.objects.all(), required=False, # WIP: broken by Django 3.1.2 -> 4.0 migration widget=AutocompleteSelectMultiple( AutocompleteTags(), AutocompleteTagsAdminStub(), ), )
# TODO: allow selecting actions for specific extractors? is this useful? # EXTRACTOR_CHOICES = [ # (name, name.title()) # for name, _, _ in get_default_archive_methods() # ] # extractor = forms.ChoiceField( # choices=EXTRACTOR_CHOICES, # required=False, # widget=forms.MultileChoiceField(attrs={'class': "form-control"}) # )
[docs] class SnapshotAdmin(SearchResultsAdminMixin, admin.ModelAdmin): list_display = ('added', 'title_str', 'files', 'size', 'url_str') sort_fields = ('title_str', 'url_str', 'added', 'files') readonly_fields = ('info', 'bookmarked', 'added', 'updated') search_fields = ('id', 'url', 'timestamp', 'title', 'tags__name') fields = ('timestamp', 'url', 'title', 'tags', *readonly_fields) list_filter = ('added', 'updated', 'tags', 'archiveresult__status') ordering = ['-added'] actions = ['add_tags', 'remove_tags', 'update_titles', 'update_snapshots', 'resnapshot_snapshot', 'overwrite_snapshots', 'delete_snapshots'] autocomplete_fields = ['tags'] inlines = [ArchiveResultInline] list_per_page = SNAPSHOTS_PER_PAGE action_form = SnapshotActionForm
[docs] def changelist_view(self, request, extra_context=None): extra_context = extra_context or {} return super().changelist_view(request, extra_context | GLOBAL_CONTEXT)
[docs] def get_urls(self): urls = super().get_urls() custom_urls = [ path('grid/', self.admin_site.admin_view(self.grid_view), name='grid') ] return custom_urls + urls
[docs] def get_queryset(self, request): self.request = request return super().get_queryset(request).prefetch_related('tags').annotate(archiveresult_count=Count('archiveresult'))
[docs] def tag_list(self, obj): return ', '.join(obj.tags.values_list('name', flat=True))
# TODO: figure out a different way to do this, you cant nest forms so this doenst work # def action(self, obj): # # csrfmiddlewaretoken: Wa8UcQ4fD3FJibzxqHN3IYrrjLo4VguWynmbzzcPYoebfVUnDovon7GEMYFRgsh0 # # action: update_snapshots # # select_across: 0 # # _selected_action: 76d29b26-2a88-439e-877c-a7cca1b72bb3 # return format_html( # ''' # <form action="/admin/core/snapshot/" method="post" onsubmit="e => e.stopPropagation()"> # <input type="hidden" name="csrfmiddlewaretoken" value="{}"> # <input type="hidden" name="_selected_action" value="{}"> # <button name="update_snapshots">Check</button> # <button name="update_titles">Pull title + favicon</button> # <button name="update_snapshots">Update</button> # <button name="overwrite_snapshots">Re-Archive (overwrite)</button> # <button name="delete_snapshots">Permanently delete</button> # </form> # ''', # csrf.get_token(self.request), #, # )
[docs] def info(self, obj): return format_html( ''' UUID: <code style="font-size: 10px; user-select: all">{}</code> &nbsp; &nbsp; Timestamp: <code style="font-size: 10px; user-select: all">{}</code> &nbsp; &nbsp; URL Hash: <code style="font-size: 10px; user-select: all">{}</code><br/> Archived: {} ({} files {}) &nbsp; &nbsp; Favicon: <img src="{}" style="height: 20px"/> &nbsp; &nbsp; Status code: {} &nbsp; &nbsp; Server: {} &nbsp; &nbsp; Content type: {} &nbsp; &nbsp; Extension: {} &nbsp; &nbsp; <br/><br/> <a href="/archive/{}">View Snapshot index ➡️</a> &nbsp; &nbsp; <a href="/admin/core/snapshot/?id__exact={}">View actions ⚙️</a> ''',, obj.timestamp, obj.url_hash, '✅' if obj.is_archived else '❌', obj.num_outputs, self.size(obj), f'/archive/{obj.timestamp}/favicon.ico', obj.status_code or '?', obj.headers and obj.headers.get('Server') or '?', obj.headers and obj.headers.get('Content-Type') or '?', obj.extension or '?', obj.timestamp,, )
[docs] def title_str(self, obj): canon = obj.as_link().canonical_outputs() tags = ''.join( format_html('<a href="/admin/core/snapshot/?tags__id__exact={}"><span class="tag">{}</span></a> ',, tag) for tag in obj.tags.all() if str(tag).strip() ) return format_html( '<a href="/{}">' '<img src="/{}/{}" class="favicon" onerror="this.remove()">' '</a>' '<a href="/{}/index.html">' '<b class="status-{}">{}</b>' '</a>', obj.archive_path, obj.archive_path, canon['favicon_path'], obj.archive_path, 'fetched' if obj.latest_title or obj.title else 'pending', urldecode(htmldecode(obj.latest_title or obj.title or ''))[:128] or 'Pending...' ) + mark_safe(f' <span class="tags">{tags}</span>')
[docs] def files(self, obj): return snapshot_icons(obj)
files.admin_order_field = 'archiveresult_count' files.short_description = 'Files Saved'
[docs] def size(self, obj): archive_size = (Path(obj.link_dir) / 'index.html').exists() and obj.archive_size if archive_size: size_txt = printable_filesize(archive_size) if archive_size > 52428800: size_txt = mark_safe(f'<b>{size_txt}</b>') else: size_txt = mark_safe('<span style="opacity: 0.3">...</span>') return format_html( '<a href="/{}" title="View all files">{}</a>', obj.archive_path, size_txt, )
size.admin_order_field = 'archiveresult_count'
[docs] def url_str(self, obj): return format_html( '<a href="{}"><code style="user-select: all;">{}</code></a>', obj.url, obj.url, )
[docs] def grid_view(self, request, extra_context=None): # cl = self.get_changelist_instance(request) # Save before monkey patching to restore for changelist list view saved_change_list_template = self.change_list_template saved_list_per_page = self.list_per_page saved_list_max_show_all = self.list_max_show_all # Monkey patch here plus self.change_list_template = 'private_index_grid.html' self.list_per_page = SNAPSHOTS_PER_PAGE self.list_max_show_all = self.list_per_page # Call monkey patched view rendered_response = self.changelist_view(request, extra_context=extra_context) # Restore values self.change_list_template = saved_change_list_template self.list_per_page = saved_list_per_page self.list_max_show_all = saved_list_max_show_all return rendered_response
# for debugging, uncomment this to print all requests: # def changelist_view(self, request, extra_context=None): # print('[*] Got request', request.method, request.POST) # return super().changelist_view(request, extra_context=None)
[docs] def update_snapshots(self, request, queryset): archive_links([ snapshot.as_link() for snapshot in queryset ], out_dir=OUTPUT_DIR)
update_snapshots.short_description = "Pull"
[docs] def update_titles(self, request, queryset): archive_links([ snapshot.as_link() for snapshot in queryset ], overwrite=True, methods=('title','favicon'), out_dir=OUTPUT_DIR)
update_titles.short_description = "⬇️ Title"
[docs] def resnapshot_snapshot(self, request, queryset): for snapshot in queryset: timestamp ='T', 'seconds') new_url = snapshot.url.split('#')[0] + f'#{timestamp}' add(new_url, tag=snapshot.tags_str())
resnapshot_snapshot.short_description = "Re-Snapshot"
[docs] def overwrite_snapshots(self, request, queryset): archive_links([ snapshot.as_link() for snapshot in queryset ], overwrite=True, out_dir=OUTPUT_DIR)
overwrite_snapshots.short_description = "Reset"
[docs] def delete_snapshots(self, request, queryset): remove(snapshots=queryset, yes=True, delete=True, out_dir=OUTPUT_DIR)
delete_snapshots.short_description = "Delete"
[docs] def add_tags(self, request, queryset): tags = request.POST.getlist('tags') print('[+] Adding tags', tags, 'to Snapshots', queryset) for obj in queryset: obj.tags.add(*tags)
add_tags.short_description = "+"
[docs] def remove_tags(self, request, queryset): tags = request.POST.getlist('tags') print('[-] Removing tags', tags, 'to Snapshots', queryset) for obj in queryset: obj.tags.remove(*tags)
remove_tags.short_description = "–" title_str.short_description = 'Title' url_str.short_description = 'Original URL' title_str.admin_order_field = 'title' url_str.admin_order_field = 'url'
[docs] class TagAdmin(admin.ModelAdmin): list_display = ('slug', 'name', 'num_snapshots', 'snapshots', 'id') sort_fields = ('id', 'name', 'slug') readonly_fields = ('id', 'num_snapshots', 'snapshots') search_fields = ('id', 'name', 'slug') fields = (*readonly_fields, 'name', 'slug') actions = ['delete_selected'] ordering = ['-id']
[docs] def num_snapshots(self, obj): return format_html( '<a href="/admin/core/snapshot/?tags__id__exact={}">{} total</a>',, obj.snapshot_set.count(), )
[docs] def snapshots(self, obj): total_count = obj.snapshot_set.count() return mark_safe('<br/>'.join( format_html( '{} <code><a href="/admin/core/snapshot/{}/change"><b>[{}]</b></a> {}</code>', snap.updated.strftime('%Y-%m-%d %H:%M') if snap.updated else 'pending...',, snap.timestamp, snap.url, ) for snap in obj.snapshot_set.order_by('-updated')[:10] ) + (f'<br/><a href="/admin/core/snapshot/?tags__id__exact={}">and {total_count-10} more...<a>' if obj.snapshot_set.count() > 10 else ''))
[docs] class ArchiveResultAdmin(admin.ModelAdmin): list_display = ('id', 'start_ts', 'extractor', 'snapshot_str', 'tags_str', 'cmd_str', 'status', 'output_str') sort_fields = ('start_ts', 'extractor', 'status') readonly_fields = ('id', 'uuid', 'snapshot_str', 'tags_str') search_fields = ('id', 'uuid', 'snapshot__url', 'extractor', 'output', 'cmd_version', 'cmd', 'snapshot__timestamp') fields = (*readonly_fields, 'snapshot', 'extractor', 'status', 'start_ts', 'end_ts', 'output', 'pwd', 'cmd', 'cmd_version') autocomplete_fields = ['snapshot'] list_filter = ('status', 'extractor', 'start_ts', 'cmd_version') ordering = ['-start_ts'] list_per_page = SNAPSHOTS_PER_PAGE
[docs] def snapshot_str(self, obj): return format_html( '<a href="/archive/{}/index.html"><b><code>[{}]</code></b></a><br/>' '<small>{}</small>', obj.snapshot.timestamp, obj.snapshot.timestamp, obj.snapshot.url[:128], )
[docs] def tags_str(self, obj): return obj.snapshot.tags_str()
[docs] def cmd_str(self, obj): return format_html( '<pre>{}</pre>', ' '.join(obj.cmd) if isinstance(obj.cmd, list) else str(obj.cmd), )
[docs] def output_str(self, obj): return format_html( '<a href="/archive/{}/{}" class="output-link">↗️</a><pre>{}</pre>', obj.snapshot.timestamp, obj.output if (obj.status == 'succeeded') and obj.extractor not in ('title', 'archive_org') else 'index.html', obj.output, )
tags_str.short_description = 'tags' snapshot_str.short_description = 'snapshot'
[docs] class ArchiveBoxAdmin(admin.AdminSite): site_header = 'ArchiveBox' index_title = 'Links' site_title = 'Index'
[docs] def get_urls(self): return [ path('core/snapshot/add/', self.add_view, name='Add'), ] + super().get_urls()
[docs] def add_view(self, request): if not request.user.is_authenticated: return redirect(f'/admin/login/?next={request.path}') request.current_app = context = { **self.each_context(request), 'title': 'Add URLs', } if request.method == 'GET': context['form'] = AddLinkForm() elif request.method == 'POST': form = AddLinkForm(request.POST) if form.is_valid(): url = form.cleaned_data["url"] print(f'[+] Adding URL: {url}') depth = 0 if form.cleaned_data["depth"] == "0" else 1 input_kwargs = { "urls": url, "depth": depth, "update_all": False, "out_dir": OUTPUT_DIR, } add_stdout = StringIO() with redirect_stdout(add_stdout): add(**input_kwargs) print(add_stdout.getvalue()) context.update({ "stdout": ansi_to_html(add_stdout.getvalue().strip()), "form": AddLinkForm() }) else: context["form"] = form return render(template_name='add.html', request=request, context=context) = ArchiveBoxAdmin(), SnapshotAdmin), TagAdmin), ArchiveResultAdmin)'delete_selected')