V1.23.28 Mise à jour de la BDD des entreprises et stratégies

scripts/inpi_bilans_backfill.py

@@ -0,0 +1,369 @@
+#!/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()