451 lines
16 KiB
Python
451 lines
16 KiB
Python
# passerelle - uniform access to multiple data sources and services
|
|
# 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 base64
|
|
|
|
from django.db import models
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from passerelle.base.models import BaseResource, HTTPResource
|
|
from passerelle.utils.api import endpoint
|
|
from passerelle.utils.jsonresponse import APIError
|
|
|
|
CONTACT_SCHEMA = {
|
|
'$schema': 'http://json-schema.org/draft-04/schema#',
|
|
"type": "object",
|
|
'properties': {
|
|
'adresseMailPerso': {
|
|
'description': 'Private mail address',
|
|
'type': 'string',
|
|
},
|
|
'numeroPortable': {
|
|
'description': 'Cell phone number',
|
|
'type': 'string',
|
|
},
|
|
'numeroTelephonePerso': {
|
|
'description': 'Private phone number',
|
|
'type': 'string',
|
|
},
|
|
'numeroTelephoneTravail': {
|
|
'description': 'Office phone number',
|
|
'type': 'string',
|
|
},
|
|
},
|
|
}
|
|
|
|
TAX_SCHEMA = {
|
|
'$schema': 'http://json-schema.org/draft-04/schema#',
|
|
"type": "object",
|
|
'properties': {
|
|
'indexImposition': {
|
|
'description': 'Tax index',
|
|
'type': 'string',
|
|
},
|
|
'anneeImposition': {
|
|
'description': 'Year of taxation (YYYY)',
|
|
'type': 'string',
|
|
},
|
|
'nombrePartImposition': {
|
|
'description': 'Number of tax shares',
|
|
'type': 'string',
|
|
},
|
|
'montantImposition': {
|
|
'description': 'Tax amount',
|
|
'type': 'string',
|
|
},
|
|
'indexIndividus': {
|
|
'description': "List of related family member indexes separated by ':'",
|
|
'type': 'string',
|
|
'pattern': r'^[0-9 :]+$',
|
|
},
|
|
},
|
|
}
|
|
|
|
DEMAND_SCHEMA = {
|
|
'$schema': 'http://json-schema.org/draft-04/schema#',
|
|
"type": "object",
|
|
'properties': {
|
|
'codeTypeAide': {
|
|
'description': 'Allowance type code',
|
|
'type': 'string',
|
|
},
|
|
'natureTypeAide': {
|
|
'description': 'Nature of the allowance (A, P or S)',
|
|
'type': 'string',
|
|
},
|
|
'individusConcernes': {
|
|
'description': "List of related family member indexes separated by ':'",
|
|
'type': 'string',
|
|
'pattern': r'^[0-9 :]+$',
|
|
},
|
|
'dateDebut': {
|
|
'description': 'Start date (YYYY-MM-DD)"',
|
|
'type': 'string',
|
|
},
|
|
'dateFin': {
|
|
'description': 'End date (YYYY-MM-DD)',
|
|
'type': 'string',
|
|
},
|
|
'montantFacture': {
|
|
'description': 'Invoice amount',
|
|
'type': 'string',
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
DOCUMENT_SCHEMA = {
|
|
'$schema': 'http://json-schema.org/draft-04/schema#',
|
|
'type': 'object',
|
|
'properties': {
|
|
'codeGedDocument': {
|
|
'description': "Code du document dans le paramétrage GED",
|
|
},
|
|
'document': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'filename': {
|
|
'description': "Nom du ficher",
|
|
'type': 'string',
|
|
},
|
|
'content_type': {
|
|
'description': "Type MIME",
|
|
'type': 'string',
|
|
},
|
|
'content': {
|
|
'description': "Contenu",
|
|
'type': 'string',
|
|
},
|
|
},
|
|
'required': ['filename', 'content_type', 'content'],
|
|
},
|
|
},
|
|
'required': ['codeGedDocument', 'document'],
|
|
}
|
|
|
|
|
|
class SolisAfiMss(BaseResource, HTTPResource):
|
|
base_url = models.CharField(
|
|
max_length=256,
|
|
blank=False,
|
|
verbose_name=_('Service URL'),
|
|
help_text=_('example: https://msppsolis.mon-application.fr/solisAPI/'),
|
|
)
|
|
ged_url = models.CharField(
|
|
max_length=256,
|
|
blank=True,
|
|
verbose_name=_('GED Service URL'),
|
|
help_text=_('example: https://solis.mon-application.fr/solisapiged/'),
|
|
)
|
|
ged_basic_auth_username = models.CharField(
|
|
max_length=128, verbose_name=_('Basic GED authentication username'), blank=True
|
|
)
|
|
ged_basic_auth_password = models.CharField(
|
|
max_length=128, verbose_name=_('Basic GED authentication password'), blank=True
|
|
)
|
|
|
|
category = _('Business Process Connectors')
|
|
|
|
class Meta:
|
|
verbose_name = _('Solis (mss-afi)')
|
|
|
|
def request(self, url, params=None, json=None, files=None, **kwargs):
|
|
headers = {'Accept': 'application/json'}
|
|
if json:
|
|
response = self.requests.post(url, json=json, headers=headers, **kwargs)
|
|
elif files:
|
|
response = self.requests.post(url, files=files, params=params, headers=headers, **kwargs)
|
|
else:
|
|
response = self.requests.get(url, params=params, headers=headers, **kwargs)
|
|
|
|
if response.status_code // 100 != 2:
|
|
try:
|
|
json_content = response.json()
|
|
except ValueError:
|
|
json_content = None
|
|
raise APIError(
|
|
'error status:%s %r, content:%r'
|
|
% (response.status_code, response.reason, response.content[:1024]),
|
|
data={'status_code': response.status_code, 'json_content': json_content},
|
|
)
|
|
|
|
if response.status_code == 204 or not response.content: # 204 No Content
|
|
return None
|
|
|
|
try:
|
|
json_response = response.json()
|
|
except ValueError:
|
|
raise APIError('invalid JSON content:%r' % response.content[:1024])
|
|
|
|
# API errors
|
|
if json_response.get('err'):
|
|
raise APIError(json_response.get('err_desc') or 'unknown error')
|
|
json_response.pop('err', None)
|
|
json_response.pop('err_desc', None)
|
|
|
|
return json_response
|
|
|
|
def check_status(self):
|
|
"""
|
|
Raise an exception if something goes wrong.
|
|
"""
|
|
return self.request(self.base_url + 'main/isAlive/')
|
|
|
|
def search_from_email(self, email):
|
|
response = self.request(
|
|
self.base_url + 'afi/agent/rechercherParEmail/', params={'adresseMail': email}
|
|
)
|
|
index = response.get('indexAgent')
|
|
|
|
adults = []
|
|
for key in ('agent', 'conjointAgent'):
|
|
if response.get(key):
|
|
adults.append(response.get(key))
|
|
children = response.get('enfantsAgent')
|
|
for record in adults + children:
|
|
record['id'] = record.get('indexIndividu')
|
|
record['text'] = '%(prenom)s %(nom)s' % record
|
|
|
|
return {'index': index, 'adults': adults, 'children': children, 'response': response}
|
|
|
|
@endpoint(
|
|
display_category=_('Agent'),
|
|
display_order=1,
|
|
perm='can_access',
|
|
methods=['get'],
|
|
description=_('Retrieve family composition'),
|
|
parameters={
|
|
'email': {'description': _("Agent's email address")},
|
|
},
|
|
)
|
|
def family(self, request, email):
|
|
results = self.search_from_email(email)
|
|
return {'data': results['adults'] + results['children']}
|
|
|
|
@endpoint(
|
|
display_category=_('Agent'),
|
|
display_order=2,
|
|
perm='can_access',
|
|
methods=['get'],
|
|
description=_('Retrieve agent'),
|
|
parameters={
|
|
'email': {'description': _("Agent's email address")},
|
|
},
|
|
)
|
|
def agent(self, request, email):
|
|
data = self.search_from_email(email)['response']
|
|
data['id'] = data['agent']['indexIndividu']
|
|
data['text'] = '%s %s' % (data['agent']['prenom'], data['agent']['nom'])
|
|
return {'data': data}
|
|
|
|
@endpoint(
|
|
display_category=_('Agent'),
|
|
display_order=3,
|
|
perm='can_access',
|
|
methods=['get'],
|
|
description=_('Retrieve adults from family composition'),
|
|
parameters={
|
|
'email': {'description': _("Agent's email address")},
|
|
},
|
|
)
|
|
def adults(self, request, email):
|
|
results = self.search_from_email(email)
|
|
return {'data': results['adults']}
|
|
|
|
@endpoint(
|
|
display_category=_('Agent'),
|
|
display_order=4,
|
|
perm='can_access',
|
|
methods=['get'],
|
|
description=_('Retrieve children from family composition'),
|
|
parameters={
|
|
'email': {'description': _("Agent's email address")},
|
|
},
|
|
)
|
|
def children(self, request, email):
|
|
results = self.search_from_email(email)
|
|
return {'data': results['children']}
|
|
|
|
@endpoint(
|
|
display_category=_('Agent'),
|
|
display_order=5,
|
|
name='update-contact',
|
|
perm='can_access',
|
|
methods=['post'],
|
|
description=_("Update contact details for an agent"),
|
|
parameters={
|
|
'email': {'description': _("Agent's email address")},
|
|
},
|
|
post={'request_body': {'schema': {'application/json': CONTACT_SCHEMA}}},
|
|
)
|
|
def update_contact(self, request, email, post_data):
|
|
post_data['indexAgent'] = str(self.search_from_email(email)['index'])
|
|
response = self.request(self.base_url + 'afi/agent/updateCoordonnees/', json=post_data)
|
|
return {'data': response}
|
|
|
|
@endpoint(
|
|
display_category=_('Budget'),
|
|
display_order=1,
|
|
perm='can_access',
|
|
methods=['get'],
|
|
description=_('Retrieve the list of charges for an agent'),
|
|
parameters={
|
|
'email': {'description': _("Agent's email address")},
|
|
'year': {'description': _('Year of taxation (YYYY)')},
|
|
},
|
|
)
|
|
def taxes(self, request, email, year=None):
|
|
index = self.search_from_email(email)['index']
|
|
params = {'indexAgent': str(index)}
|
|
|
|
if year:
|
|
params['anneeImposition'] = year
|
|
response = self.request(self.base_url + 'afi/budget/getImposition/', params=params)
|
|
data = [response] if response else []
|
|
else:
|
|
response = self.request(self.base_url + 'afi/budget/getImpositionsParAgent/', params=params)
|
|
data = response.get('listeImposition')
|
|
|
|
for record in data:
|
|
record['id'] = record.get('anneeImposition')
|
|
record['text'] = '%(anneeImposition)s: %(montantImposition)s' % record
|
|
return {'data': data}
|
|
|
|
@endpoint(
|
|
display_category=_('Budget'),
|
|
display_order=2,
|
|
name='declare-tax',
|
|
perm='can_access',
|
|
methods=['post'],
|
|
description=_("Register an agent's tax for one year"),
|
|
parameters={
|
|
'email': {'description': _("Agent's email address")},
|
|
},
|
|
post={'request_body': {'schema': {'application/json': TAX_SCHEMA}}},
|
|
)
|
|
def declare_tax(self, request, email, post_data):
|
|
results = self.search_from_email(email)
|
|
post_data['indexAgent'] = str(results['index'])
|
|
if post_data.get('indexIndividus'):
|
|
related_persons = []
|
|
for person in results['adults'] + results['children']:
|
|
for index in [x.strip() for x in post_data['indexIndividus'].split(':') if x.strip()]:
|
|
if str(person['indexIndividu']) == index:
|
|
related_persons.append(index)
|
|
post_data['indexIndividus'] = related_persons
|
|
response = self.request(self.base_url + 'afi/budget/declarerImpot/', json=post_data)
|
|
return {'data': response}
|
|
|
|
@endpoint(
|
|
display_category=_('Budget'),
|
|
display_order=3,
|
|
name='simulate-quotient',
|
|
perm='can_access',
|
|
methods=['get'],
|
|
description=_('Simulate the calculation of a Quotient from the tax amount and the number of shares'),
|
|
parameters={
|
|
'email': {'description': _("Agent's email address")},
|
|
'year': {'description': _('Year of taxation (YYYY)')},
|
|
},
|
|
)
|
|
def simulate_quotient(self, request, code, nb_parts, amount):
|
|
params = {
|
|
'codeCalcul': code,
|
|
'nbrPartImposition': nb_parts,
|
|
'mntImposition': amount,
|
|
}
|
|
response = self.request(self.base_url + 'afi/budget/calculer/', params=params)
|
|
return {'data': response}
|
|
|
|
@endpoint(
|
|
display_category=_('Allowance'),
|
|
display_order=1,
|
|
perm='can_access',
|
|
methods=['get'],
|
|
description=_('Retrieve the list of allowance from an agent'),
|
|
parameters={
|
|
'email': {'description': _("Agent's email address")},
|
|
'text_template': {'description': _('Text template')},
|
|
},
|
|
)
|
|
def helps(self, request, email):
|
|
params = {'indexAgent': str(self.search_from_email(email)['index'])}
|
|
response = self.request(self.base_url + 'afi/aide/getAidesParAgent/', params=params)
|
|
|
|
for record in response['aidesFinancieres']:
|
|
record['id'] = record.get('indexAideFinanciere')
|
|
record['text'] = '%(dateDemandeAide)s (%(suiviAide)s)' % record
|
|
return {'data': response['aidesFinancieres']}
|
|
|
|
@endpoint(
|
|
display_category=_('Allowance'),
|
|
display_order=2,
|
|
name='demand-help',
|
|
perm='can_access',
|
|
methods=['post'],
|
|
description=_('Submit allowance for an agent'),
|
|
parameters={
|
|
'email': {'description': _("Agent's email address")},
|
|
},
|
|
post={'request_body': {'schema': {'application/json': DEMAND_SCHEMA}}},
|
|
)
|
|
def demand_help(self, request, email, post_data):
|
|
results = self.search_from_email(email)
|
|
|
|
related_persons = []
|
|
for person in results['adults'] + results['children']:
|
|
for index in [x.strip() for x in post_data['individusConcernes'].split(':') if x.strip()]:
|
|
if str(person['indexIndividu']) == index:
|
|
related_persons.append({"indexIndividu": index})
|
|
|
|
post_data['indexAgent'] = str(results['index'])
|
|
post_data['individusConcernes'] = related_persons
|
|
response = self.request(self.base_url + 'afi/aide/deposer/', json=post_data)
|
|
return {'data': response}
|
|
|
|
@endpoint(
|
|
display_category=_('Allowance'),
|
|
display_order=3,
|
|
name='add-document',
|
|
perm='can_access',
|
|
methods=['post'],
|
|
description=_('Submit a document to the GED'),
|
|
parameters={
|
|
'email': {'description': _("Agent's email address"), 'optional': True},
|
|
'indexAideFinanciere': {'description': _("Allowance index"), 'optional': True},
|
|
},
|
|
post={'request_body': {'schema': {'application/json': DOCUMENT_SCHEMA}}},
|
|
)
|
|
def add_document(self, request, post_data, email=None, indexAideFinanciere=None):
|
|
params = {'codeGedDocument': post_data['codeGedDocument']}
|
|
if indexAideFinanciere:
|
|
params['typeIdMetierClassement'] = 'AIDEFINANCIERE'
|
|
params['idMetierClassement'] = indexAideFinanciere
|
|
elif email:
|
|
params['typeIdMetierClassement'] = 'INDIVIDU'
|
|
params['idMetierClassement'] = str(self.search_from_email(email)['index'])
|
|
else:
|
|
raise APIError("'email' or 'indexAideFinanciere' is a required property", http_status=400)
|
|
|
|
document = post_data['document']
|
|
content = base64.b64decode(document['content'])
|
|
files = {'document': (document['filename'], content, document['content_type'])}
|
|
auth = (self.ged_basic_auth_username, self.ged_basic_auth_password)
|
|
response = self.request(
|
|
self.ged_url + 'ged/web/document/deposer', params=params, files=files, auth=auth
|
|
)
|
|
return {'data': response}
|