From b3b91ad1d8332b0d27497e66d17fc75786886abf Mon Sep 17 00:00:00 2001 From: Abdelkouddous LHACHIMI Date: Mon, 15 Dec 2025 00:00:22 +0100 Subject: [PATCH] Config Env / Phases --- get_all_users.py | 204 ++++++++++++++++++++++++++--------------------- 1 file changed, 114 insertions(+), 90 deletions(-) diff --git a/get_all_users.py b/get_all_users.py index 3b4b375..72811af 100644 --- a/get_all_users.py +++ b/get_all_users.py @@ -28,9 +28,17 @@ from rich.console import Console # CONFIGURATION - CREDENTIALS # ============================================================================ -DEFAULT_USER_NAME = "admin" -DEFAULT_PASSWORD = "+J3/rw..'ynxXDHwt?bAvn_>" -REALME = "ziwig-pro" +ENV = "pre-prod" # "prod" or "pre-prod" +if "prod" in sys.argv: + ENV = "prod" + sys.argv.remove("prod") +elif "pre-prod" in sys.argv: + ENV = "pre-prod" + sys.argv.remove("pre-prod") + +DEFAULT_USER_NAME = "admin" if ENV == "prod" else "admin" +DEFAULT_PASSWORD = "+J3/rw..'ynxXDHwt?bAvn_>" if ENV == "prod" else "123@Admin" +REALME = "ziwig-pro" if ENV == "prod" else "ziwig-pro" # ============================================================================ @@ -41,7 +49,7 @@ REALME = "ziwig-pro" MICROSERVICES = { "IAM": { "app_id": None, # IAM doesn't use app_id - "base_url": "https://api-auth.ziwig-connect.com", + "base_url": "https://api-auth.ziwig-connect.com" if ENV == "prod" else "https://api-iam.pp.ziwig-platform.com", # PRE-PROD "endpoints": { "login": "/api/auth/{REALME}/login", # POST : Body = {"username": "{user_name}", "password": "{pass}"} "refresh": "/api/auth/refreshToken", # POST : Body = {"refresh_token": "{refresh_token}"} @@ -75,8 +83,8 @@ MICROSERVICES = { # } # }, "HRD": { - "app_id": "93bc44fd-c64b-4fff-a450-f3cba956e934", - "base_url": "https://api-resources.ziwig-connect.com", + "app_id": "93bc44fd-c64b-4fff-a450-f3cba956e934" if ENV == "prod" else "234d72f1-509a-4e99-b2b9-ee12da1e336d", + "base_url": "https://api-resources.ziwig-connect.com" if ENV == "prod" else "https://api-hrd.pp.ziwig-platform.com", "endpoints": { "config_token": "/api/auth/config-token", # POST : Body = {"userId": "{user_id}", "clientId": "{app_id}", "userAgent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36"}} "refresh": "/api/auth/refreshToken", # POST : Body = {"refresh_token": "{refresh_token}"} @@ -773,6 +781,19 @@ def main(): """ global main_thread_pool, subtasks_thread_pool, number_of_threads + phase1 = True + phase2 = True + phase3 = True + + if len(sys.argv) > 1: + if "1" not in sys.argv: + phase1 = False + if "2" not in sys.argv: + phase2 = False + if "3" not in sys.argv: + phase3 = False + + # ========== AUTHENTICATION ========== print() login_status = login() @@ -786,7 +807,7 @@ def main(): number_of_threads = int( questionary.text( "Number of threads:", - default="12", + default="5", validate=lambda x: x.isdigit() and 0 < int(x) <= MAX_THREADS ).ask() ) @@ -799,100 +820,103 @@ def main(): init_subtasks_pool() # ========== PHASE 1: ROLES ========== - print() - console.print("==================================================") - console.print("[bold cyan]PHASE 1: Processing Roles[/bold cyan]") - console.print("==================================================") - - console.print("Fetching roles...") - roles_response = get_roles() - roles_data = roles_response.get("data", []) - - output_data_roles = {} - - console.print("Initializing user list from roles...") - for role in roles_data: - role_info = {"id": role.get("id"), "name": role.get("name")} - users = role.get("users", []) - - for user_id in users: - if user_id not in output_data_roles: - output_data_roles[user_id] = { - "roles": [role_info], - "user": {}, - "professional": {} - } - else: - existing_role_ids = [r["id"] for r in output_data_roles[user_id]["roles"]] - if role_info["id"] not in existing_role_ids: - output_data_roles[user_id]["roles"].append(role_info) - - process_user_list(output_data_roles, "Roles") - - - # ========== PHASE 2: APPLICATIONS ========== - print() - console.print("==================================================") - console.print("[bold cyan]PHASE 2: Processing Applications[/bold cyan]") - console.print("==================================================") - - console.print("Fetching applications...") - apps_response = get_applications() - # apps_response is a list of dicts - - for app in apps_response: - app_name = app.get("label") or app.get("name") or "Unknown" - client_id = app.get("clientId") - - if not client_id: - continue - + if phase1: print() - console.print(f"[bold magenta]--- Application: {app_name} ---[/bold magenta]") + console.print("==================================================") + console.print("[bold cyan]PHASE 1: Processing Roles[/bold cyan]") + console.print("==================================================") + + console.print("Fetching roles...") + roles_response = get_roles() + roles_data = roles_response.get("data", []) - console.print(f"Fetching profiles for {app_name}...") - profiles_response = get_profiles_by_app_id(client_id) - profiles_data = profiles_response.get("data", []) - - output_data_app = {} - - console.print(f"Initializing user list for {app_name}...") - for profile in profiles_data: - profile_info = {"id": profile.get("id"), "name": profile.get("name")} - profile_id = profile.get("id") + output_data_roles = {} + + console.print("Initializing user list from roles...") + for role in roles_data: + role_info = {"id": role.get("id"), "name": role.get("name")} + users = role.get("users", []) - # Fetch users for this profile - # Note: The API returns a list of user IDs directly? - # Spec check: "get_users_by_profile_id : la réponse est un simple array de user_id" - try: - users_list = get_users_by_profile_id(profile_id) - # Ensure it's a list - if not isinstance(users_list, list): - users_list = [] - except Exception as e: - logging.error(f"Error fetching users for profile {profile_id}: {e}") - users_list = [] - - for user_id in users_list: - if user_id not in output_data_app: - output_data_app[user_id] = { - "profiles": [profile_info], + for user_id in users: + if user_id not in output_data_roles: + output_data_roles[user_id] = { + "roles": [role_info], "user": {}, "professional": {} } else: - existing_profile_ids = [p["id"] for p in output_data_app[user_id]["profiles"]] - if profile_info["id"] not in existing_profile_ids: - output_data_app[user_id]["profiles"].append(profile_info) + existing_role_ids = [r["id"] for r in output_data_roles[user_id]["roles"]] + if role_info["id"] not in existing_role_ids: + output_data_roles[user_id]["roles"].append(role_info) + + process_user_list(output_data_roles, "Roles") + + + # ========== PHASE 2: APPLICATIONS ========== + if phase2: + print() + console.print("==================================================") + console.print("[bold cyan]PHASE 2: Processing Applications[/bold cyan]") + console.print("==================================================") + + console.print("Fetching applications...") + apps_response = get_applications() + # apps_response is a list of dicts + + for app in apps_response: + app_name = app.get("label") or app.get("name") or "Unknown" + client_id = app.get("clientId") - # Process this application's users - # Sanitize app_name for filename - safe_app_name = "".join([c if c.isalnum() else "_" for c in app_name]) - process_user_list(output_data_app, f"Application {app_name}", f"_{safe_app_name}") + if not client_id: + continue + + print() + console.print(f"[bold magenta]--- Application: {app_name} ---[/bold magenta]") + + console.print(f"Fetching profiles for {app_name}...") + profiles_response = get_profiles_by_app_id(client_id) + profiles_data = profiles_response.get("data", []) + + output_data_app = {} + + console.print(f"Initializing user list for {app_name}...") + for profile in profiles_data: + profile_info = {"id": profile.get("id"), "name": profile.get("name")} + profile_id = profile.get("id") + + # Fetch users for this profile + # Note: The API returns a list of user IDs directly? + # Spec check: "get_users_by_profile_id : la réponse est un simple array de user_id" + try: + users_list = get_users_by_profile_id(profile_id) + # Ensure it's a list + if not isinstance(users_list, list): + users_list = [] + except Exception as e: + logging.error(f"Error fetching users for profile {profile_id}: {e}") + users_list = [] + + for user_id in users_list: + if user_id not in output_data_app: + output_data_app[user_id] = { + "profiles": [profile_info], + "user": {}, + "professional": {} + } + else: + existing_profile_ids = [p["id"] for p in output_data_app[user_id]["profiles"]] + if profile_info["id"] not in existing_profile_ids: + output_data_app[user_id]["profiles"].append(profile_info) + + # Process this application's users + # Sanitize app_name for filename + safe_app_name = "".join([c if c.isalnum() else "_" for c in app_name]) + process_user_list(output_data_app, f"Application {app_name}", f"_{safe_app_name}") # ========== PHASE 3: ENDOBEST CENTERS ========== - process_endobest_centers() + if phase3: + process_endobest_centers() # ========== FINALIZATION ==========