104 lines
2.8 KiB
Python
104 lines
2.8 KiB
Python
"""
|
|
Migration script to update existing admin users from old roles to new roles.
|
|
|
|
Old roles -> New roles:
|
|
superadmin -> sysadmin
|
|
melody_editor -> editor (with melodies full access)
|
|
device_manager -> editor (with devices + equipment full access)
|
|
user_manager -> editor (with app_users full access)
|
|
viewer -> user (view-only)
|
|
|
|
Usage:
|
|
python migrate_roles.py
|
|
python migrate_roles.py --dry-run (preview changes without applying)
|
|
"""
|
|
import argparse
|
|
import sys
|
|
from shared.firebase import init_firebase, get_db
|
|
|
|
|
|
ROLE_MAPPING = {
|
|
"superadmin": "sysadmin",
|
|
"melody_editor": "editor",
|
|
"device_manager": "editor",
|
|
"user_manager": "editor",
|
|
"viewer": "user",
|
|
}
|
|
|
|
FULL = {"view": True, "add": True, "edit": True, "delete": True}
|
|
VIEW_ONLY = {"view": True, "add": False, "edit": False, "delete": False}
|
|
|
|
PERMISSION_MAPPING = {
|
|
"melody_editor": {
|
|
"melodies": FULL,
|
|
"devices": VIEW_ONLY,
|
|
"app_users": VIEW_ONLY,
|
|
"equipment": VIEW_ONLY,
|
|
"mqtt": False,
|
|
},
|
|
"device_manager": {
|
|
"melodies": VIEW_ONLY,
|
|
"devices": FULL,
|
|
"app_users": VIEW_ONLY,
|
|
"equipment": FULL,
|
|
"mqtt": True,
|
|
},
|
|
"user_manager": {
|
|
"melodies": VIEW_ONLY,
|
|
"devices": VIEW_ONLY,
|
|
"app_users": FULL,
|
|
"equipment": VIEW_ONLY,
|
|
"mqtt": False,
|
|
},
|
|
"viewer": {
|
|
"melodies": VIEW_ONLY,
|
|
"devices": VIEW_ONLY,
|
|
"app_users": VIEW_ONLY,
|
|
"equipment": VIEW_ONLY,
|
|
"mqtt": False,
|
|
},
|
|
}
|
|
|
|
|
|
def migrate(dry_run=False):
|
|
init_firebase()
|
|
db = get_db()
|
|
if not db:
|
|
print("ERROR: Firebase initialization failed.")
|
|
sys.exit(1)
|
|
|
|
docs = db.collection("admin_users").get()
|
|
migrated = 0
|
|
|
|
for doc in docs:
|
|
data = doc.to_dict()
|
|
old_role = data.get("role", "")
|
|
|
|
if old_role in ROLE_MAPPING:
|
|
new_role = ROLE_MAPPING[old_role]
|
|
permissions = PERMISSION_MAPPING.get(old_role)
|
|
|
|
print(f" {data.get('email', '?'):30s} {old_role:20s} -> {new_role}")
|
|
|
|
if not dry_run:
|
|
update = {"role": new_role}
|
|
if permissions:
|
|
update["permissions"] = permissions
|
|
doc.reference.update(update)
|
|
|
|
migrated += 1
|
|
else:
|
|
print(f" {data.get('email', '?'):30s} {old_role:20s} (already migrated, skipping)")
|
|
|
|
action = "would be" if dry_run else "were"
|
|
print(f"\n{migrated} user(s) {action} migrated.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Migrate admin user roles")
|
|
parser.add_argument("--dry-run", action="store_true", help="Preview changes without applying")
|
|
args = parser.parse_args()
|
|
|
|
print("Migrating admin user roles...\n")
|
|
migrate(dry_run=args.dry_run)
|