ASPIC/scripts/inpi_bilans_backfill.py

#!/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()