Completed
Push — master ( 180df1...ad4d03 )
by Bart
02:15 queued 01:00
created

LocationElementSchemaNode.validator()   F

Complexity

Conditions 9

Size

Total Lines 27

Duplication

Lines 0
Ratio 0 %

Importance

Changes 3
Bugs 0 Features 0
Metric Value
cc 9
dl 0
loc 27
rs 3
c 3
b 0
f 0
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
7
import colander
8
import pycountry
9
from colander import null
10
from crabpy.gateway.exception import GatewayRuntimeException, GatewayResourceNotFoundException
11
from oe_geoutils.validation.validators_address import process_huisnummer, process_subadres
12
from oe_geoutils.validation.validators_cadaster_parcels import CadasterSchemaNode
13
14
15
class ProvincieSchemaNode(colander.MappingSchema):
16
    naam = colander.SchemaNode(
17
        colander.String(),
18
        validator=colander.Length(1, 50),
19
        missing=None
20
    )
21
    niscode = colander.SchemaNode(
22
        colander.Integer(),
23
        missing=None
24
    )
25
26
27
class GemeenteSchemaNode(colander.MappingSchema):
28
    naam = colander.SchemaNode(
29
        colander.String(),
30
        validator=colander.Length(1, 255),
31
        missing=None
32
    )
33
    id = colander.SchemaNode(
34
        colander.Integer(),
35
        missing=None
36
    )
37
    niscode = colander.SchemaNode(
38
        colander.Integer(),
39
        missing=None
40
    )
41
42
43
class LocationElementSchemaNode(colander.MappingSchema):
44
    id = colander.SchemaNode(
45
        colander.Integer(),
46
        missing=None
47
    )
48
49
    type = colander.SchemaNode(
50
        colander.String(),
51
        validator=colander.Length(1, 250)
52
    )
53
54
    provincie = ProvincieSchemaNode(
55
        missing={}
56
    )
57
58
    gemeente = GemeenteSchemaNode(
59
        missing={}
60
    )
61
62
    straat = colander.SchemaNode(
63
        colander.String(),
64
        validator=colander.Length(1, 100),
65
        missing=None
66
    )
67
68
    straat_id = colander.SchemaNode(
69
        colander.Integer(),
70
        missing=None
71
    )
72
73
    huisnummer = colander.SchemaNode(
74
        colander.String(),
75
        validator=colander.Length(1, 20),
76
        missing=None
77
    )
78
79
    huisnummer_id = colander.SchemaNode(
80
        colander.Integer(),
81
        missing=None
82
    )
83
84
    subadres = colander.SchemaNode(
85
        colander.String(),
86
        validator=colander.Length(1, 20),
87
        missing=None
88
    )
89
90
    subadres_id = colander.SchemaNode(
91
        colander.Integer(),
92
        missing=None
93
    )
94
95
    postcode = colander.SchemaNode(
96
        colander.String(),
97
        validator=colander.Length(1, 20),
98
        missing=None
99
    )
100
101
    land = colander.SchemaNode(
102
        colander.String(),
103
        validator=colander.Length(1, 100),
104
        missing='BE'
105
    )
106
107
    perceel = CadasterSchemaNode(
108
        missing=None
109
    )
110
111
    omschrijving = colander.SchemaNode(
112
        colander.String(),
113
        validator=colander.Length(1, 250),
114
        missing=None
115
    )
116
117
    def preparer(self, locatie_element):
118
        if locatie_element is None or not locatie_element:
119
            return null  # pragma: no cover
120
        request = self.bindings['request']
121
        crab_gateway = request.crab_gateway()
122
        capakey_gateway = request.capakey_gateway()
123
        locatie_element_type = ''
124
        if 'type' in locatie_element:
125
            locatie_element_type = locatie_element.get('type')
126
        if 'land' in locatie_element and locatie_element.get('land').upper() == 'BE':
127
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementAdres':
128
                locatie_element = self._prepare_locatie_adres(crab_gateway, locatie_element)
129
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementPerceel':
130
                locatie_element = self._prepare_locatie_perceel(capakey_gateway, crab_gateway, locatie_element)
131
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementOpenbaarDomein':
132
                locatie_element = self._prepare_locatie(crab_gateway, locatie_element)
133
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElement':
134
                locatie_element = self._prepare_locatie(crab_gateway, locatie_element)
135
        else:
136
            locatie_element['gemeente']['id'] = None
137
            locatie_element['straat_id'] = None
138
            locatie_element['huisnummer_id'] = None
139
            locatie_element['subadres_id'] = None
140
        if 'land' in locatie_element:
141
            locatie_element['land'] = locatie_element.get('land').upper()
142
        return locatie_element
143
144
    def validator(self, node, locatie_element):
145
        request = self.bindings['request']
146
        crab_gateway = request.crab_gateway()
147
        capakey_gateway = request.capakey_gateway()
148
        locatie_element_type = ''
149
        land = None
150
        if 'type' in locatie_element:
151
            locatie_element_type = locatie_element.get('type')
152
        if 'land' in locatie_element:
153
            land = locatie_element.get('land')
154
            try:
155
                pycountry.countries.get(alpha2=land)
156
            except KeyError:
157
                raise colander.Invalid(
158
                    node,
159
                    'ongeldige landcode %s, dit is geen ISO 3166 code' %
160
                    land
161
                )
162
        if land == 'BE':
163
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementAdres':
164
                self._validate_locatie_adres(crab_gateway, locatie_element, node)
165
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementPerceel':
166
                self._validate_locatie_perceel(capakey_gateway, locatie_element, node)
167
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementOpenbaarDomein':
168
                self._validate_locatie_openbaar_domein(locatie_element, node)
169
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElement':
170
                self._validate_locatie(locatie_element, node)
171
172
    @staticmethod
173
    def _prepare_locatie_adres(crab_gateway, locatie_element):
174
        if locatie_element.get('gemeente', {}).get('id', None) is None:
175
            if locatie_element.get('gemeente', {}).get('naam', None) is None:
176
                return locatie_element
177
            gemeente = locatie_element.get('gemeente', {}).get('naam')
178
            gewest_ids = [2, 1, 3]
179
            for gewest_id in gewest_ids:
180
                gemeenten = crab_gateway.list_gemeenten(gewest_id)
181
                gemeente_val = next((g for g in gemeenten if g.naam.lower() == gemeente.lower()), None)
182
                if gemeente_val:
183
                    locatie_element['gemeente']['id'] = gemeente_val.id
184
                    break
185
            if locatie_element.get('gemeente', {}).get('id', None) is None:
186
                return locatie_element
187
        gemeente_id = locatie_element.get('gemeente', {}).get('id', None)
188
        straat_id = locatie_element.get('straat_id', None)
189
        straat = locatie_element.get('straat', None)
190
        huisnummer_id = locatie_element.get('huisnummer_id', None)
191
        huisnummer = locatie_element.get('huisnummer', None)
192
        subadres_id = locatie_element.get('subadres_id', None)
193
        subadres = locatie_element.get('subadres', None)
194
        try:
195
            gemeente = crab_gateway.get_gemeente_by_id(gemeente_id)
196
        except (GatewayRuntimeException, AttributeError):
197
            locatie_element['gemeente']['naam'] = None
198
            return locatie_element
199
        if gemeente:
200
            locatie_element['gemeente']['naam'] = "" + gemeente.naam
201
            locatie_element['gemeente']['niscode'] = gemeente.niscode
202
            locatie_element['provincie']['niscode'] = gemeente.provincie.niscode
203
            locatie_element['provincie']['naam'] = gemeente.provincie.naam
204
            if straat_id:
205
                straat_val = next((s for s in gemeente.straten if s.id == straat_id), None)
206
                if straat_val:
207
                    locatie_element['straat'] = "" + straat_val.label
208
                    num_val = process_huisnummer(locatie_element, huisnummer_id, huisnummer, straat_val)
209
                    if num_val:
210
                        process_subadres(locatie_element, subadres_id, subadres, num_val)
211
            if not straat_id and straat:
212
                straat_val = next((s for s in gemeente.straten if s.label.lower() == straat.lower()), None)
213
                if straat_val:
214
                    locatie_element['straat_id'] = straat_val.id
215
                    num_val = process_huisnummer(locatie_element, huisnummer_id, huisnummer, straat_val)
216
                    if num_val:
217
                        process_subadres(locatie_element, subadres_id, subadres, num_val)
218
        return locatie_element
219
220
    @staticmethod
221
    def _prepare_locatie_perceel(capakey_gateway, crab_gateway, locatie_element):
222
        try:
223
            kadastrale_afdelingen = capakey_gateway.list_kadastrale_afdelingen()
224
            capakey = locatie_element['perceel']['capakey']
225
            afdeling_niscode = int(capakey[0:5])
226
            afdeling = next((afd for afd in kadastrale_afdelingen if afd.id == afdeling_niscode), None)
227
            if afdeling:
228
                locatie_element['perceel']['sectie'] = capakey[5:6]
229
                locatie_element['perceel']['perceel'] = capakey[6:]
230
                locatie_element['perceel']['afdeling'] = afdeling.naam
231
                gemeente = crab_gateway.get_gemeente_by_niscode(afdeling.gemeente.id)
232
                locatie_element['gemeente']['id'] = gemeente.id
233
                locatie_element['gemeente']['naam'] = gemeente.naam
234
                locatie_element['gemeente']['niscode'] = afdeling.gemeente.id
235
                locatie_element['provincie']['niscode'] = gemeente.provincie.niscode
236
                locatie_element['provincie']['naam'] = gemeente.provincie.naam
237
        except (GatewayRuntimeException, AttributeError, GatewayResourceNotFoundException) as e:
238
            # couldn't prepare data , skipping
239
            pass
240
        return locatie_element
241
242
    @staticmethod
243
    def _prepare_locatie(crab_gateway, locatie_element):
244
        gemeente = None
245
        gemeente_id = locatie_element.get('gemeente', {}).get('id', None)
246
        gemeente_niscode = locatie_element.get('gemeente', {}).get('niscode', None)
247
        gemeente_naam = locatie_element.get('gemeente', {}).get('naam', None)
248
        if gemeente_id:
249
            try:
250
                gemeente = crab_gateway.get_gemeente_by_id(gemeente_id)
251
            except (GatewayRuntimeException, AttributeError):
252
                gemeente = None
253
        if gemeente is None and gemeente_niscode:
254
            try:
255
                gemeente = crab_gateway.get_gemeente_by_niscode(gemeente_niscode)
256
            except (GatewayRuntimeException, AttributeError):
257
                gemeente = None
258
        if gemeente is None and gemeente_naam:
259
            gewest_ids = [2, 1, 3]
260
            for gewest_id in gewest_ids:
261
                try:
262
                    gemeenten = crab_gateway.list_gemeenten(gewest_id)
263
                    gemeente = next((g for g in gemeenten if g.naam.lower() == gemeente_naam.lower()), None)
264
                    if gemeente:
265
                        break
266
                except (GatewayRuntimeException, AttributeError):  # pragma no cover
267
                    gemeente = None
268
        if gemeente:
269
            locatie_element['gemeente']['id'] = gemeente.id
270
            locatie_element['gemeente']['naam'] = gemeente.naam
271
            locatie_element['gemeente']['niscode'] = gemeente.niscode
272
            locatie_element['provincie']['niscode'] = gemeente.provincie.niscode
273
            locatie_element['provincie']['naam'] = gemeente.provincie.naam
274
        else:
275
            locatie_element['gemeente']['id'] = None
276
        return locatie_element
277
278
    @staticmethod
279
    def _validate_provincie_gemeente(locatie_element, node):
280
        gemeente = locatie_element.get('gemeente', {}).get('naam', None)
281
        gemeente_id = locatie_element.get('gemeente', {}).get('id', None)
282
        gemeente_niscode = locatie_element.get('gemeente', {}).get('niscode', None)
283
        provincie = locatie_element.get('provincie', {}).get('naam', None)
284
        provincie_niscode = locatie_element.get('provincie', {}).get('niscode', None)
285
        if gemeente_id is None:
286
            raise colander.Invalid(
287
                node,
288
                'geen correcte gemeente_id gevonden voor de gemeente {0}'.format(gemeente)
289
            )
290
        if gemeente is None:  # if gemeente is still None here, the gemeente_id was incorrect
291
            raise colander.Invalid(
292
                node,
293
                'ongeldig gemeente_id {0}'.format(gemeente_id)
294
            )
295
        if gemeente_niscode is None:  # pragma no cover  # normally the first 2 checks will fail in this case
296
            raise colander.Invalid(
297
                node,
298
                'ongeldige gemeente_niscode {0}'.format(gemeente_niscode)
299
            )
300
        if provincie is None:  # pragma no cover   # normally the first 2 checks will fail in this case
301
            raise colander.Invalid(
302
                node,
303
                'ongeldige provincie {0}'.format(provincie)
304
            )
305
        if provincie_niscode is None:  # pragma no cover   # normally the first 2 checks will fail in this case
306
            raise colander.Invalid(
307
                node,
308
                'ongeldige provincie_niscode {0}'.format(provincie_niscode)
309
            )
310
311
    def _validate_locatie_perceel(self, capakey_gateway, locatie_element, node):
312
        self._validate_provincie_gemeente(locatie_element, node)
313
        kadastrale_afdelingen = capakey_gateway.list_kadastrale_afdelingen()
314
        capakey = locatie_element['perceel']['capakey']
315
        afdeling_niscode = int(capakey[0:5])
316
        afdeling = next((afd for afd in kadastrale_afdelingen if afd.id == afdeling_niscode), None)
317
        if afdeling is None:
318
            raise colander.Invalid(
319
                node,
320
                'ongeldige kadastrale afdeling voor capakey {0}'.format(capakey)
321
            )
322
323
    def _validate_locatie_openbaar_domein(self, locatie_element, node):
324
        self._validate_provincie_gemeente(locatie_element, node)
325
326
    def _validate_locatie(self, locatie_element, node):
327
        self._validate_provincie_gemeente(locatie_element, node)
328
329
    def _validate_locatie_adres(self, crab_gateway, locatie_element, node):
330
        self._validate_provincie_gemeente(locatie_element, node)
331
        gemeente_id = locatie_element.get('gemeente', {}).get('id', None)
332
        straat_id = locatie_element.get('straat_id', None)
333
        huisnummer_id = locatie_element.get('huisnummer_id', None)
334
        postcode = locatie_element.get('postcode', None)
335
        subadres_id = locatie_element.get('subadres_id', None)
336 View Code Duplication
        if straat_id is not None:
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
337
            gemeente = crab_gateway.get_gemeente_by_id(gemeente_id)
338
            try:
339
                straat = crab_gateway.get_straat_by_id(straat_id)
340
            except (GatewayRuntimeException, AttributeError):
341
                raise colander.Invalid(
342
                    node,
343
                    'ongeldig straat_id'
344
                )
345
            if straat.gemeente_id != gemeente_id:
346
                raise colander.Invalid(
347
                    node,
348
                    'de straat %s met id %s ligt niet in gemeente %s' %
349
                    (locatie_element.get('straat', ''), straat_id, gemeente.naam)
350
                )
351
            if huisnummer_id is not None:
352
                try:
353
                    huisnummer = crab_gateway.get_huisnummer_by_id(huisnummer_id)
354
                except (GatewayRuntimeException, AttributeError):
355
                    raise colander.Invalid(
356
                        node,
357
                        'ongeldig huisnummer_id'
358
                    )
359
                if huisnummer.straat_id != straat_id:
360
                    raise colander.Invalid(
361
                        node,
362
                        'het huisnummer %s met id %s ligt niet in straat %s' %
363
                        (locatie_element.get('huisnummer', ''), huisnummer_id, straat.label)
364
                    )
365
                if postcode is not None:
366
                    postkanton = crab_gateway.get_postkanton_by_huisnummer(huisnummer_id)
367
                    if postcode != str(postkanton.id):
368
                        raise colander.Invalid(
369
                            node,
370
                            'postcode %s is niet correct voor dit adres, mogelijke postcode is %s' %
371
                            (postcode, postkanton.id)
372
                        )
373
                if subadres_id is not None:
374
                    try:
375
                        subadres = crab_gateway.get_subadres_by_id(subadres_id)
376
                    except (GatewayRuntimeException, AttributeError):
377
                        raise colander.Invalid(
378
                            node,
379
                            'ongeldig subadres_id'
380
                        )
381
                    if subadres.huisnummer_id != huisnummer_id:
382
                        raise colander.Invalid(
383
                            node,
384
                            'het subadres %s met id %s ligt niet op huisnummer %s' %
385
                            (locatie_element.get('subadres', ''), subadres_id, huisnummer.huisnummer)
386
                        )
387
        if straat_id is None and huisnummer_id is not None:
388
            raise colander.Invalid(
389
                node,
390
                'als er een huisnummer_id wordt gegeven, moet men ook het straat_id invullen'
391
            )
392
        if huisnummer_id is None and postcode is not None:
393
            postkantons = crab_gateway.list_postkantons_by_gemeente(gemeente_id)
394
            postkantons = [str(pk.id) for pk in postkantons]
395
            if postcode not in postkantons:
396
                raise colander.Invalid(
397
                    node,
398
                    'postcode %s is niet correct voor dit adres, mogelijke postcode(s) zijn %s' %
399
                    (postcode, postkantons)
400
                )
401