149 lines
5.9 KiB
Python
149 lines
5.9 KiB
Python
# 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
|
|
|
|
import re
|
|
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')))
|
|
|
|
FNI_PLATE_PATTERN = re.compile(r'(\d{2,4})([A-Z]{2,3})(\d{2,3})')
|
|
|
|
|
|
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()
|
|
fin_match = FNI_PLATE_PATTERN.match(immat)
|
|
if fin_match:
|
|
immat = f'{fin_match.group(3)}{fin_match.group(2)}{fin_match.group(1).zfill(4)}'
|
|
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': '747XT01'}},
|
|
)
|
|
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': '747XT01'}},
|
|
)
|
|
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': '747XT01'}},
|
|
)
|
|
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': '747XT01'}},
|
|
)
|
|
def consulterfinitiontheoriqueparimmat(self, request, immat, codesra=None):
|
|
return self.get_infos_by_immat('consulterfinitiontheoriqueparimmat', immat, codesra)
|