abdelkouddous/Get_All_Users
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Config Env / Phases

Browse Source
This commit is contained in:
Abdelkouddous LHACHIMI
2025-12-15 00:00:22 +01:00
parent ef174f3276
commit b3b91ad1d8
1 changed files with 114 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

200
get_all_users.py
View File

@@ -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(f"Fetching profiles for {app_name}...")
profiles_response = get_profiles_by_app_id(client_id)
profiles_data = profiles_response.get("data", [])
console.print("Fetching roles...")
roles_response = get_roles()
roles_data = roles_response.get("data", [])
output_data_app = {}
output_data_roles = {}
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")
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 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}")
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")
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 ==========