src/contact/admin.py

   1 # This file is part of Wolnelektury, licensed under GNU Affero GPLv3 or later.
   2 # Copyright © Fundacja Nowoczesna Polska. See NOTICE for more information.
   3 #
   4 import csv
   5 import json
   6
   7 from django.contrib import admin
   8 from django.utils.translation import ugettext as _
   9 from django.utils.safestring import mark_safe
  10 from django.conf.urls import url
  11 from django.http import HttpResponse, Http404
  12
  13 from wolnelektury.utils import UnicodeCSVWriter
  14 from .forms import contact_forms, admin_list_width
  15 from .models import Contact
  16
  17
  18 class ContactAdminMeta(admin.ModelAdmin.__class__):
  19     def __getattr__(cls, name):
  20         if name.startswith('admin_list_'):
  21             return lambda self: ""
  22         raise AttributeError(name)
  23
  24
  25 class ContactAdmin(admin.ModelAdmin, metaclass=ContactAdminMeta):
  26     date_hierarchy = 'created_at'
  27     list_display = ['created_at', 'contact', 'form_tag'] + \
  28         ["admin_list_%d" % i for i in range(admin_list_width)]
  29     fields = ['form_tag', 'created_at', 'contact', 'ip']
  30     readonly_fields = ['form_tag', 'created_at', 'contact', 'ip']
  31     list_filter = ['form_tag']
  32
  33     @staticmethod
  34     def admin_list(obj, nr):
  35         try:
  36             field_name = contact_forms[obj.form_tag].admin_list[nr]
  37         except BaseException:
  38             return ''
  39         else:
  40             return Contact.pretty_print(obj.get_body_json().get(field_name, ''), for_html=True)
  41
  42     def __getattr__(self, name):
  43         if name.startswith('admin_list_'):
  44             nr = int(name[len('admin_list_'):])
  45             return lambda obj: self.admin_list(obj, nr)
  46         raise AttributeError(name)
  47
  48     def change_view(self, request, object_id, form_url='', extra_context=None):
  49         if object_id:
  50             try:
  51                 instance = Contact.objects.get(pk=object_id)
  52                 body = instance.get_body_json()
  53                 assert isinstance(body, dict)
  54             except (Contact.DoesNotExist, AssertionError):
  55                 pass
  56             else:
  57                 # Create readonly fields from the body JSON.
  58                 attachments = list(instance.attachment_set.all())
  59                 body_keys = list(body.keys()) + [a.tag for a in attachments]
  60
  61                 # Find the original form.
  62                 try:
  63                     orig_fields = contact_forms[instance.form_tag]().fields
  64                 except KeyError:
  65                     orig_fields = {}
  66
  67                 # Try to preserve the original order.
  68                 orig_keys = list(orig_fields.keys())
  69                 admin_keys = [key for key in orig_keys if key in body_keys] + \
  70                              [key for key in body_keys if key not in orig_keys]
  71                 admin_fields = ['body__%s' % key for key in admin_keys]
  72
  73                 self.readonly_fields.extend(admin_fields)
  74
  75                 self.fieldsets = [
  76                     (None, {'fields': self.fields}),
  77                     (_('Body'), {'fields': admin_fields}),
  78                 ]
  79
  80                 # Create field getters for fields and attachments.
  81                 def attach_getter(key, value):
  82                     def f(self):
  83                         return value
  84                     f.short_description = orig_fields[key].label if key in orig_fields else _(key)
  85                     setattr(self, "body__%s" % key, f)
  86
  87                 for k, v in body.items():
  88                     attach_getter(k, Contact.pretty_print(v, for_html=True))
  89
  90                 download_link = "<a href='%(url)s'>%(url)s</a>"
  91                 for attachment in attachments:
  92                     link = mark_safe(download_link % {
  93                             'url': attachment.get_absolute_url()})
  94                     attach_getter(attachment.tag, link)
  95         return super(ContactAdmin, self).change_view(
  96             request, object_id, form_url=form_url, extra_context=extra_context)
  97
  98     def changelist_view(self, request, extra_context=None):
  99         context = dict()
 100         if 'form_tag' in request.GET:
 101             form = contact_forms.get(request.GET['form_tag'])
 102             context['extract_types'] = [
 103                 {'slug': 'all', 'label': _('all')},
 104                 {'slug': 'contacts', 'label': _('contacts')}]
 105             context['extract_types'] += [type for type in getattr(form, 'extract_types', [])]
 106         return super(ContactAdmin, self).changelist_view(request, extra_context=context)
 107
 108     def get_urls(self):
 109         return [
 110             url(r'^extract/(?P<form_tag>[\w-]+)/(?P<extract_type_slug>[\w-]+)/$',
 111                 self.admin_site.admin_view(extract_view), name='contact_extract')
 112         ] + super(ContactAdmin, self).get_urls()
 113
 114
 115 def extract_view(request, form_tag, extract_type_slug):
 116     contacts_by_spec = dict()
 117     form = contact_forms.get(form_tag)
 118     if form is None and extract_type_slug not in ('contacts', 'all'):
 119         raise Http404
 120
 121     q = Contact.objects.filter(form_tag=form_tag)
 122     at_year = request.GET.get('created_at__year')
 123     at_month = request.GET.get('created_at__month')
 124     if at_year:
 125         q = q.filter(created_at__year=at_year)
 126         if at_month:
 127             q = q.filter(created_at__month=at_month)
 128
 129     # Segregate contacts by body key sets
 130     if form:
 131         orig_keys = list(form().fields.keys())
 132     else:
 133         orig_keys = []
 134     for contact in q.all():
 135         if extract_type_slug == 'contacts':
 136             keys = ['contact']
 137         elif extract_type_slug == 'all':
 138             keys = contact.get_body_json().keys() + ['contact']
 139             keys = [key for key in orig_keys if key in keys] + [key for key in keys if key not in orig_keys]
 140         else:
 141             keys = form.get_extract_fields(contact, extract_type_slug)
 142         contacts_by_spec.setdefault(tuple(keys), []).append(contact)
 143
 144     response = HttpResponse(content_type='text/csv')
 145     csv_writer = UnicodeCSVWriter(response)
 146
 147     # Generate list for each body key set
 148     for keys, contacts in contacts_by_spec.items():
 149         csv_writer.writerow(keys)
 150         for contact in contacts:
 151             if extract_type_slug == 'contacts':
 152                 records = [dict(contact=contact.contact)]
 153             elif extract_type_slug == 'all':
 154                 records = [dict(contact=contact.contact, **contact.get_body_json())]
 155             else:
 156                 records = form.get_extract_records(keys, contact, extract_type_slug)
 157
 158             for record in records:
 159                 for key in keys:
 160                     if key not in record:
 161                         record[key] = ''
 162                     if isinstance(record[key], str):
 163                         pass
 164                     elif isinstance(record[key], bool):
 165                         record[key] = 'tak' if record[key] else 'nie'
 166                     elif isinstance(record[key], (list, tuple)) and all(isinstance(v, str) for v in record[key]):
 167                         record[key] = ', '.join(record[key])
 168                     else:
 169                         record[key] = json.dumps(record[key])
 170
 171                 csv_writer.writerow([record[key] for key in keys])
 172         csv_writer.writerow([])
 173
 174     response['Content-Disposition'] = 'attachment; filename="kontakt.csv"'
 175     return response
 176
 177 admin.site.register(Contact, ContactAdmin)