passerelle/passerelle/contrib/lille_urban_card/models.py

111 lines
4.3 KiB
Python

# passerelle - uniform access to multiple data sources and services
# Copyright (C) 2019 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 json
import re
from django.db import models
from django.utils.six.moves.urllib_parse import urljoin
from django.utils.translation import ugettext_lazy as _
from passerelle.base.models import BaseResource
from passerelle.utils.api import endpoint
from passerelle.utils.http_authenticators import HttpBearerAuth
from passerelle.utils.jsonresponse import APIError
from passerelle.sms import SMSGatewayMixin
class TokenError(APIError):
pass
class LilleUrbanCard(BaseResource):
base_url = models.URLField(max_length=256, blank=False,
verbose_name=_('Base URL'),
help_text=_('API base URL'))
username = models.CharField(max_length=128, verbose_name=_('Username'))
password = models.CharField(max_length=128, verbose_name=_('Password'))
category = 'Lille'
class Meta:
verbose_name = _('Lille Urban Card')
@classmethod
def get_verbose_name(cls):
return cls._meta.verbose_name
def check_status(self):
self.get_token()
def get_token(self):
response = self.requests.post(
urljoin(self.base_url, '/clu/ws/auth/connexion'),
json={'login': self.username, 'password': self.password}).json()
if response.get('erreur'):
self.logger.error('error getting token (%r)', response['erreur'])
raise TokenError(response['erreur'])
return response['token']
@endpoint(perm='can_access', description=_('Card Request'), methods=['post'])
def card_request(self, request, *args, **kwargs):
data = json.loads(request.body)
for kind_of_optional_field in ('nom_naissance', 'telephone', 'complement_numero_voie'):
if not data.get(kind_of_optional_field):
data[kind_of_optional_field] = ''
for boolean_field in ('recevoir_journal_senior', 'recevoir_msg_info_senior',
'acceptation_reg_int', 'acceptation_reg_int_resp_legal',
):
if data.get(boolean_field) == 'Oui':
data[boolean_field] = 1
else:
data[boolean_field] = 0
if data.get('telephone'):
data['telephone'] = ''.join(re.findall(r'\d', data['telephone']))
if data['civilite'] == 'Monsieur':
data['civilite'] = 1
else:
data['civilite'] = 2
data['code_postal'] = int(data['code_postal'])
data['ville'] = data['ville'].upper()
data['photo'] = data['photo']['content']
services = {}
for key in sorted(data.keys()):
if key.startswith('service_'):
service = key.split('_')[1]
if key.count('_') == 1: # ex: service_zoo
if data[key] == 'Oui':
services[service] = {'service': service, 'newsletter': 0}
elif service in services: # ex: service_zoo_newsletter
attr = key.split('_')[2]
if data[key] == 'Oui':
services[service][attr] = 1
else:
services[service][attr] = 0
del data[key]
if services:
data['services'] = services.values()
response = self.requests.post(
urljoin(self.base_url, '/clu/ws/demanderCarte'),
json=data,
auth=HttpBearerAuth(self.get_token())).json()
if response.get('erreur'):
self.logger.error('error requesting card (%r)', response['erreur'])
raise APIError(response['erreur'])
return {'data': response} # {"n_demande_clu":10000005}