1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
|
from django.contrib.auth.models import User
from django.contrib.contenttypes.models import ContentType
from django.contrib.gis.db import models
from django.contrib.postgres.fields import ArrayField
from ishtar_common.utils import ugettext_lazy as _
MAIN_CONTENT_TYPES = (("archaeological_operations", "operation"),)
class ApiUser(models.Model):
user_ptr = models.OneToOneField(
User, primary_key=True, related_name="apiuser", on_delete=models.CASCADE
)
ip = models.GenericIPAddressField(verbose_name=_("IP"))
class Meta:
verbose_name = _("API - User")
verbose_name_plural = _("API - Users")
def __str__(self):
return self.user_ptr.username
class ApiSearchModel(models.Model):
user = models.ForeignKey(ApiUser, on_delete=models.CASCADE)
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
limit_query = models.TextField(
verbose_name=_("Limit query"),
blank=True,
null=True,
help_text=_("Search query add to each request"),
)
class Meta:
verbose_name = _("API - Search model")
verbose_name_plural = _("API - Search models")
class ApiExternalSource(models.Model):
url = models.URLField(verbose_name=_("URL"))
name = models.CharField(verbose_name=_("Name"), max_length=200)
key = models.CharField(_("Key"), max_length=40)
class Meta:
verbose_name = _("API - External source")
verbose_name_plural = _("API - External sources")
def update_matches(self, content):
result = {
"created": 0,
"updated": 0,
"deleted": 0,
"search_model do not exist": [],
"type do not exist": [],
}
for search_model in content:
app, model_name = search_model.split(".")
try:
ct = ContentType.objects.get(app_label=app, model=model_name)
except ContentType.DoesNotExist:
result["search_model do not exist"].append(search_model)
continue
for ct_type, keys, values in content[search_model]:
tapp, tmodel_name = ct_type.split(".")
try:
ct_type = ContentType.objects.get(app_label=tapp, model=tmodel_name)
except ContentType.DoesNotExist:
result["type do not exist"].append(ct_type)
continue
t_model = ct_type.model_class()
current_matches = []
for slug, label in values:
current_matches.append(slug)
m, created = ApiKeyMatch.objects.get_or_create(
source=self,
search_model=ct,
associated_type=ct_type,
search_keys=keys,
distant_slug=slug,
defaults={"distant_label": label},
)
updated = False
if not created and m.distant_label != label:
updated = True
m.distant_label = label
if not m.do_not_match and not m.local_slug:
slug_key = "txt_idx"
if hasattr(t_model, "slug"):
slug_key = "slug"
q = t_model.objects.filter(**{slug_key: m.distant_slug})
if q.count():
local_value = q.all()[0]
setattr(m, "local_slug", getattr(local_value, slug_key))
m.local_value = str(local_value)
updated = True
if updated:
m.save()
if not created:
result["updated"] += 1
if created:
result["created"] += 1
# delete removed keys
q = ApiKeyMatch.objects.filter(
source=self, search_model=ct, associated_type=ct_type,
).exclude(distant_slug__in=current_matches)
result["deleted"] += q.count()
q.delete()
return result
def generate_match_csv(self):
pass
class ApiKeyMatch(models.Model):
source = models.ForeignKey(ApiExternalSource, on_delete=models.CASCADE)
search_model = models.ForeignKey(
ContentType, on_delete=models.CASCADE, verbose_name=_("Search model"),
related_name="key_match_search_models"
)
associated_type = models.ForeignKey(
ContentType, on_delete=models.CASCADE, verbose_name=_("Associated type"),
related_name="key_match_types"
)
search_keys = ArrayField(
models.CharField(max_length=200), verbose_name=_("Search keys"), blank=True
)
distant_slug = models.SlugField(
verbose_name=_("Distant key"), max_length=200, allow_unicode=True
)
distant_label = models.TextField(
verbose_name=_("Distant value"), blank=True, default=""
)
local_slug = models.SlugField(
verbose_name=_("Local key"), max_length=200, allow_unicode=True
)
local_label = models.TextField(
verbose_name=_("Local value"), blank=True, default=""
)
do_not_match = models.BooleanField(
verbose_name=_("Disable match for this search"), default=False
)
class Meta:
verbose_name = _("API - Key match")
verbose_name_plural = _("API - Keys matches")
|
