Chorus piste api implementation

.github/scripts/release.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+import json
+import subprocess
+
+
+def get_last_version() -> str:
+    """Return the version number of the last release."""
+    json_string = (
+        subprocess.run(
+            ["gh", "release", "view", "--json", "tagName"],
+            check=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        .stdout.decode("utf8")
+        .strip()
+    )
+
+    return json.loads(json_string)["tagName"]
+
+
+def bump_patch_number(version_number: str) -> str:
+    """Return a copy of `version_number` with the patch number incremented."""
+    major, minor, patch = version_number.split(".")
+    return f"{major}.{minor}.{int(patch) + 1}"
+
+
+def create_new_patch_release():
+    """Create a new patch release on GitHub."""
+    try:
+        last_version_number = get_last_version()
+    except subprocess.CalledProcessError as err:
+        if err.stderr.decode("utf8").startswith("HTTP 404:"):
+            # The project doesn't have any releases yet.
+            new_version_number = "1.0.1"
+        else:
+            raise
+    else:
+        new_version_number = bump_patch_number(last_version_number)
+
+    subprocess.run(
+        ["gh", "release", "create", "--generate-notes", new_version_number],
+        check=True,
+    )
+
+
+if __name__ == "__main__":
+    create_new_patch_release()

.github/workflows/publish.yml

@@ -0,0 +1,17 @@
+name: Publish to PyPI.org
+on:
+  release:
+    types: [published]
+jobs:
+  pypi:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+        with:
+          fetch-depth: 0
+      - run: mkdir -p dist && python3 -m pip install --upgrade pip && python3 -m pip install --upgrade build && python3 -m build
+      - name: Publish package
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          password: ${{ secrets.PYPI_API_TOKEN }}

.github/workflows/release.yml

@@ -0,0 +1,12 @@
+name: Create a new patch release
+on: workflow_dispatch
+jobs:
+  github:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+      - name: Create new patch release
+        run: .github/scripts/release.py
+        env:
+          GITHUB_TOKEN: ${{ secrets.PERSONAL_ACCESS_TOKEN }}

CHANGES.txt

@@ -0,0 +1,5 @@
+# Change Log / journal des changements
+
+# 22/08/2023
+## IMPLEMENTATION OF CHORUS API
+*

README.md

@@ -1 +1,49 @@
-# chorus-api
+# CHORUS API
+
+# chorusapi [![Downloads](https://static.pepy.tech/personalized-badge/chorusapi?period=total&units=international_system&left_color=black&right_color=orange&left_text=Downloads)](https://pepy.tech/project/izi18n)
+
+Chorus API Package [https://piste.gouv.fr](https://piste.gouv.fr)
+
+## How to install chorus-api
+
+# Install
+
+```shell
+pip install chorus-api
+```
+
+# Usage
+
+* [GET TOKEN](exemples/get_token.py)
+
+```python
+from chorusapi.client import ChorusAPI
+from exemples.env import CLIENT_ID, CLIENT_SECRET
+
+chorus_api = ChorusAPI(
+    client_id=CLIENT_ID,
+    client_secret=CLIENT_SECRET
+)
+
+token = chorus_api.auth().get_token()
+print(token)
+```
+
+* [Consulter les informations liées au dépôt d'un flux](exemples/status_depot.py)
+
+```python
+from chorusapi.client import ChorusAPI
+from exemples.env import CLIENT_ID, CLIENT_SECRET, TECH_USERNAME, TECH_PASSWORD
+
+chorus_api = ChorusAPI(
+    client_id=CLIENT_ID,
+    client_secret=CLIENT_SECRET,
+    tech_username=TECH_USERNAME,
+    tech_password=TECH_PASSWORD
+)
+
+status_depot = chorus_api.auth().consulter_cr("CPP0XXXXXXXXXXXXX")
+print(status_depot.__dict__)
+```
+
+## More details in [examples](exemples) folders

chorusapi/client.py

@@ -0,0 +1,15 @@
+from chorusapi.models.factures import Factures
+from chorusapi.models.structures import Structures
+from chorusapi.models.transverses import Transverses
+
+
+class ChorusAPI(Factures, Structures, Transverses):
+    def __init__(self, client_id, client_secret, tech_username=None, tech_password=None, sandbox=True):
+        """
+        :param client_id:
+        :param client_secret:
+        :param tech_username:
+        :param tech_password:
+        :param sandbox:
+        """
+        super().__init__(client_id, client_secret, tech_username, tech_password, sandbox)

chorusapi/constants/api_constant.py

@@ -0,0 +1,4 @@
+AUTH_SANDBOX_URL = 'https://sandbox-oauth.piste.gouv.fr/api/oauth/token'
+AUTH_PRODUCTION_URL = 'https://oauth.piste.gouv.fr/api/oauth/token'
+API_SANDBOX_URL = 'https://sandbox-api.piste.gouv.fr'
+API_PRODUCTION_URL = 'https://api.piste.gouv.fr'

chorusapi/constants/syntax_flux_constant.py

@@ -0,0 +1,16 @@
+class SyntaxFlux:
+    # Flux type
+    IN_DP_E1_UBL_INVOICE = 'IN_DP_E1_UBL_INVOICE'
+    IN_DP_E1_CII_16B = 'IN_DP_E1_CII_16B'
+    IN_DP_E2_UBL_INVOICE_MIN = 'IN_DP_E2_UBL_INVOICE_MIN'
+    IN_DP_E2_CPP_FACTURE_MIN = 'IN_DP_E2_CPP_FACTURE_MIN'
+    IN_DP_E2_CII_MIN_16B = 'IN_DP_E2_CII_MIN_16B'
+    IN_DP_E2_CII_FACTURX = 'IN_DP_E2_CII_FACTURX'
+
+    # IN_DP_E2_CII_MIN = 'IN_DP_E2_CII_MIN'
+    # IN_DP_E2_PES_FACTURE_MIN = 'IN_DP_E2_PES_FACTURE_MIN'
+    # IN_DP_PDFARCHIVE_UBL_INVOICE_MUG = 'IN_DP_PDFARCHIVE_UBL_INVOICE_MUG'
+    # IN_DP_PDFARCHIVE_CII_MUG = 'IN_DP_PDFARCHIVE_CII_MUG'
+    # IN_DP_E1_CII = 'IN_DP_E1_CII'
+    # IN_DP_E1_PES_FACTURE = 'IN_DP_E1_PES_FACTURE'
+    # IN_DP_E1_XCBL = 'IN_DP_E1_XCBL'

chorusapi/exceptions/exception.py

@@ -0,0 +1,6 @@
+class ChorusApiException(Exception):
+    def __init__(self, msg):
+        self.msg = msg
+
+    def __str__(self):
+        return repr("Erreur Chorus API: {}".format(str(self.msg)))

chorusapi/models/api.py

@@ -0,0 +1,74 @@
+import datetime
+
+import requests
+
+from chorusapi.utils.chorus_utils import base_64_encode
+from chorusapi.constants.api_constant import AUTH_SANDBOX_URL, AUTH_PRODUCTION_URL, API_SANDBOX_URL, API_PRODUCTION_URL
+from chorusapi.exceptions.exception import ChorusApiException
+
+
+class API:
+    def __init__(self, client_id, client_secret, tech_username=None, tech_password=None, sandbox=True):
+        """
+        :param client_id:
+        :param client_secret:
+        :param tech_username:
+        :param tech_password:
+        :param sandbox:
+        """
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.tech_username = tech_username
+        self.tech_password = tech_password
+        self.sandbox = sandbox
+        self.auth_url = AUTH_SANDBOX_URL if self.sandbox else AUTH_PRODUCTION_URL
+        self.api_url = API_SANDBOX_URL if self.sandbox else API_PRODUCTION_URL
+        self.grant_type = "client_credentials"
+        self.scope = "openid"
+        self.token = None
+        self._token_date_expiration = None
+
+    def _token_has_expire(self):
+        if self.token and self._token_date_expiration:
+            return self._token_date_expiration < datetime.datetime.now()
+
+        return True
+
+    def auth(self):
+        """
+        :return: {API}
+        """
+        if self._token_has_expire():
+            headers = {
+                'content-type': 'application/x-www-form-urlencoded',
+            }
+
+            data = f'grant_type={self.grant_type}&client_id={self.client_id}&client_secret={self.client_secret}&scope={self.scope}'
+            req = requests.post(self.auth_url, headers=headers, data=data)
+            if req.status_code != 200:
+                raise ChorusApiException(req.text)
+
+            json_response = req.json()
+
+            self.token = json_response['access_token']
+            self._token_date_expiration = datetime.datetime.now() + datetime.timedelta(
+                seconds=json_response.get('expires_in', 0))
+        return self
+
+    def _make_headers(self):
+        cpro_account = base_64_encode(f"{self.tech_username}:{self.tech_password}")
+
+        if not self.token:
+            raise ChorusApiException("Pas de jeton token, veuillez appeler la méthode auth() pour générer un token")
+
+        return {
+            'cpro-account': f"{cpro_account}",
+            'Authorization': f'Bearer {self.token}',
+            'Content-Type': 'application/json'
+        }
+
+    def _make_error_msg(self, req):
+        return f"Code: {req.status_code} Reason: {req.reason}  Message: {req.text}"
+
+    def get_token(self):
+        return self.token

chorusapi/models/factures.py

@@ -0,0 +1,59 @@
+from chorusapi.responses.api_response import DeposerFluxResponse, RechercheFactureResponse
+from chorusapi.constants.syntax_flux_constant import SyntaxFlux
+from chorusapi.exceptions.exception import ChorusApiException
+from chorusapi.models.api import API
+import requests
+import json
+
+
+class Factures(API):
+    def __init__(self, client_id, client_secret, tech_username=None, tech_password=None, sandbox=True):
+        super().__init__(client_id, client_secret, tech_username, tech_password, sandbox)
+
+    def deposer_flux(self, fichier_flux, file_name, avec_signature=False,
+                     syntaxe_flux=SyntaxFlux.IN_DP_E2_CII_FACTURX):
+        """
+        : La méthode deposerFluxFacture permet de déposer un fichier XML ou PDF/A3 permettant de renseigner les données
+        nécessaires à la constitution d'un flux facture.
+        :param fichier_flux: Le fichier facture encodé en base64
+        :param file_name: Le nom du fichier + extension
+        :param avec_signature:
+        :param syntaxe_flux:
+        :return: {DeposerFluxResponse}
+        """
+        headers = self._make_headers()
+        data = {
+            'idUtilisateurCourant': 0,
+            'fichierFlux': fichier_flux,
+            'nomFichier': file_name,
+            'syntaxeFlux': syntaxe_flux,
+            'avecSignature': avec_signature,
+        }
+        _url = "{}/cpro/factures/v1/deposer/flux".format(self.api_url)
+
+        req = requests.post(_url, data=json.dumps(data), headers=headers, allow_redirects=True)
+
+        if req.status_code != 200:
+            raise ChorusApiException(self._make_error_msg(req))
+
+        return DeposerFluxResponse(req.json())
+
+    def rechercher_facture(self, numero_flux_depot):
+        """
+        : La méthode rechercherFactureParFournisseur permet d'afficher les factures émises correspondant aux paramètres
+        de recherche renseignés.
+        :param numero_flux_depot:
+        :return: {RechercheFactureResponse}
+        """
+
+        _url = "{}/cpro/factures/v1/rechercher/fournisseur".format(self.api_url)
+
+        headers = self._make_headers()
+        payload = {"numeroFluxDepot": numero_flux_depot, "typeFacture": "FACTURE"}
+
+        req = requests.post(_url, headers=headers, data=json.dumps(payload))
+
+        if req.status_code != 200:
+            raise ChorusApiException(self._make_error_msg(req))
+
+        return RechercheFactureResponse(req.json())