Fix error handling.
[redakcja.git] / 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                     setattr(self, attname, wdvalue)
168
169     def wikidata_link(self):
170         if self.wikidata:
171             return format_html(
172                 '<a href="https://www.wikidata.org/wiki/{wd}" target="_blank">{wd}</a>',
173                 wd=self.wikidata,
174             )
175         else:
176             return ""
177
178     wikidata_link.admin_order_field = "wikidata"
179
180
181 class WikidataAdminMixin:
182     class Media:
183         css = {"screen": ("catalogue/wikidata_admin.css",)}
184         js = ("catalogue/wikidata_admin.js",)
185
186     def save_related(self, request, form, formsets, change):
187         super().save_related(request, form, formsets, change)
188         form.instance.save()