127 lines
4.0 KiB
Python
127 lines
4.0 KiB
Python
# barbacompta - invoicing for dummies
|
|
# Copyright (C) 2010-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
|
|
import logging
|
|
|
|
import requests
|
|
from django.conf import settings
|
|
from django.utils.encoding import force_text
|
|
|
|
from eo_gestion.decorators import cache
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
PLATFORMS = {
|
|
'qualif': {
|
|
'TOKEN_URL': 'https://sandbox-oauth.aife.economie.gouv.fr/api/oauth/token',
|
|
'FLUX_URL': 'https://sandbox-api.aife.economie.gouv.fr/cpro/factures/v1/deposer/flux',
|
|
'ANNUAIRE_URL': 'https://chorus-pro.gouv.fr/qualif/static/far0037/courant/FAR0037.zip',
|
|
},
|
|
'prod': {
|
|
'TOKEN_URL': 'https://oauth.aife.economie.gouv.fr/api/oauth/token',
|
|
'FLUX_URL': 'https://api.aife.economie.gouv.fr/cpro/factures/v1/deposer/flux',
|
|
'ANNUAIRE_URL': 'https://chorus-pro.gouv.fr/cpp/static/far0037/courant/FAR0037.zip',
|
|
},
|
|
}
|
|
|
|
|
|
def get_platform():
|
|
platform = settings.CHORUS_PLATFORM
|
|
return PLATFORMS[platform]
|
|
|
|
|
|
def get_annuaire_url():
|
|
return get_platform()['ANNUAIRE_URL']
|
|
|
|
|
|
def get_token_url():
|
|
return get_platform()['TOKEN_URL']
|
|
|
|
|
|
def get_flux_url():
|
|
return get_platform()['FLUX_URL']
|
|
|
|
|
|
@cache
|
|
def get_piste_access_token():
|
|
client_id = settings.CHORUS_PISTE_CLIENT_ID
|
|
client_secret = settings.CHORUS_PISTE_CLIENT_SECRET
|
|
url = get_token_url()
|
|
|
|
logger.debug('getting piste oauth token with context: %s', locals())
|
|
|
|
response = requests.post(
|
|
url,
|
|
data={
|
|
'grant_type': 'client_credentials',
|
|
'client_id': client_id,
|
|
'client_secret': client_secret,
|
|
'scope': 'openid',
|
|
},
|
|
)
|
|
|
|
logger.debug('piste oauth token %s', response.text)
|
|
|
|
return response.json()['access_token']
|
|
|
|
|
|
def push_to_chorus(pdf_bytes: bytes, pdf_name: str):
|
|
tech_user_login = settings.CHORUS_TECH_USER_LOGIN
|
|
tech_user_password = settings.CHORUS_TECH_USER_PASSWORD
|
|
|
|
token = base64.b64encode((tech_user_login + ':' + tech_user_password).encode('utf-8')).decode('ascii')
|
|
|
|
logger.debug('sending invoice to ChorusPro: %s', dict(locals(), pdf_bytes='<removed>'))
|
|
|
|
payload = {
|
|
'fichierFlux': base64.b64encode(pdf_bytes).decode('ascii'),
|
|
'nomFichier': pdf_name,
|
|
'syntaxeFlux': 'IN_DP_E2_CII_FACTURX',
|
|
'avecSignature': False,
|
|
}
|
|
|
|
data = {}
|
|
|
|
try:
|
|
response = requests.post(
|
|
get_flux_url(),
|
|
headers={
|
|
'cpro-account': token,
|
|
'authorization': 'Bearer ' + get_piste_access_token(),
|
|
'content-type': 'application/json;charset=utf-8',
|
|
'accept': 'application/json;charset=utf-8',
|
|
},
|
|
json=payload,
|
|
)
|
|
except requests.RequestException as e:
|
|
logger.warning('invoice %s sent, failure: %s', pdf_name, e)
|
|
data['http.requests_error'] = str(e)
|
|
else:
|
|
data['http.response.status-code'] = response.status_code
|
|
try:
|
|
data.update(response.json())
|
|
logger.debug('invoice %s sent, reveiced %s', pdf_name, response.text)
|
|
except ValueError:
|
|
logger.warning('response from chorus is not JSON: %r', response.content[:1024])
|
|
data['http.response.headers'] = {
|
|
force_text(key, errors='replace'): force_text(value, errors='replace')
|
|
for key, value in response.headers.items()
|
|
}
|
|
data['http.response.body'] = repr(response.content[:1024])
|
|
return data
|