__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.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 ( OUTPUT_DIR, SNAPSHOTS_PER_PAGE, VERSION, VERSIONS_AVAILABLE, CAN_UPGRADE ) GLOBAL_CONTEXT = {'VERSION': VERSION, 'VERSIONS_AVAILABLE': VERSIONS_AVAILABLE, 'CAN_UPGRADE': CAN_UPGRADE} # Admin URLs # /admin/ # /admin/login/ # /admin/core/ # /admin/core/snapshot/ # /admin/core/snapshot/:uuid/ # /admin/core/tag/ # /admin/core/tag/:uuid/ # TODO: https://stackoverflow.com/questions/40760880/add-custom-button-to-django-admin-panel class ArchiveResultInline(admin.TabularInline): model = ArchiveResult 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 class AutocompleteTags: model = Tag search_fields = ['name'] name = 'tags' class AutocompleteTagsAdminStub: name = 'admin' 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"}) # ) 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 def changelist_view(self, request, extra_context=None): extra_context = extra_context or {} return super().changelist_view(request, extra_context | GLOBAL_CONTEXT) 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 def get_queryset(self, request): self.request = request return super().get_queryset(request).prefetch_related('tags') 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( # ''' #
# # # # # # # #
# ''', # csrf.get_token(self.request), # obj.id, # ) def info(self, obj): return format_html( ''' UUID: {}     Timestamp: {}     URL Hash: {}
Archived: {} ({} files {})     Favicon:     Status code: {}     Server: {}     Content type: {}     Extension: {}    

View Snapshot index ➡️     View actions ⚙️ ''', obj.id, 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, obj.id, ) def title_str(self, obj): canon = obj.as_link().canonical_outputs() tags = ''.join( format_html('{} ', tag.id, tag) for tag in obj.tags.all() if str(tag).strip() ) return format_html( '' '' '' '' '{}' '', 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' {tags}') def files(self, obj): return snapshot_icons(obj) files.admin_order_field = 'updated' files.short_description = 'Files Saved' 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'{size_txt}') else: size_txt = mark_safe('...') return format_html( '{}', obj.archive_path, size_txt, ) size.admin_order_field = 'archiveresult__count' def url_str(self, obj): return format_html( '{}', obj.url, obj.url, ) 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 core_tags.py 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) def update_snapshots(self, request, queryset): archive_links([ snapshot.as_link() for snapshot in queryset ], out_dir=OUTPUT_DIR) update_snapshots.short_description = "Pull" 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" def resnapshot_snapshot(self, request, queryset): for snapshot in queryset: timestamp = datetime.now(timezone.utc).isoformat('T', 'seconds') new_url = snapshot.url.split('#')[0] + f'#{timestamp}' add(new_url, tag=snapshot.tags_str()) resnapshot_snapshot.short_description = "Re-Snapshot" 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" def delete_snapshots(self, request, queryset): remove(snapshots=queryset, yes=True, delete=True, out_dir=OUTPUT_DIR) delete_snapshots.short_description = "Delete" 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 = "+" 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' 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'] def num_snapshots(self, obj): return format_html( '{} total', obj.id, obj.snapshot_set.count(), ) def snapshots(self, obj): total_count = obj.snapshot_set.count() return mark_safe('
'.join( format_html( '{} [{}] {}', snap.updated.strftime('%Y-%m-%d %H:%M') if snap.updated else 'pending...', snap.id, snap.timestamp, snap.url, ) for snap in obj.snapshot_set.order_by('-updated')[:10] ) + (f'
and {total_count-10} more...' if obj.snapshot_set.count() > 10 else '')) 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 def snapshot_str(self, obj): return format_html( '[{}]
' '{}', obj.snapshot.timestamp, obj.snapshot.timestamp, obj.snapshot.url[:128], ) def tags_str(self, obj): return obj.snapshot.tags_str() def cmd_str(self, obj): return format_html( '
{}
', ' '.join(obj.cmd) if isinstance(obj.cmd, list) else str(obj.cmd), ) def output_str(self, obj): return format_html( '↗️
{}
', 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' class ArchiveBoxAdmin(admin.AdminSite): site_header = 'ArchiveBox' index_title = 'Links' site_title = 'Index' def get_urls(self): return [ path('core/snapshot/add/', self.add_view, name='Add'), ] + super().get_urls() def add_view(self, request): if not request.user.is_authenticated: return redirect(f'/admin/login/?next={request.path}') request.current_app = self.name 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) admin.site = ArchiveBoxAdmin() admin.site.register(get_user_model()) admin.site.register(Snapshot, SnapshotAdmin) admin.site.register(Tag, TagAdmin) admin.site.register(ArchiveResult, ArchiveResultAdmin) admin.site.disable_action('delete_selected')