-
Notifications
You must be signed in to change notification settings - Fork 27
/
documents.py
154 lines (132 loc) · 4.57 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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
import json
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, format_url_for_search
from core.sources.models import Source
@registry.register_document
class SourceDocument(Document):
class Index:
name = 'sources'
settings = {'number_of_shards': 1, 'number_of_replicas': 0}
locale = fields.ListField(fields.KeywordField())
last_update = fields.DateField(attr='updated_at')
updated_by = fields.KeywordField(attr='updated_by.username')
owner = fields.KeywordField(attr='parent_resource', normalizer='lowercase')
owner_type = fields.KeywordField(attr='parent_resource_type')
public_can_view = fields.TextField(attr='public_can_view')
source_type = fields.KeywordField(attr='source_type', normalizer='lowercase')
version = fields.KeywordField(attr='version')
name = fields.TextField(attr='name')
_name = fields.KeywordField(attr='name', normalizer='lowercase')
canonical_url = fields.TextField(attr='canonical_url')
_canonical_url = fields.TextField()
mnemonic = fields.TextField(attr='mnemonic')
_mnemonic = fields.KeywordField(attr='mnemonic', normalizer='lowercase')
extras = fields.ObjectField(dynamic=True)
identifier = fields.ObjectField()
jurisdiction = fields.ObjectField()
publisher = fields.KeywordField(attr='publisher', normalizer='lowercase')
content_type = fields.KeywordField(attr='content_type', normalizer='lowercase')
custom_validation_schema = fields.KeywordField(attr='custom_validation_schema', normalizer='lowercase')
hierarchy_meaning = fields.KeywordField()
created_by = fields.KeywordField()
class Django:
model = Source
fields = [
'full_name',
'revision_date',
'retired',
'experimental',
'case_sensitive',
'compositional',
'version_needed',
'external_id',
]
@staticmethod
def get_match_phrase_attrs():
return ['name', 'external_id', 'canonical_url']
@staticmethod
def get_exact_match_attrs():
return {
'mnemonic': {
'boost': 4,
},
'name': {
'boost': 3.5,
},
'canonical_url': {
'boost': 3,
},
}
@staticmethod
def get_wildcard_search_attrs():
return {
'mnemonic': {
'boost': 1,
'lower': True,
'wildcard': True
},
'name': {
'boost': 0.8,
'lower': True,
'wildcard': True
},
'canonical_url': {
'boost': 0.6,
'lower': True,
'wildcard': True
}
}
@staticmethod
def get_fuzzy_search_attrs():
return {
'name': {
'boost': 0.8,
},
}
@staticmethod
def prepare_locale(instance):
return get(instance.supported_locales, [])
@staticmethod
def prepare_extras(instance):
value = {}
if instance.extras:
value = jsonify_safe(instance.extras)
if isinstance(value, dict):
value = flatten_dict(value)
if value:
value = json.loads(json.dumps(value))
return value or {}
@staticmethod
def prepare_identifier(instance):
value = {}
if instance.identifier:
value = jsonify_safe(instance.identifier)
if isinstance(value, dict):
value = flatten_dict(value)
if isinstance(value, str):
value = {'value': value}
return value or {}
@staticmethod
def prepare_jurisdiction(instance):
value = {}
if instance.jurisdiction:
value = jsonify_safe(instance.jurisdiction)
if isinstance(value, dict):
value = flatten_dict(value)
if isinstance(value, str):
value = {'value': value}
return value or {}
@staticmethod
def prepare_created_by(instance):
return instance.created_by.username
@staticmethod
def prepare_hierarchy_meaning(instance):
hierarchy_meaning = instance.hierarchy_meaning
if hierarchy_meaning:
return hierarchy_meaning.lower()
return 'None'
@staticmethod
def prepare__canonical_url(instance):
return format_url_for_search(instance.canonical_url)