{"id":12303,"date":"2024-01-21T01:21:23","date_gmt":"2024-01-21T00:21:23","guid":{"rendered":"https:\/\/www.auditsi.eu\/?p=12303"},"modified":"2024-07-19T11:12:29","modified_gmt":"2024-07-19T09:12:29","slug":"python-automatiser-le-telechargement-en-masse-des-documents-juridiques-et-financiers-des-entreprises-au-format-pdf-a-laide-de-lapi-de-linpi","status":"publish","type":"post","link":"https:\/\/www.auditsi.eu\/?p=12303","title":{"rendered":"Python : automatiser le t\u00e9l\u00e9chargement en masse des documents juridiques et financiers des entreprises au format PDF \u00e0 l’aide de l’API de l’INPI"},"content":{"rendered":"

Le besoin en information sur les entreprises<\/strong> est croissant. Que ce soit pour \u00e9valuer la solvabilit\u00e9<\/strong> ou la fiabilit\u00e9 de clients et de fournisseurs avant d’entamer une relation commerciale, se conformer \u00e0 la loi Sapin 2<\/strong>, mener un audit d’acquisition<\/strong>..<\/p>\n

\"PYTHON<\/p>\n

L’INPI propose diff\u00e9rentes solutions pour acc\u00e9der aux informations juridiques et financi\u00e8res<\/strong> publi\u00e9es par les entreprises : portail<\/a>, FTP et API.<\/p>\n

L’objet de cet article est de fournir quelques clefs pour d\u00e9montrer tout l’int\u00e9r\u00eat d’utiliser un script Python<\/strong> pour automatiser la collecte en masse de la documentation juridique et financi\u00e8res<\/strong> des entreprises \u00e0 l’aide de requ\u00eates sur l’API<\/strong> (Application Programming Interface<\/em> ou interface de programmation applicative) de l’INPI<\/strong>.<\/p>\n

Pour fonctionner le script Python n\u00e9cessite des identifiants de connexion \u00e0 l’API INPI stock\u00e9s dans un fichier texte nomm\u00e9 Identifiants_API_INPI.txt et une liste de num\u00e9ros SIREN enregistr\u00e9s dans un fichier intitul\u00e9 SIREN \u00e0 interroger.txt.<\/p>\n

\"API<\/p>\n

Pour obtenir ses identifiants, il faut au pr\u00e9alable s’inscrire sur le site de l’INPI : https:\/\/data.inpi.fr\/login<\/a>.<\/p>\n

\"INPI<\/p>\n

Une fois l’inscription r\u00e9alis\u00e9e, se rendre dans l’espace personnel :<\/p>\n

\"INPI<\/p>\n

Puis cliquer sur Mes acc\u00e8s APIS \/ SFTP puis Acc\u00e8s APIs RNE :<\/p>\n

\"INPI<\/p>\n

Ensuite, cliquer sur le bouton :<\/p>\n

\"INPIEnfin, d\u00e9crire le projet dans la zone de saisie et cocher les API voulues (Actes et Comptes annuels).<\/p>\n

Le script Python \u00e0 r\u00e9diger doit reprendre les \u00e9tapes d\u00e9crites ci-apr\u00e8s :<\/p>\n

1\u00e8re \u00e9tape : lecture des identifiants de connexion \u00e0 l’API<\/em><\/p>\n

Dans notre exemple, les identifiants sont lus \u00e0 partir d’un fichier et stock\u00e9s dans les variables username<\/em> et password<\/em>.<\/p>\n

# Lecture des identifiants depuis un fichier\r\ndef lecture_des_identifiants(fichier_ID_API_INPI):\r\n    try:\r\n        with open(fichier_ID_API_INPI, 'r') as file:\r\n            lignes = file.readlines()\r\n            if len(lignes) >= 2:\r\n                username = lignes[0].strip()\r\n                password = lignes[1].strip()\r\n                return username, password\r\n            else:\r\n                raise Exception(\"Le fichier ne contient pas les informations d'identification n\u00e9cessaires.\")\r\n    except Exception as e:\r\n        raise Exception(f\"Erreur lors de la lecture du fichier d'identifiants : {str(e)}\")<\/pre>\n
2\u00e8me \u00e9tape : lecture de la liste des num\u00e9ros SIREN \u00e0 interroger<\/em><\/p>\n
La liste des SIREN \u00e0 interroger est stock\u00e9e dans un fichier texte (un SIREN par ligne). Le code suivant en assure la lecture (open<\/em>) et le stockage (liste_siren = [line.strip() for line in file]<\/em>) dans la variable liste_siren<\/em>.<\/p>\n
La variable nb_siren<\/em> d\u00e9nombre les SIREN \u00e0 interroger.<\/p>\n
# Lecture des SIREN \u00e0 interroger \u00e0 partir d'un fichier\r\ndef lecture_liste_siren(fichier_siren):\r\n    liste_siren = []\r\n    with open(fichier_siren, 'r') as file:\r\n        liste_siren = [line.strip() for line in file]\r\n    with open(fichier_siren, 'r') as file:\r\n        nb_siren = len(file.readlines())\r\n    return liste_siren, nb_siren<\/pre>\n
3\u00e8me \u00e9tape : connexion \u00e0 l’API<\/em><\/p>\n
Avant d’interroger les fiches SIREN, il est n\u00e9cessaire d’obtenir un token<\/em><\/strong>. Ce dernier est obtenu apr\u00e8s communication des identifiants<\/strong> (username<\/em> et password<\/em>) au registre national des entreprises g\u00e9r\u00e9 par l’INPI (url = “https:\/\/registre-national-entreprises.inpi.fr\/api\/sso\/login”<\/em>).<\/p>\n
# Connexion API et obtention token\r\ndef collecte_token(username, password):\r\n    url = \"https:\/\/registre-national-entreprises.inpi.fr\/api\/sso\/login\"\r\n    headers = {\"Content-Type\": \"application\/json\"}\r\n    data = {\"username\": username, \"password\": password}\r\n\r\n    response = requests.post(url, json=data, headers=headers)\r\n\r\n    if response.status_code == 200:\r\n        return response.json()[\"token\"]\r\n    else:\r\n        raise Exception(f\"\u00c9chec de l'authentification. Code d'erreur : {response.status_code}\")<\/pre>\n
La variable response<\/em> lit le statut du serveur<\/strong>. Par exemple, si l’authentification aboutit, cette variable \u00e9gale \u00e0 200 (statut OK)<\/strong>. Les principaux statuts (ou codes d’erreur) sont list\u00e9s plus loin.<\/p>\n
4\u00e8me \u00e9tape : collecte de la liste des documents disponibles<\/em><\/p>\n
Une requ\u00eate est envoy\u00e9e \u00e0 l’API (url = f”https:\/\/registre-national-entreprises.inpi.fr\/api\/companies\/{siren}\/attachments”<\/em>) incluant le token<\/em> (headers = {“Authorization”: f”Bearer {token}”}<\/em>).<\/p>\n
Si la requ\u00eate aboutit (if response.status_code == 200<\/em>), l’API retourne la liste des documents<\/strong> (documents = response.json()<\/em>) disponibles au t\u00e9l\u00e9chargement.<\/p>\n
Ensuite, la boucle for document_type in [“bilans”, “actes”]<\/em> passe en revue tous les documents financiers et juridiques. Le nom du fichier \u00e0 t\u00e9l\u00e9charger (variable nom_fichier<\/em>) est calcul\u00e9 fonction de la nature du document (match document_type<\/em>).<\/p>\n
# Liste les documents et lance leur t\u00e9l\u00e9chargement\r\ndef telecharge_documents(siren, token, dossier_siren):\r\n    # Param\u00e8tres requ\u00eate API\r\n    url = f\"https:\/\/registre-national-entreprises.inpi.fr\/api\/companies\/{siren}\/attachments\"\r\n    headers = {\"Authorization\": f\"Bearer {token}\"}\r\n\r\n    # Interroge le serveur INPI\r\n    response = requests.get(url, headers=headers)\r\n\r\n    # Statut API ok...\r\n    if response.status_code == 200:\r\n        # Collecte liste des documents disponibles au t\u00e9l\u00e9chargement\r\n        documents = response.json()\r\n\r\n        # Passe en revue tous les documents disponibles pour \u00e9ventuel t\u00e9l\u00e9chargement\r\n        for document_type in [\"bilans\", \"actes\"]:\r\n            for document in documents.get(document_type, []):\r\n                document_id = document[\"id\"]\r\n                document_date_depot = document[\"dateDepot\"]\r\n\r\n                match document_type:\r\n                    case \"bilans\":\r\n                        document_type_bilan = document[\"typeBilan\"]\r\n                        document_date_cloture = document[\"dateCloture\"]\r\n                        nom_fichier = f\"{document_type}_{siren}_{document_date_cloture}_{document_type_bilan}_{document_date_depot}_{document_id}.pdf\"\r\n                        telecharge_document(document_type, document_id, nom_fichier, siren, token, dossier_siren)\r\n                    case \"actes\":\r\n                        type_acte = \"type acte inconnu\"\r\n                        if \"typeRdd\" in document and document[\"typeRdd\"]:\r\n                            type_acte = document[\"typeRdd\"][0][\"typeActe\"]\r\n                        nom_fichier = f\"{document_type}_{siren}_{type_acte}_{document_date_depot}_{document_id}.pdf\"\r\n                        telecharge_document(document_type, document_id, nom_fichier, siren, token, dossier_siren)\r\n                    case _:\r\n                        print(f\"Type de document non g\u00e9r\u00e9 : {document_type}\")\r\n    # ... ou Statut API = erreur\r\n    else:\r\n        raise Exception(f\"\u00c9chec de la r\u00e9cup\u00e9ration des documents. Code d'erreur : {response.status_code}\")<\/pre>\n
Ceci fait, le programme appelle la fonction <\/strong>telecharge_document<\/strong>(document_type, document_id, nom_fichier, siren, token, dossier_siren)<\/em> afin de lancer le t\u00e9l\u00e9chargement du document :<\/p>\n
# T\u00e9l\u00e9charge un document (bilan, bilan saisi, ou acte) \u00e0 partir de son identifiant\r\ndef telecharge_document(document_type, document_id, nom_fichier, siren, token, dossier_siren):\r\n    # Param\u00e8tres requ\u00eate API\r\n    url = f\"https:\/\/registre-national-entreprises.inpi.fr\/api\/{document_type}\/{document_id}\/download\"\r\n    headers = {\"Authorization\": f\"Bearer {token}\"}\r\n\r\n    try:\r\n        # Affiche le nom du fichier en cours\r\n        message=f\"\\tDocument {nom_fichier} : \"\r\n        message_complet=message\r\n        print(message,end=\"\")\r\n\r\n        # V\u00e9rifie si le fichier existe d\u00e9j\u00e0\r\n        # Si existe d\u00e9j\u00e0 : affiche le message \"Le document existe d\u00e9j\u00e0. T\u00e9l\u00e9chargement abandonn\u00e9...\"...\r\n        chemin_fichier = os.path.join(dossier_siren, nom_fichier)\r\n        if fichier_existe(chemin_fichier):\r\n            message=\"Le document existe d\u00e9j\u00e0. T\u00e9l\u00e9chargement abandonn\u00e9...\"\r\n            message_complet=message_complet + message\r\n            print(message)\r\n        # ... sinon interrogation de L'API\r\n        else:\r\n            # Interroge le serveur INPI\r\n            response = requests.get(url, headers=headers)\r\n\r\n            # Statut API ok\r\n            if response.status_code == 200:\r\n                # Sauvegarde le contenu dans le dossier correspondant au SIREN\r\n                with open(chemin_fichier, \"wb\") as pdf_file:\r\n                    pdf_file.write(response.content)\r\n                    message=\"Le document est t\u00e9l\u00e9charg\u00e9.\"\r\n                    message_complet=message_complet + message\r\n                    print(message)\r\n            elif response.status_code == 404:\r\n                message=\"Le document n'a pas \u00e9t\u00e9 trouv\u00e9.\"\r\n                message_complet=message_complet + message\r\n                print(message)\r\n            else:\r\n                message=f\"\u00c9chec du t\u00e9l\u00e9chargement du document. Code d'erreur : {response.status_code}\"\r\n                message_complet=message_complet + message\r\n                raise Exception(message)\r\n        journalisation_message(siren, message_complet, dossier_siren)\r\n    except Exception as e:\r\n        message=f\"Une exception s'est produite lors du t\u00e9l\u00e9chargement du document : {str(e)}\"\r\n        message_complet=message_complet + message\r\n        print(message)\r\n        journalisation_message(siren, message_complet, dossier_siren)<\/pre>\n
Le t\u00e9l\u00e9chargement passe, une fois encore, par l’interrogation du serveur (avec authentification)<\/strong>. Chaque r\u00e9ponse du serveur est journalis\u00e9e dans un fichier texte<\/strong> enregistr\u00e9 dans le m\u00eame dossier (\\{siren}) que les documents t\u00e9l\u00e9charg\u00e9s afin de pouvoir v\u00e9rifier ult\u00e9rieurement l’exhaustivit\u00e9 des t\u00e9l\u00e9chargements.<\/p>\n
Les documents d\u00e9j\u00e0 t\u00e9l\u00e9charg\u00e9s (if fichier_existe(chemin_fichier)<\/em>) ne sont pas t\u00e9l\u00e9charg\u00e9s une nouvelle fois.<\/p>\n
Exemple de contenu d’un dossier SIREN \u00e0 l’issue de la boucle de t\u00e9l\u00e9chargement :<\/p>\n
\"API<\/p>\n
Fonction de journalisation des \u00e9v\u00e9nements de t\u00e9l\u00e9chargement :<\/p>\n
# Journalisation des messages\r\ndef journalisation_message(siren, message, dossier_siren):\r\n    nom_fichier_journalisation = f\"Journalisation_API_INPI_{siren}.txt\"\r\n    chemin_journalisation = os.path.join(dossier_siren, nom_fichier_journalisation)\r\n    \r\n    # Test existence du fichier de journalisation\r\n    fichier_existe = os.path.exists(chemin_journalisation)\r\n\r\n    # Ouverture fichier en mode \"a\" (ajout) s'il existe, sinon cr\u00e9ation et ouverture en mode \"w\" (\u00e9criture)\r\n    with open(chemin_journalisation, \"a\" if fichier_existe else \"w\", encoding=\"utf-8\") as journal_file:\r\n        journal_file.write(message + \"\\n\")<\/pre>\n
Exemple de fichier de journalisation :<\/p>\n
\"API<\/p>\n
Liste des codes d’erreur (statuts) de l’API :<\/p>\n
\"API<\/p>\n
A noter : les transferts de donn\u00e9es depuis l’API de l’INPI sont limit\u00e9s par des quotas quotidiens<\/strong> : 10 000 requ\u00eates et 10 Go.<\/p>\n
___<\/p>\n
Pour approfondir le sujet : autres API<\/strong> utiles avec des exemples de codes VBA :<\/p>\n