LocationElementSchemaNode.validator()   F
last analyzed

Complexity

Conditions 10

Size

Total Lines 30

Duplication

Lines 0
Ratio 0 %

Importance

Changes 3
Bugs 0 Features 0
Metric Value
cc 10
c 3
b 0
f 0
dl 0
loc 30
rs 3.1304

How to fix   Complexity   

Complexity

Complex classes like LocationElementSchemaNode.validator() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

1
# -*- coding: utf-8 -*-
2
"""
3
This module validates an location elements.
4
For an address in Belgium the validation uses the CRAB principles.
5
"""
6
import logging
7
8
import re
9
import colander
10
import pycountry
11
from colander import null
12
from crabpy.gateway.exception import GatewayRuntimeException, GatewayResourceNotFoundException
13
from oe_geoutils.validation.validators_address import process_huisnummer, process_subadres, process_straat
14
from oe_geoutils.validation.validators_cadaster_parcels import CadasterSchemaNode
15
import copy
16
17
log = logging.getLogger(__name__)
18
19
20
class ProvincieSchemaNode(colander.MappingSchema):
21
    naam = colander.SchemaNode(
22
        colander.String(),
23
        validator=colander.Length(1, 50),
24
        missing=None
25
    )
26
    niscode = colander.SchemaNode(
27
        colander.Integer(),
28
        missing=None
29
    )
30
31
32
class GemeenteSchemaNode(colander.MappingSchema):
33
    naam = colander.SchemaNode(
34
        colander.String(),
35
        validator=colander.Length(1, 255),
36
        missing=None
37
    )
38
    id = colander.SchemaNode(
39
        colander.Integer(),
40
        missing=None
41
    )
42
    niscode = colander.SchemaNode(
43
        colander.Integer(),
44
        missing=None
45
    )
46
47
48
class DeelGemeenteSchemaNode(colander.MappingSchema):
49
    naam = colander.SchemaNode(
50
        colander.String(),
51
        validator=colander.Length(1, 255),
52
        missing=None
53
    )
54
55
    niscode = colander.SchemaNode(
56
        colander.String(),
57
        validator=colander.Length(1, 10),
58
        missing=None
59
    )
60
61
62
def _get_hoofd_niscode(deelgemeente_niscode):
63
    """
64
    :param string niscode van deelgemeente e.g.("24062A")
65
    :return: int niscode fusiegemeente e.g.(24062)
66
    """
67
    if re.match('^[0-9]{5}[a-zA-Z]{1,1}$', deelgemeente_niscode):
68
        return int(deelgemeente_niscode[:-1])
69
    elif re.match('^[0-9]{5}', deelgemeente_niscode):
70
        return int(deelgemeente_niscode)
71
    else:
72
        raise Exception('invalid deelgemeente niscode')
73
74
75
class LocationElementSchemaNode(colander.MappingSchema):
76
    id = colander.SchemaNode(
77
        colander.Integer(),
78
        missing=None
79
    )
80
81
    type = colander.SchemaNode(
82
        colander.String(),
83
        validator=colander.Length(1, 250)
84
    )
85
86
    provincie = ProvincieSchemaNode(
87
        missing={}
88
    )
89
90
    gemeente = GemeenteSchemaNode(
91
        missing={}
92
    )
93
94
    deelgemeente = DeelGemeenteSchemaNode(
95
        missing={}
96
    )
97
98
    straat = colander.SchemaNode(
99
        colander.String(),
100
        validator=colander.Length(1, 100),
101
        missing=None
102
    )
103
104
    straat_id = colander.SchemaNode(
105
        colander.Integer(),
106
        missing=None
107
    )
108
109
    huisnummer = colander.SchemaNode(
110
        colander.String(),
111
        validator=colander.Length(1, 20),
112
        missing=None
113
    )
114
115
    huisnummer_id = colander.SchemaNode(
116
        colander.Integer(),
117
        missing=None
118
    )
119
120
    subadres = colander.SchemaNode(
121
        colander.String(),
122
        validator=colander.Length(1, 20),
123
        missing=None
124
    )
125
126
    subadres_id = colander.SchemaNode(
127
        colander.Integer(),
128
        missing=None
129
    )
130
131
    postcode = colander.SchemaNode(
132
        colander.String(),
133
        validator=colander.Length(1, 20),
134
        missing=None
135
    )
136
137
    land = colander.SchemaNode(
138
        colander.String(),
139
        validator=colander.Length(1, 100),
140
        missing='BE'
141
    )
142
143
    perceel = CadasterSchemaNode(
144
        missing=None
145
    )
146
147
    omschrijving = colander.SchemaNode(
148
        colander.String(),
149
        validator=colander.Length(1, 250),
150
        missing=None
151
    )
152
153
    def preparer(self, input_locatie_element):
154
        locatie_element = copy.deepcopy(input_locatie_element)
155
        if locatie_element is None or not locatie_element:
156
            return null  # pragma: no cover
157
        request = self.bindings['request']
158
        crab_gateway = request.crab_gateway()
159
        capakey_gateway = request.capakey_gateway()
160
        locatie_element_type = ''
161
        if 'type' in locatie_element:
162
            locatie_element_type = locatie_element.get('type')
163
        if 'land' in locatie_element and locatie_element.get('land').upper() == 'BE':
164
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementAdres':
165
                locatie_element = self._prepare_locatie_adres(crab_gateway, locatie_element)
166
            elif locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementPerceel':
167
                locatie_element = self._prepare_locatie_perceel(capakey_gateway, crab_gateway, locatie_element)
168
            elif locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementOpenbaarDomein':
169
                locatie_element = self._prepare_locatie(crab_gateway, locatie_element)
170
            elif locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElement':
171
                locatie_element = self._prepare_locatie(crab_gateway, locatie_element)
172
        else:
173
            locatie_element['gemeente']['id'] = None
174
            locatie_element['straat_id'] = None
175
            locatie_element['huisnummer_id'] = None
176
            locatie_element['subadres_id'] = None
177
        if 'land' in locatie_element:
178
            locatie_element['land'] = locatie_element.get('land').upper()
179
        return locatie_element
180
181
    def validator(self, node, locatie_element):
182
        request = self.bindings['request']
183
        crab_gateway = request.crab_gateway()
184
        capakey_gateway = request.capakey_gateway()
185
        locatie_element_type = ''
186
        land = None
187
        if 'type' in locatie_element:
188
            locatie_element_type = locatie_element.get('type')
189
        if 'land' in locatie_element:
190
            land = locatie_element.get('land')
191
            try:
192
                try:
193
                    pycountry.countries.get(alpha2=land)
194
                except KeyError:
195
                    pycountry.countries.get(alpha_2=land)
196
            except KeyError:
197
                raise colander.Invalid(
198
                    node,
199
                    'ongeldige landcode %s, dit is geen ISO 3166 code' %
200
                    land
201
                )
202
        if land == 'BE':
203
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementAdres':
204
                self._validate_locatie_adres(crab_gateway, locatie_element, node)
205
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementPerceel':
206
                self._validate_locatie_perceel(capakey_gateway, locatie_element, node)
207
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementOpenbaarDomein':
208
                self._validate_locatie_openbaar_domein(locatie_element, node)
209
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElement':
210
                self._validate_locatie(locatie_element, node)
211
212
    @staticmethod
213
    def _prepare_locatie_adres(crab_gateway, locatie_element):
214
        # Gemeente
215
        if locatie_element.get('gemeente', {}).get('id', None) is None:
216
            if locatie_element.get('gemeente', {}).get('naam', None) is None:
217
                return locatie_element
218
219
            gemeente = locatie_element.get('gemeente', {}).get('naam')
220
            gewest_ids = [2, 1, 3]
221
            for gewest_id in gewest_ids:
222
                gemeenten = crab_gateway.list_gemeenten(gewest_id)
223
                gemeente_val = next((g for g in gemeenten if g.naam.lower() == gemeente.lower()), None)
224
                if gemeente_val:
225
                    locatie_element['gemeente']['id'] = gemeente_val.id
226
                    break
227
            if locatie_element.get('gemeente', {}).get('id', None) is None:
228
                return locatie_element
229
230
        # gemeente id gekend
231
        gemeente_id = locatie_element.get('gemeente', {}).get('id', None)
232
        straat_id = locatie_element.get('straat_id', None)
233
        straat = locatie_element.get('straat', None)
234
        huisnummer_id = locatie_element.get('huisnummer_id', None)
235
        huisnummer = locatie_element.get('huisnummer', None)
236
        subadres_id = locatie_element.get('subadres_id', None)
237
        subadres = locatie_element.get('subadres', None)
238
        try:
239
            gemeente = crab_gateway.get_gemeente_by_id(gemeente_id)
240
        except (GatewayRuntimeException, AttributeError):
241
            locatie_element['gemeente']['naam'] = None
242
            return locatie_element
243
244
        if gemeente:
245
            locatie_element['gemeente']['naam'] = "" + gemeente.naam
246
            locatie_element['gemeente']['niscode'] = gemeente.niscode
247
            locatie_element['provincie']['niscode'] = gemeente.provincie.niscode
248
            locatie_element['provincie']['naam'] = gemeente.provincie.naam
249
            if straat_id:
250
                straat_val = next((s for s in gemeente.straten if s.id == straat_id), None)
251
                if straat_val:
252
                    locatie_element['straat'] = "" + straat_val.label
253
                    num_val = process_huisnummer(locatie_element, huisnummer_id, huisnummer, straat_val)
254
                    if num_val:
255
                        process_subadres(locatie_element, subadres_id, subadres, num_val)
256
            if not straat_id and straat:
257
                straat_val = process_straat(locatie_element, straat, gemeente, crab_gateway)
258
                if straat_val:
259
                    locatie_element['straat_id'] = straat_val.id
260
                    num_val = process_huisnummer(locatie_element, huisnummer_id, huisnummer, straat_val)
261
                    if num_val:
262
                        process_subadres(locatie_element, subadres_id, subadres, num_val)
263
        _set_deelgemeente(locatie_element, gemeente, crab_gateway)
264
        return locatie_element
265
266
    @staticmethod
267
    def _prepare_locatie_perceel(capakey_gateway, crab_gateway, locatie_element):
268
        try:
269
            kadastrale_afdelingen = capakey_gateway.list_kadastrale_afdelingen()
270
            capakey = locatie_element['perceel']['capakey']
271
            afdeling_niscode = int(capakey[0:5])
272
            afdeling = next((afd for afd in kadastrale_afdelingen if afd.id == afdeling_niscode), None)
273
            if afdeling:
274
                locatie_element['perceel']['sectie'] = capakey[5:6]
275
                locatie_element['perceel']['perceel'] = capakey[6:]
276
                locatie_element['perceel']['afdeling'] = afdeling.naam
277
                gemeente = crab_gateway.get_gemeente_by_niscode(afdeling.gemeente.id)
278
                locatie_element['gemeente']['id'] = gemeente.id
279
                locatie_element['gemeente']['naam'] = gemeente.naam
280
                locatie_element['gemeente']['niscode'] = afdeling.gemeente.id
281
                _set_deelgemeente(locatie_element, gemeente, crab_gateway)
282
                locatie_element['provincie']['niscode'] = gemeente.provincie.niscode
283
                locatie_element['provincie']['naam'] = gemeente.provincie.naam
284
        except (GatewayRuntimeException, AttributeError, GatewayResourceNotFoundException) as e:
285
            log.warning("Failed to prepare the locatie perceel. Skipping...")
286
        return locatie_element
287
288
    @staticmethod
289
    def _prepare_locatie(crab_gateway, locatie_element):
290
        gemeente = None
291
        gemeente_id = locatie_element.get('gemeente', {}).get('id', None)
292
        gemeente_niscode = locatie_element.get('gemeente', {}).get('niscode', None)
293
        gemeente_naam = locatie_element.get('gemeente', {}).get('naam', None)
294
        if gemeente_id:
295
            try:
296
                gemeente = crab_gateway.get_gemeente_by_id(gemeente_id)
297
            except (GatewayRuntimeException, AttributeError):
298
                gemeente = None
299
        if gemeente is None and gemeente_niscode:
300
            try:
301
                gemeente = crab_gateway.get_gemeente_by_niscode(gemeente_niscode)
302
            except (GatewayRuntimeException, AttributeError):
303
                gemeente = None
304
        if gemeente is None and gemeente_naam:
305
            gewest_ids = [2, 1, 3]
306
            for gewest_id in gewest_ids:
307
                try:
308
                    gemeenten = crab_gateway.list_gemeenten(gewest_id)
309
                    gemeente = next((g for g in gemeenten if g.naam.lower() == gemeente_naam.lower()), None)
310
                    if gemeente:
311
                        break
312
                except (GatewayRuntimeException, AttributeError):  # pragma no cover
313
                    gemeente = None
314
        if gemeente:
315
            locatie_element['gemeente']['id'] = gemeente.id
316
            locatie_element['gemeente']['naam'] = gemeente.naam
317
            locatie_element['gemeente']['niscode'] = gemeente.niscode
318
            locatie_element['provincie']['niscode'] = gemeente.provincie.niscode
319
            locatie_element['provincie']['naam'] = gemeente.provincie.naam
320
            _set_deelgemeente(locatie_element, gemeente, crab_gateway)
321
        else:
322
            locatie_element['gemeente']['id'] = None
323
        return locatie_element
324
325
    @staticmethod
326
    def _validate_provincie_gemeente(locatie_element, node):
327
        gemeente = locatie_element.get('gemeente', {}).get('naam', None)
328
        gemeente_id = locatie_element.get('gemeente', {}).get('id', None)
329
        gemeente_niscode = locatie_element.get('gemeente', {}).get('niscode', None)
330
        provincie = locatie_element.get('provincie', {}).get('naam', None)
331
        provincie_niscode = locatie_element.get('provincie', {}).get('niscode', None)
332
        if gemeente_id is None:
333
            raise colander.Invalid(
334
                node,
335
                'geen correcte gemeente_id gevonden voor de gemeente {0}'.format(gemeente)
336
            )
337
        if gemeente is None:  # if gemeente is still None here, the gemeente_id was incorrect
338
            raise colander.Invalid(
339
                node,
340
                'ongeldig gemeente_id {0}'.format(gemeente_id)
341
            )
342
        if gemeente_niscode is None:  # pragma no cover  # normally the first 2 checks will fail in this case
343 View Code Duplication
            raise colander.Invalid(
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
344
                node,
345
                'ongeldige gemeente_niscode {0}'.format(gemeente_niscode)
346
            )
347
        if provincie is None:  # pragma no cover   # normally the first 2 checks will fail in this case
348
            raise colander.Invalid(
349
                node,
350
                'ongeldige provincie {0}'.format(provincie)
351
            )
352
        if provincie_niscode is None:  # pragma no cover   # normally the first 2 checks will fail in this case
353
            raise colander.Invalid(
354
                node,
355
                'ongeldige provincie_niscode {0}'.format(provincie_niscode)
356
            )
357
358
    def _validate_locatie_perceel(self, capakey_gateway, locatie_element, node):
359
        self._validate_provincie_gemeente(locatie_element, node)
360
        kadastrale_afdelingen = capakey_gateway.list_kadastrale_afdelingen()
361
        capakey = locatie_element['perceel']['capakey']
362
        afdeling_niscode = int(capakey[0:5])
363
        afdeling = next((afd for afd in kadastrale_afdelingen if afd.id == afdeling_niscode), None)
364
        if afdeling is None:
365
            raise colander.Invalid(
366
                node,
367
                'ongeldige kadastrale afdeling voor capakey {0}'.format(capakey)
368
            )
369
370
    def _validate_locatie_openbaar_domein(self, locatie_element, node):
371
        self._validate_provincie_gemeente(locatie_element, node)
372
373
    def _validate_locatie(self, locatie_element, node):
374
        self._validate_provincie_gemeente(locatie_element, node)
375
376
    def _validate_locatie_adres(self, crab_gateway, locatie_element, node):
377
        self._validate_provincie_gemeente(locatie_element, node)
378
        gemeente_id = locatie_element.get('gemeente', {}).get('id', None)
379
        straat_id = locatie_element.get('straat_id', None)
380
        huisnummer_id = locatie_element.get('huisnummer_id', None)
381
        postcode = locatie_element.get('postcode', None)
382
        subadres_id = locatie_element.get('subadres_id', None)
383
384
        if locatie_element.get('deelgemeente', False):
385
            # if has deelgemeente naam and no niscode
386
            if locatie_element.get('deelgemeente').get('naam', False) and \
387
                    locatie_element.get('deelgemeente').get('niscode', None) is None:
388
                raise colander.Invalid(node, 'deelgemeente moet een niscode hebben')
389
390
        if locatie_element.get('deelgemeente', {}).get('niscode', False) and \
391
                locatie_element.get('gemeente', {}).get('niscode', False):  # if gemeente and deelgemeente has niscode
392
            if _get_hoofd_niscode(locatie_element.get('deelgemeente', {}).get('niscode')) != \
393
                    locatie_element.get('gemeente', {}).get('niscode'):
394
                raise colander.Invalid(node,
395
                                       'niscode van gemeente en deelgemeente moet hetzelfde zijn '
396
                                       '(uitgezonderd, toegevoegde letters van deelgemeentes)')
397
        if straat_id is not None:
398
            gemeente = crab_gateway.get_gemeente_by_id(gemeente_id)
399
            try:
400
                straat = crab_gateway.get_straat_by_id(straat_id)
401
            except (GatewayRuntimeException, AttributeError):
402
                raise colander.Invalid(
403
                    node,
404
                    'ongeldig straat_id'
405
                )
406
            if straat.gemeente_id != gemeente_id:
407
                raise colander.Invalid(
408
                    node,
409
                    'de straat %s met id %s ligt niet in gemeente %s' %
410
                    (locatie_element.get('straat', ''), straat_id, gemeente.naam)
411
                )
412
            if huisnummer_id is not None:
413
                try:
414
                    huisnummer = crab_gateway.get_huisnummer_by_id(huisnummer_id)
415
                except (GatewayRuntimeException, AttributeError):
416
                    raise colander.Invalid(
417
                        node,
418
                        'ongeldig huisnummer_id'
419
                    )
420
                if huisnummer.straat_id != straat_id:
421
                    raise colander.Invalid(
422
                        node,
423
                        'het huisnummer %s met id %s ligt niet in straat %s' %
424
                        (locatie_element.get('huisnummer', ''), huisnummer_id, straat.label)
425
                    )
426
                if postcode is not None:
427
                    postkanton = crab_gateway.get_postkanton_by_huisnummer(huisnummer_id)
428
                    if postcode != str(postkanton.id):
429
                        raise colander.Invalid(
430
                            node,
431
                            'postcode %s is niet correct voor dit adres, mogelijke postcode is %s' %
432
                            (postcode, postkanton.id)
433
                        )
434
                if subadres_id is not None:
435
                    try:
436
                        subadres = crab_gateway.get_subadres_by_id(subadres_id)
437
                    except (GatewayRuntimeException, AttributeError):
438
                        raise colander.Invalid(
439
                            node,
440
                            'ongeldig subadres_id'
441
                        )
442
                    if subadres.huisnummer_id != huisnummer_id:
443
                        raise colander.Invalid(
444
                            node,
445
                            'het subadres %s met id %s ligt niet op huisnummer %s' %
446
                            (locatie_element.get('subadres', ''), subadres_id, huisnummer.huisnummer)
447
                        )
448
        if straat_id is None and huisnummer_id is not None:
449
            raise colander.Invalid(
450
                node,
451
                'als er een huisnummer_id wordt gegeven, moet men ook het straat_id invullen'
452
            )
453
        if huisnummer_id is None and postcode is not None:
454
            postkantons = crab_gateway.list_postkantons_by_gemeente(gemeente_id)
455
            postkantons = [str(pk.id) for pk in postkantons]
456
            if postcode not in postkantons:
457
                raise colander.Invalid(
458
                    node,
459
                    'postcode %s is niet correct voor dit adres, mogelijke postcode(s) zijn %s' %
460
                    (postcode, postkantons)
461
                )
462
463
464
def _set_deelgemeente(locatie_element, gemeente, crab_gateway):
465
    if locatie_element.get('deelgemeente', {}).get('niscode', None) is not None:
466
        deelgemeente = crab_gateway.deelgemeenten.get(locatie_element.get('deelgemeente', {}).get('niscode'), False)
467
        if deelgemeente:
468
            locatie_element['deelgemeente']['naam'] = deelgemeente['naam']
469
470
    elif locatie_element.get('deelgemeente', {}).get('naam', None) is not None and gemeente:
471
        deelgemeentes = crab_gateway.list_deelgemeenten_by_gemeente(gemeente)
472
        for gem in deelgemeentes:
473
            if gem.naam == locatie_element.get('deelgemeente', {}).get('naam'):
474
                locatie_element['deelgemeente']['niscode'] = gem.id
475