295 lines
11 KiB
Python
295 lines
11 KiB
Python
# Copyright (C) 2020 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/>.
|
|
|
|
|
|
import binascii
|
|
import json
|
|
import re
|
|
from base64 import b64decode
|
|
from datetime import datetime
|
|
|
|
from django.db import models
|
|
from django.http import HttpResponse
|
|
from django.utils.text import slugify
|
|
from django.utils.translation import gettext_lazy as _
|
|
from requests import RequestException
|
|
|
|
from passerelle.base.models import BaseResource, HTTPResource
|
|
from passerelle.utils.api import endpoint
|
|
from passerelle.utils.jsonresponse import APIError
|
|
|
|
REQUEST_SCHEMA = {
|
|
'type': 'object',
|
|
'$schema': 'http://json-schema.org/draft-04/schema#',
|
|
'title': 'Signal Arretes',
|
|
'description': 'Public Occupation Request Schema',
|
|
'required': [
|
|
'declarant_organisation',
|
|
'declarant_quality',
|
|
'declarant_civility',
|
|
'declarant_name',
|
|
'declarant_surname',
|
|
'declarant_email',
|
|
'occupation_lane',
|
|
'occupation_city',
|
|
'occupation_type',
|
|
'occupation_start_date',
|
|
'occupation_end_date',
|
|
],
|
|
'properties': {
|
|
'declarant_organisation': {
|
|
'description': _('"Individual" or enterprise name'),
|
|
'type': 'string',
|
|
},
|
|
'declarant_siret': {
|
|
'description': _('Entreprise SIRET number'),
|
|
'type': 'string',
|
|
'pattern': '(\\d{14})?',
|
|
'pattern_description': _('14-digits siret number'),
|
|
},
|
|
'declarant_quality': {
|
|
'description': _('Declarant quality'),
|
|
'type': 'string',
|
|
'enum': ['Particulier', 'Entreprise', 'Association'],
|
|
},
|
|
'file_number': {'description': _('Declarant reference'), 'type': 'string'},
|
|
'declarant_civility': {
|
|
'description': _('Declarant civility'),
|
|
'type': 'string',
|
|
'enum': ['MONSIEUR', 'MADAME'],
|
|
},
|
|
'declarant_name': {'description': _('Declarant name'), 'type': 'string'},
|
|
'declarant_surname': {'description': _('Declarant surname'), 'type': 'string'},
|
|
'declarant_address': {'description': _('Declarant address'), 'type': 'string'},
|
|
'declarant_zip': {'description': _('Declarant ZIP code'), 'type': 'string'},
|
|
'declarant_city': {'description': _('Declarant city'), 'type': 'string'},
|
|
'declarant_email': {'description': _('Declarant email address'), 'type': 'string'},
|
|
'declarant_phone': {'description': _('Declarant phone number'), 'type': 'string'},
|
|
'occupation_lane': {'description': _('Occupation lane'), 'type': 'string'},
|
|
'occupation_number': {'description': _('Occupation lane number'), 'type': 'string'},
|
|
'occupation_city': {'description': _('Occupation city'), 'type': 'string'},
|
|
'occupation_type': {'description': _('Occupation type'), 'type': 'string'},
|
|
'occupation_start_date': {
|
|
'description': _('Occupation start date'),
|
|
'type': 'string',
|
|
'format': 'date',
|
|
},
|
|
'occupation_end_date': {'description': _('Occupation end date'), 'type': 'string', 'format': 'date'},
|
|
'comment': {'description': _('Comment'), 'type': 'string'},
|
|
},
|
|
}
|
|
|
|
|
|
class SignalArretes(BaseResource, HTTPResource):
|
|
base_url = models.URLField(_('Base API URL'))
|
|
category = _('Business Process Connectors')
|
|
|
|
class Meta:
|
|
verbose_name = 'Signal Arrêtés ™'
|
|
|
|
@classmethod
|
|
def get_verbose_name(cls):
|
|
return cls._meta.verbose_name
|
|
|
|
def _call(self, endpoint, post_data=None):
|
|
url = f'{self.base_url}/CreationDemandeService.svc/{endpoint}'
|
|
|
|
try:
|
|
response = None
|
|
if not post_data:
|
|
response = self.requests.get(url)
|
|
else:
|
|
response = self.requests.post(url, json=post_data)
|
|
|
|
response.raise_for_status()
|
|
except RequestException as e:
|
|
if response is not None and response.status_code == 400:
|
|
error_msg_match = re.search(
|
|
'Le message d\'exception est \'(.*)\'\\. Pour plus d\'informations', response.text
|
|
)
|
|
if error_msg_match:
|
|
error_message = error_msg_match.group(1)
|
|
raise APIError(
|
|
'An error occured during the request to Signal Arrêtés: %s' % error_message
|
|
)
|
|
|
|
raise APIError('An error occured during the request to Signal Arrêtés: %s' % e)
|
|
|
|
try:
|
|
return response.json()
|
|
except ValueError:
|
|
raise APIError('Expected valid json')
|
|
|
|
def _get_value(self, endpoint, post_data=None, request_id=None):
|
|
if request_id:
|
|
url = f'{endpoint}/{request_id}'
|
|
else:
|
|
url = endpoint
|
|
|
|
response = self._call(url, post_data=post_data)
|
|
result_key = f'{endpoint}Result'
|
|
if not isinstance(response, dict) or result_key not in response:
|
|
raise APIError('Expected a dictionary with a %s key' % result_key)
|
|
|
|
result_str = response[result_key]
|
|
try:
|
|
return json.loads(result_str)
|
|
except ValueError:
|
|
raise APIError('Expected valid json string at %s key' % result_key)
|
|
|
|
def _get_list(self, endpoint, post_data=None, q=None, id=None):
|
|
result = self._get_value(endpoint, post_data=post_data)
|
|
if not isinstance(result, list):
|
|
raise APIError('Expected a list')
|
|
|
|
if q is not None:
|
|
q = q.lower()
|
|
result = filter(lambda it: q in it.lower(), result)
|
|
|
|
if id is not None:
|
|
result = list(filter(lambda it: slugify(it) == id, result))
|
|
|
|
return {'data': [{'id': slugify(it), 'text': it} for it in result]}
|
|
|
|
@endpoint(
|
|
description=_('Get cities available in Signal Arrêtés'),
|
|
parameters={
|
|
'id': {
|
|
'description': _('Get exactly one city from it\'s id'),
|
|
'example_value': 'base-de-vie',
|
|
},
|
|
'q': {'description': _('Search text'), 'example_value': 'Angou'},
|
|
},
|
|
)
|
|
def cities(self, request, q=None, id=None, **kwargs):
|
|
return self._get_list('GetCommunes', post_data=None, q=q, id=id)
|
|
|
|
@endpoint(
|
|
description=_('Get lanes available in Signal Arrêtés'),
|
|
parameters={
|
|
'city': {'description': _('Get lanes for this city')},
|
|
'id': {
|
|
'description': _('Get exactly one lane from it\'s id'),
|
|
'example_value': 'rue-nicolas-appert',
|
|
},
|
|
'q': {'description': _('Search text'), 'example_value': 'Rue Nic'},
|
|
},
|
|
)
|
|
def lanes(self, request, city, q=None, id=None):
|
|
return self._get_list('GetVoies', {'Commune': city}, q=q, id=id)
|
|
|
|
@endpoint(
|
|
description=_('Get available occupation types in Signal Arrêtés'),
|
|
parameters={
|
|
'id': {
|
|
'description': _('Get exactly one occupation type from it\'s id'),
|
|
'example_value': 'base-de-vie',
|
|
},
|
|
'q': {'description': _('Search text'), 'example_value': 'Base de'},
|
|
},
|
|
)
|
|
def occupation_types(self, request, q=None, id=None):
|
|
return self._get_list('GetNaturesOccupation', q=q, id=id)
|
|
|
|
@endpoint(
|
|
description=_('Create a public domain occupation request'),
|
|
post={'request_body': {'schema': {'application/json': REQUEST_SCHEMA}}},
|
|
)
|
|
def create_request(self, request, post_data):
|
|
def _format_date(date_string):
|
|
return datetime.strptime(date_string, '%d/%m/%Y').strftime('%Y-%m-%d')
|
|
|
|
query_data = {
|
|
'organisationDeclarante': post_data['declarant_organisation'],
|
|
'qualite': post_data['declarant_quality'],
|
|
'SIRET': post_data['declarant_siret'],
|
|
'numeroDossier': post_data['file_number'],
|
|
'commentaire': post_data['comment'],
|
|
'contact': {
|
|
'civilite': post_data['declarant_civility'],
|
|
'nom': post_data['declarant_name'],
|
|
'prenom': post_data['declarant_surname'],
|
|
'email': post_data['declarant_email'],
|
|
'adresseLigne1': post_data['declarant_address'],
|
|
'CP': post_data['declarant_zip'],
|
|
'ville': post_data['declarant_city'],
|
|
'telephone': post_data['declarant_phone'],
|
|
},
|
|
'localisation': {
|
|
'nomVoie': post_data['occupation_lane'],
|
|
'commune': post_data['occupation_city'],
|
|
'natureOccupation': post_data['occupation_type'],
|
|
'dateDebut': datetime.strptime(post_data['occupation_start_date'], '%d/%m/%Y').strftime(
|
|
'%Y-%m-%d'
|
|
),
|
|
'dateFin': datetime.strptime(post_data['occupation_end_date'], '%d/%m/%Y').strftime(
|
|
'%Y-%m-%d'
|
|
),
|
|
'numeroVoie': post_data['occupation_number'],
|
|
},
|
|
}
|
|
|
|
query_data = {k: v for k, v in query_data.items() if v}
|
|
query_data['contact'] = {k: v for k, v in query_data['contact'].items() if v}
|
|
query_data['localisation'] = {k: v for k, v in query_data['localisation'].items() if v}
|
|
|
|
result_string = self._call('CreationDODP', query_data)
|
|
if not isinstance(result_string, str):
|
|
raise APIError('Expected a string')
|
|
|
|
try:
|
|
result = json.loads(result_string)
|
|
except ValueError:
|
|
raise APIError('Returned string should be valid json')
|
|
|
|
if not isinstance(result, dict) or len(result) != 1:
|
|
raise APIError('Expected a dictionary with one element')
|
|
|
|
return {'request_id': list(result.keys())[0], 'request_status': list(result.values())[0]}
|
|
|
|
@endpoint(
|
|
description=_('Get status of given request in Signal Arrêtés'),
|
|
parameters={
|
|
'request_id': {'description': _('The occupation request id returned by create_request')},
|
|
},
|
|
)
|
|
def request_status(self, request, request_id):
|
|
return {'request_status': self._get_value('GetStatutDemande', request_id=request_id)}
|
|
|
|
@endpoint(
|
|
description=_('Get document associated with given request in Signal Arrêtés'),
|
|
parameters={
|
|
'request_id': {'description': _('The occupation request id returned by create_request')},
|
|
},
|
|
)
|
|
def request_document(self, request, request_id):
|
|
result = self._get_value('GetDocumentDemande', request_id=request_id)
|
|
|
|
if isinstance(result, str):
|
|
raise APIError(result)
|
|
|
|
filename = result['name']
|
|
content_type = result['contentType']
|
|
|
|
try:
|
|
content = b64decode(result['content'], validate=True)
|
|
except binascii.Error as e:
|
|
raise APIError(f'Corrupted base64 content {e}')
|
|
|
|
response = HttpResponse(content, content_type=content_type)
|
|
response['Content-Disposition'] = f'attachment; filename="{filename}"'
|
|
|
|
return response
|