-
Notifications
You must be signed in to change notification settings - Fork 27
/
documents.py
137 lines (117 loc) · 4.75 KB
/
documents.py
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
from django_elasticsearch_dsl import Document, fields
from django_elasticsearch_dsl.registries import registry
from pydash import get
from core.common.utils import jsonify_safe, flatten_dict
from core.mappings.models import Mapping
@registry.register_document
class MappingDocument(Document):
class Index:
name = 'mappings'
settings = {'number_of_shards': 1, 'number_of_replicas': 0}
class Django:
model = Mapping
fields = [
'external_id'
]
last_update = fields.DateField(attr='updated_at')
updated_by = fields.KeywordField(attr='updated_by.username')
owner = fields.KeywordField(attr='owner_name', normalizer="lowercase")
owner_type = fields.KeywordField(attr='owner_type')
source = fields.KeywordField(attr='source', normalizer="lowercase")
retired = fields.KeywordField(attr='retired')
is_latest_version = fields.KeywordField(attr='is_latest_version')
is_in_latest_source_version = fields.KeywordField(attr='is_in_latest_source_version')
map_type = fields.KeywordField(attr='map_type', normalizer="lowercase")
from_concept = fields.ListField(fields.TextField())
to_concept = fields.ListField(fields.TextField())
from_concept_owner = fields.KeywordField(attr='from_source_owner')
to_concept_owner = fields.KeywordField(attr='to_source_owner')
from_concept_owner_type = fields.KeywordField(attr='from_source_owner_type')
to_concept_owner_type = fields.KeywordField(attr='to_source_owner_type')
from_concept_source = fields.KeywordField(attr='from_source_name')
to_concept_source = fields.KeywordField(attr='to_source_name')
source_version = fields.ListField(fields.KeywordField())
collection_version = fields.ListField(fields.KeywordField())
expansion = fields.ListField(fields.KeywordField())
collection = fields.ListField(fields.KeywordField())
collection_url = fields.ListField(fields.KeywordField())
collection_owner_url = fields.ListField(fields.KeywordField())
public_can_view = fields.BooleanField(attr='public_can_view')
id_lowercase = fields.KeywordField(attr='mnemonic', normalizer="lowercase")
id = fields.TextField(attr='mnemonic')
extras = fields.ObjectField(dynamic=True)
created_by = fields.KeywordField(attr='created_by.username')
@staticmethod
def get_match_phrase_attrs():
return ['from_concept', 'to_concept', 'external_id']
@staticmethod
def get_exact_match_attrs():
return {
'id': {
'boost': 4
},
'to_concept': {
'boost': 3.5,
},
'from_concept': {
'boost': 3,
}
}
@staticmethod
def get_wildcard_search_attrs():
return {
'id': {
'boost': 1
},
'from_concept': {
'boost': 0.8,
'lower': False
},
'to_concept': {
'boost': 0.6,
'lower': False
}
}
@staticmethod
def get_fuzzy_search_attrs():
return {
'from_concept': {
'boost': 0.8,
},
'to_concept': {
'boost': 0.6,
}
}
@staticmethod
def prepare_from_concept(instance):
from_concept_name = get(instance, 'from_concept_name') or get(instance, 'from_concept.display_name')
return [instance.from_concept_code, from_concept_name]
@staticmethod
def prepare_to_concept(instance):
return [instance.get_to_concept_code(), instance.get_to_concept_name()]
@staticmethod
def prepare_source_version(instance):
return list(instance.sources.values_list('version', flat=True))
@staticmethod
def prepare_collection_version(instance):
return list(set(instance.expansion_set.values_list('collection_version__version', flat=True)))
@staticmethod
def prepare_expansion(instance):
return list(instance.expansion_set.values_list('mnemonic', flat=True))
@staticmethod
def prepare_collection(instance):
return list(set(instance.expansion_set.values_list('collection_version__mnemonic', flat=True)))
@staticmethod
def prepare_collection_url(instance):
return list(set(list(instance.expansion_set.values_list('collection_version__uri', flat=True))))
@staticmethod
def prepare_collection_owner_url(instance):
return list(set(expansion.owner_url for expansion in instance.expansion_set.all()))
@staticmethod
def prepare_extras(instance):
value = {}
if instance.extras:
value = jsonify_safe(instance.extras)
if isinstance(value, dict):
value = flatten_dict(value)
return value or {}