import os import requests import time import sys METABASE_URL = os.getenv("METABASE_URL", "http://127.0.0.1:3000") # Admin Metabase ADMIN_FIRST_NAME = os.getenv("MB_FIRST_NAME", "Admin") ADMIN_LAST_NAME = os.getenv("MB_LAST_NAME", "User") ADMIN_EMAIL = os.getenv("MB_EMAIL", "admin@example.com") ADMIN_PASSWORD = os.getenv("MB_PASSWORD", "sUperm0tdep@ss3") # Site prefs SITE_NAME = os.getenv("MB_SITE_NAME", "My Metabase") SITE_LOCALE = os.getenv("MB_SITE_LOCALE", "fr") # Database config DB_ENGINE = os.getenv("MB_DB_ENGINE", "postgres") DB_NAME = os.getenv("MB_DB_NAME", "sql") DB_HOST = os.getenv("MB_DB_HOST", "database") DB_PORT = os.getenv("MB_DB_PORT", "5432") DB_USER = os.getenv("MB_DB_USER", "metabase_user") DB_PASS = os.getenv("MB_DB_PASS", "supermotdepasse") ADMIN = { "first_name": ADMIN_FIRST_NAME, "last_name": ADMIN_LAST_NAME, "email": ADMIN_EMAIL, "password": ADMIN_PASSWORD } DB = { "engine": DB_ENGINE, "name": DB_NAME, "is_on_demand": False, "is_full_sync": True, "is_sample": False, "details": { "host": DB_HOST, "port": DB_PORT, "dbname": DB_NAME, "user": DB_USER, "password": DB_PASS, "ssl": False } } def wait_for_metabase(timeout=300): print("⏳ Attente de Metabase...") start = time.time() while time.time() - start < timeout: try: r = requests.get(f"{METABASE_URL}/api/health", timeout=2) if r.ok: print("✅ Metabase est prêt") return True except requests.RequestException: pass time.sleep(5) print("❌ Timeout en attendant Metabase") return False def get_setup_token(): r = requests.get(f"{METABASE_URL}/api/session/properties") r.raise_for_status() print("✅ setup-token") return r.json().get("setup-token") def run_setup(token): payload = { "token": token, "user": ADMIN, "prefs": {"site_name": "IUT", "site_locale": "fr"} } r = requests.post(f"{METABASE_URL}/api/setup", json=payload) if r.status_code == 403 and "existe" in r.text.lower(): return login(ADMIN_EMAIL, ADMIN_PASSWORD) r.raise_for_status() return r.json()["id"] def login(email, password): r = requests.post(f"{METABASE_URL}/api/session", json={"username": email, "password": password}) r.raise_for_status() return r.json()["id"] def database_exists(session_id, name): headers = {"X-Metabase-Session": session_id} r = requests.get(f"{METABASE_URL}/api/database", headers=headers) r.raise_for_status() for db in r.json()["data"]: if db["name"].lower() == name.lower(): print(db["id"]) return db["id"] return None def add_database(session_id, db_config): db_id = database_exists(session_id, db_config["name"]) if db_id: print(f"ℹ️ Base '{db_config['name']}' existe déjà (id={db_id})") return db_id headers = {"X-Metabase-Session": session_id, "Content-Type": "application/json"} r = requests.post(f"{METABASE_URL}/api/database", headers=headers, json=DB) r.raise_for_status() db_id = r.json()["id"] print(f"🗄️ Base '{db_config['name']}' ajoutée (id={db_id})") return db_id def get_or_create_collection(session_id, name, parent_id=None): headers = {"X-Metabase-Session": session_id} r = requests.get(f"{METABASE_URL}/api/collection", headers=headers) r.raise_for_status() for coll in r.json(): if coll["name"].lower() == name.lower(): print(f"ℹ️ Collection '{name}' existe déjà (id={coll['id']})") return coll["id"] payload = {"name": name, "color": "#509EE3"} if parent_id: payload["parent_id"] = parent_id r = requests.post(f"{METABASE_URL}/api/collection", headers=headers, json=payload) r.raise_for_status() print(f"📁 Collection '{name}' créée (id={r.json()['id']})") return r.json()["id"] def get_or_create_question(session_id, name, db_id, sql, collection_id): headers = {"X-Metabase-Session": session_id} r = requests.get(f"{METABASE_URL}/api/card", headers=headers) r.raise_for_status() for q in r.json(): if q["name"].lower() == name.lower(): print(f"ℹ️ Question '{name}' existe déjà (id={q['id']})") return q["id"] payload = { "name": name, "dataset_query": { "database": db_id, "type": "native", "native": {"query": sql} }, "display": "table", "collection_id": collection_id } r = requests.post(f"{METABASE_URL}/api/card", headers=headers, json=payload) r.raise_for_status() print(f"❓ Question '{name}' créée (id={r.json()['id']})") return r.json()["id"] def get_or_create_dashboard(session_id, name, collection_id): headers = {"X-Metabase-Session": session_id} r = requests.get(f"{METABASE_URL}/api/dashboard", headers=headers) r.raise_for_status() for d in r.json(): if d["name"].lower() == name.lower(): print(f"ℹ️ Dashboard '{name}' existe déjà (id={d['id']})") return d["id"] payload = {"name": name, "collection_id": collection_id} r = requests.post(f"{METABASE_URL}/api/dashboard", headers=headers, json=payload) r.raise_for_status() print(f"📊 Dashboard '{name}' créé (id={r.json()['id']})") return r.json()["id"] def add_question_to_dashboard(session_id, dashboard_id, card_id): headers = {"X-Metabase-Session": session_id} payload = {"cardId": card_id} r = requests.post(f"{METABASE_URL}/api/dashboard/{dashboard_id}/cards", headers=headers, json=payload) if r.status_code == 400 and "already" in r.text.lower(): print(f"ℹ️ Question {card_id} déjà liée au dashboard {dashboard_id}") return r.raise_for_status() print(f"➕ Question {card_id} ajoutée au dashboard {dashboard_id}") if __name__ == "__main__": if not wait_for_metabase(): sys.exit(1) token = get_setup_token() session = run_setup(token) db_id = add_database(session, DB) coll_id = get_or_create_collection(session, "IUT Dashboard") # Exemple : une question SQL q1 = get_or_create_question( session, "Nombre de lignes", db_id, "SELECT COUNT(*) AS total FROM information_schema.tables;", coll_id ) # Exemple : un dashboard dash_id = get_or_create_dashboard(session, "Vue d'ensemble", coll_id) add_question_to_dashboard(session, dash_id, q1) print("🎉 Initialisation Metabase terminée avec dashboards/questions")