#!/usr/bin/env python3 """ Backfill historique des comptes annuels (API RNE/INPI, /api/bilans-saisis) dans EntreprisesXXPhase3, pour toutes les entreprises deja connues (une ligne par SIREN+annee, jamais ecrasee -- voir la conception Phase3). Contrairement a l'extraction Sirene (sirene_extract.py), cet appel API n'est pas filtrable par departement : on l'interroge par lot de SIREN (max ~300 avant de depasser la limite de longueur d'URL du serveur, on utilise 250 par securite), sans limite de date, pour recuperer tout l'historique disponible (comptes annuels dispo depuis 2017 selon l'espace personnel INPI). Les donnees saisies sont une liasse fiscale brute (codes CERFA), pas un CA/ resultat net direct -- la table TYPE_BILAN_MAPPING ci-dessous traduit les codes pertinents pour chaque type de bilan (C, S, K, AC, AS, B), etablie le 12 septembre 2026 a partir de la doc officielle INPI (v5, juin 2025) et verifiee empiriquement sur un vrai bilan (EDF, SIREN 552081317). Regle de colonne (verifiee empiriquement) : quand un code liasse a une colonne m3, elle contient le total de l'exercice en cours (m4 = exercice precedent) ; sinon c'est m1 qui est l'exercice en cours. Les bilans en confidentialite non publique ne sont pas exploites (CA/ resultat net non extraits), par respect du choix de confidentialite declare. Usage : python3 inpi_bilans_backfill.py python3 inpi_bilans_backfill.py --departements 56,29 Identifiants lus depuis .env (ENTREPRISES_DB_*, INPI_USERNAME, INPI_PASSWORD) dans le meme dossier, jamais en argument. """ import argparse import json import os import sys import time import urllib.error import urllib.parse import urllib.request import pymysql class QuotaExceededError(Exception): """Quota de l'API INPI depasse (distinct d'une simple limitation de debit passagere) -- inutile de retenter, il faut arreter et laisser le quota se regenerer (journalier selon toute vraisemblance) avant de reprendre.""" API_LOGIN = "https://registre-national-entreprises.inpi.fr/api/sso/login" API_BILANS_SAISIS = "https://registre-national-entreprises.inpi.fr/api/bilans-saisis" BATCH_SIZE = 250 # 300 confirme fonctionnel, marge de securite prise PAGE_SIZE = 100 RATE_DELAY = 0.2 # code liasse -> [codes a sommer pour le CA] / [codes a sommer pour le resultat net] TYPE_BILAN_MAPPING = { "C": {"ca": ["FJ"], "rn": ["HN"]}, "S": {"ca": ["209", "210", "214", "215", "217", "218"], "rn": ["310"]}, "K": {"ca": ["FJ"], "rn": ["R6"]}, "AC": {"ca": ["FP"], "rn": ["HO"]}, "AS": {"ca": ["EA", "EB", "EC", "ED", "EE"], "rn": ["FJ"]}, "B": {"ca": [], "rn": ["R3"]}, } # Priorite quand plusieurs bilans existent pour le meme SIREN+annee (frequent : # une societe mere depose a la fois ses comptes individuels et consolides). # On privilegie systematiquement les comptes de l'entite elle-meme (C, AC, AS, # S) a un chiffre de groupe consolide (K) qui ne represente plus la societe # precise -- et B (banque) en tout dernier recours seulement. TYPE_PRIORITY = {"C": 0, "AC": 0, "AS": 1, "S": 2, "K": 3, "B": 4} def log(msg): print(msg, file=sys.stderr, flush=True) def load_env(path): env = {} with open(path) as f: for line in f: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) env[k.strip()] = v.strip() return env def http_get_json(url, auth, retries=8): # "auth" est un dict mutable {"token", "username", "password"} plutot qu'un # jeton brut, pour pouvoir se reconnecter tout seul sur un 401 (jeton perime # en cours de traitement, constate le 14/09/2026 sur une execution de plusieurs # heures -- l'ancien code retentait 8 fois avec le meme jeton mort, ce qui ne # pouvait jamais reussir, et perdait tous les lots restants du departement). for attempt in range(retries): req = urllib.request.Request(url, headers={"Authorization": f"Bearer {auth['token']}"}) try: with urllib.request.urlopen(req, timeout=30) as resp: return json.load(resp) except urllib.error.HTTPError as e: if e.code == 401: if attempt == retries - 1: raise log(" !! 401 Unauthorized -- jeton probablement expire, reconnexion") auth["token"] = login(auth["username"], auth["password"]) continue if e.code == 429: # Deux cas bien distincts derriere le meme code HTTP 429, constates # le 14/09/2026 : une limitation de debit passagere (retentable avec # patience) et un vrai quota de service depasse (corps de reponse # {"errorCode":"quota_exceeded","webserviceCode":"QUOTA_SERVICE"} -- # la, retenter ne sert a rien, il faut arreter tout de suite. Avant # cette distinction, le script s'est acharne des heures a retenter # un quota epuise, perdant lot apres lot pour rien. try: body = json.loads(e.read()) except Exception: body = {} if body.get("errorCode") == "quota_exceeded": raise QuotaExceededError(body.get("message", "quota INPI depasse")) if attempt == retries - 1: raise retry_after = e.headers.get("Retry-After") if e.headers else None wait = int(retry_after) if retry_after and retry_after.isdigit() else min(10 * (2 ** attempt), 300) log(f" !! 429 Too Many Requests -- pause de {wait}s avant nouvelle tentative " f"({attempt + 1}/{retries})") time.sleep(wait) continue if attempt == retries - 1: raise log(f" retry apres erreur reseau ({e})") time.sleep(2 + attempt) except (urllib.error.URLError, TimeoutError) as e: if attempt == retries - 1: raise log(f" retry apres erreur reseau ({e})") time.sleep(2 + attempt) def login(username, password): req = urllib.request.Request( API_LOGIN, data=json.dumps({"username": username, "password": password}).encode(), headers={"Content-Type": "application/json"}, method="POST", ) with urllib.request.urlopen(req, timeout=20) as resp: return json.load(resp)["token"] def get_liasse_value(pages, code): for page in pages: for l in page.get("liasses", []): if l.get("code") == code: raw = l.get("m3") if l.get("m3") not in (None, "") else l.get("m1") if raw in (None, ""): continue try: return int(raw) except ValueError: continue return None def sum_codes(pages, codes): if not codes: return None total = 0 found_any = False for code in codes: v = get_liasse_value(pages, code) if v is not None: total += v found_any = True return total if found_any else None def parse_bilan(entry): """Retourne (annee, type_bilan, ca, resultat_net) ou None si non exploitable.""" if entry.get("deleted"): return None if entry.get("confidentiality") not in (None, "Public"): return None bilan_saisi = entry.get("bilanSaisi") if not bilan_saisi: return None bilan = bilan_saisi.get("bilan", {}) identite = bilan.get("identite", {}) type_bilan = identite.get("codeTypeBilan") mapping = TYPE_BILAN_MAPPING.get(type_bilan) if mapping is None: return None date_cloture = identite.get("dateClotureExercice") if not date_cloture or len(date_cloture) < 4: return None annee = date_cloture[:4] pages = bilan.get("detail", {}).get("pages", []) ca = sum_codes(pages, mapping["ca"]) rn = sum_codes(pages, mapping["rn"]) if ca is None and rn is None: return None return annee, type_bilan, ca, rn MAX_PAGES_PAR_LOT = 100 # garde-fou : 100*PAGE_SIZE = 10 000 resultats max par lot, # au-dela quelque chose ne tourne pas rond (curseur bloque # ou volume anormal) -- ne jamais boucler indefiniment. def fetch_bilans_for_batch(sirens, auth): # Le curseur searchAfter doit porter sur l'identifiant unique du document # (champ "id", ex. un ObjectId), jamais sur le SIREN : un SIREN a en general # plusieurs bilans (plusieurs annees, parfois individuel + consolide), donc # il revient plusieurs fois dans les resultats -- l'utiliser comme curseur # provoquait un faux "curseur bloque" des qu'un SIREN se trouvait en # dernniere position d'une page, abandonnant le lot prematurement et # perdant les bilans non encore recuperes (bug trouve le 14/09/2026 : # ~14% des lots du departement 31 perdaient ainsi une partie de leur # historique). Verifie empiriquement : le champ "id" donne une pagination # propre, sans doublon ni repli, contrairement au SIREN. out = [] search_after = None pages_lues = 0 while True: pages_lues += 1 if pages_lues > MAX_PAGES_PAR_LOT: log(f" !! lot bloque apres {MAX_PAGES_PAR_LOT} pages " f"(searchAfter={search_after}) -- abandon de ce lot pour eviter une " f"boucle infinie, {len(out)} resultats deja recuperes conserves") break params = [("siren[]", s) for s in sirens] params.append(("pageSize", str(PAGE_SIZE))) if search_after: params.append(("searchAfter", search_after)) url = API_BILANS_SAISIS + "?" + urllib.parse.urlencode(params) data = http_get_json(url, auth) time.sleep(RATE_DELAY) if not data: break out.extend(data) if len(data) < PAGE_SIZE: break nouveau_search_after = data[-1].get("id") if not nouveau_search_after or nouveau_search_after == search_after: log(f" !! curseur searchAfter bloque sur {nouveau_search_after} -- " f"abandon de ce lot, {len(out)} resultats deja recuperes conserves") break search_after = nouveau_search_after return out def upsert_finance(conn, table, siren, annee, ca, rn): sql = ( f"INSERT INTO `{table}` (ENT_Siren, ENT_FinancesAnnee, ENT_FinancesCa, ENT_FinancesResultatNet) " f"VALUES (%s,%s,%s,%s) " f"ON DUPLICATE KEY UPDATE ENT_FinancesCa=VALUES(ENT_FinancesCa), " f"ENT_FinancesResultatNet=VALUES(ENT_FinancesResultatNet)" ) with conn.cursor() as cur: cur.execute(sql, (siren, annee, ca, rn)) def backfill_departement(conn, dep, auth): table1 = f"Entreprises{dep}Phase1" table3 = f"Entreprises{dep}Phase3" with conn.cursor() as cur: cur.execute(f"SELECT ENT_Siren FROM `{table1}`") sirens = [row[0] for row in cur.fetchall()] log(f"=== Departement {dep} : {len(sirens)} entreprises a traiter ===") total_rows = 0 total_ignores_confidentiel = 0 total_types_inconnus = 0 unknown_types = {} for i in range(0, len(sirens), BATCH_SIZE): batch = sirens[i:i + BATCH_SIZE] try: entries = fetch_bilans_for_batch(batch, auth) except QuotaExceededError: log(f" !! quota INPI depasse au lot {i}-{i+len(batch)} -- arret immediat du " f"departement {dep} (inutile de continuer, le quota est global au compte, " f"pas au departement)") raise except Exception as e: log(f" !! lot {i}-{i+len(batch)} : erreur {e}, on continue") continue # best[(siren, annee)] = (priorite, ca, rn) -- on ne garde que le # bilan de plus haute priorite (cf TYPE_PRIORITY) par SIREN+annee, # pour ne jamais laisser un bilan consolide (K) ecraser au hasard # le bilan individuel (C) de la meme annee, ou l'inverse. best = {} for entry in entries: siren = entry.get("siren") if not siren: continue if entry.get("confidentiality") not in (None, "Public"): total_ignores_confidentiel += 1 continue bilan_saisi = entry.get("bilanSaisi") if bilan_saisi: type_bilan = bilan_saisi.get("bilan", {}).get("identite", {}).get("codeTypeBilan") if type_bilan and type_bilan not in TYPE_BILAN_MAPPING: total_types_inconnus += 1 unknown_types[type_bilan] = unknown_types.get(type_bilan, 0) + 1 parsed = parse_bilan(entry) if parsed is None: continue annee, type_bilan, ca, rn = parsed prio = TYPE_PRIORITY.get(type_bilan, 99) key = (siren, annee) current = best.get(key) if current is None or prio < current[0]: best[key] = (prio, ca, rn) for (siren, annee), (_prio, ca, rn) in best.items(): upsert_finance(conn, table3, siren, annee, ca, rn) total_rows += 1 if (i // BATCH_SIZE) % 20 == 0: log(f" ... {i+len(batch)}/{len(sirens)} entreprises traitees, " f"{total_rows} lignes financieres upsertees jusqu'ici") log(f"Departement {dep} termine : {total_rows} lignes financieres upsertees dans {table3} " f"({total_ignores_confidentiel} bilans confidentiels ignores, " f"{total_types_inconnus} types de bilan non mappes : {unknown_types})") def main(): ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--departements", default=None, help="Liste separee par des virgules (defaut : tous ceux de ENT_Departements)") ap.add_argument("--env-file", default=os.path.join(os.path.dirname(__file__), ".env")) args = ap.parse_args() env = load_env(args.env_file) conn = pymysql.connect( host=env["ENTREPRISES_DB_HOST"], port=int(env.get("ENTREPRISES_DB_PORT", 3306)), user=env["ENTREPRISES_DB_USER"], password=env["ENTREPRISES_DB_PASS"], database=env["ENTREPRISES_DB_NAME"], charset="utf8mb4", autocommit=True, ) if args.departements: departements = [d.strip() for d in args.departements.split(",") if d.strip()] else: with conn.cursor() as cur: cur.execute("SELECT DEP_Code FROM ENT_Departements ORDER BY DEP_Code") departements = [row[0] for row in cur.fetchall()] log(f"Departements a traiter : {departements}") auth = { "token": login(env["INPI_USERNAME"], env["INPI_PASSWORD"]), "username": env["INPI_USERNAME"], "password": env["INPI_PASSWORD"], } log("Connexion INPI reussie") try: for dep in departements: backfill_departement(conn, dep, auth) finally: conn.close() if __name__ == "__main__": main()