26fb17fe-5a2d-422c-9747-c0adef10e362

Committed 25 Mar 2025 10:41AM UTC coverage: 85.751% (-0.03%) from 85.785%

Build # 26fb17fe-5a2d-422c-9747-c0adef10e362

Build Type

push

circleci

Committed by

ctueck

Commit Message

add platforms to Meilisearch

Run Details

23 of 24 new or added lines in 5 files covered. (95.83%)

17 existing lines in 3 files now uncovered.

9725 of 11341 relevant lines covered (85.75%)

0.86 hits per line

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

93.41

/submissionapi/csv_functions/csv_handler.py

import csv

import itertools
import re

from submissionapi.csv_functions.csv_insensitive_dict_reader import DictReaderInsensitive


class CSVHandler:
    """
        Class to handle CSV upload, transform it to a submission request object
    """
    FIELDS = {
        'reports': [
            r'agency',
            r'contributing_agencies\[\d+\]',
            r'report_id',
            r'local_identifier',
            r'status',
            r'decision',
            r'summary',
            r'valid_from',
            r'valid_to',
            r'date_format',
            r'other_comment'
        ],
        'activities': [
            r'activities\[\d+\]\.id',
            r'activities\[\d+\]\.local_identifier',
            r'activities\[\d+\]\.agency',
            r'activities\[\d+\]\.group',
        ],
        'report_links': [
            r'link\[\d+\]',
            r'link_display_name\[\d+\]'
        ],
        'report_files': [
            r'file\[\d+\]\.original_location',
            r'file\[\d+\]\.display_name',
        ],
        'report_files__report_language': [
            r'file\[\d+\]\.report_language\[\d+\]',
        ],
        'institutions': [
            r'institution\[\d+\]\.deqar_id',
            r'institution\[\d+\]\.eter_id',
            r'institution\[\d+\]\.identifier',
            r'institution\[\d+\]\.resource'
        ],
        'platforms': [
            r'platform\[\d+\]\.deqar_id',
            r'platform\[\d+\]\.eter_id',
            r'platform\[\d+\]\.identifier',
            r'platform\[\d+\]\.resource'
        ],
        'platforms': [
            r'platform\[\d+\]\.deqar_id',
            r'platform\[\d+\]\.eter_id',
            r'platform\[\d+\]\.name_official',
            r'platform\[\d+\]\.name_official_transliterated',
            r'platform\[\d+\]\.name_english',
            r'platform\[\d+\]\.acronym',
            r'platform\[\d+\]\.website_link'
        ],
        'platforms__identifiers': [
            r'platform\[\d+\]\.identifier\[\d+\]',
            r'platform\[\d+\]\.resource\[\d+\]',
        ],
        'platforms__alternative_names': [
            r'platform\[\d+\]\.name_alternative\[\d+\]',
            r'platform\[\d+\]\.name_alternative_transliterated\[\d+\]',
        ],
        'platforms__locations': [
            r'platform\[\d+\]\.country\[\d+\]',
            r'platform\[\d+\]\.city\[\d+\]',
            r'platform\[\d+\]\.latitude\[\d+\]',
            r'platform\[\d+\]\.longitude\[\d+\]',
        ],
        'platforms__qf_ehea_levels': [
            r'platform\[\d+\]\.qf_ehea_level\[\d+\]',
        ],
        'programmes': [
            r'programme\[\d+\]\.name_primary',
            r'programme\[\d+\]\.qualification_primary',
            r'programme\[\d+\]\.nqf_level',
            r'programme\[\d+\]\.qf_ehea_level',
            r'programme\[\d+\]\.degree_outcome',
            r'programme\[\d+\]\.workload_ects',
            r'programme\[\d+\]\.learning_outcome_description',
            r'programme\[\d+\]\.field_study',
            r'programme\[\d+\]\.assessment_certification',
        ],
        'programmes__identifiers': [
            r'programme\[\d+\]\.identifier\[\d+\]',
            r'programme\[\d+\]\.resource\[\d+\]',
        ],
        'programmes__alternative_names': [
            r'programme\[\d+\]\.name_alternative\[\d+\]',
            r'programme\[\d+\]\.qualification_alternative\[\d+\]',
        ],
        'programmes__countries': [
            r'programme\[\d+\]\.country\[\d+\]',
        ],
        'programmes__learning_outcomes': [
            r'programme\[\d+\]\.learning_outcome\[\d+\]',
        ]
    }

    def __init__(self, csvfile):
        self.csvfile = csvfile
        self.submission_data = []
        self.report_record = {}
        self.error = False
        self.error_message = ""
        self.dialect = None
        self.reader = None

    def handle(self):
        if self._csv_is_valid():
            self._read_csv()
            for row in self.reader:
                self._create_report(row)
                self._create_activities(row)
                self._create_report_links(row)
                self._create_report_files(row)
                self._create_institutions(row)
                self._create_platforms(row)
                self._create_programmes(row)
                self._create_programmes_alternative_names(row)
                self._create_programmes_identifiers(row)
                self._create_programmes_countries(row)
                self._create_learning_outcomes(row)
                self._clear_submission_data()
        else:
            self.error = True
            self.error_message = 'The CSV file appears to be invalid.'

    def _csv_is_valid(self):
        try:
            self.csvfile.seek(0)
            self.dialect = csv.Sniffer().sniff(self.csvfile.read(), delimiters=['\t', ',', ';'])
            return True
        except csv.Error:
            return False

    def _read_csv(self):
        self.csvfile.seek(0)
        self.reader = DictReaderInsensitive(self.csvfile)

    def _create_report(self, row):
        csv_fields = self.reader.fieldnames
        for field in self.FIELDS['reports']:
            r = re.compile(field)
            rematch = sorted(list(filter(r.match, csv_fields)), key=str.lower)

            if len(rematch) > 0:
                if 'contributing_agencies' in field:
                    self.report_record['contributing_agencies'] = []
                    for column in rematch:
                        self.report_record['contributing_agencies'].append(row[column])
                else:
                    self.report_record[rematch[0]] = row[rematch[0]]

    def _create_activities(self, row):
        self._create_first_level_placeholder(['activities'])
        self._create_first_level_values('activities', row, dotted=True)

    def _create_institutions(self, row):
        self._create_first_level_placeholder(['institutions'])
        self._create_first_level_values('institutions', row, dotted=True)

    def _create_platforms(self, row):
        self._create_first_level_placeholder(['platforms'])
        self._create_first_level_values('platforms', row, dotted=True)

    def _create_platforms(self, row):
        self._create_first_level_placeholder(['platforms',
                                              'platforms__identifiers',
                                              'platforms__alternative_names',
                                              'platforms__locations',
                                              'platforms__qf_ehea_levels'])
        self._create_first_level_values('platforms', row, dotted=True)

    def _create_platforms_identifiers(self, row):
        self._create_second_level_placeholder('platforms__identifiers', dictkey=True)
        self._create_second_level_values('platforms__identifiers', row, dictkey=True)

    def _create_platforms_alternative_names(self, row):
        self._create_second_level_placeholder('platforms__alternative_names', dictkey=True)
        self._create_second_level_values('platforms__alternative_names', row, dictkey=True)

    def _create_platforms_locations(self, row):
        self._create_second_level_placeholder('platforms__locations', dictkey=True)
        self._create_second_level_values('platforms__locations', row, dictkey=True)

    def _create_platforms_qf_ehea_levels(self, row):
        self._create_second_level_placeholder('platforms__qf_ehea_levels')
        self._create_second_level_values('platforms__qf_ehea_levels', row)

    def _create_programmes(self, row):
        self._create_first_level_placeholder(['programmes',
                                              'programmes__identifiers',
                                              'programmes__alternative_names',
                                              'programmes__countries',
                                              'programmes__learning_outcomes'])
        self._create_first_level_values('programmes', row, dotted=True)

    def _create_programmes_identifiers(self, row):
        self._create_second_level_placeholder('programmes__identifiers', dictkey=True)
        self._create_second_level_values('programmes__identifiers', row, dictkey=True)

    def _create_programmes_alternative_names(self, row):
        self._create_second_level_placeholder('programmes__alternative_names', dictkey=True)
        self._create_second_level_values('programmes__alternative_names', row, dictkey=True)

    def _create_programmes_countries(self, row):
        self._create_second_level_placeholder('programmes__countries')
        self._create_second_level_values('programmes__countries', row)

    def _create_learning_outcomes(self, row):
        self._create_second_level_placeholder('programmes__learning_outcomes')
        self._create_second_level_values('programmes__learning_outcomes', row)

    def _create_report_links(self, row):
        self._create_first_level_placeholder(['report_links'])
        self._create_first_level_values('report_links', row)

    def _create_report_files(self, row):
        self._create_first_level_placeholder(['report_files', 'report_files__report_language'])
        self._create_first_level_values('report_files', row, dotted=True)

        self._create_second_level_placeholder('report_files__report_language')
        self._create_second_level_values('report_files__report_language', row)

    def _create_first_level_placeholder(self, field_key_array):
        fields = []
        wrapper = field_key_array[0].split('__')[0]
        csv_fields = self.reader.fieldnames

        for fk in field_key_array:
            fields += self.FIELDS[fk]

        # Create wrapper
        self.report_record[wrapper] = []

        for field in fields:
            r = re.compile(field)

            max_index = 0
            for csv_field in csv_fields:
                match = r.match(csv_field)
                if match:
                    groups = re.search(r"\d+", csv_field)
                    index = int(groups.group(0))
                    if index > max_index:
                        max_index = index

            # Create plaholder if it doesn't exists yet
            if max_index > 0:
                for i in range(0, max_index):
                    if len(self.report_record[wrapper]) < i+1:
                        self.report_record[wrapper].append({})

    def _create_first_level_values(self, wrapper, row, dotted=False):
        csv_fields = self.reader.fieldnames
        for field in self.FIELDS[wrapper]:
            r = re.compile(field)
            rematch = sorted(list(filter(r.match, csv_fields)), key=str.lower)

            if len(rematch) > 0:
                for fld in rematch:
                    if row[fld] != '-':
                        index = re.search(r"\[\d+\]", fld).group()
                        field = fld.replace(index, "")
                        index = int(re.search(r"\d+", index).group())-1
                        if dotted:
                            field = field.split('.')[1]
                        self.report_record[wrapper][index][field] = row[fld]

    def _create_second_level_placeholder(self, field_key, dictkey=None):
        csv_fields = self.reader.fieldnames
        first_level_wrapper_name, wrapper = field_key.split('__')

        # Create second level wrapper
        first_level_wrapper = self.report_record[first_level_wrapper_name]

        for first_level_wrapper_item in first_level_wrapper:
            first_level_wrapper_item[wrapper] = []

            if dictkey:
                for field in self.FIELDS[field_key]:
                    r = re.compile(field)

                    max_index = 0
                    for csv_field in csv_fields:
                        match = r.match(csv_field)
                        if match:
                            groups = re.search(r"\d+", csv_field)
                            index = int(groups.group(0))
                            if index > max_index:
                                max_index = index

                    # Create plaholder if it doesn't exists yet
                    if max_index > 0:
                        for i in range(0, max_index):
                            if len(first_level_wrapper_item[wrapper]) < i+1:
                                first_level_wrapper_item[wrapper].append({})

    def _create_second_level_values(self, field_key, row, dictkey=None):
        csv_fields = self.reader.fieldnames
        first_level_wrapper_name, wrapper = field_key.split('__')
        first_level_wrapper = self.report_record[first_level_wrapper_name]

        for field in self.FIELDS[field_key]:
            r = re.compile(field)
            rematch = sorted(list(filter(r.match, csv_fields)), key=str.lower)

            if len(rematch) > 0:
                for fld in rematch:
                    if row[fld] != '-':
                        [field01, field02] = fld.split('.')
                        index01 = int(re.search(r"\d+", field01).group())

                        index02 = re.search(r"\[\d+\]", field02).group()
                        field02 = field02.replace(index02, "")
                        index02 = int(re.search(r"\d+", index02).group())
                        if dictkey:
                            first_level_wrapper[index01-1][wrapper][index02-1][field02] = row[fld]
                        else:
                            first_level_wrapper[index01-1][wrapper].append(row[fld])

    def _clear_submission_data(self):
        self.submission_data.append(self.clean_empty(self.report_record))

    def clean_empty(self, d):
        if not isinstance(d, (dict, list)):
            return d
        if isinstance(d, list):
            return [v for v in (self.clean_empty(v) for v in d) if v]
        return {k: v for k, v in ((k, self.clean_empty(v)) for k, v in d.items()) if v}

1	import csv	1✔
2
3	import itertools	1✔
4	import re	1✔
5
6	from submissionapi.csv_functions.csv_insensitive_dict_reader import DictReaderInsensitive	1✔
7
8
9	class CSVHandler:	1✔
10	"""
11	Class to handle CSV upload, transform it to a submission request object
12	"""
13	FIELDS = {	1✔
14	'reports': [
15	r'agency',
16	r'contributing_agencies\[\d+\]',
17	r'report_id',
18	r'local_identifier',
19	r'status',
20	r'decision',
21	r'summary',
22	r'valid_from',
23	r'valid_to',
24	r'date_format',
25	r'other_comment'
26	],
27	'activities': [
28	r'activities\[\d+\]\.id',
29	r'activities\[\d+\]\.local_identifier',
30	r'activities\[\d+\]\.agency',
31	r'activities\[\d+\]\.group',
32	],
33	'report_links': [
34	r'link\[\d+\]',
35	r'link_display_name\[\d+\]'
36	],
37	'report_files': [
38	r'file\[\d+\]\.original_location',
39	r'file\[\d+\]\.display_name',
40	],
41	'report_files__report_language': [
42	r'file\[\d+\]\.report_language\[\d+\]',
43	],
44	'institutions': [
45	r'institution\[\d+\]\.deqar_id',
46	r'institution\[\d+\]\.eter_id',
47	r'institution\[\d+\]\.identifier',
48	r'institution\[\d+\]\.resource'
49	],
50	'platforms': [
51	r'platform\[\d+\]\.deqar_id',
52	r'platform\[\d+\]\.eter_id',
53	r'platform\[\d+\]\.identifier',
54	r'platform\[\d+\]\.resource'
55	],
56	'platforms': [
57	r'platform\[\d+\]\.deqar_id',
58	r'platform\[\d+\]\.eter_id',
59	r'platform\[\d+\]\.name_official',
60	r'platform\[\d+\]\.name_official_transliterated',
61	r'platform\[\d+\]\.name_english',
62	r'platform\[\d+\]\.acronym',
63	r'platform\[\d+\]\.website_link'
64	],
65	'platforms__identifiers': [
66	r'platform\[\d+\]\.identifier\[\d+\]',
67	r'platform\[\d+\]\.resource\[\d+\]',
68	],
69	'platforms__alternative_names': [
70	r'platform\[\d+\]\.name_alternative\[\d+\]',
71	r'platform\[\d+\]\.name_alternative_transliterated\[\d+\]',
72	],
73	'platforms__locations': [
74	r'platform\[\d+\]\.country\[\d+\]',
75	r'platform\[\d+\]\.city\[\d+\]',
76	r'platform\[\d+\]\.latitude\[\d+\]',
77	r'platform\[\d+\]\.longitude\[\d+\]',
78	],
79	'platforms__qf_ehea_levels': [
80	r'platform\[\d+\]\.qf_ehea_level\[\d+\]',
81	],
82	'programmes': [
83	r'programme\[\d+\]\.name_primary',
84	r'programme\[\d+\]\.qualification_primary',
85	r'programme\[\d+\]\.nqf_level',
86	r'programme\[\d+\]\.qf_ehea_level',
87	r'programme\[\d+\]\.degree_outcome',
88	r'programme\[\d+\]\.workload_ects',
89	r'programme\[\d+\]\.learning_outcome_description',
90	r'programme\[\d+\]\.field_study',
91	r'programme\[\d+\]\.assessment_certification',
92	],
93	'programmes__identifiers': [
94	r'programme\[\d+\]\.identifier\[\d+\]',
95	r'programme\[\d+\]\.resource\[\d+\]',
96	],
97	'programmes__alternative_names': [
98	r'programme\[\d+\]\.name_alternative\[\d+\]',
99	r'programme\[\d+\]\.qualification_alternative\[\d+\]',
100	],
101	'programmes__countries': [
102	r'programme\[\d+\]\.country\[\d+\]',
103	],
104	'programmes__learning_outcomes': [
105	r'programme\[\d+\]\.learning_outcome\[\d+\]',
106	]
107	}
108
109	def __init__(self, csvfile):	1✔
110	self.csvfile = csvfile	1✔
111	self.submission_data = []	1✔
112	self.report_record = {}	1✔
113	self.error = False	1✔
114	self.error_message = ""	1✔
115	self.dialect = None	1✔
116	self.reader = None	1✔
117
118	def handle(self):	1✔
119	if self._csv_is_valid():	1✔
120	self._read_csv()	1✔
121	for row in self.reader:	1✔
122	self._create_report(row)	1✔
123	self._create_activities(row)	1✔
124	self._create_report_links(row)	1✔
125	self._create_report_files(row)	1✔
126	self._create_institutions(row)	1✔
127	self._create_platforms(row)	1✔
128	self._create_programmes(row)	1✔
129	self._create_programmes_alternative_names(row)	1✔
130	self._create_programmes_identifiers(row)	1✔
131	self._create_programmes_countries(row)	1✔
132	self._create_learning_outcomes(row)	1✔
133	self._clear_submission_data()	1✔
134	else:
UNCOV 135	self.error = True	×
UNCOV 136	self.error_message = 'The CSV file appears to be invalid.'	×
137
138	def _csv_is_valid(self):	1✔
139	try:	1✔
140	self.csvfile.seek(0)	1✔
141	self.dialect = csv.Sniffer().sniff(self.csvfile.read(), delimiters=['\t', ',', ';'])	1✔
142	return True	1✔
143	except csv.Error:	1✔
144	return False	1✔
145
146	def _read_csv(self):	1✔
147	self.csvfile.seek(0)	1✔
148	self.reader = DictReaderInsensitive(self.csvfile)	1✔
149
150	def _create_report(self, row):	1✔
151	csv_fields = self.reader.fieldnames	1✔
152	for field in self.FIELDS['reports']:	1✔
153	r = re.compile(field)	1✔
154	rematch = sorted(list(filter(r.match, csv_fields)), key=str.lower)	1✔
155
156	if len(rematch) > 0:	1✔
157	if 'contributing_agencies' in field:	1✔
158	self.report_record['contributing_agencies'] = []	1✔
159	for column in rematch:	1✔
160	self.report_record['contributing_agencies'].append(row[column])	1✔
161	else:
162	self.report_record[rematch[0]] = row[rematch[0]]	1✔
163
164	def _create_activities(self, row):	1✔
165	self._create_first_level_placeholder(['activities'])	1✔
166	self._create_first_level_values('activities', row, dotted=True)	1✔
167
168	def _create_institutions(self, row):	1✔
169	self._create_first_level_placeholder(['institutions'])	1✔
170	self._create_first_level_values('institutions', row, dotted=True)	1✔
171
172	def _create_platforms(self, row):	1✔
UNCOV 173	self._create_first_level_placeholder(['platforms'])	×
UNCOV 174	self._create_first_level_values('platforms', row, dotted=True)	×
175
176	def _create_platforms(self, row):	1✔
177	self._create_first_level_placeholder(['platforms',	1✔
178	'platforms__identifiers',
179	'platforms__alternative_names',
180	'platforms__locations',
181	'platforms__qf_ehea_levels'])
182	self._create_first_level_values('platforms', row, dotted=True)	1✔
183
184	def _create_platforms_identifiers(self, row):	1✔
UNCOV 185	self._create_second_level_placeholder('platforms__identifiers', dictkey=True)	×
UNCOV 186	self._create_second_level_values('platforms__identifiers', row, dictkey=True)	×
187
188	def _create_platforms_alternative_names(self, row):	1✔
UNCOV 189	self._create_second_level_placeholder('platforms__alternative_names', dictkey=True)	×
UNCOV 190	self._create_second_level_values('platforms__alternative_names', row, dictkey=True)	×
191
192	def _create_platforms_locations(self, row):	1✔
UNCOV 193	self._create_second_level_placeholder('platforms__locations', dictkey=True)	×
UNCOV 194	self._create_second_level_values('platforms__locations', row, dictkey=True)	×
195
196	def _create_platforms_qf_ehea_levels(self, row):	1✔
UNCOV 197	self._create_second_level_placeholder('platforms__qf_ehea_levels')	×
UNCOV 198	self._create_second_level_values('platforms__qf_ehea_levels', row)	×
199
200	def _create_programmes(self, row):	1✔
201	self._create_first_level_placeholder(['programmes',	1✔
202	'programmes__identifiers',
203	'programmes__alternative_names',
204	'programmes__countries',
205	'programmes__learning_outcomes'])
206	self._create_first_level_values('programmes', row, dotted=True)	1✔
207
208	def _create_programmes_identifiers(self, row):	1✔
209	self._create_second_level_placeholder('programmes__identifiers', dictkey=True)	1✔
210	self._create_second_level_values('programmes__identifiers', row, dictkey=True)	1✔
211
212	def _create_programmes_alternative_names(self, row):	1✔
213	self._create_second_level_placeholder('programmes__alternative_names', dictkey=True)	1✔
214	self._create_second_level_values('programmes__alternative_names', row, dictkey=True)	1✔
215
216	def _create_programmes_countries(self, row):	1✔
217	self._create_second_level_placeholder('programmes__countries')	1✔
218	self._create_second_level_values('programmes__countries', row)	1✔
219
220	def _create_learning_outcomes(self, row):	1✔
221	self._create_second_level_placeholder('programmes__learning_outcomes')	1✔
222	self._create_second_level_values('programmes__learning_outcomes', row)	1✔
223
224	def _create_report_links(self, row):	1✔
225	self._create_first_level_placeholder(['report_links'])	1✔
226	self._create_first_level_values('report_links', row)	1✔
227
228	def _create_report_files(self, row):	1✔
229	self._create_first_level_placeholder(['report_files', 'report_files__report_language'])	1✔
230	self._create_first_level_values('report_files', row, dotted=True)	1✔
231
232	self._create_second_level_placeholder('report_files__report_language')	1✔
233	self._create_second_level_values('report_files__report_language', row)	1✔
234
235	def _create_first_level_placeholder(self, field_key_array):	1✔
236	fields = []	1✔
237	wrapper = field_key_array[0].split('__')[0]	1✔
238	csv_fields = self.reader.fieldnames	1✔
239
240	for fk in field_key_array:	1✔
241	fields += self.FIELDS[fk]	1✔
242
243	# Create wrapper
244	self.report_record[wrapper] = []	1✔
245
246	for field in fields:	1✔
247	r = re.compile(field)	1✔
248
249	max_index = 0	1✔
250	for csv_field in csv_fields:	1✔
251	match = r.match(csv_field)	1✔
252	if match:	1✔
253	groups = re.search(r"\d+", csv_field)	1✔
254	index = int(groups.group(0))	1✔
255	if index > max_index:	1✔
256	max_index = index	1✔
257
258	# Create plaholder if it doesn't exists yet
259	if max_index > 0:	1✔
260	for i in range(0, max_index):	1✔
261	if len(self.report_record[wrapper]) < i+1:	1✔
262	self.report_record[wrapper].append({})	1✔
263
264	def _create_first_level_values(self, wrapper, row, dotted=False):	1✔
265	csv_fields = self.reader.fieldnames	1✔
266	for field in self.FIELDS[wrapper]:	1✔
267	r = re.compile(field)	1✔
268	rematch = sorted(list(filter(r.match, csv_fields)), key=str.lower)	1✔
269
270	if len(rematch) > 0:	1✔
271	for fld in rematch:	1✔
272	if row[fld] != '-':	1✔
273	index = re.search(r"\[\d+\]", fld).group()	1✔
274	field = fld.replace(index, "")	1✔
275	index = int(re.search(r"\d+", index).group())-1	1✔
276	if dotted:	1✔
277	field = field.split('.')[1]	1✔
278	self.report_record[wrapper][index][field] = row[fld]	1✔
279
280	def _create_second_level_placeholder(self, field_key, dictkey=None):	1✔
281	csv_fields = self.reader.fieldnames	1✔
282	first_level_wrapper_name, wrapper = field_key.split('__')	1✔
283
284	# Create second level wrapper
285	first_level_wrapper = self.report_record[first_level_wrapper_name]	1✔
286
287	for first_level_wrapper_item in first_level_wrapper:	1✔
288	first_level_wrapper_item[wrapper] = []	1✔
289
290	if dictkey:	1✔
291	for field in self.FIELDS[field_key]:	1✔
292	r = re.compile(field)	1✔
293
294	max_index = 0	1✔
295	for csv_field in csv_fields:	1✔
296	match = r.match(csv_field)	1✔
297	if match:	1✔
298	groups = re.search(r"\d+", csv_field)	1✔
299	index = int(groups.group(0))	1✔
300	if index > max_index:	1✔
301	max_index = index	1✔
302
303	# Create plaholder if it doesn't exists yet
304	if max_index > 0:	1✔
305	for i in range(0, max_index):	1✔
306	if len(first_level_wrapper_item[wrapper]) < i+1:	1✔
307	first_level_wrapper_item[wrapper].append({})	1✔
308
309	def _create_second_level_values(self, field_key, row, dictkey=None):	1✔
310	csv_fields = self.reader.fieldnames	1✔
311	first_level_wrapper_name, wrapper = field_key.split('__')	1✔
312	first_level_wrapper = self.report_record[first_level_wrapper_name]	1✔
313
314	for field in self.FIELDS[field_key]:	1✔
315	r = re.compile(field)	1✔
316	rematch = sorted(list(filter(r.match, csv_fields)), key=str.lower)	1✔
317
318	if len(rematch) > 0:	1✔
319	for fld in rematch:	1✔
320	if row[fld] != '-':	1✔
321	[field01, field02] = fld.split('.')	1✔
322	index01 = int(re.search(r"\d+", field01).group())	1✔
323
324	index02 = re.search(r"\[\d+\]", field02).group()	1✔
325	field02 = field02.replace(index02, "")	1✔
326	index02 = int(re.search(r"\d+", index02).group())	1✔
327	if dictkey:	1✔
328	first_level_wrapper[index01-1][wrapper][index02-1][field02] = row[fld]	1✔
329	else:
330	first_level_wrapper[index01-1][wrapper].append(row[fld])	1✔
331
332	def _clear_submission_data(self):	1✔
333	self.submission_data.append(self.clean_empty(self.report_record))	1✔
334
335	def clean_empty(self, d):	1✔
336	if not isinstance(d, (dict, list)):	1✔
337	return d	1✔
338	if isinstance(d, list):	1✔
339	return [v for v in (self.clean_empty(v) for v in d) if v]	1✔
340	return {k: v for k, v in ((k, self.clean_empty(v)) for k, v in d.items()) if v}	1✔

EQAR / eqar_backend / 26fb17fe-5a2d-422c-9747-c0adef10e362

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