ASPIC/scripts/ent_communes_stats.py

#!/usr/bin/env python3
"""
Calcule et stocke, pour chaque commune de ENT_Communes, le nombre d'entreprises
actives par categorie de taille (EI, TPE, PME, ETI, GE) -- pre-calcule pour que
le futur module de recherche par perimetre geographique n'ait jamais a compter
en temps reel sur les tables Entreprises*Phase1 (qui grossissent).

Regle de classification (une entreprise tombe dans une seule categorie) :
  - ENT_CategorieEntreprise = 'GE'  -> GE
  - ENT_CategorieEntreprise = 'ETI' -> ETI
  - ENT_CategorieEntreprise = 'PME' -> PME
  - categorie NULL et nature juridique = entrepreneur individuel (1er
    chiffre de ENT_NatureJuridique = '1') -> EI
  - categorie NULL, tout le reste -> TPE
(ENT_CategorieEntreprise, quand renseignee par l'INSEE, est deja fiable et
prioritaire ; les entreprises non categorisees par l'INSEE -- la grande
majorite -- sont reparties entre EI et TPE selon leur forme juridique.)

Concu pour tourner une fois par jour (cf ENT_Communes.COM_DateMajStats) : idem
potent, recalcule et ecrase les compteurs a chaque execution, jamais un delta.

Usage :
  python3 ent_communes_stats.py
  python3 ent_communes_stats.py --departements 56,29
"""
import argparse
import os
import sys

import pymysql


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 compute_departement(conn, dep):
    table1 = f"Entreprises{dep}Phase1"
    # IMPORTANT : Entreprises<dep>Phase1 contient aussi des entreprises dont le
    # siege est dans un AUTRE departement (etablissement secondaire dans <dep> --
    # la "pollution" deja identifiee). On restreint donc explicitement aux
    # communes qui appartiennent reellement a <dep> (via ENT_Communes), pour
    # qu'un departement ne puisse jamais ecraser les stats d'une commune qui ne
    # lui appartient pas avec un sous-ensemble incomplet.
    sql = f"""
        SELECT
            t.ENT_CodeCommune,
            COALESCE(SUM(t.ENT_CategorieEntreprise = 'GE'), 0) AS ge,
            COALESCE(SUM(t.ENT_CategorieEntreprise = 'ETI'), 0) AS eti,
            COALESCE(SUM(t.ENT_CategorieEntreprise = 'PME'), 0) AS pme,
            COALESCE(SUM(t.ENT_CategorieEntreprise IS NULL AND LEFT(t.ENT_NatureJuridique, 1) = '1'), 0) AS ei,
            COALESCE(SUM(t.ENT_CategorieEntreprise IS NULL AND
                (t.ENT_NatureJuridique IS NULL OR LEFT(t.ENT_NatureJuridique, 1) != '1')), 0) AS tpe
        FROM `{table1}` t
        INNER JOIN ENT_Communes c
            ON c.COM_CodeInsee COLLATE utf8mb4_0900_ai_ci = t.ENT_CodeCommune AND c.COM_Departement = %s
        WHERE t.ENT_CodeCommune IS NOT NULL
        GROUP BY t.ENT_CodeCommune
    """
    with conn.cursor() as cur:
        cur.execute(sql, (dep,))
        rows = cur.fetchall()

    with conn.cursor() as cur:
        for code_commune, ge, eti, pme, ei, tpe in rows:
            cur.execute(
                "UPDATE ENT_Communes SET COM_EI=%s, COM_TPE=%s, COM_PME=%s, COM_ETI=%s, "
                "COM_GE=%s, COM_DateMajStats=NOW() WHERE COM_CodeInsee=%s",
                (ei, tpe, pme, eti, ge, code_commune),
            )
    log(f"  {dep} : {len(rows)} communes mises a jour")
    return len(rows)


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}")
    total = 0
    try:
        for dep in departements:
            total += compute_departement(conn, dep)
    finally:
        conn.close()
    log(f"Termine : {total} communes mises a jour au total")


if __name__ == "__main__":
    main()