__package__ = 'archivebox.core' import os import json from io import StringIO from pathlib import Path from contextlib import redirect_stdout from datetime import datetime, timezone from typing import Dict, Any from django.contrib import admin from django.db.models import Count, Q from django.urls import path, reverse 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.core.exceptions import ValidationError from django.conf import settings from django import forms from signal_webhooks.admin import WebhookAdmin from signal_webhooks.utils import get_webhook_model # from plugantic.admin import CustomPlugin from ..util import htmldecode, urldecode, ansi_to_html from core.models import Snapshot, ArchiveResult, Tag, SnapshotTag from core.forms import AddLinkForm from core.mixins import SearchResultsAdminMixin from api.models import APIToken from abid_utils.models import get_or_create_system_user_pk from abid_utils.admin import ABIDModelAdmin from index.html import snapshot_icons from logging_util import printable_filesize from main import add, remove from extractors import archive_links CONFIG = settings.CONFIG GLOBAL_CONTEXT = {'VERSION': CONFIG.VERSION, 'VERSIONS_AVAILABLE': CONFIG.VERSIONS_AVAILABLE, 'CAN_UPGRADE': CONFIG.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 ArchiveBoxAdmin(admin.AdminSite): site_header = 'ArchiveBox' index_title = 'Links' site_title = 'Index' namespace = 'admin' 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: Dict[str, Any] = { **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": CONFIG.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) archivebox_admin = ArchiveBoxAdmin() archivebox_admin.register(get_user_model()) archivebox_admin.disable_action('delete_selected') # archivebox_admin.register(CustomPlugin) # patch admin with methods to add data views (implemented by admin_data_views package) # https://github.com/MrThearMan/django-admin-data-views # https://mrthearman.github.io/django-admin-data-views/setup/ ############### Additional sections are defined in settings.ADMIN_DATA_VIEWS ######### from admin_data_views.admin import get_app_list, admin_data_index_view, get_admin_data_urls, get_urls archivebox_admin.get_app_list = get_app_list.__get__(archivebox_admin, ArchiveBoxAdmin) archivebox_admin.admin_data_index_view = admin_data_index_view.__get__(archivebox_admin, ArchiveBoxAdmin) # type: ignore archivebox_admin.get_admin_data_urls = get_admin_data_urls.__get__(archivebox_admin, ArchiveBoxAdmin) # type: ignore archivebox_admin.get_urls = get_urls(archivebox_admin.get_urls).__get__(archivebox_admin, ArchiveBoxAdmin) class ArchiveResultInline(admin.TabularInline): name = 'Archive Results Log' model = ArchiveResult # fk_name = 'snapshot' extra = 1 readonly_fields = ('result_id', 'start_ts', 'end_ts', 'extractor', 'command', 'cmd_version') fields = ('id', *readonly_fields, 'status', 'output') show_change_link = True # # classes = ['collapse'] # # list_display_links = ['abid'] def result_id(self, obj): return format_html('[{}]', reverse('admin:core_archiveresult_change', args=(obj.id,)), obj.abid) def command(self, obj): return format_html('{}', " ".join(obj.cmd or [])) class TagInline(admin.TabularInline): model = Tag.snapshot_set.through # type: ignore # fk_name = 'snapshot' fields = ('id', 'tag') extra = 1 # min_num = 1 max_num = 1000 autocomplete_fields = ( 'tag', ) from django.contrib.admin.helpers import ActionForm from django.contrib.admin.widgets import FilteredSelectMultiple # class AutocompleteTags: # model = Tag # search_fields = ['name'] # name = 'name' # # source_field = 'name' # remote_field = Tag._meta.get_field('name') # class AutocompleteTagsAdminStub: # name = 'admin' class SnapshotActionForm(ActionForm): tags = forms.ModelMultipleChoiceField( queryset=Tag.objects.all(), required=False, widget=FilteredSelectMultiple( 'core_tag__name', False, ), ) # 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"}) # ) def get_abid_info(self, obj): return format_html( # URL Hash: {}
''' {}     📖 API DOCS
    .abid:                   {}
    .abid.uuid:           {}
    .id:                       {}
    TS:                  {}   {}        {}: {}
    URI:                 {}     {}           {}: {}   SALT:   {}
    SUBTYPE:       {}           {}                           {}: {}
    RAND:             {}       {}                 {}: {}
    .old_id:                {}
''', obj.api_url, obj.api_url, obj.api_docs_url, str(obj.abid), str(obj.ABID.uuid), str(obj.id), obj.ABID.ts, str(obj.ABID.uuid)[0:14], obj.abid_ts_src, obj.abid_values['ts'].isoformat() if isinstance(obj.abid_values['ts'], datetime) else obj.abid_values['ts'], obj.ABID.uri, str(obj.ABID.uuid)[14:26], obj.abid_uri_src, str(obj.abid_values['uri']), obj.ABID.uri_salt, obj.ABID.subtype, str(obj.ABID.uuid)[26:28], obj.abid_subtype_src, str(obj.abid_values['subtype']), obj.ABID.rand, str(obj.ABID.uuid)[28:36], obj.abid_rand_src, str(obj.abid_values['rand'])[-7:], str(getattr(obj, 'old_id', '')), ) @admin.register(Snapshot, site=archivebox_admin) class SnapshotAdmin(SearchResultsAdminMixin, ABIDModelAdmin): class Meta: model = Snapshot list_display = ('added', 'title_str', 'files', 'size', 'url_str') # list_editable = ('title',) sort_fields = ('title_str', 'url_str', 'added', 'files') readonly_fields = ('tags', 'timestamp', 'admin_actions', 'status_info', 'bookmarked', 'added', 'updated', 'created', 'modified', 'API', 'link_dir') search_fields = ('id', 'url', 'abid', 'old_id', 'timestamp', 'title', 'tags__name') list_filter = ('added', 'updated', 'archiveresult__status', 'created_by', 'tags') fields = ('url', 'created_by', 'title', *readonly_fields) ordering = ['-added'] actions = ['add_tags', 'remove_tags', 'update_titles', 'update_snapshots', 'resnapshot_snapshot', 'overwrite_snapshots', 'delete_snapshots'] autocomplete_fields = ['tags'] inlines = [TagInline, ArchiveResultInline] list_per_page = CONFIG.SNAPSHOTS_PER_PAGE action_form = SnapshotActionForm save_on_top = True def changelist_view(self, request, extra_context=None): extra_context = extra_context or {} try: return super().changelist_view(request, extra_context | GLOBAL_CONTEXT) except Exception as e: self.message_user(request, f'Error occurred while loading the page: {str(e)} {request.GET} {request.POST}') return super().changelist_view(request, GLOBAL_CONTEXT) def change_view(self, request, object_id, form_url="", extra_context=None): snapshot = None try: snapshot = snapshot or Snapshot.objects.get(id=object_id) except (Snapshot.DoesNotExist, Snapshot.MultipleObjectsReturned, ValidationError): pass try: snapshot = snapshot or Snapshot.objects.get(abid=Snapshot.abid_prefix + object_id.split('_', 1)[-1]) except (Snapshot.DoesNotExist, ValidationError): pass try: snapshot = snapshot or Snapshot.objects.get(old_id=object_id) except (Snapshot.DoesNotExist, Snapshot.MultipleObjectsReturned, ValidationError): pass if snapshot: object_id = str(snapshot.id) return super().change_view( request, object_id, form_url, extra_context=extra_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', 'archiveresult_set').annotate(archiveresult_count=Count('archiveresult')) 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.pk, # ) def admin_actions(self, obj): return format_html( # URL Hash: {}
''' Summary page ➡️     Result files 📑     Admin actions ⚙️ ''', obj.timestamp, obj.timestamp, obj.pk, ) def status_info(self, obj): return format_html( # URL Hash: {}
''' Archived: {} ({} files {})     Favicon:     Status code: {}    
Server: {}     Content type: {}     Extension: {}     ''', '✅' if obj.is_archived else '❌', obj.num_outputs, self.size(obj) or '0kb', 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 '-', ) def API(self, obj): try: return get_abid_info(self, obj) except Exception as e: return str(e) @admin.display( description='Title', ordering='title', ) 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}') @admin.display( description='Files Saved', ordering='archiveresult_count', ) def files(self, obj): return snapshot_icons(obj) @admin.display( ordering='archiveresult_count' ) 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, ) @admin.display( description='Original URL', ordering='url', ) def url_str(self, obj): return format_html( '{}', obj.url, obj.url[:128], ) 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 = CONFIG.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) @admin.action( description="Pull" ) def update_snapshots(self, request, queryset): archive_links([ snapshot.as_link() for snapshot in queryset ], out_dir=CONFIG.OUTPUT_DIR) @admin.action( description="⬇️ Title" ) def update_titles(self, request, queryset): archive_links([ snapshot.as_link() for snapshot in queryset ], overwrite=True, methods=('title','favicon'), out_dir=CONFIG.OUTPUT_DIR) @admin.action( description="Re-Snapshot" ) 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()) @admin.action( description="Reset" ) def overwrite_snapshots(self, request, queryset): archive_links([ snapshot.as_link() for snapshot in queryset ], overwrite=True, out_dir=CONFIG.OUTPUT_DIR) @admin.action( description="Delete" ) def delete_snapshots(self, request, queryset): remove(snapshots=queryset, yes=True, delete=True, out_dir=CONFIG.OUTPUT_DIR) @admin.action( description="+" ) 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) @admin.action( 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) # @admin.register(SnapshotTag, site=archivebox_admin) # class SnapshotTagAdmin(ABIDModelAdmin): # list_display = ('id', 'snapshot', 'tag') # sort_fields = ('id', 'snapshot', 'tag') # search_fields = ('id', 'snapshot_id', 'tag_id') # fields = ('snapshot', 'id') # actions = ['delete_selected'] # ordering = ['-id'] # def API(self, obj): # return get_abid_info(self, obj) @admin.register(Tag, site=archivebox_admin) class TagAdmin(ABIDModelAdmin): list_display = ('created', 'created_by', 'abid', 'name', 'num_snapshots', 'snapshots') sort_fields = ('name', 'slug', 'abid', 'created_by', 'created') readonly_fields = ('slug', 'abid', 'created', 'modified', 'API', 'num_snapshots', 'snapshots') search_fields = ('abid', 'name', 'slug') fields = ('name', 'created_by', *readonly_fields) actions = ['delete_selected'] ordering = ['-created'] def API(self, obj): try: return get_abid_info(self, obj) except Exception as e: return str(e) def num_snapshots(self, tag): return format_html( '{} total', tag.id, tag.snapshot_set.count(), ) def snapshots(self, tag): total_count = tag.snapshot_set.count() return mark_safe('
'.join( format_html( '[{}] {}', snap.pk, snap.updated.strftime('%Y-%m-%d %H:%M') if snap.updated else 'pending...', snap.url[:64], ) for snap in tag.snapshot_set.order_by('-updated')[:10] ) + (f'
and {total_count-10} more...' if tag.snapshot_set.count() > 10 else '')) @admin.register(ArchiveResult, site=archivebox_admin) class ArchiveResultAdmin(ABIDModelAdmin): list_display = ('start_ts', 'snapshot_info', 'tags_str', 'extractor', 'cmd_str', 'status', 'output_str') sort_fields = ('start_ts', 'extractor', 'status') readonly_fields = ('cmd_str', 'snapshot_info', 'tags_str', 'created', 'modified', 'API', 'output_summary') search_fields = ('id', 'old_id', 'abid', 'snapshot__url', 'extractor', 'output', 'cmd_version', 'cmd', 'snapshot__timestamp') fields = ('snapshot', 'extractor', 'status', 'output', 'pwd', 'start_ts', 'end_ts', 'created_by', 'cmd_version', 'cmd', *readonly_fields) autocomplete_fields = ['snapshot'] list_filter = ('status', 'extractor', 'start_ts', 'cmd_version') ordering = ['-start_ts'] list_per_page = CONFIG.SNAPSHOTS_PER_PAGE @admin.display( description='Snapshot Info' ) def snapshot_info(self, result): return format_html( '[{}]   {}   {}
', result.snapshot.timestamp, result.snapshot.abid, result.snapshot.added.strftime('%Y-%m-%d %H:%M'), result.snapshot.url[:128], ) def API(self, obj): try: return get_abid_info(self, obj) except Exception as e: raise e return str(e) @admin.display( description='Snapshot Tags' ) def tags_str(self, result): return result.snapshot.tags_str() def cmd_str(self, result): return format_html( '
{}
', ' '.join(result.cmd) if isinstance(result.cmd, list) else str(result.cmd), ) def output_str(self, result): return format_html( '↗️
{}
', result.snapshot.timestamp, result.output if (result.status == 'succeeded') and result.extractor not in ('title', 'archive_org') else 'index.html', result.output, ) def output_summary(self, result): snapshot_dir = Path(CONFIG.OUTPUT_DIR) / str(result.pwd).split('data/', 1)[-1] output_str = format_html( '
{}

', result.output, ) output_str += format_html('See result files ...
', str(result.snapshot.timestamp))
        path_from_output_str = (snapshot_dir / result.output)
        output_str += format_html('{}/{}
', str(snapshot_dir), str(result.output))
        if path_from_output_str.exists():
            root_dir = str(path_from_output_str)
        else:
            root_dir = str(snapshot_dir)


        # print(root_dir, str(list(os.walk(root_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_str += format_html('{}{}/
', indent, os.path.basename(root))
            indentation_str = ' ' * 4 * (depth + 1)
            for filename in sorted(files):
                is_hidden = filename.startswith('.')
                output_str += format_html('{}{}
', int(not is_hidden), indentation_str, filename.strip())

        return output_str + format_html('
') @admin.register(APIToken, site=archivebox_admin) class APITokenAdmin(ABIDModelAdmin): list_display = ('created', 'abid', 'created_by', 'token_redacted', 'expires') sort_fields = ('abid', 'created', 'created_by', 'expires') readonly_fields = ('abid', 'created') search_fields = ('id', 'abid', 'created_by__username', 'token') fields = ('created_by', 'token', 'expires', *readonly_fields) list_filter = ('created_by',) ordering = ['-created'] list_per_page = 100 @admin.register(get_webhook_model(), site=archivebox_admin) class CustomWebhookAdmin(WebhookAdmin, ABIDModelAdmin): list_display = ('created', 'created_by', 'abid', *WebhookAdmin.list_display) sort_fields = ('created', 'created_by', 'abid', 'referenced_model', 'endpoint', 'last_success', 'last_error') readonly_fields = ('abid', 'created', *WebhookAdmin.readonly_fields)