src/catalogue/wikidata.py

   1 # This file is part of FNP-Redakcja, licensed under GNU Affero GPLv3 or later.
   2 # Copyright © Fundacja Nowoczesna Polska. See NOTICE for more information.
   3 #
   4 from datetime import date
   5 from django.conf import settings
   6 from django.db import models
   7 from django.db.models.signals import m2m_changed
   8 from django.utils.html import format_html
   9 from django.utils.translation import gettext_lazy as _
  10 from wikidata.client import Client
  11 from wikidata.datavalue import DatavalueError
  12 from modeltranslation.translator import translator
  13 from modeltranslation.settings import AVAILABLE_LANGUAGES
  14 from .wikimedia import Downloadable
  15
  16
  17 class WikidataModel(models.Model):
  18     wikidata = models.CharField(
  19         max_length=255,
  20         blank=True,
  21         help_text=_('If you have a Wikidata ID, like "Q1337", enter it and save.'),
  22     )
  23
  24     class Meta:
  25         abstract = True
  26
  27     def get_wikidata_property(self, client, entity, wd, lang):
  28         wdvalue = None
  29
  30         if callable(wd):
  31             return wd(
  32                 lambda next_arg:
  33                 self.get_wikidata_property(
  34                     client, entity, next_arg, lang
  35                 )
  36             )
  37         elif wd == "description":
  38             wdvalue = entity.description.get(lang, str(entity.description))
  39         elif wd == "label":
  40             wdvalue = entity.label.get(lang, str(entity.label))
  41         elif wd[0] == 'P':
  42             try:
  43                 # TODO: lang?
  44                 wdvalue = entity.get(client.get(wd))
  45             except DatavalueError:
  46                 pass
  47         else:
  48             try:
  49                 # wiki links identified as 'plwiki' etc.
  50                 wdvalue = entity.attributes['sitelinks'][wd]['url']
  51             except KeyError:
  52                 pass
  53
  54         return wdvalue
  55
  56
  57     def wikidata_populate_field(self, client, entity, attname, wd, save, lang, force=False):
  58         if not force:
  59             model_field = self._meta.get_field(attname)
  60             if isinstance(model_field, models.ManyToManyField):
  61                 if getattr(self, attname).all().exists():
  62                     return
  63             else:
  64                 if getattr(self, attname):
  65                     return
  66
  67         wdvalue = self.get_wikidata_property(client, entity, wd, lang)
  68
  69         self.set_field_from_wikidata(attname, wdvalue, save=save)
  70
  71     def wikidata_populate(self, save=True, force=False):
  72         Wikidata = type(self).Wikidata
  73         client = Client()
  74         # Probably should getlist
  75         entity = client.get(self.wikidata)
  76         for attname in dir(Wikidata):
  77             if attname.startswith("_"):
  78                 continue
  79             wd = getattr(Wikidata, attname)
  80
  81             self.wikidata_populate_attribute(client, entity, attname, wd, save=save, force=force)
  82         if hasattr(Wikidata, '_supplement'):
  83             for attname, wd in Wikidata._supplement(self):
  84                 self.wikidata_populate_attribute(client, entity, attname, wd, save=save, force=force)
  85
  86     def wikidata_fields_for_attribute(self, attname):
  87         field = getattr(type(self), attname)
  88         if type(self) in translator._registry:
  89             try:
  90                 opts = translator.get_options_for_model(type(self))
  91             except:
  92                 pass
  93             else:
  94                 if attname in opts.fields:
  95                     tfields = opts.fields[attname]
  96                     for tf in tfields:
  97                         yield tf.name, tf.language
  98                     return
  99
 100         yield attname, settings.LANGUAGE_CODE
 101
 102     def wikidata_populate_attribute(self, client, entity, attname, wd, save, force=False):
 103         for fieldname, lang in self.wikidata_fields_for_attribute(attname):
 104             self.wikidata_populate_field(client, entity, fieldname, wd, save, lang, force=force)
 105
 106     def save(self, **kwargs):
 107         am_new = self.pk is None
 108
 109         super().save()
 110         if am_new and self.wikidata and hasattr(self, "Wikidata"):
 111             self.wikidata_populate()
 112
 113         kwargs.update(force_insert=False, force_update=True)
 114         super().save(**kwargs)
 115
 116     def set_field_from_wikidata(self, attname, wdvalue, save, language='pl'):
 117         if not wdvalue:
 118             return
 119         # Find out what this model field is
 120         model_field = self._meta.get_field(attname)
 121         skip_set = False
 122         if isinstance(model_field, models.ForeignKey):
 123             rel_model = model_field.related_model
 124             if issubclass(rel_model, WikidataModel):
 125                 label = wdvalue.label.get(language, str(wdvalue.label))
 126                 try:
 127                     wdvalue = rel_model.objects.get(wikidata=wdvalue.id)
 128                 except rel_model.DoesNotExist:
 129                     wdvalue = rel_model(wikidata=wdvalue.id)
 130                     if save:
 131                         wdvalue.save()
 132                 wdvalue._wikidata_label = label
 133                 setattr(self, attname, wdvalue)
 134         elif isinstance(model_field, models.ManyToManyField):
 135             rel_model = model_field.related_model
 136             if issubclass(rel_model, WikidataModel):
 137                 label = wdvalue.label.get(language, str(wdvalue.label))
 138                 try:
 139                     wdvalue = rel_model.objects.get(wikidata=wdvalue.id)
 140                 except rel_model.DoesNotExist:
 141                     wdvalue = rel_model(wikidata=wdvalue.id)
 142                     if save:
 143                         wdvalue.save()
 144                 wdvalue._wikidata_label = label
 145                 getattr(self, attname).set([wdvalue])
 146         else:
 147             # How to get original title?
 148             if isinstance(wdvalue, date):
 149                 if isinstance(model_field, models.IntegerField):
 150                     wdvalue = wdvalue.year
 151
 152             # If downloadable (and not save)?
 153             elif isinstance(wdvalue, Downloadable):
 154                 if save:
 155                     wdvalue.apply_to_field(self, attname)
 156                     skip_set = True
 157
 158             elif hasattr(wdvalue, 'label'):
 159                 wdvalue = wdvalue.label.get(language, str(wdvalue.label))
 160
 161             if not skip_set:
 162                 try:
 163                     wdvalue = model_field.to_python(wdvalue)
 164                 except:
 165                     pass
 166                 else:
 167                     max_length = getattr(model_field, 'max_length', None)
 168                     if max_length:
 169                         wdvalue = wdvalue[:max_length]
 170                     setattr(self, attname, wdvalue)
 171
 172     def wikidata_link(self):
 173         if self.wikidata:
 174             return format_html(
 175                 '<a href="https://www.wikidata.org/wiki/{wd}" target="_blank">{wd}</a>',
 176                 wd=self.wikidata,
 177             )
 178         else:
 179             return ""
 180
 181     wikidata_link.admin_order_field = "wikidata"
 182
 183
 184 class WikidataAdminMixin:
 185     class Media:
 186         css = {"screen": ("catalogue/wikidata_admin.css",)}
 187         js = ("catalogue/wikidata_admin.js",)
 188
 189     def save_related(self, request, form, formsets, change):
 190         super().save_related(request, form, formsets, change)
 191         form.instance.save()