LocationElementSchemaNode

Complexity

Total Complexity

90

Size/Duplication

Total Lines	357
Duplicated Lines	14.01 %

Importance

Changes	3
Bugs	0	Features	0

10 Methods

Rating	Name	Duplication	Size	Complexity
B	_prepare_locatie_perceel()	0	21	5
F	_prepare_locatie_adres()	0	47	21
F	_prepare_locatie()	0	35	14
A	_validate_locatie_openbaar_domein()	0	2	1
F	_validate_locatie_adres()	50	71	18
F	validator()	0	27	9
F	preparer()	0	26	11
A	_validate_locatie_perceel()	0	10	4
B	_validate_provincie_gemeente()	0	31	6
A	_validate_locatie()	0	2	1

# -*- coding: utf-8 -*-
"""
This module validates an location elements.
For an address in Belgium the validation uses the CRAB principles.
"""

import colander
import pycountry
from colander import null
from crabpy.gateway.exception import GatewayRuntimeException, GatewayResourceNotFoundException
from oe_geoutils.validation.validators_address import process_huisnummer, process_subadres
from oe_geoutils.validation.validators_cadaster_parcels import CadasterSchemaNode


class ProvincieSchemaNode(colander.MappingSchema):
    naam = colander.SchemaNode(
        colander.String(),
        validator=colander.Length(1, 50),
        missing=None
    )
    niscode = colander.SchemaNode(
        colander.Integer(),
        missing=None
    )


class GemeenteSchemaNode(colander.MappingSchema):
    naam = colander.SchemaNode(
        colander.String(),
        validator=colander.Length(1, 255),
        missing=None
    )
    id = colander.SchemaNode(
        colander.Integer(),
        missing=None
    )
    niscode = colander.SchemaNode(
        colander.Integer(),
        missing=None
    )


class LocationElementSchemaNode(colander.MappingSchema):
    id = colander.SchemaNode(
        colander.Integer(),
        missing=None
    )

    type = colander.SchemaNode(
        colander.String(),
        validator=colander.Length(1, 250)
    )

    provincie = ProvincieSchemaNode(
        missing={}
    )

    gemeente = GemeenteSchemaNode(
        missing={}
    )

    straat = colander.SchemaNode(
        colander.String(),
        validator=colander.Length(1, 100),
        missing=None
    )

    straat_id = colander.SchemaNode(
        colander.Integer(),
        missing=None
    )

    huisnummer = colander.SchemaNode(
        colander.String(),
        validator=colander.Length(1, 20),
        missing=None
    )

    huisnummer_id = colander.SchemaNode(
        colander.Integer(),
        missing=None
    )

    subadres = colander.SchemaNode(
        colander.String(),
        validator=colander.Length(1, 20),
        missing=None
    )

    subadres_id = colander.SchemaNode(
        colander.Integer(),
        missing=None
    )

    postcode = colander.SchemaNode(
        colander.String(),
        validator=colander.Length(1, 20),
        missing=None
    )

    land = colander.SchemaNode(
        colander.String(),
        validator=colander.Length(1, 100),
        missing='BE'
    )

    perceel = CadasterSchemaNode(
        missing=None
    )

    omschrijving = colander.SchemaNode(
        colander.String(),
        validator=colander.Length(1, 250),
        missing=None
    )

    def preparer(self, locatie_element):
        if locatie_element is None or not locatie_element:
            return null  # pragma: no cover
        request = self.bindings['request']
        crab_gateway = request.crab_gateway()
        capakey_gateway = request.capakey_gateway()
        locatie_element_type = ''
        if 'type' in locatie_element:
            locatie_element_type = locatie_element.get('type')
        if 'land' in locatie_element and locatie_element.get('land').upper() == 'BE':
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementAdres':
                locatie_element = self._prepare_locatie_adres(crab_gateway, locatie_element)
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementPerceel':
                locatie_element = self._prepare_locatie_perceel(capakey_gateway, crab_gateway, locatie_element)
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementOpenbaarDomein':
                locatie_element = self._prepare_locatie(crab_gateway, locatie_element)
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElement':
                locatie_element = self._prepare_locatie(crab_gateway, locatie_element)
        else:
            locatie_element['gemeente']['id'] = None
            locatie_element['straat_id'] = None
            locatie_element['huisnummer_id'] = None
            locatie_element['subadres_id'] = None
        if 'land' in locatie_element:
            locatie_element['land'] = locatie_element.get('land').upper()
        return locatie_element

    def validator(self, node, locatie_element):
        request = self.bindings['request']
        crab_gateway = request.crab_gateway()
        capakey_gateway = request.capakey_gateway()
        locatie_element_type = ''
        land = None
        if 'type' in locatie_element:
            locatie_element_type = locatie_element.get('type')
        if 'land' in locatie_element:
            land = locatie_element.get('land')
            try:
                pycountry.countries.get(alpha2=land)
            except KeyError:
                raise colander.Invalid(
                    node,
                    'ongeldige landcode %s, dit is geen ISO 3166 code' %
                    land
                )
        if land == 'BE':
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementAdres':
                self._validate_locatie_adres(crab_gateway, locatie_element, node)
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementPerceel':
                self._validate_locatie_perceel(capakey_gateway, locatie_element, node)
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElementOpenbaarDomein':
                self._validate_locatie_openbaar_domein(locatie_element, node)
            if locatie_element_type == 'https://id.erfgoed.net/vocab/ontology#LocatieElement':
                self._validate_locatie(locatie_element, node)

    @staticmethod
    def _prepare_locatie_adres(crab_gateway, locatie_element):
        if locatie_element.get('gemeente', {}).get('id', None) is None:
            if locatie_element.get('gemeente', {}).get('naam', None) is None:
                return locatie_element
            gemeente = locatie_element.get('gemeente', {}).get('naam')
            gewest_ids = [2, 1, 3]
            for gewest_id in gewest_ids:
                gemeenten = crab_gateway.list_gemeenten(gewest_id)
                gemeente_val = next((g for g in gemeenten if g.naam.lower() == gemeente.lower()), None)
                if gemeente_val:
                    locatie_element['gemeente']['id'] = gemeente_val.id
                    break
            if locatie_element.get('gemeente', {}).get('id', None) is None:
                return locatie_element
        gemeente_id = locatie_element.get('gemeente', {}).get('id', None)
        straat_id = locatie_element.get('straat_id', None)
        straat = locatie_element.get('straat', None)
        huisnummer_id = locatie_element.get('huisnummer_id', None)
        huisnummer = locatie_element.get('huisnummer', None)
        subadres_id = locatie_element.get('subadres_id', None)
        subadres = locatie_element.get('subadres', None)
        try:
            gemeente = crab_gateway.get_gemeente_by_id(gemeente_id)
        except (GatewayRuntimeException, AttributeError):
            locatie_element['gemeente']['naam'] = None
            return locatie_element
        if gemeente:
            locatie_element['gemeente']['naam'] = "" + gemeente.naam
            locatie_element['gemeente']['niscode'] = gemeente.niscode
            locatie_element['provincie']['niscode'] = gemeente.provincie.niscode
            locatie_element['provincie']['naam'] = gemeente.provincie.naam
            if straat_id:
                straat_val = next((s for s in gemeente.straten if s.id == straat_id), None)
                if straat_val:
                    locatie_element['straat'] = "" + straat_val.label
                    num_val = process_huisnummer(locatie_element, huisnummer_id, huisnummer, straat_val)
                    if num_val:
                        process_subadres(locatie_element, subadres_id, subadres, num_val)
            if not straat_id and straat:
                straat_val = next((s for s in gemeente.straten if s.label.lower() == straat.lower()), None)
                if straat_val:
                    locatie_element['straat_id'] = straat_val.id
                    num_val = process_huisnummer(locatie_element, huisnummer_id, huisnummer, straat_val)
                    if num_val:
                        process_subadres(locatie_element, subadres_id, subadres, num_val)
        return locatie_element

    @staticmethod
    def _prepare_locatie_perceel(capakey_gateway, crab_gateway, locatie_element):
        try:
            kadastrale_afdelingen = capakey_gateway.list_kadastrale_afdelingen()
            capakey = locatie_element['perceel']['capakey']
            afdeling_niscode = int(capakey[0:5])
            afdeling = next((afd for afd in kadastrale_afdelingen if afd.id == afdeling_niscode), None)
            if afdeling:
                locatie_element['perceel']['sectie'] = capakey[5:6]
                locatie_element['perceel']['perceel'] = capakey[6:]
                locatie_element['perceel']['afdeling'] = afdeling.naam
                gemeente = crab_gateway.get_gemeente_by_niscode(afdeling.gemeente.id)
                locatie_element['gemeente']['id'] = gemeente.id
                locatie_element['gemeente']['naam'] = gemeente.naam
                locatie_element['gemeente']['niscode'] = afdeling.gemeente.id
                locatie_element['provincie']['niscode'] = gemeente.provincie.niscode
                locatie_element['provincie']['naam'] = gemeente.provincie.naam
        except (GatewayRuntimeException, AttributeError, GatewayResourceNotFoundException) as e:
            # couldn't prepare data , skipping
            pass
        return locatie_element

    @staticmethod
    def _prepare_locatie(crab_gateway, locatie_element):
        gemeente = None
        gemeente_id = locatie_element.get('gemeente', {}).get('id', None)
        gemeente_niscode = locatie_element.get('gemeente', {}).get('niscode', None)
        gemeente_naam = locatie_element.get('gemeente', {}).get('naam', None)
        if gemeente_id:
            try:
                gemeente = crab_gateway.get_gemeente_by_id(gemeente_id)
            except (GatewayRuntimeException, AttributeError):
                gemeente = None
        if gemeente is None and gemeente_niscode:
            try:
                gemeente = crab_gateway.get_gemeente_by_niscode(gemeente_niscode)
            except (GatewayRuntimeException, AttributeError):
                gemeente = None
        if gemeente is None and gemeente_naam:
            gewest_ids = [2, 1, 3]
            for gewest_id in gewest_ids:
                try:
                    gemeenten = crab_gateway.list_gemeenten(gewest_id)
                    gemeente = next((g for g in gemeenten if g.naam.lower() == gemeente_naam.lower()), None)
                    if gemeente:
                        break
                except (GatewayRuntimeException, AttributeError):  # pragma no cover
                    gemeente = None
        if gemeente:
            locatie_element['gemeente']['id'] = gemeente.id
            locatie_element['gemeente']['naam'] = gemeente.naam
            locatie_element['gemeente']['niscode'] = gemeente.niscode
            locatie_element['provincie']['niscode'] = gemeente.provincie.niscode
            locatie_element['provincie']['naam'] = gemeente.provincie.naam
        else:
            locatie_element['gemeente']['id'] = None
        return locatie_element

    @staticmethod
    def _validate_provincie_gemeente(locatie_element, node):
        gemeente = locatie_element.get('gemeente', {}).get('naam', None)
        gemeente_id = locatie_element.get('gemeente', {}).get('id', None)
        gemeente_niscode = locatie_element.get('gemeente', {}).get('niscode', None)
        provincie = locatie_element.get('provincie', {}).get('naam', None)
        provincie_niscode = locatie_element.get('provincie', {}).get('niscode', None)
        if gemeente_id is None:
            raise colander.Invalid(
                node,
                'geen correcte gemeente_id gevonden voor de gemeente {0}'.format(gemeente)
            )
        if gemeente is None:  # if gemeente is still None here, the gemeente_id was incorrect
            raise colander.Invalid(
                node,
                'ongeldig gemeente_id {0}'.format(gemeente_id)
            )
        if gemeente_niscode is None:  # pragma no cover  # normally the first 2 checks will fail in this case
            raise colander.Invalid(
                node,
                'ongeldige gemeente_niscode {0}'.format(gemeente_niscode)
            )
        if provincie is None:  # pragma no cover   # normally the first 2 checks will fail in this case
            raise colander.Invalid(
                node,
                'ongeldige provincie {0}'.format(provincie)
            )
        if provincie_niscode is None:  # pragma no cover   # normally the first 2 checks will fail in this case
            raise colander.Invalid(
                node,
                'ongeldige provincie_niscode {0}'.format(provincie_niscode)
            )

    def _validate_locatie_perceel(self, capakey_gateway, locatie_element, node):
        self._validate_provincie_gemeente(locatie_element, node)
        kadastrale_afdelingen = capakey_gateway.list_kadastrale_afdelingen()
        capakey = locatie_element['perceel']['capakey']
        afdeling_niscode = int(capakey[0:5])
        afdeling = next((afd for afd in kadastrale_afdelingen if afd.id == afdeling_niscode), None)
        if afdeling is None:
            raise colander.Invalid(
                node,
                'ongeldige kadastrale afdeling voor capakey {0}'.format(capakey)
            )

    def _validate_locatie_openbaar_domein(self, locatie_element, node):
        self._validate_provincie_gemeente(locatie_element, node)

    def _validate_locatie(self, locatie_element, node):
        self._validate_provincie_gemeente(locatie_element, node)

    def _validate_locatie_adres(self, crab_gateway, locatie_element, node):
        self._validate_provincie_gemeente(locatie_element, node)
        gemeente_id = locatie_element.get('gemeente', {}).get('id', None)
        straat_id = locatie_element.get('straat_id', None)
        huisnummer_id = locatie_element.get('huisnummer_id', None)
        postcode = locatie_element.get('postcode', None)
        subadres_id = locatie_element.get('subadres_id', None)
        if straat_id is not None:

This code seems to be duplicated in your project.

            gemeente = crab_gateway.get_gemeente_by_id(gemeente_id)
            try:
                straat = crab_gateway.get_straat_by_id(straat_id)
            except (GatewayRuntimeException, AttributeError):
                raise colander.Invalid(
                    node,
                    'ongeldig straat_id'
                )
            if straat.gemeente_id != gemeente_id:
                raise colander.Invalid(
                    node,
                    'de straat %s met id %s ligt niet in gemeente %s' %
                    (locatie_element.get('straat', ''), straat_id, gemeente.naam)
                )
            if huisnummer_id is not None:
                try:
                    huisnummer = crab_gateway.get_huisnummer_by_id(huisnummer_id)
                except (GatewayRuntimeException, AttributeError):
                    raise colander.Invalid(
                        node,
                        'ongeldig huisnummer_id'
                    )
                if huisnummer.straat_id != straat_id:
                    raise colander.Invalid(
                        node,
                        'het huisnummer %s met id %s ligt niet in straat %s' %
                        (locatie_element.get('huisnummer', ''), huisnummer_id, straat.label)
                    )
                if postcode is not None:
                    postkanton = crab_gateway.get_postkanton_by_huisnummer(huisnummer_id)
                    if postcode != str(postkanton.id):
                        raise colander.Invalid(
                            node,
                            'postcode %s is niet correct voor dit adres, mogelijke postcode is %s' %
                            (postcode, postkanton.id)
                        )
                if subadres_id is not None:
                    try:
                        subadres = crab_gateway.get_subadres_by_id(subadres_id)
                    except (GatewayRuntimeException, AttributeError):
                        raise colander.Invalid(
                            node,
                            'ongeldig subadres_id'
                        )
                    if subadres.huisnummer_id != huisnummer_id:
                        raise colander.Invalid(
                            node,
                            'het subadres %s met id %s ligt niet op huisnummer %s' %
                            (locatie_element.get('subadres', ''), subadres_id, huisnummer.huisnummer)
                        )
        if straat_id is None and huisnummer_id is not None:
            raise colander.Invalid(
                node,
                'als er een huisnummer_id wordt gegeven, moet men ook het straat_id invullen'
            )
        if huisnummer_id is None and postcode is not None:
            postkantons = crab_gateway.list_postkantons_by_gemeente(gemeente_id)
            postkantons = [str(pk.id) for pk in postkantons]
            if postcode not in postkantons:
                raise colander.Invalid(
                    node,
                    'postcode %s is niet correct voor dit adres, mogelijke postcode(s) zijn %s' %
                    (postcode, postkantons)
                )