Coverage for discuss_data/core/helpers.py : 58%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import logging
3from django.core.exceptions import PermissionDenied
4from guardian.core import ObjectPermissionChecker
5from elasticsearch_dsl import Q
6from discuss_data.core.documents import DataSetDocument, UserDocument
7from discuss_data.pages.models import ManualPage
10logger = logging.getLogger(__name__)
13SEARCH_FIELDS_USER = [
14 "first_name",
15 "last_name^3",
16 "countries.name",
17 "interests.name",
18 "affiliation_user.place_of_institution",
19 "affiliation_user.country_of_institution",
20 "affiliation_user.name_of_institution",
21 "affiliation_user.position",
22]
23SEARCH_FIELDS_DATASET = [
24 "title^3",
25 "subtitle^2",
26 "dataset_creator.name",
27 "countries.name",
28 "keywords.name",
29 "published_categories.name",
30 "published_main_category.name",
31 "owner.first_name",
32 "owner.last_name^2",
33 "description",
34 "related_dataset_text",
35 "sources_of_data",
36 "funding",
37 "institutional_affiliation",
38 "languages_of_data.name",
39 "methods_of_data_collection.name",
40 "analysis_of_data_collection.name",
41 "disciplines.name",
42]
44# SEARCH_FILTERS = {
45# "countries": list(),
46# "categories": list(),
47# "keywords": list(),
48# "languages": list(),
49# "methods_of_data_collection": list(),
50# "methods_of_data_analysis": list(),
51# "disciplines": list(),
52# }
55# Commented out, as filters are not applied using this code.
56# Switching back to older version below.
57# def add_tag_filter(search, filter_name, filter_list, filter_field):
58# """
59# Add specific filters dynamically to search object using the alternative elasticsearch-dsl syntax
60# see: https://elasticsearch-dsl.readthedocs.io/en/latest/search_dsl.html#dotted-fields
61# """
62# if filter_list:
63# query_list = list()
64# # construct list of Q match clauses for filtering elements
65# for elem in filter_list:
66# if elem != "":
67# query_list.append(Q("match", **{"{}.name".format(filter_name): elem}))
68# # add the list to the search object
69# return search.filter("terms", **{filter_name + "." + filter_field: query_list})
70# else:
71# # return original search object if filter_list is None
72# return search
75def add_tag_filter(search, filter_name, filter_list):
76 """
77 Add specific filters dynamically using the alternative elasticsearch-dsl syntax
78 see: https://elasticsearch-dsl.readthedocs.io/en/latest/search_dsl.html#dotted-fields
79 """
80 if filter_list:
81 for elem in filter_list:
82 if elem != "":
83 search = search.query("match", **{"{}.name".format(filter_name): elem})
84 return search
87def index_search(index, term, **kwargs):
88 """
89 Using elastic simple_query_string with default wildcard attached, default AND
90 operator and a limited set of weighted fields. Parallel use of wildcards and
91 fuzziness is not possible. See fields and options:
92 https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-simple-query-string-query.html
93 """
95 if term:
96 # add fuzziness to each word in query for catching misspellings
97 # term_list = term.split(" ")
98 # wc_term = "~1 ".join(term_list)
99 # wc_term = wc_term + "~1"
100 # add wildcard to each word
101 term_list = term.split(" ")
102 wc_term = "* ".join(term_list)
103 wc_term = wc_term + "*"
104 else:
105 # use wildcard search for returning all objects in index if no term is provided (default on list pages)
106 wc_term = "*"
108 countries = kwargs.get("countries", None)
109 categories = kwargs.get("categories", None)
110 keywords = kwargs.get("keywords", None)
111 languages = kwargs.get("languages", None)
112 methods_of_data_collection = kwargs.get("methods_of_data_collection", None)
113 methods_of_data_analysis = kwargs.get("methods_of_data_analysis", None)
114 disciplines = kwargs.get("disciplines", None)
116 if index == "user_index":
117 document = UserDocument
118 search_fields = SEARCH_FIELDS_USER
120 # do only index published datasets!!
121 if index == "dataset_index":
122 document = DataSetDocument
123 search_fields = SEARCH_FIELDS_DATASET
125 # create an elastic simple query clause
126 q = Q(
127 "simple_query_string",
128 query=wc_term,
129 fields=search_fields,
130 default_operator="AND",
131 )
133 # "filter" does not work with categories as they come from 2 different fields.
134 # Using "query" clause instead, but only AND concatenation with q produces reasonable results
135 if categories:
136 for elem in categories:
137 if elem != "":
138 q_cat = Q("match", published_main_category__name=elem) | Q(
139 "match", published_categories__name=elem
140 )
141 # using & instead of | to add to query
142 q = q & q_cat
144 # create elastic bool query and include q query
145 filter_query = Q("bool", must=q)
147 # create search object from document
148 search = document.search().query(filter_query)
150 # add filters for both dataset_index and user_index to search
151 search = add_tag_filter(search, "countries", countries)
153 # add filters for dataset_index
154 if index == "dataset_index":
155 search = add_tag_filter(search, "keywords", keywords)
156 search = add_tag_filter(search, "languages_of_data", languages)
157 search = add_tag_filter(search, "disciplines", disciplines)
158 search = add_tag_filter(
159 search, "methods_of_data_collection", methods_of_data_collection
160 )
161 search = add_tag_filter(
162 search, "methods_of_data_analysis", methods_of_data_analysis
163 )
165 # add filters for user_index
166 if index == "user_index":
167 search = add_tag_filter(search, "interests", keywords)
169 # return all results, not just 10 (elastic default)
170 total = search.count()
171 search = search[0:total]
172 return search.to_queryset()
175def split_items_2(items):
176 output = list()
177 items_len = len(items)
178 for i in range(0, items_len, 2):
179 try:
180 output.append((items[i], items[i + 1]))
181 except IndexError:
182 output.append((items[i], None))
183 return output
186def split_items_3(items):
187 output = list()
188 items_len = len(items)
189 for i in range(0, items_len, 3):
190 try:
191 output.append((items[i], items[i + 1], items[i + 2]))
192 except IndexError:
193 try:
194 output.append((items[i], items[i + 1], None))
195 except IndexError:
196 output.append((items[i], None))
197 return output
200def check_perms_403(permission, user, dd_obj):
201 if not check_perms(permission, user, dd_obj):
202 raise PermissionDenied
205def is_curator_or_check_perms_for_ds(permission, user, ds):
206 if user in ds.get_curators():
207 return True
208 else:
209 check = check_perms(permission, user, ds)
210 return check
213def get_oep_parameters(dtype, outer_panel, add_text):
214 return {
215 "type": dtype,
216 "title": "{}s".format(dtype.capitalize()),
217 "edit_url": "{}_add".format(dtype),
218 "list_url": "get_user_{}s".format(dtype),
219 "outer_panel": outer_panel,
220 "add_text": add_text,
221 }
224def check_perms(permission, user, dd_obj):
225 try:
226 if dd_obj.owner == user:
227 logger.debug("is owner")
228 return True
229 except Exception:
230 logger.debug("is not owner")
232 checker = ObjectPermissionChecker(user)
234 if checker.has_perm(permission, dd_obj):
235 logger.debug("%s has perm %s on %s" % (user, permission, dd_obj))
236 return True
237 else:
238 logger.debug("%s lacks perm %s on %s" % (user, permission, dd_obj))
239 return False
240 return False
243# this is not a view function
244def get_help_texts(slug):
245 try: # try block because this code part gets somehow called before migrations are executed.
246 # Querying for a ManualPage thus triggers a db error in an empty db
247 # ("django.db.utils.ProgrammingError: relation "pages_manualpage" does not exist")
248 # as the db schema is not defined yet
249 manpage = ManualPage.objects.get(slug=slug)
250 help_text_dict = dict()
251 help_text_dict["manpage"] = manpage
252 for block in manpage.body:
253 help_text_dict[block.value["field_slug"]] = block.value["paragraph"]
254 return help_text_dict
255 except Exception:
256 return None