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
|
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": [],
}
updated = []
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(search_model)
continue
t_model = ct_type.model_class()
for slug, label in values:
m, created = ApiKeyMatch.objects.get_or_create(
source=self, search_model=ct, 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
return result
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"))
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")
|
