seize-infra-doc/scripts/furious-to-zitadel/merge-duplicates.py

#!/usr/bin/env python3
"""
merge-duplicates.py — Fusionne les doublons après import.

Pour chaque doublon :
  1. Lit les metadata Furious sur le compte "nouveau"
  2. Les copie sur le compte "ancien" (à garder)
  3. Supprime le compte "nouveau"

⚠️ Adapter la variable DUPLICATES pour ton cas.

Auteur : Hedi Kalboussi
Date : 5 juin 2026
"""

import base64
import json
import logging
import time
from pathlib import Path

import jwt
import requests
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend

# ============================================================================
# CONFIGURATION — À ADAPTER
# ============================================================================

# Liste des doublons à fusionner
# Format : { "keep_user_id": <id_ancien>, "delete_email": <email_du_nouveau>, "furious_id": <id_furious> }
DUPLICATES = [
    # Exemple :
    # {
    #     "keep_user_id": "375119988404518915",
    #     "delete_email": "hedi.kalboussi@seizeconsulting.com",
    #     "furious_id": 5,
    # },
]

SCRIPT_DIR = Path(__file__).parent
KEY_FILE = SCRIPT_DIR / "zitadelfuriousimportkey.json"
RAW_DATA_FILE = SCRIPT_DIR / "furious-users-raw.json"

ZITADEL_DOMAIN = "https://sso.seizeconsulting.com"

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger(__name__)

# ============================================================================
# AUTH
# ============================================================================

def get_zitadel_token():
    with open(KEY_FILE) as f:
        key_data = json.load(f)
    private_key = serialization.load_pem_private_key(
        key_data["key"].encode(), password=None, backend=default_backend()
    )
    now = int(time.time())
    payload = {
        "iss": key_data["userId"],
        "sub": key_data["userId"],
        "aud": ZITADEL_DOMAIN,
        "iat": now,
        "exp": now + 3600,
    }
    assertion = jwt.encode(
        payload, private_key, algorithm="RS256", headers={"kid": key_data["keyId"]}
    )
    response = requests.post(
        f"{ZITADEL_DOMAIN}/oauth/v2/token",
        data={
            "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
            "scope": "openid profile urn:zitadel:iam:org:project:id:zitadel:aud",
            "assertion": assertion,
        },
    )
    response.raise_for_status()
    return response.json()["access_token"]

# ============================================================================
# OPÉRATIONS
# ============================================================================

def find_user_by_email(email, token):
    response = requests.post(
        f"{ZITADEL_DOMAIN}/v2/users",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "queries": [
                {"emailQuery": {"emailAddress": email, "method": "TEXT_QUERY_METHOD_EQUALS"}}
            ]
        },
    )
    response.raise_for_status()
    users = response.json().get("result", [])
    return users[0].get("userId") if users else None


def get_user_metadata(user_id, token):
    """Récupère toutes les metadata d'un user."""
    response = requests.get(
        f"{ZITADEL_DOMAIN}/management/v1/users/{user_id}/metadata/_search",
        headers={"Authorization": f"Bearer {token}"},
    )
    if not response.ok:
        return {}
    result = response.json().get("result", [])
    return {
        m["key"]: base64.b64decode(m["value"]).decode()
        for m in result
    }


def set_user_metadata(user_id, key, value, token):
    value_b64 = base64.b64encode(str(value).encode()).decode()
    response = requests.post(
        f"{ZITADEL_DOMAIN}/management/v1/users/{user_id}/metadata/{key}",
        headers={"Authorization": f"Bearer {token}"},
        json={"value": value_b64},
    )
    return response.ok


def delete_user(user_id, token):
    response = requests.delete(
        f"{ZITADEL_DOMAIN}/v2/users/{user_id}",
        headers={"Authorization": f"Bearer {token}"},
    )
    return response.ok

# ============================================================================
# MAIN
# ============================================================================

def main():
    if not DUPLICATES:
        log.error("Variable DUPLICATES vide. Adapter le script avant exécution.")
        return
    
    token = get_zitadel_token()
    log.info("✅ Token Zitadel OK")
    
    for dup in DUPLICATES:
        keep_id = dup["keep_user_id"]
        delete_email = dup["delete_email"]
        
        log.info(f"\n=== Doublon : keep={keep_id}, delete={delete_email} ===")
        
        # 1. Trouver le user "nouveau" à supprimer
        delete_id = find_user_by_email(delete_email, token)
        if not delete_id:
            log.warning(f"User {delete_email} pas trouvé (peut-être déjà fusionné)")
            continue
        
        # 2. Récupérer ses metadata
        metadata = get_user_metadata(delete_id, token)
        log.info(f"  Metadata récupérées : {list(metadata.keys())}")
        
        # 3. Les copier sur l'ancien
        for key, value in metadata.items():
            if value and set_user_metadata(keep_id, key, value, token):
                log.info(f"  → Copié {key}={value[:30]}...")
        
        # 4. Supprimer le nouveau
        if delete_user(delete_id, token):
            log.info(f"  ✅ Supprimé : {delete_email}")
        else:
            log.error(f"  💥 Échec suppression : {delete_email}")
        
        time.sleep(0.5)


if __name__ == "__main__":
    main()