""" Management command: migrate_legacy_staff ========================================= Creates missing RUFORUM Secretariat Staff accounts in the new system and assigns the correct role based on their legacy auth_group membership. Role mapping (legacy group → new UserRole): Secretariat Admins → admin Grants Managers → grants_manager Scholarship Managers → grants_manager Grant Application Reviewers → reviewer Scholarship Reviewers → reviewer Planning, M&E → mel_officer Administrative/HR → admin (closest available until HRM module built) Training Team → programme_officer Contacts managers → programme_officer Project managers → program_manager Staffs → (staff — no matching role, set to programme_officer) Superuser (is_superuser) → admin If a user belongs to multiple groups the highest-privilege role wins (admin > grants_manager > mel_officer > program_manager > programme_officer > reviewer). Usage: docker compose run --rm web python manage.py migrate_legacy_staff docker compose run --rm web python manage.py migrate_legacy_staff --dry-run docker compose run --rm web python manage.py migrate_legacy_staff --fix-existing """ from __future__ import annotations import pymysql import pymysql.cursors from django.contrib.auth.hashers import make_password from django.core.management.base import BaseCommand from django.db import transaction from apps.core.legacy_db import LEGACY_DB # Role priority — higher index wins when a user has multiple groups ROLE_PRIORITY = [ "reviewer", "programme_officer", "program_manager", "mel_officer", "grants_manager", "admin", ] GROUP_TO_ROLE = { "Secretariat Admins": "admin", "Grants Managers": "grants_manager", "Scholarship Managers": "grants_manager", "Grants Team": "grants_manager", "Grant Application Reviewers": "reviewer", "Scholarship Reviewers": "reviewer", "Scholarship Validators": "reviewer", "Planning, Monitoring & Evaluation": "mel_officer", "Administrative/HR": "admin", "Training Team": "programme_officer", "Contacts managers": "programme_officer", "Project managers": "program_manager", "Staffs": "programme_officer", "Scholarships Team": "programme_officer", "PIs": "applicant", "Applicants": "applicant", "Students": "scholar", } def _best_role(groups: list[str], is_superuser: bool, job_title: str = "") -> str: if is_superuser: return "admin" # Job-title overrides take precedence over group-derived roles jt = (job_title or "").lower() if any(kw in jt for kw in ["grants manager", "grants officer", "program officer grants", "grants programme officer", "programme officer, grants"]): return "grants_manager" if any(kw in jt for kw in ["finance manager", "accountant", "finance and administration", "budget controller", "management accountant"]): return "admin" if any(kw in jt for kw in ["training officer", "programme officer", "program officer", "communication", "knowledge management", "technical specialist", "executive secretary", "executive assistant", "technical assistant", "administrative", "hr/procurement"]): return "programme_officer" if any(kw in jt for kw in ["m&e officer", "monitoring and evaluation"]): return "mel_officer" roles = [GROUP_TO_ROLE[g] for g in groups if g in GROUP_TO_ROLE] # Filter out applicant/scholar roles for secretariat staff — they shouldn't # be demoted to student/applicant just because they're in those groups roles = [r for r in roles if r not in ("applicant", "scholar")] if not roles: return "programme_officer" best = max(roles, key=lambda r: ROLE_PRIORITY.index(r) if r in ROLE_PRIORITY else -1) return best class Command(BaseCommand): help = "Migrate RUFORUM Secretariat Staff accounts to the new system with correct roles" def add_arguments(self, parser): parser.add_argument("--dry-run", action="store_true", help="Simulate without writing.") parser.add_argument("--fix-existing", action="store_true", help="Also fix role on existing accounts that have wrong/no role.") def handle(self, *args, **options): self.dry_run = options["dry_run"] self.fix_existing = options["fix_existing"] if self.dry_run: self.stdout.write(self.style.WARNING(" DRY RUN — no data will be written.\n")) self.conn = pymysql.connect(cursorclass=pymysql.cursors.DictCursor, **LEGACY_DB) self.cur = self.conn.cursor() self._run() self.cur.close() self.conn.close() def _run(self): from apps.core.authentication.models import CustomUser from apps.core.permissions.roles import UserRole # ── Load all staff + their groups from legacy ───────────────────────── self.cur.execute(""" SELECT cu.id, cu.business_email, cu.first_name, cu.last_name, cu.contact_type, cu.is_staff, cu.is_superuser, cu.job_title, cu.is_active, GROUP_CONCAT(ag.name ORDER BY ag.name SEPARATOR '||') as grp_list FROM contacts_user cu LEFT JOIN contacts_user_groups cug ON cug.user_id = cu.id LEFT JOIN auth_group ag ON ag.id = cug.group_id WHERE cu.contact_type IN ( 'RUFORUM Secretariat Staff', 'Former RUFORUM Secretariat Staff', 'RUFORUM Staff' ) OR cu.is_staff = 1 GROUP BY cu.id, cu.business_email, cu.first_name, cu.last_name, cu.contact_type, cu.is_staff, cu.is_superuser, cu.job_title, cu.is_active """) rows = self.cur.fetchall() # Filter to genuine secretariat staff (exclude test/external accounts) # Lowercase all excludes so comparison always works EXCLUDE_EMAILS = {e.lower() for e in { "ruforummis-team@aptivate.org", "developer@gmail.com", "chris+ruforum@aptivate.org", "markos@aptivate.org", "ff@yahoo.com", "fassad9@yahoo.com", "jenifferikirimat@gmail.com", # test PI account # External partners/government — not RUFORUM staff "B.Ouzrourou@ofid.org", "mirjamblaak@gmail.com", "sameh@badea.org", "a.tenkouano@coraf.org", "psecretaryuer@gmail.com", "Diana.atwine@health.go.ug", "ekamba@agriculture.go.ug", "arthur.makara@mosti.go.ug", "janeaceng@gmail.com", "fsungo@ucm.ac.mz", }} staff_rows = [ r for r in rows if (r["business_email"] or "").lower().strip() not in EXCLUDE_EMAILS and (r["business_email"] or "").strip() ] self.stdout.write(f" Legacy staff to process: {len(staff_rows)}") existing_users = { u.email.lower(): u for u in CustomUser.objects.filter( email__in=[r["business_email"].strip().lower() for r in staff_rows if r["business_email"]] ) } created = fixed = skipped = errors = 0 for row in staff_rows: email = (row["business_email"] or "").strip().lower() if not email: errors += 1 continue groups = [g.strip() for g in (row["grp_list"] or "").split("||") if g.strip()] intended_role = _best_role(groups, bool(row["is_superuser"]), row.get("job_title") or "") is_former = row["contact_type"] == "Former RUFORUM Secretariat Staff" existing = existing_users.get(email) if existing: # Fix role if wrong or missing if self.fix_existing and existing.role != intended_role: if self.dry_run: self.stdout.write( f" [FIX] {email} — role {existing.role!r} → {intended_role!r}" ) fixed += 1 else: try: existing.role = intended_role existing.save(update_fields=["role"]) self.stdout.write( self.style.SUCCESS( f" [FIX] {email} — role → {intended_role}" ) ) fixed += 1 except Exception as exc: self.stderr.write(f" [ERR] {email}: {exc}") errors += 1 else: skipped += 1 continue # Create new account first = (row["first_name"] or "").strip()[:150] last = (row["last_name"] or "").strip()[:150] if self.dry_run: self.stdout.write( f" [CREATE] {email} | {first} {last} | role={intended_role} " f"| groups={', '.join(groups) or 'none'}" ) created += 1 continue try: with transaction.atomic(): user = CustomUser.objects.create( email=email, first_name=first, last_name=last, role=intended_role, is_active=not is_former, # former staff → inactive is_staff=False, password=make_password(None), # unusable — must reset ) self.stdout.write( self.style.SUCCESS( f" [CREATE] {email} | role={intended_role}" f"{' (inactive — former staff)' if is_former else ''}" ) ) created += 1 except Exception as exc: self.stderr.write(f" [ERR] {email}: {exc}") errors += 1 self.stdout.write( self.style.SUCCESS( f"\n Done — created: {created}, fixed: {fixed}, " f"skipped (already correct): {skipped}, errors: {errors}" ) ) if not self.dry_run: self.stdout.write("\n Staff accounts summary:") # Show final state of all staff for row in sorted(staff_rows, key=lambda r: (r["contact_type"] or "", r["last_name"] or "")): email = (row["business_email"] or "").strip().lower() user = CustomUser.objects.filter(email=email).first() if user: groups_str = (row["grp_list"] or "none")[:60] self.stdout.write( f" {'✓' if user.role else '?'} {email:<40} " f"role={user.role or 'NONE':<20} active={user.is_active}" )