#5950

Committed 11 Dec 2024 12:12PM UTC coverage: 74.961% (+0.08%) from 74.883%

Build # #5950

Build Type

Pull #6496

coveralls-python

Committed by

danmihaila

Commit Message

HDX-10398 only sysadmins can access the url

Pull Request Pull Request #6496: HDX-10398 export data completeness and rename "completness"

Run Details

57 of 68 new or added lines in 5 files covered. (83.82%)

62 existing lines in 3 files now uncovered.

12565 of 16762 relevant lines covered (74.96%)

0.75 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.86

/ckanext-hdx_org_group/ckanext/hdx_org_group/helpers/country_helper.py

import ckan
import logging
import ckan.model as model
import ckan.lib.base as base
import ckan.plugins.toolkit as tk
import ckanext.hdx_org_group.helpers.caching as caching
# import ckanext.hdx_package.helpers.screenshot as screenshot
from datetime import datetime
from ckanext.hdx_org_group.controller_logic.group_search_logic import GroupSearchLogic
import ckan.lib.helpers as h
from ckanext.hdx_search.helpers.constants import DEFAULT_SORTING
from ckan.common import config

c = tk.c
render = tk.render
abort = tk.abort
NotFound = NotFound = tk.ObjectNotFound
NotAuthorized = tk.NotAuthorized
get_action = tk.get_action
request = tk.request
_ = tk._
log = logging.getLogger(__name__)

lookup_group_plugin = ckan.lib.plugins.lookup_group_plugin
GROUP_TYPES = ['group']

# def get_latest_cod_dataset(country_name):
#     context = {'model': model, 'session': model.Session,
#                'user': c.user or c.author, 'for_view': True,
#                'auth_user_obj': c.userobj}
#
#     search = search_controller.HDXSearchController()
#
#     fq = 'groups:"{}" tags:cod +dataset_type:dataset'.format(country_name)
#     query_result = search._performing_search(u'', fq, ['organization', 'tags'], 1, 1, DEFAULT_SORTING, None,
#                                              None, context)
#
#     return next(iter(query_result.get('results', [])), None)


def _sort_datasets_by_is_good(data_completeness):
    categories = data_completeness.get('categories')
    for cat in categories:
        if cat.get('data_series'):
            for ds in cat.get('data_series'):
                datasets_list = ds.get('datasets')
                if datasets_list:
                    datasets_sorted_list = sorted(datasets_list, key=lambda item: item['is_good'] == False)
                    ds['datasets'] = datasets_sorted_list

    return data_completeness


def country_topline(id):
    log.info('The id of the page is: ' + id)

    country_dict = get_country(id)
    top_line_data_list = caching.cached_topline_numbers(id)
    template_data = {
        'data': {
            'country_dict': country_dict,
            'widgets': {
                'top_line_data_list': top_line_data_list
            }
        }
    }
    return base.render('country/country_topline.html', extra_vars=template_data)

    # return country_read(id=id, get_only_toplines=True)


def get_template_data(country_dict, not_filtered_facet_info):

    # latest_cod_dataset = get_latest_cod_dataset(country_dict.get('name'))

    follower_count = get_action('group_follower_count')(
        {'model': model, 'session': model.Session},
        {'id': country_dict['id']}
    )

    top_line_data_list = caching.cached_topline_numbers(country_dict['id'], country_dict.get('activity_level'))

    organization_list = _get_org_list_for_menu_from_facets(not_filtered_facet_info)
    # f_event_list = _get_event_list_for_featured(country_dict['id'])
    # f_thumbnail_list = _get_thumbnail_list_for_featured(country_dict, f_event_list,
    #                                                     not_filtered_facet_info.get('results'),
    #                                                     latest_cod_dataset)
    # f_organization_list = _get_org_list_for_featured_from_facets(not_filtered_facet_info)
    # f_tag_list = _get_tag_list_for_featured_from_facets(not_filtered_facet_info)

    data_completeness = _get_data_completeness(country_dict.get('name')) \
        if country_dict.get('data_completeness') == 'active' else None

    if data_completeness:
        data_completeness = _sort_datasets_by_is_good(data_completeness)

    template_data = {
        'data': {
            'country_dict': country_dict,
            'stats_section': {
                'organization_list': organization_list,
                'num_of_organizations':
                    len(not_filtered_facet_info.get('facets', {}).get('organization', {}).get('items', [])),
                'num_of_cods': not_filtered_facet_info.get('num_of_cods', 0),
                'num_of_datasets': not_filtered_facet_info.get('num_of_total_items'),
                'num_of_followers': follower_count
            },
            'widgets': {
                'top_line_data_list': top_line_data_list,
                # 'chart_data_list': chart_data_list,
                'show': len(top_line_data_list) > 0  # or len(chart_data_list) > 0
            },
            # 'featured_section': {
            #     'thumbnail_list': f_thumbnail_list,
            #     'event_list': f_event_list,
            #     'organization_list': f_organization_list[:5],
            #     'tag_list': f_tag_list[:10],
            #     'show': len(f_organization_list) > 0 or len(f_tag_list) > 0
            # },
            'data_completeness': data_completeness,

        },
        'errors': None,
        'error_summary': '',
    }

    return template_data


def _get_org_list_for_menu_from_facets(full_facet_info):
    org_list = [
        {
            'display_name': org.get('display_name'),
            'name': org.get('name'),
            'url': h.url_for('organization_read', id=org.get('name'))
        }
        for org in full_facet_info.get('facets', {}).get('organization', {}).get('items', [])
    ]
    return org_list


# def _get_org_list_for_featured_from_facets(full_facet_info):
#     org_list = [
#         {
#             'display_name': org.get('display_name'),
#             'name': org.get('name'),
#             'count': org.get('count'),
#             'url': h.url_for('organization_read', id=org.get('name'))
#         }
#         for org in full_facet_info.get('facets', {}).get('organization', {}).get('items', []) if
#         org.get('name') != 'hdx'
#     ]
#     result = sorted(org_list, key=itemgetter('count'), reverse=True)
#     return result


# def _get_tag_list_for_featured_from_facets(full_facet_info):
#     tag_list = [
#         {
#             'display_name': tag.get('display_name'),
#             'count': tag.get('count'),
#             'name': tag.get('name'),
#             'url': '?tags=' + tag.get('name') + '#dataset-filter-start'
#         }
#         for tag in full_facet_info.get('facets', {}).get('vocab_Topics', {}).get('items', [])
#     ]
#     tag_list_by_count = sorted(tag_list, key=itemgetter('count'), reverse=True)
#     return tag_list_by_count


# def _get_thumbnail_list_for_featured(country_dict, event_list, latest_datasets, latest_cod_dataset):
#     '''
#     :param country_dict:
#     :type country_dict: dict
#     :param event_list: if this was already fetched in the controller we can reuse it
#     :type event_list: list
#     :param latest_datasets: a list of the latest updated datasets for the country
#     :type latest_datasets: list
#     :return: list of dicts containing display_name, type and url for the featured thumbnails
#     :rtype: list
#     '''
#
#     cloned_latest_datasets = latest_datasets[:]
#     default_thumbnail_url = '/images/featured_locs_placeholder1.png'
#     thumbnail_list = [None, None]
#     if event_list:
#         thumbnail_list[0] = __event_as_thumbnail_dict(event_list[0], default_thumbnail_url)
#     elif cloned_latest_datasets:
#         thumbnail_list[0] = __dataset_as_thumbnail_dict(cloned_latest_datasets[0], default_thumbnail_url)
#         del cloned_latest_datasets[0]
#     if latest_cod_dataset:
#         cod_thumbnail_url = screenshot.create_download_link(latest_cod_dataset, default_thumbnail_url)
#         thumbnail_list[1] = __dataset_as_thumbnail_dict(latest_cod_dataset, cod_thumbnail_url, True)
#     elif cloned_latest_datasets:
#         thumbnail_list[1] = __dataset_as_thumbnail_dict(cloned_latest_datasets[0], default_thumbnail_url)
#         del cloned_latest_datasets[0]
#
#     if thumbnail_list[0] and thumbnail_list[1] \
#         and thumbnail_list[0].get('url') == thumbnail_list[1].get('url') and cloned_latest_datasets:
#         thumbnail_list[0] = __dataset_as_thumbnail_dict(cloned_latest_datasets[0], default_thumbnail_url)
#         del cloned_latest_datasets[0]
#
#     # thumbnail_list[0]['thumbnail_url'] = '/images/featured_locs_placeholder1.png'
#     # thumbnail_list[1]['thumbnail_url'] = '/images/featured_locs_placeholder2.png'
#     return thumbnail_list


# def __dataset_as_thumbnail_dict(dataset_dict, thumbnail_url, is_cod=False):
#     return {
#         'display_name': dataset_dict.get('title'),
#         'type': 'COD' if is_cod else 'Dataset',
#         'url': h.url_for('dataset_read', id=dataset_dict.get('name')),
#         'thumbnail_url': thumbnail_url
#     }


# def __event_as_thumbnail_dict(event_dict, thumbnail_url, is_cod=False):
#     return {
#         'display_name': event_dict.get('title'),
#         'type': 'Event',
#         'url': h.url_for('read_event', id=event_dict.get('name')),
#         'thumbnail_url': thumbnail_url
#     }


# def _get_event_list_for_featured(group_id):
#     context = {'model': model, 'session': model.Session, 'user': c.user or c.author, 'auth_user_obj': c.userobj}
#     pages_list = get_action('group_page_list')(context, {'id': group_id})
#     return pages_list


def get_country(id):
    context = {'model': model, 'session': model.Session,
               'user': c.user,
               'schema': _db_to_form_schema(group_type='group'),
               'for_view': True}
    try:
        context['include_datasets'] = False
        group_dict = get_action('hdx_light_group_show')(context, {'id': id})
        if group_dict.get('type') not in GROUP_TYPES:
            abort(404, _('Incorrect group type'))
        return group_dict

    except NotFound:
        abort(404, _('Group not found'))
    except NotAuthorized:
        abort(403, _('Unauthorized to read group %s') % id)


def _db_to_form_schema(group_type=None):
    """This is an interface to manipulate data from the database
    into a format suitable for the form (optional)"""
    return lookup_group_plugin(group_type).db_to_form_schema()


def _get_data_completeness(location_code):
    cached = config.get('hdx.datagrid.prod')
    data = None
    if cached:
        data = caching.cached_data_completeness(location_code)
    else:
        data = caching.cached_data_completeness.original(location_code)
    return data


def get_not_filtered_facet_info(country_dict):
    context = {'model': model, 'session': model.Session,
               'user': c.user or c.author, 'for_view': True,
               'auth_user_obj': c.userobj}

    fq = 'groups:"{}" +dataset_type:dataset'.format(country_dict.get('name'))

    # The facet titles are not really needed in this case but we need to follow the process
    facets = {'vocab_Topics': 'tags', 'organization': 'organization'}
    search_logic = GroupSearchLogic(country_dict.get('name'), None)
    query_result = search_logic._performing_search(u'', fq, list(facets.keys()), 2, 1, DEFAULT_SORTING, None,
                                                   None, context)
    non_filtered_facet_info = search_logic._prepare_facets_info(query_result.get('search_facets'), {}, {},
                                                                facets, query_result.get('count'), u'')

    non_filtered_facet_info['results'] = query_result.get('results', [])

    return non_filtered_facet_info

def hdx_datagrid_org_get_display_text(dataseries_dict):
    if 'stats' in dataseries_dict:
        if 'state' in dataseries_dict.get('stats'):
            if dataseries_dict.get('stats').get('state') == 'not_applicable':
                for flag in dataseries_dict.get('flags'):
                    if flag.get('key', '') == 'not_applicable' and flag.get('display_text'):
                        return flag.get('display_text')
    return 'Not applicable'

DATA_COMPLETENESS_LABELS_DICT = {
    'data_series': 'subcategories',
    'good_datasets_num': 'available_datasets_num',
    'dataset_goodness_percentage':'dataset_availability_percentage',
    'good_dataseries_num':'available_dataseries_num',
    'not_good_dataseries_num': 'outdated_dataseries_num',
    'good_dataseries_text': 'available_dataseries_text',
    'not_good_dataseries_text': 'outdated_dataseries_text',
    'dataseries_good_percentage':'dataseries_available_percentage',
    'dataseries_not_good_percentage':'dataseries_outdated_percentage',
    'is_good': 'is_available',
}

DATA_COMPLETENESS_STATE_DICT = {
    'good':'available',
    'not_good':'outdated',
    'empty':'empty',
}

DATA_COMPLETENESS_KEYS_TBR_LIST = ['rules']

def _transform_data_completeness(data):
    if isinstance(data, dict):
        new_dict = {}
        for key, value in data.items():
            # Replace keys if present in DATA_COMPLETENESS_LABELS_DICT
            new_key = DATA_COMPLETENESS_LABELS_DICT.get(key, key)

            # Replace state values if applicable
            if key == 'state' and isinstance(value, str):
                value = DATA_COMPLETENESS_STATE_DICT.get(value, value)

            # Recursively transform values
            new_dict[new_key] = _transform_data_completeness(value)
        return new_dict
    elif isinstance(data, list):
        return [_transform_data_completeness(item) for item in data]
    return data


def remove_keys_from_dict(data, keys_to_remove):
    """
    Recursively removes specified keys from a dictionary.

    :param data: The dictionary to process.
    :param keys_to_remove: List of keys to remove.
    :return: The modified dictionary.
    """
    if isinstance(data, dict):
        return {
            key: remove_keys_from_dict(value, keys_to_remove)
            for key, value in data.items()
            if key not in keys_to_remove
        }
    elif isinstance(data, list):
        return [remove_keys_from_dict(item, keys_to_remove) for item in data]
    else:
        return data

def hdx_replace_datagrid_labels(data_completeness, grp_dict):
    # Replace keys in the dictionary
    updated_data = _transform_data_completeness(data_completeness)
    # Remove specified keys
    updated_data = remove_keys_from_dict(updated_data, DATA_COMPLETENESS_KEYS_TBR_LIST)
    updated_data.pop('inherits_from', None)
    updated_data.pop('name', None)
    updated_data.pop('title', None)
    updated_data.pop('description', None)
    updated_data['stats'].pop('state', None)
    updated_data['date'] = datetime.now()
    updated_data['iso3'] = grp_dict.get('name')
    updated_data['title'] = grp_dict.get('title') or grp_dict.get('display_name')
    return updated_data

1	import ckan	1✔
2	import logging	1✔
3	import ckan.model as model	1✔
4	import ckan.lib.base as base	1✔
5	import ckan.plugins.toolkit as tk	1✔
6	import ckanext.hdx_org_group.helpers.caching as caching	1✔
7	# import ckanext.hdx_package.helpers.screenshot as screenshot
8	from datetime import datetime	1✔
9	from ckanext.hdx_org_group.controller_logic.group_search_logic import GroupSearchLogic	1✔
10	import ckan.lib.helpers as h	1✔
11	from ckanext.hdx_search.helpers.constants import DEFAULT_SORTING	1✔
12	from ckan.common import config	1✔
13
14	c = tk.c	1✔
15	render = tk.render	1✔
16	abort = tk.abort	1✔
17	NotFound = NotFound = tk.ObjectNotFound	1✔
18	NotAuthorized = tk.NotAuthorized	1✔
19	get_action = tk.get_action	1✔
20	request = tk.request	1✔
21	_ = tk._	1✔
22	log = logging.getLogger(__name__)	1✔
23
24	lookup_group_plugin = ckan.lib.plugins.lookup_group_plugin	1✔
25	GROUP_TYPES = ['group']	1✔
26
27	# def get_latest_cod_dataset(country_name):
28	# context = {'model': model, 'session': model.Session,
29	# 'user': c.user or c.author, 'for_view': True,
30	# 'auth_user_obj': c.userobj}
31	#
32	# search = search_controller.HDXSearchController()
33	#
34	# fq = 'groups:"{}" tags:cod +dataset_type:dataset'.format(country_name)
35	# query_result = search._performing_search(u'', fq, ['organization', 'tags'], 1, 1, DEFAULT_SORTING, None,
36	# None, context)
37	#
38	# return next(iter(query_result.get('results', [])), None)
39
40
41	def _sort_datasets_by_is_good(data_completeness):	1✔
NEW 42	categories = data_completeness.get('categories')	×
43	for cat in categories:	×
NEW 44	if cat.get('data_series'):	×
NEW 45	for ds in cat.get('data_series'):	×
NEW 46	datasets_list = ds.get('datasets')	×
47	if datasets_list:	×
48	datasets_sorted_list = sorted(datasets_list, key=lambda item: item['is_good'] == False)	×
49	ds['datasets'] = datasets_sorted_list	×
50
51	return data_completeness	×
52
53
54	def country_topline(id):	1✔
NEW 55	log.info('The id of the page is: ' + id)	×
56
57	country_dict = get_country(id)	×
58	top_line_data_list = caching.cached_topline_numbers(id)	×
59	template_data = {	×
60	'data': {
61	'country_dict': country_dict,
62	'widgets': {
63	'top_line_data_list': top_line_data_list
64	}
65	}
66	}
67	return base.render('country/country_topline.html', extra_vars=template_data)	×
68
69	# return country_read(id=id, get_only_toplines=True)
70
71
72	def get_template_data(country_dict, not_filtered_facet_info):	1✔
73
74	# latest_cod_dataset = get_latest_cod_dataset(country_dict.get('name'))
75
76	follower_count = get_action('group_follower_count')(	1✔
77	{'model': model, 'session': model.Session},
78	{'id': country_dict['id']}
79	)
80
81	top_line_data_list = caching.cached_topline_numbers(country_dict['id'], country_dict.get('activity_level'))	1✔
82
83	organization_list = _get_org_list_for_menu_from_facets(not_filtered_facet_info)	1✔
84	# f_event_list = _get_event_list_for_featured(country_dict['id'])
85	# f_thumbnail_list = _get_thumbnail_list_for_featured(country_dict, f_event_list,
86	# not_filtered_facet_info.get('results'),
87	# latest_cod_dataset)
88	# f_organization_list = _get_org_list_for_featured_from_facets(not_filtered_facet_info)
89	# f_tag_list = _get_tag_list_for_featured_from_facets(not_filtered_facet_info)
90
91	data_completeness = _get_data_completeness(country_dict.get('name')) \	1✔
92	if country_dict.get('data_completeness') == 'active' else None
93
94	if data_completeness:	1✔
NEW 95	data_completeness = _sort_datasets_by_is_good(data_completeness)	×
96
97	template_data = {	1✔
98	'data': {
99	'country_dict': country_dict,
100	'stats_section': {
101	'organization_list': organization_list,
102	'num_of_organizations':
103	len(not_filtered_facet_info.get('facets', {}).get('organization', {}).get('items', [])),
104	'num_of_cods': not_filtered_facet_info.get('num_of_cods', 0),
105	'num_of_datasets': not_filtered_facet_info.get('num_of_total_items'),
106	'num_of_followers': follower_count
107	},
108	'widgets': {
109	'top_line_data_list': top_line_data_list,
110	# 'chart_data_list': chart_data_list,
111	'show': len(top_line_data_list) > 0 # or len(chart_data_list) > 0
112	},
113	# 'featured_section': {
114	# 'thumbnail_list': f_thumbnail_list,
115	# 'event_list': f_event_list,
116	# 'organization_list': f_organization_list[:5],
117	# 'tag_list': f_tag_list[:10],
118	# 'show': len(f_organization_list) > 0 or len(f_tag_list) > 0
119	# },
120	'data_completeness': data_completeness,
121
122	},
123	'errors': None,
124	'error_summary': '',
125	}
126
127	return template_data	1✔
128
129
130	def _get_org_list_for_menu_from_facets(full_facet_info):	1✔
131	org_list = [	1✔
132	{
133	'display_name': org.get('display_name'),
134	'name': org.get('name'),
135	'url': h.url_for('organization_read', id=org.get('name'))
136	}
137	for org in full_facet_info.get('facets', {}).get('organization', {}).get('items', [])
138	]
139	return org_list	1✔
140
141
142	# def _get_org_list_for_featured_from_facets(full_facet_info):
143	# org_list = [
144	# {
145	# 'display_name': org.get('display_name'),
146	# 'name': org.get('name'),
147	# 'count': org.get('count'),
148	# 'url': h.url_for('organization_read', id=org.get('name'))
149	# }
150	# for org in full_facet_info.get('facets', {}).get('organization', {}).get('items', []) if
151	# org.get('name') != 'hdx'
152	# ]
153	# result = sorted(org_list, key=itemgetter('count'), reverse=True)
154	# return result
155
156
157	# def _get_tag_list_for_featured_from_facets(full_facet_info):
158	# tag_list = [
159	# {
160	# 'display_name': tag.get('display_name'),
161	# 'count': tag.get('count'),
162	# 'name': tag.get('name'),
163	# 'url': '?tags=' + tag.get('name') + '#dataset-filter-start'
164	# }
165	# for tag in full_facet_info.get('facets', {}).get('vocab_Topics', {}).get('items', [])
166	# ]
167	# tag_list_by_count = sorted(tag_list, key=itemgetter('count'), reverse=True)
168	# return tag_list_by_count
169
170
171	# def _get_thumbnail_list_for_featured(country_dict, event_list, latest_datasets, latest_cod_dataset):
172	# '''
173	# :param country_dict:
174	# :type country_dict: dict
175	# :param event_list: if this was already fetched in the controller we can reuse it
176	# :type event_list: list
177	# :param latest_datasets: a list of the latest updated datasets for the country
178	# :type latest_datasets: list
179	# :return: list of dicts containing display_name, type and url for the featured thumbnails
180	# :rtype: list
181	# '''
182	#
183	# cloned_latest_datasets = latest_datasets[:]
184	# default_thumbnail_url = '/images/featured_locs_placeholder1.png'
185	# thumbnail_list = [None, None]
186	# if event_list:
187	# thumbnail_list[0] = __event_as_thumbnail_dict(event_list[0], default_thumbnail_url)
188	# elif cloned_latest_datasets:
189	# thumbnail_list[0] = __dataset_as_thumbnail_dict(cloned_latest_datasets[0], default_thumbnail_url)
190	# del cloned_latest_datasets[0]
191	# if latest_cod_dataset:
192	# cod_thumbnail_url = screenshot.create_download_link(latest_cod_dataset, default_thumbnail_url)
193	# thumbnail_list[1] = __dataset_as_thumbnail_dict(latest_cod_dataset, cod_thumbnail_url, True)
194	# elif cloned_latest_datasets:
195	# thumbnail_list[1] = __dataset_as_thumbnail_dict(cloned_latest_datasets[0], default_thumbnail_url)
196	# del cloned_latest_datasets[0]
197	#
198	# if thumbnail_list[0] and thumbnail_list[1] \
199	# and thumbnail_list[0].get('url') == thumbnail_list[1].get('url') and cloned_latest_datasets:
200	# thumbnail_list[0] = __dataset_as_thumbnail_dict(cloned_latest_datasets[0], default_thumbnail_url)
201	# del cloned_latest_datasets[0]
202	#
203	# # thumbnail_list[0]['thumbnail_url'] = '/images/featured_locs_placeholder1.png'
204	# # thumbnail_list[1]['thumbnail_url'] = '/images/featured_locs_placeholder2.png'
205	# return thumbnail_list
206
207
208	# def __dataset_as_thumbnail_dict(dataset_dict, thumbnail_url, is_cod=False):
209	# return {
210	# 'display_name': dataset_dict.get('title'),
211	# 'type': 'COD' if is_cod else 'Dataset',
212	# 'url': h.url_for('dataset_read', id=dataset_dict.get('name')),
213	# 'thumbnail_url': thumbnail_url
214	# }
215
216
217	# def __event_as_thumbnail_dict(event_dict, thumbnail_url, is_cod=False):
218	# return {
219	# 'display_name': event_dict.get('title'),
220	# 'type': 'Event',
221	# 'url': h.url_for('read_event', id=event_dict.get('name')),
222	# 'thumbnail_url': thumbnail_url
223	# }
224
225
226	# def _get_event_list_for_featured(group_id):
227	# context = {'model': model, 'session': model.Session, 'user': c.user or c.author, 'auth_user_obj': c.userobj}
228	# pages_list = get_action('group_page_list')(context, {'id': group_id})
229	# return pages_list
230
231
232	def get_country(id):	1✔
233	context = {'model': model, 'session': model.Session,	1✔
234	'user': c.user,
235	'schema': _db_to_form_schema(group_type='group'),
236	'for_view': True}
237	try:	1✔
238	context['include_datasets'] = False	1✔
239	group_dict = get_action('hdx_light_group_show')(context, {'id': id})	1✔
240	if group_dict.get('type') not in GROUP_TYPES:	1✔
241	abort(404, _('Incorrect group type'))	×
242	return group_dict	1✔
243
244	except NotFound:	×
245	abort(404, _('Group not found'))	×
246	except NotAuthorized:	×
247	abort(403, _('Unauthorized to read group %s') % id)	×
248
249
250	def _db_to_form_schema(group_type=None):	1✔
251	"""This is an interface to manipulate data from the database
252	into a format suitable for the form (optional)"""
253	return lookup_group_plugin(group_type).db_to_form_schema()	1✔
254
255
256	def _get_data_completeness(location_code):	1✔
257	cached = config.get('hdx.datagrid.prod')	1✔
258	data = None	1✔
259	if cached:	1✔
260	data = caching.cached_data_completeness(location_code)	×
261	else:
262	data = caching.cached_data_completeness.original(location_code)	1✔
263	return data	1✔
264
265
266	def get_not_filtered_facet_info(country_dict):	1✔
267	context = {'model': model, 'session': model.Session,	1✔
268	'user': c.user or c.author, 'for_view': True,
269	'auth_user_obj': c.userobj}
270
271	fq = 'groups:"{}" +dataset_type:dataset'.format(country_dict.get('name'))	1✔
272
273	# The facet titles are not really needed in this case but we need to follow the process
274	facets = {'vocab_Topics': 'tags', 'organization': 'organization'}	1✔
275	search_logic = GroupSearchLogic(country_dict.get('name'), None)	1✔
276	query_result = search_logic._performing_search(u'', fq, list(facets.keys()), 2, 1, DEFAULT_SORTING, None,	1✔
277	None, context)
278	non_filtered_facet_info = search_logic._prepare_facets_info(query_result.get('search_facets'), {}, {},	1✔
279	facets, query_result.get('count'), u'')
280
281	non_filtered_facet_info['results'] = query_result.get('results', [])	1✔
282
283	return non_filtered_facet_info	1✔
284
285	def hdx_datagrid_org_get_display_text(dataseries_dict):	1✔
286	if 'stats' in dataseries_dict:	×
287	if 'state' in dataseries_dict.get('stats'):	×
288	if dataseries_dict.get('stats').get('state') == 'not_applicable':	×
289	for flag in dataseries_dict.get('flags'):	×
290	if flag.get('key', '') == 'not_applicable' and flag.get('display_text'):	×
291	return flag.get('display_text')	×
292	return 'Not applicable'	×
293
294	DATA_COMPLETENESS_LABELS_DICT = {	1✔
295	'data_series': 'subcategories',
296	'good_datasets_num': 'available_datasets_num',
297	'dataset_goodness_percentage':'dataset_availability_percentage',
298	'good_dataseries_num':'available_dataseries_num',
299	'not_good_dataseries_num': 'outdated_dataseries_num',
300	'good_dataseries_text': 'available_dataseries_text',
301	'not_good_dataseries_text': 'outdated_dataseries_text',
302	'dataseries_good_percentage':'dataseries_available_percentage',
303	'dataseries_not_good_percentage':'dataseries_outdated_percentage',
304	'is_good': 'is_available',
305	}
306
307	DATA_COMPLETENESS_STATE_DICT = {	1✔
308	'good':'available',
309	'not_good':'outdated',
310	'empty':'empty',
311	}
312
313	DATA_COMPLETENESS_KEYS_TBR_LIST = ['rules']	1✔
314
315	def _transform_data_completeness(data):	1✔
316	if isinstance(data, dict):	1✔
317	new_dict = {}	1✔
318	for key, value in data.items():	1✔
319	# Replace keys if present in DATA_COMPLETENESS_LABELS_DICT
320	new_key = DATA_COMPLETENESS_LABELS_DICT.get(key, key)	1✔
321
322	# Replace state values if applicable
323	if key == 'state' and isinstance(value, str):	1✔
324	value = DATA_COMPLETENESS_STATE_DICT.get(value, value)	1✔
325
326	# Recursively transform values
327	new_dict[new_key] = _transform_data_completeness(value)	1✔
328	return new_dict	1✔
329	elif isinstance(data, list):	1✔
330	return [_transform_data_completeness(item) for item in data]	1✔
331	return data	1✔
332
333
334	def remove_keys_from_dict(data, keys_to_remove):	1✔
335	"""
336	Recursively removes specified keys from a dictionary.
337
338	:param data: The dictionary to process.
339	:param keys_to_remove: List of keys to remove.
340	:return: The modified dictionary.
341	"""
342	if isinstance(data, dict):	1✔
343	return {	1✔
344	key: remove_keys_from_dict(value, keys_to_remove)
345	for key, value in data.items()
346	if key not in keys_to_remove
347	}
348	elif isinstance(data, list):	1✔
349	return [remove_keys_from_dict(item, keys_to_remove) for item in data]	1✔
350	else:
351	return data	1✔
352
353	def hdx_replace_datagrid_labels(data_completeness, grp_dict):	1✔
354	# Replace keys in the dictionary
355	updated_data = _transform_data_completeness(data_completeness)	1✔
356	# Remove specified keys
357	updated_data = remove_keys_from_dict(updated_data, DATA_COMPLETENESS_KEYS_TBR_LIST)	1✔
358	updated_data.pop('inherits_from', None)	1✔
359	updated_data.pop('name', None)	1✔
360	updated_data.pop('title', None)	1✔
361	updated_data.pop('description', None)	1✔
362	updated_data['stats'].pop('state', None)	1✔
363	updated_data['date'] = datetime.now()	1✔
364	updated_data['iso3'] = grp_dict.get('name')	1✔
365	updated_data['title'] = grp_dict.get('title') or grp_dict.get('display_name')	1✔
366	return updated_data	1✔

OCHA-DAP / hdx-ckan / #5950

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous