🏦 Data Vault 2.0 — modéliser un entrepôt qui encaisse le changement

%pip install -q duckdb pandas

import re
import duckdb

con = duckdb.connect()  # base en mémoire : rien à installer, rien à nettoyer


def voir(sql):
    """Exécute une requête et affiche le résultat en tableau."""
    try:
        return con.sql(sql).df()
    except duckdb.CatalogException as e:  # table pas encore créée par un exercice
        print("⏳", str(e).splitlines()[0])


def val(sql):
    """Renvoie la première valeur d'une requête (pour les vérifications)."""
    ligne = con.sql(sql).fetchone()
    return ligne[0] if ligne else None


def executer(sql):
    """Exécute ton SQL — ou te prévient si l'exercice n'est pas encore rempli."""
    sans_commentaires = re.sub(r"--[^\n]*", "", sql).strip()
    if not sans_commentaires:
        print("⏳ Exercice pas encore fait : écris ton SQL à la place des commentaires,\n"
              "   puis relance la cellule.")
        return
    con.execute(sql)


def verifier(titre, tests):
    """Lance une liste de (description, test) et affiche ✅ / ❌ pour chacun."""
    tout_ok = True
    for description, test in tests:
        try:
            ok = bool(test())
        except Exception as e:  # table absente, colonne manquante…
            ok, description = False, f"{description}  ({type(e).__name__})"
        print(("✅ " if ok else "❌ ") + description)
        tout_ok &= ok
    if tout_ok:
        print(f"\n🎉 {titre} réussi !")
    else:
        print(f"\n↪️ {titre} : pas encore. Relis l'énoncé, ou ouvre la solution.")


print("✅ DuckDB", duckdb.__version__, "prêt")

con.execute("""
CREATE OR REPLACE TABLE stg_agences_clients AS
SELECT * FROM (VALUES
    -- lot du 5 janvier
    ('C-1042', 'Kouassi Ama',   'Abidjan',      DATE '2026-01-05'),
    ('C-1043', 'Traoré Moussa', 'Yamoussoukro', DATE '2026-01-05'),
    ('C-1045', 'Koné Awa',      'San-Pédro',    DATE '2026-01-05'),
    -- lot du 18 juin : Kouassi Ama a déménagé, les deux autres n'ont pas changé
    ('C-1042', 'Kouassi Ama',   'Bouaké',       DATE '2026-06-18'),
    ('C-1043', 'Traoré Moussa', 'Yamoussoukro', DATE '2026-06-18'),
    ('C-1045', 'Koné Awa',      'San-Pédro',    DATE '2026-06-18')
) AS t(numero_client, nom, ville, date_chargement);

CREATE OR REPLACE TABLE stg_mobile_clients AS
SELECT * FROM (VALUES
    -- lot du 11 février : numéros saisis à la main dans l'appli
    (' c-1042', '+225 07 00 00 00 01', 'ama.kouassi@mail.test',  DATE '2026-02-11'),
    ('C-1043',  '+225 05 00 00 00 02', NULL,                     DATE '2026-02-11'),
    ('C-1044 ', '+225 01 00 00 00 03', 'yao.bamba@mail.test',    DATE '2026-02-11')
) AS t(numero_client, telephone, email, date_chargement);

CREATE OR REPLACE TABLE stg_agences_comptes AS
SELECT * FROM (VALUES
    -- lot du 5 janvier : qui détient quel compte, et à quel titre
    ('CPT-501', 'C-1042', 'courant', 'actif', 'titulaire',    DATE '2026-01-05'),
    ('CPT-502', 'C-1042', 'épargne', 'actif', 'titulaire',    DATE '2026-01-05'),
    ('CPT-503', 'C-1045', 'courant', 'actif', 'titulaire',    DATE '2026-01-05'),
    -- lot du 2 avril : CPT-502 devient un compte joint, C-1043 en est co-titulaire
    ('CPT-501', 'C-1042', 'courant', 'actif', 'titulaire',    DATE '2026-04-02'),
    ('CPT-502', 'C-1042', 'épargne', 'actif', 'titulaire',    DATE '2026-04-02'),
    ('CPT-502', 'C-1043', 'épargne', 'actif', 'co-titulaire', DATE '2026-04-02'),
    ('CPT-503', 'C-1045', 'courant', 'actif', 'titulaire',    DATE '2026-04-02'),
    -- lot du 1er septembre : C-1043 a quitté le compte joint
    ('CPT-501', 'C-1042', 'courant', 'actif', 'titulaire',    DATE '2026-09-01'),
    ('CPT-502', 'C-1042', 'épargne', 'actif', 'titulaire',    DATE '2026-09-01'),
    ('CPT-503', 'C-1045', 'courant', 'actif', 'titulaire',    DATE '2026-09-01')
) AS t(numero_compte, numero_client, type_compte, statut, role, date_chargement);
""")

voir("SELECT * FROM stg_mobile_clients")

import hashlib


def md5_py(texte):
    return hashlib.md5(texte.encode()).hexdigest()


def lignes(sql):
    return con.sql(sql).fetchall()


# états attendus, réutilisés pour vérifier qu'une table n'a PAS bougé
HUB_CLIENTS = [("C-1042", "2026-01-05", "agences"), ("C-1043", "2026-01-05", "agences"),
               ("C-1044", "2026-02-11", "mobile"),  ("C-1045", "2026-01-05", "agences")]
VERSIONS_AGENCES = [("C-1042", "2026-01-05", "Kouassi Ama", "Abidjan"),
                    ("C-1042", "2026-06-18", "Kouassi Ama", "Bouaké"),
                    ("C-1043", "2026-01-05", "Traoré Moussa", "Yamoussoukro"),
                    ("C-1045", "2026-01-05", "Koné Awa", "San-Pédro")]


ETAT_1ER_MARS = [("C-1042", "Kouassi Ama", "Abidjan", "+225 07 00 00 00 01", "ama.kouassi@mail.test"),
                 ("C-1043", "Traoré Moussa", "Yamoussoukro", "+225 05 00 00 00 02", None),
                 ("C-1044", None, None, "+225 01 00 00 00 03", "yao.bamba@mail.test"),
                 ("C-1045", "Koné Awa", "San-Pédro", None, None)]


def versions_agences():
    return lignes("""SELECT h.numero_client, s.date_chargement::VARCHAR, s.nom, s.ville
                     FROM sat_client_agences s JOIN hub_client h USING (hk_client) ORDER BY 1, 2""")


def colonnes(table):
    """Les colonnes d'une table ou d'une vue, dans l'ordre."""
    return [r[0] for r in con.sql(f"SELECT column_name FROM information_schema.columns "
                                  f"WHERE table_name = '{table}' ORDER BY ordinal_position").fetchall()]


def _exercice_1():
    return [
        ("la table hub_client existe",
         lambda: val("SELECT COUNT(*) FROM information_schema.tables WHERE table_name = 'hub_client'") == 1),
        ("colonnes : hk_client, numero_client, date_chargement, source",
         lambda: colonnes("hub_client") == ["hk_client", "numero_client", "date_chargement", "source"]),
        ("hk_client est la clé primaire",
         lambda: val("""SELECT COUNT(*) FROM duckdb_constraints()
                        WHERE table_name = 'hub_client' AND constraint_type = 'PRIMARY KEY'
                          AND constraint_column_names = ['hk_client']""") == 1),
        ("aucune colonne n'accepte NULL",
         lambda: val("""SELECT COUNT(*) FROM information_schema.columns
                        WHERE table_name = 'hub_client' AND is_nullable = 'YES'""") == 0),
    ]


def _exercice_2():
    return [
        ("v_stg_agences_clients : bonnes colonnes",
         lambda: colonnes("v_stg_agences_clients") ==
                 ["hk_client", "numero_client", "nom", "ville", "date_chargement", "source"]),
        ("v_stg_mobile_clients : bonnes colonnes",
         lambda: colonnes("v_stg_mobile_clients") ==
                 ["hk_client", "numero_client", "telephone", "email", "date_chargement", "source"]),
        ("toutes les clés métier sont normalisées (C-xxxx)",
         lambda: val("""SELECT COUNT(*) FROM (SELECT numero_client FROM v_stg_agences_clients
                        UNION ALL SELECT numero_client FROM v_stg_mobile_clients)
                        WHERE NOT regexp_full_match(numero_client, 'C-[0-9]{4}')""") == 0),
        ("' c-1042' (mobile) et 'C-1042' (agences) ont la même clé de hachage",
         lambda: val("SELECT hk_client FROM v_stg_mobile_clients WHERE numero_client = 'C-1042'") ==
                 val("SELECT DISTINCT hk_client FROM v_stg_agences_clients WHERE numero_client = 'C-1042'")),
        ("la clé de hachage est bien le MD5 de la clé normalisée",
         lambda: val("SELECT COUNT(*) FROM v_stg_mobile_clients WHERE hk_client <> md5(numero_client)") == 0),
        ("la colonne source vaut 'agences' / 'mobile'",
         lambda: val("SELECT string_agg(DISTINCT source, ',') FROM v_stg_agences_clients") == "agences"
                 and val("SELECT string_agg(DISTINCT source, ',') FROM v_stg_mobile_clients") == "mobile"),
    ]


def _exercice_3():
    def hub():
        return con.sql("SELECT numero_client, date_chargement::VARCHAR, source "
                       "FROM hub_client ORDER BY numero_client").fetchall()

    def rejouer_tout():
        avant = hub()
        for vue, lot in LOTS_CLIENTS:
            charger_hub_client(vue, lot)
        return hub() == avant

    return [
        ("4 clients dans le hub, ni plus ni moins", lambda: len(hub()) == 4),
        ("chaque client garde la date et la source de sa première apparition", lambda: hub() == HUB_CLIENTS),
        ("rejouer les trois lots ne change rien (chargement idempotent)", rejouer_tout),
    ]


def _exercice_4():
    return [
        ("bonnes colonnes, dans le bon ordre",
         lambda: colonnes("v_stg_agences_comptes") ==
                 ["hk_lien", "hk_client", "hk_compte", "numero_client", "numero_compte",
                  "type_compte", "statut", "role", "date_chargement", "source"]),
        ("une ligne par ligne de staging",
         lambda: val("SELECT COUNT(*) FROM v_stg_agences_comptes") == val("SELECT COUNT(*) FROM stg_agences_comptes")),
        ("hk_lien = md5('C-1042||CPT-501') pour ce couple",
         lambda: val("""SELECT DISTINCT hk_lien FROM v_stg_agences_comptes
                        WHERE numero_client = 'C-1042' AND numero_compte = 'CPT-501'""") == md5_py("C-1042||CPT-501")),
        ("hk_client identique à celui du staging des clients",
         lambda: val("SELECT DISTINCT hk_client FROM v_stg_agences_comptes WHERE numero_client = 'C-1042'") ==
                 val("SELECT DISTINCT hk_client FROM v_stg_agences_clients WHERE numero_client = 'C-1042'")),
        ("hk_compte = md5 du numéro de compte",
         lambda: val("SELECT COUNT(*) FROM v_stg_agences_comptes WHERE hk_compte <> md5(numero_compte)") == 0),
        ("source = 'agences'",
         lambda: val("SELECT string_agg(DISTINCT source, ',') FROM v_stg_agences_comptes") == "agences"),
    ]


def _exercice_5():
    def couples():
        return lignes("""SELECT c.numero_client, k.numero_compte, l.date_chargement::VARCHAR
                         FROM link_client_compte l
                         JOIN hub_client c USING (hk_client) JOIN hub_compte k USING (hk_compte)
                         ORDER BY 1, 2""")

    def rejouer_tout():
        avant = (couples(), lignes("SELECT * FROM hub_compte ORDER BY 1"))
        for lot in LOTS_COMPTES:
            charger_hub_compte(lot)
            charger_link_client_compte(lot)
        return (couples(), lignes("SELECT * FROM hub_compte ORDER BY 1")) == avant

    return [
        ("hub_compte : 3 comptes, découverts le 5 janvier",
         lambda: lignes("SELECT numero_compte, date_chargement::VARCHAR FROM hub_compte ORDER BY 1") ==
                 [("CPT-501", "2026-01-05"), ("CPT-502", "2026-01-05"), ("CPT-503", "2026-01-05")]),
        ("link_client_compte ne contient que des clés (5 colonnes)",
         lambda: colonnes("link_client_compte") == ["hk_lien", "hk_client", "hk_compte", "date_chargement", "source"]),
        ("4 relations, dont le compte joint C-1043 ↔ CPT-502 découvert le 2 avril",
         lambda: couples() == [("C-1042", "CPT-501", "2026-01-05"), ("C-1042", "CPT-502", "2026-01-05"),
                               ("C-1043", "CPT-502", "2026-04-02"), ("C-1045", "CPT-503", "2026-01-05")]),
        ("chaque clé du link existe dans son hub",
         lambda: val("""SELECT COUNT(*) FROM link_client_compte l
                        WHERE l.hk_client NOT IN (SELECT hk_client FROM hub_client)
                           OR l.hk_compte NOT IN (SELECT hk_compte FROM hub_compte)""") == 0),
        ("rejouer les trois lots ne change rien", rejouer_tout),
    ]


def _exercice_6():
    return [
        ("empreinte('Kouassi Ama', 'Abidjan') = md5('KOUASSI AMA||ABIDJAN')",
         lambda: val("SELECT empreinte('Kouassi Ama', 'Abidjan')") == md5_py("KOUASSI AMA||ABIDJAN")),
        ("insensible aux espaces et à la casse",
         lambda: val("SELECT empreinte('  kouassi ama ', 'ABIDJAN') = empreinte('Kouassi Ama', 'Abidjan')")),
        ("jamais NULL, même si un attribut l'est",
         lambda: val("SELECT empreinte('+225 05 00 00 00 02', NULL) IS NOT NULL")),
        ("la position compte : empreinte('a', NULL) ≠ empreinte(NULL, 'a')",
         lambda: val("SELECT empreinte('a', NULL) <> empreinte(NULL, 'a')")),
        ("le séparateur est là : empreinte('ab', 'c') ≠ empreinte('a', 'bc')",
         lambda: val("SELECT empreinte('ab', 'c') <> empreinte('a', 'bc')")),
        ("accepte une date et un nombre",
         lambda: val("SELECT empreinte(DATE '2026-01-05', 3) IS NOT NULL")),
    ]


def _exercice_7():
    def rejouer_tout():
        avant = versions_agences()
        for lot in LOTS_AGENCES:
            charger_sat_client_agences(lot)
        return versions_agences() == avant

    return [
        ("colonnes : hk_client, date_chargement, hash_diff, nom, ville, source",
         lambda: colonnes("sat_client_agences") == ["hk_client", "date_chargement", "hash_diff", "nom", "ville", "source"]),
        ("clé primaire (hk_client, date_chargement)",
         lambda: val("""SELECT COUNT(*) FROM duckdb_constraints()
                        WHERE table_name = 'sat_client_agences' AND constraint_type = 'PRIMARY KEY'
                          AND constraint_column_names = ['hk_client', 'date_chargement']""") == 1),
        ("4 versions : le déménagement de C-1042 en ajoute une, les autres clients aucune",
         lambda: versions_agences() == VERSIONS_AGENCES),
        ("chaque hash_diff vaut empreinte(nom, ville)",
         lambda: val("SELECT COUNT(*) FROM sat_client_agences WHERE hash_diff <> empreinte(nom, ville)") == 0),
        ("rejouer les deux lots ne change rien", rejouer_tout),
    ]


def _exercice_8():
    def versions_mobile():
        return lignes("""SELECT h.numero_client, s.date_chargement::VARCHAR, s.telephone, s.email
                         FROM sat_client_mobile s JOIN hub_client h USING (hk_client) ORDER BY 1, 2""")

    def rejouer_tout():
        avant = versions_mobile()
        for lot in LOTS_MOBILE:
            charger_sat_client_mobile(lot)
        return versions_mobile() == avant

    return [
        ("colonnes : hk_client, date_chargement, hash_diff, telephone, email, source",
         lambda: colonnes("sat_client_mobile") == ["hk_client", "date_chargement", "hash_diff", "telephone", "email", "source"]),
        ("3 versions, une par client de l'appli (rattachées au hub)", lambda: len(versions_mobile()) == 3),
        ("aucune empreinte NULL (C-1043 n'a pas d'e-mail)",
         lambda: val("SELECT COUNT(*) FROM sat_client_mobile WHERE hash_diff IS NULL") == 0),
        ("hub_client n'a pas bougé", lambda: lignes("""SELECT numero_client, date_chargement::VARCHAR, source
                                                      FROM hub_client ORDER BY 1""") == HUB_CLIENTS),
        ("sat_client_agences n'a pas bougé", lambda: versions_agences() == VERSIONS_AGENCES),
        ("rejouer le lot ne change rien", rejouer_tout),
    ]


def _exercice_9():
    def roles():
        return lignes("""SELECT c.numero_client, k.numero_compte, s.date_chargement::VARCHAR, s.role
                         FROM sat_lien_client_compte s JOIN link_client_compte l USING (hk_lien)
                         JOIN hub_client c USING (hk_client) JOIN hub_compte k USING (hk_compte)
                         ORDER BY 1, 2, 3""")

    def rejouer_tout():
        avant = roles()
        for lot in LOTS_COMPTES:
            charger_sat_lien_client_compte(lot)
        return roles() == avant

    return [
        ("colonnes : hk_lien, date_chargement, hash_diff, role, source",
         lambda: colonnes("sat_lien_client_compte") == ["hk_lien", "date_chargement", "hash_diff", "role", "source"]),
        ("4 versions, dont C-1043 co-titulaire de CPT-502 depuis le 2 avril",
         lambda: roles() == [("C-1042", "CPT-501", "2026-01-05", "titulaire"),
                             ("C-1042", "CPT-502", "2026-01-05", "titulaire"),
                             ("C-1043", "CPT-502", "2026-04-02", "co-titulaire"),
                             ("C-1045", "CPT-503", "2026-01-05", "titulaire")]),
        ("chaque version est rattachée à une relation du link",
         lambda: val("""SELECT COUNT(*) FROM sat_lien_client_compte
                        WHERE hk_lien NOT IN (SELECT hk_lien FROM link_client_compte)""") == 0),
        ("chaque hash_diff vaut empreinte(role, NULL)",
         lambda: val("SELECT COUNT(*) FROM sat_lien_client_compte WHERE hash_diff <> empreinte(role, NULL)") == 0),
        ("rejouer les trois lots ne change rien", rejouer_tout),
    ]


def _exercice_10():
    def historique():
        return lignes("""SELECT c.numero_client, k.numero_compte, e.date_chargement::VARCHAR, e.actif
                         FROM sat_eff_client_compte e JOIN link_client_compte l USING (hk_lien)
                         JOIN hub_client c USING (hk_client) JOIN hub_compte k USING (hk_compte)
                         ORDER BY 1, 2, 3""")

    def actives_aujourdhui():
        return lignes("""SELECT c.numero_client, k.numero_compte
                         FROM (SELECT hk_lien, actif FROM sat_eff_client_compte
                               QUALIFY ROW_NUMBER() OVER (PARTITION BY hk_lien ORDER BY date_chargement DESC) = 1) e
                         JOIN link_client_compte l USING (hk_lien)
                         JOIN hub_client c USING (hk_client) JOIN hub_compte k USING (hk_compte)
                         WHERE e.actif ORDER BY 1, 2""")

    def rejouer_tout():
        avant = historique()
        for lot in LOTS_COMPTES:
            charger_sat_eff_client_compte(lot)
        return historique() == avant

    return [
        ("colonnes : hk_lien, date_chargement, actif, source",
         lambda: colonnes("sat_eff_client_compte") == ["hk_lien", "date_chargement", "actif", "source"]),
        ("C-1043 ↔ CPT-502 : actif le 2 avril, inactif le 1er septembre",
         lambda: [r for r in historique() if r[0] == "C-1043"] ==
                 [("C-1043", "CPT-502", "2026-04-02", True), ("C-1043", "CPT-502", "2026-09-01", False)]),
        ("5 lignes en tout : rien n'est réinséré quand le statut ne change pas", lambda: len(historique()) == 5),
        ("3 relations actives aujourd'hui",
         lambda: actives_aujourdhui() == [("C-1042", "CPT-501"), ("C-1042", "CPT-502"), ("C-1045", "CPT-503")]),
        ("le link a gardé ses 4 relations : rien n'a été supprimé",
         lambda: val("SELECT COUNT(*) FROM link_client_compte") == 4),
        ("rejouer les trois lots ne change rien", rejouer_tout),
    ]


def _exercice_11():
    def etat(d):
        return lignes(f"SELECT * FROM etat_clients_au(DATE '{d}') ORDER BY numero_client")

    return [
        ("colonnes : numero_client, nom, ville, telephone, email",
         lambda: [c[0] for c in con.sql("SELECT * FROM etat_clients_au(DATE '2026-03-01')").description] ==
                 ["numero_client", "nom", "ville", "telephone", "email"]),
        ("au 1er mars : 4 clients, Kouassi Ama à Abidjan, C-1044 connu par l'appli seulement",
         lambda: etat("2026-03-01") == ETAT_1ER_MARS),
        ("au 1er juillet : Kouassi Ama est à Bouaké",
         lambda: [r[2] for r in etat("2026-07-01") if r[0] == "C-1042"] == ["Bouaké"]),
        ("au 10 janvier : C-1044 n'est pas encore connu, et aucun téléphone n'est arrivé",
         lambda: [r[0] for r in etat("2026-01-10")] == ["C-1042", "C-1043", "C-1045"]
                 and all(r[3] is None for r in etat("2026-01-10"))),
    ]


def _exercice_12():
    def via_pit(d):
        return lignes(f"""SELECT h.numero_client, a.nom, a.ville, m.telephone, m.email
                          FROM pit_client p JOIN hub_client h USING (hk_client)
                          LEFT JOIN sat_client_agences a
                                 ON a.hk_client = p.hk_client AND a.date_chargement = p.date_agences
                          LEFT JOIN sat_client_mobile m
                                 ON m.hk_client = p.hk_client AND m.date_chargement = p.date_mobile
                          WHERE p.date_snapshot = DATE '{d}' ORDER BY 1""")

    def pareil_partout():
        for (d,) in lignes("SELECT date_snapshot::VARCHAR FROM snapshots"):
            if via_pit(d) != lignes(f"SELECT * FROM etat_clients_au(DATE '{d}') ORDER BY 1"):
                return False
        return True

    return [
        ("colonnes : hk_client, date_snapshot, date_agences, date_mobile",
         lambda: colonnes("pit_client") == ["hk_client", "date_snapshot", "date_agences", "date_mobile"]),
        ("35 lignes : chaque client, à chaque date où il est déjà connu",
         lambda: val("SELECT COUNT(*) FROM pit_client") == 35),
        ("C-1042 au 1er juillet : version des agences du 18 juin, version mobile du 11 février",
         lambda: lignes("""SELECT p.date_agences::VARCHAR, p.date_mobile::VARCHAR FROM pit_client p
                           JOIN hub_client h USING (hk_client)
                           WHERE h.numero_client = 'C-1042' AND p.date_snapshot = DATE '2026-07-01'""") ==
                 [("2026-06-18", "2026-02-11")]),
        ("C-1045 au 1er juillet : aucune version mobile (NULL)",
         lambda: val("""SELECT p.date_mobile IS NULL FROM pit_client p JOIN hub_client h USING (hk_client)
                        WHERE h.numero_client = 'C-1045' AND p.date_snapshot = DATE '2026-07-01'""")),
        ("à chaque date, la PIT donne le même état que ta macro etat_clients_au (exercice 11)", pareil_partout),
    ]


def _exercice_13():
    return [
        ("colonnes : numero_client, nom, ville, telephone, email, nb_comptes_actifs",
         lambda: colonnes("dim_client") ==
                 ["numero_client", "nom", "ville", "telephone", "email", "nb_comptes_actifs"]),
        ("une ligne par client (4)",
         lambda: val("SELECT COUNT(*) FROM dim_client") == 4
                 and val("SELECT COUNT(DISTINCT numero_client) FROM dim_client") == 4),
        ("Kouassi Ama : à Bouaké, avec son téléphone, et 2 comptes actifs",
         lambda: lignes("SELECT ville, telephone, nb_comptes_actifs FROM dim_client WHERE numero_client = 'C-1042'") ==
                 [("Bouaké", "+225 07 00 00 00 01", 2)]),
        ("C-1043 a quitté le compte joint : 0 compte actif",
         lambda: val("SELECT nb_comptes_actifs FROM dim_client WHERE numero_client = 'C-1043'") == 0),
        ("C-1044, connu par l'appli seulement : pas de nom, 0 compte",
         lambda: lignes("SELECT nom, nb_comptes_actifs FROM dim_client WHERE numero_client = 'C-1044'") == [(None, 0)]),
    ]


def _exercice_14():
    def historique():
        return lignes("""SELECT numero_client, ville, valide_du::VARCHAR, valide_au::VARCHAR, est_courant
                         FROM dim_client_historique ORDER BY numero_client, valide_du""")

    return [
        ("colonnes : numero_client, nom, ville, valide_du, valide_au, est_courant",
         lambda: colonnes("dim_client_historique") ==
                 ["numero_client", "nom", "ville", "valide_du", "valide_au", "est_courant"]),
        ("C-1042 : Abidjan jusqu'au 18 juin, puis Bouaké (version courante)",
         lambda: [r for r in historique() if r[0] == "C-1042"] ==
                 [("C-1042", "Abidjan", "2026-01-05", "2026-06-18", False),
                  ("C-1042", "Bouaké", "2026-06-18", None, True)]),
        ("une seule version courante par client",
         lambda: val("""SELECT COUNT(*) FROM (SELECT numero_client FROM dim_client_historique
                        WHERE est_courant GROUP BY 1 HAVING COUNT(*) <> 1)""") == 0
                 and val("SELECT COUNT(*) FROM dim_client_historique WHERE est_courant") == 3),
        ("une version courante n'a pas de fin, une ancienne en a une",
         lambda: val("""SELECT COUNT(*) FROM dim_client_historique
                        WHERE est_courant <> (valide_au IS NULL)""") == 0),
    ]


def verifier_exercice(n):
    verifier(f"Exercice {n}", globals()[f"_exercice_{n}"]())


print("✅ Vérifications chargées")

executer("""
-- À TOI : CREATE TABLE hub_client ( ... )
""")

verifier_exercice(1)

voir("""
SELECT 'C-1042'  AS saisie, md5('C-1042')  AS hk UNION ALL
SELECT ' c-1042',           md5(' c-1042')
""")

executer("""
-- À TOI : CREATE OR REPLACE VIEW v_stg_agences_clients AS SELECT ... ;
-- À TOI : CREATE OR REPLACE VIEW v_stg_mobile_clients  AS SELECT ... ;
""")

verifier_exercice(2)

def charger_hub_client(vue, lot):
    executer(f"""
    -- À TOI : INSERT INTO hub_client SELECT ...
    --   depuis {vue}, pour le lot DATE '{lot}',
    --   uniquement les clés absentes du hub
    """)


LOTS_CLIENTS = [
    ("v_stg_agences_clients", "2026-01-05"),
    ("v_stg_mobile_clients",  "2026-02-11"),
    ("v_stg_agences_clients", "2026-06-18"),
]
for vue, lot in LOTS_CLIENTS:
    charger_hub_client(vue, lot)

voir("SELECT * FROM hub_client ORDER BY numero_client")

verifier_exercice(3)

voir("""
SELECT md5('AB' || 'C')         = md5('A' || 'BC')         AS sans_separateur_meme_cle,
       md5('AB' || '||' || 'C') = md5('A' || '||' || 'BC') AS avec_separateur_meme_cle
""")

executer("""
-- À TOI : CREATE OR REPLACE VIEW v_stg_agences_comptes AS SELECT ...
""")

verifier_exercice(4)

executer("""
-- À TOI : CREATE OR REPLACE TABLE hub_compte (...)
--         (même recette que hub_client)
-- À TOI : CREATE OR REPLACE TABLE link_client_compte (...)
--         (5 colonnes, que des clés)
""")


def charger_hub_compte(lot):
    executer(f"""
    -- À TOI : les comptes absents de hub_compte, pour le lot DATE '{lot}'
    """)


def charger_link_client_compte(lot):
    executer(f"""
    -- À TOI : les couples (client, compte) absents du link,
    --         pour le lot DATE '{lot}'
    """)


LOTS_COMPTES = ["2026-01-05", "2026-04-02", "2026-09-01"]
for lot in LOTS_COMPTES:
    charger_hub_compte(lot)
    charger_link_client_compte(lot)

voir("""
SELECT c.numero_client, k.numero_compte, l.date_chargement, l.source
FROM link_client_compte l
JOIN hub_client c USING (hk_client)
JOIN hub_compte k USING (hk_compte)
ORDER BY 1, 2
""")

verifier_exercice(5)

voir("""
SELECT
    -- avec || : un seul NULL et toute l'empreinte devient NULL
    'Abidjan' || NULL                  AS avec_barres,
    -- avec concat_ws : deux lignes différentes donnent le même texte
    concat_ws('||', 'a', NULL, 'b')    AS concat_ws_1,
    concat_ws('||', 'a', 'b', NULL)    AS concat_ws_2
""")

executer("""
-- À TOI : CREATE OR REPLACE MACRO empreinte(a, b) AS ...
""")

verifier_exercice(6)

executer("""
-- À TOI : CREATE OR REPLACE TABLE sat_client_agences (...)
--         clé primaire : PRIMARY KEY (hk_client, date_chargement)
""")


def charger_sat_client_agences(lot):
    executer(f"""
    -- À TOI : INSERT INTO sat_client_agences
    --   les clients de v_stg_agences_clients pour le lot DATE '{lot}'
    --   dont l'empreinte diffère de leur dernière version connue à cette date
    """)


LOTS_AGENCES = ["2026-01-05", "2026-06-18"]
for lot in LOTS_AGENCES:
    charger_sat_client_agences(lot)

voir("""
SELECT h.numero_client, s.date_chargement, s.nom, s.ville, s.hash_diff
FROM sat_client_agences s JOIN hub_client h USING (hk_client)
ORDER BY 1, 2
""")

verifier_exercice(7)

executer("""
-- À TOI : CREATE OR REPLACE TABLE sat_client_mobile (...)
""")


def charger_sat_client_mobile(lot):
    executer(f"""
    -- À TOI : même recette que charger_sat_client_agences,
    --         sur v_stg_mobile_clients
    """)


LOTS_MOBILE = ["2026-02-11"]
for lot in LOTS_MOBILE:
    charger_sat_client_mobile(lot)

voir("""
SELECT h.numero_client, s.date_chargement, s.telephone, s.email, s.hash_diff
FROM sat_client_mobile s JOIN hub_client h USING (hk_client)
ORDER BY 1
""")

verifier_exercice(8)

con.execute("""
CREATE OR REPLACE TABLE sat_compte (
    hk_compte       VARCHAR NOT NULL,
    date_chargement DATE    NOT NULL,
    hash_diff       VARCHAR NOT NULL,
    type_compte     VARCHAR,
    statut          VARCHAR,
    source          VARCHAR NOT NULL,
    PRIMARY KEY (hk_compte, date_chargement)
)
""")


def charger_sat_compte(lot):
    con.execute(f"""
    INSERT INTO sat_compte
    SELECT s.hk_compte, s.date_chargement, empreinte(s.type_compte, s.statut),
           s.type_compte, s.statut, s.source
    FROM (SELECT DISTINCT hk_compte, date_chargement, type_compte, statut, source
          FROM v_stg_agences_comptes) s
    LEFT JOIN (
        SELECT hk_compte, hash_diff
        FROM sat_compte
        WHERE date_chargement <= DATE '{lot}'
        QUALIFY ROW_NUMBER() OVER (
            PARTITION BY hk_compte ORDER BY date_chargement DESC) = 1
    ) derniere USING (hk_compte)
    WHERE s.date_chargement = DATE '{lot}'
      AND (derniere.hash_diff IS NULL
           OR derniere.hash_diff <> empreinte(s.type_compte, s.statut))
    """)


try:
    for lot in LOTS_COMPTES:
        charger_sat_compte(lot)
except duckdb.Error:
    print("⏳ Cette cellule s'appuie sur les exercices 4 à 6 :\n"
          "   termine-les, puis relance-la.")

voir("""
SELECT k.numero_compte, s.* EXCLUDE (hk_compte)
FROM sat_compte s JOIN hub_compte k USING (hk_compte)
ORDER BY 1
""")

executer("""
-- À TOI : CREATE OR REPLACE TABLE sat_lien_client_compte (...)
""")


def charger_sat_lien_client_compte(lot):
    executer(f"""
    -- À TOI : même recette que les satellites de hub,
    --         avec hk_lien comme clé du parent
    """)


for lot in LOTS_COMPTES:
    charger_sat_lien_client_compte(lot)

voir("""
SELECT c.numero_client, k.numero_compte, s.date_chargement, s.role
FROM sat_lien_client_compte s
JOIN link_client_compte l USING (hk_lien)
JOIN hub_client c USING (hk_client)
JOIN hub_compte k USING (hk_compte)
ORDER BY 1, 2
""")

verifier_exercice(9)

executer("""
-- À TOI : CREATE OR REPLACE TABLE sat_eff_client_compte (...)
""")


def charger_sat_eff_client_compte(lot):
    executer(f"""
    -- À TOI : pour chaque relation du link connue au DATE '{lot}',
    --   actif = la relation est-elle dans le lot ?
    --   insérer seulement si ce statut diffère du dernier connu
    """)


for lot in LOTS_COMPTES:
    charger_sat_eff_client_compte(lot)

voir("""
SELECT c.numero_client, k.numero_compte, e.date_chargement, e.actif
FROM sat_eff_client_compte e
JOIN link_client_compte l USING (hk_lien)
JOIN hub_client c USING (hk_client)
JOIN hub_compte k USING (hk_compte)
ORDER BY 1, 2, 3
""")

verifier_exercice(10)

executer("""
-- À TOI : CREATE OR REPLACE MACRO etat_clients_au(d) AS TABLE
--         WITH agences AS (...), mobile AS (...)
--         SELECT ... FROM hub_client ...
""")

voir("SELECT * FROM etat_clients_au(DATE '2026-03-01')")

verifier_exercice(11)

con.execute("""
CREATE OR REPLACE TABLE snapshots AS
SELECT CAST(generate_series AS DATE) AS date_snapshot
FROM generate_series(DATE '2026-01-01', DATE '2026-10-01', INTERVAL 1 MONTH)
""")
voir("SELECT * FROM snapshots")

executer("""
-- À TOI : CREATE OR REPLACE TABLE pit_client AS
--         SELECT h.hk_client, s.date_snapshot, (...) AS date_agences, (...) AS date_mobile
--         FROM hub_client h JOIN snapshots s ON ...
""")

# L'état au 1er juillet, avec la PIT : rien que des égalités
voir("""
SELECT h.numero_client, a.nom, a.ville, m.telephone, m.email
FROM pit_client p
JOIN hub_client h USING (hk_client)
LEFT JOIN sat_client_agences a
       ON a.hk_client = p.hk_client AND a.date_chargement = p.date_agences
LEFT JOIN sat_client_mobile m
       ON m.hk_client = p.hk_client AND m.date_chargement = p.date_mobile
WHERE p.date_snapshot = DATE '2026-07-01'
ORDER BY 1
""")

verifier_exercice(12)

voir("""
SELECT table_name
FROM information_schema.tables
WHERE regexp_matches(table_name, '^(hub|link|sat)_')
ORDER BY table_name
""")

executer("""
-- À TOI : CREATE OR REPLACE VIEW dim_client AS
--         l'état d'aujourd'hui + le nombre de comptes actifs de chaque client
""")

voir("SELECT * FROM dim_client")

verifier_exercice(13)

executer("""
-- À TOI : CREATE OR REPLACE VIEW dim_client_historique AS
--         une ligne par version de sat_client_agences, avec ses dates de validité
""")

voir("SELECT * FROM dim_client_historique ORDER BY numero_client, valide_du")

verifier_exercice(14)