passerelle/passerelle/apps/sivin/models.py

# passerelle - uniform access to multiple data sources and services
# Copyright (C) 2022 Entr'ouvert
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.xs

from urllib.parse import urljoin

from django.core.cache import cache
from django.db import models
from django.utils.translation import ugettext_lazy as _
from requests import RequestException

from passerelle.base.models import BaseResource
from passerelle.utils.api import APIError, endpoint

PLATFORMS = {
    'test': {
        'token_url': 'https://api.rec.sivin.fr/token',
        'api_base_url': 'https://api.rec.sivin.fr/sivin/v2/',
    },
    'prod': {'token_url': 'https://api.sivin.fr/token', 'api_base_url': 'https://api.sivin.fr/sivin/v1/'},
}


ENVS = (('test', _('Test')), ('prod', _('Production')))


class Resource(BaseResource):
    consumer_key = models.CharField(_('Consumer key'), max_length=128)
    consumer_secret = models.CharField(_('Consumer secret'), max_length=128)
    environment = models.CharField(_('Environment'), choices=ENVS, max_length=4)

    category = _('Data Sources')

    log_requests_errors = False

    class Meta:
        verbose_name = _('SIvin')

    def check_status(self):
        self.get_token(renew=True)

    def get_token(self, renew=False):
        cache_key = 'sivin-%s-token' % self.pk
        token = cache.get(cache_key)
        if not renew and token:
            return token

        resp = self.requests.post(
            PLATFORMS[self.environment]['token_url'],
            auth=(self.consumer_key, self.consumer_secret),
            data={'grant_type': 'client_credentials'},
        )
        if resp.status_code >= 400:
            raise APIError('Failed to get token. Error: %s' % resp.status_code)
        resp = resp.json()
        token = resp['access_token']
        timeout = int(resp['expires_in'])
        cache.set(cache_key, token, timeout)
        self.logger.debug('new token: %s (timeout %ss)', token, timeout)
        return token

    def call(self, endpoint, payload):
        url = urljoin(PLATFORMS[self.environment]['api_base_url'], endpoint)
        try:
            resp = self.requests.post(
                url, headers={'Authorization': 'Bearer %s' % self.get_token()}, json=payload
            )
            if resp.status_code >= 400:
                raise APIError(resp.content)
        except RequestException as e:
            raise APIError('failed to call %s: %s' % (url, e))
        return resp.json()

    def get_infos_by_immat(self, endpoint, immat, codesra=None):
        # remove dashes / spaces in immat to avoid lookup issues
        immat = immat.strip().replace('-', '').replace(' ', '').upper()
        payload = {'immat': immat}
        if codesra is not None:
            payload['codesra'] = codesra
        return {'data': self.call(endpoint, payload)}

    @endpoint(
        perm='can_access',
        description=_('Get vehicle informations by VIN number'),
        parameters={'vin': {'description': _('VIN number'), 'example_value': 'VF1BA0E0514143067'}},
    )
    def consultervehiculeparvin(self, request, vin):
        return {'data': self.call('consultervehiculeparvin', {'vin': vin})}

    @endpoint(
        perm='can_access',
        description=_('Get VIN vehicles list of a SIREN'),
        parameters={'siren': {'description': _('SIREN Number'), 'example_value': '000399634'}},
    )
    def consulterflotteparsiren(self, request, siren):
        result = self.call('consulterflotteparsiren', {'siren': siren})
        return {'data': [{'id': vin, 'text': vin} for vin in result.get('vins', [])]}

    @endpoint(
        perm='can_access',
        description=_('Get vehicle details by registration plate'),
        parameters={'immat': {'description': _('Registration plate number'), 'example_value': '01XT0747'}},
    )
    def consultervehiculeparimmat(self, request, immat, codesra=None):
        return self.get_infos_by_immat('consultervehiculeparimmat', immat, codesra)

    @endpoint(
        perm='can_access',
        description=_('Get vehicle "finition" by registration plate'),
        parameters={'immat': {'description': _('Registration plate number'), 'example_value': '01XT0747'}},
    )
    def consulterfinitionparimmat(self, request, immat, codesra=None):
        result = self.get_infos_by_immat('consulterfinitionparimmat', immat, codesra)
        return {'data': result['data']['finitions']}

    @endpoint(
        perm='can_access',
        description=_('Get vehicle "finition" by registration plate, ordered by rangs'),
        parameters={'immat': {'description': _('Registration plate number'), 'example_value': '01XT0747'}},
    )
    def consulterfinitionscoresparimmat(self, request, immat, codesra=None):
        return self.get_infos_by_immat('consulterfinitionscoresparimmat', immat, codesra)

    @endpoint(
        perm='can_access',
        description=_('Get vehicle theorical "finition" by registration plate'),
        parameters={'immat': {'description': _('Registration plate number'), 'example_value': '01XT0747'}},
    )
    def consulterfinitiontheoriqueparimmat(self, request, immat, codesra=None):
        return self.get_infos_by_immat('consulterfinitiontheoriqueparimmat', immat, codesra)