""" Management command: migrate_legacy_students =========================================== Migrates student data from the legacy ruforum_rims_db (MySQL) into the new IILMP system (PostgreSQL) in 4 phases: Phase 1 — Scholars contacts_student + contacts_user → CustomUser + Scholar (academic profile: degree, thesis, supervisors, reg no, etc.) Phase 2 — Progress reports student_reports_studentreport → ProgressReport + sub-tables: supervisordetails, skillsimprovement, manuscript, conferencepaper, presentation Phase 3 — Graduation records contacts_student where grad_actual IS NOT NULL → GraduationRecord (signal auto-creates AlumniProfile) Phase 4 — Link scholars to existing Awards / Scholarships grants_studentmembership already migrated in migrate_legacy_p1 — this phase re-links Scholar rows to Award where missing Usage: docker compose run --rm web python manage.py migrate_legacy_students docker compose run --rm web python manage.py migrate_legacy_students --phase 1 docker compose run --rm web python manage.py migrate_legacy_students --phase 1 2 docker compose run --rm web python manage.py migrate_legacy_students --dry-run docker compose run --rm web python manage.py migrate_legacy_students --reset """ from __future__ import annotations import logging from datetime import date, datetime from typing import Any 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 django.utils import timezone logger = logging.getLogger(__name__) from apps.core.legacy_db import LEGACY_DB LEGACY_TAG = "[migrated_from:ruforum_rims_db]" # Map legacy degree_program_level → Scholar.DegreeLevel values DEGREE_MAP = { "msc": "msc", "phd": "phd", "mphil": "mphil", "bachelor": "bachelor", "pgdip": "pgdip", "": "", } # Map legacy student_type → Scholar.StudentType STUDENT_TYPE_MAP = { "competitive_grants": "competitive_grant", "regional_programs": "regional_programme", "other": "other", "": "other", } def _date(val) -> date | None: if val is None: return None if isinstance(val, date): return val try: return datetime.strptime(str(val)[:10], "%Y-%m-%d").date() except (ValueError, TypeError): return None def _int(val) -> int | None: try: return int(val) if val is not None else None except (ValueError, TypeError): return None def _str(val, maxlen: int = 0) -> str: s = (val or "").strip() return s[:maxlen] if maxlen else s class Command(BaseCommand): help = "Migrate legacy contacts_student data into RIMS Scholar / ProgressReport / GraduationRecord" def add_arguments(self, parser): parser.add_argument( "--phase", nargs="+", type=int, default=[1, 2, 3, 4], help="Phases to run (default: all). e.g. --phase 1 2", ) parser.add_argument("--dry-run", action="store_true", help="Simulate without writing.") parser.add_argument( "--reset", action="store_true", help="Delete all legacy-tagged Scholar / ProgressReport / GraduationRecord rows first.", ) def handle(self, *args, **options): self.dry_run = options["dry_run"] self.verbosity = options.get("verbosity", 1) if self.dry_run: self.stdout.write(self.style.WARNING(" DRY RUN — no data will be written.\n")) # Connect to legacy MySQL try: self.conn = pymysql.connect( cursorclass=pymysql.cursors.DictCursor, **LEGACY_DB, ) self.cur = self.conn.cursor() except Exception as exc: self.stderr.write(self.style.ERROR(f"Cannot connect to legacy DB: {exc}")) return if options["reset"] and not self.dry_run: self._reset() runners = { 1: self._phase1_scholars, 2: self._phase2_progress_reports, 3: self._phase3_graduation, 4: self._phase4_relink_awards, } for phase in sorted(options["phase"]): self.stdout.write(self.style.MIGRATE_HEADING(f"\n=== Phase {phase} ===")) runners[phase]() self.cur.close() self.conn.close() self.stdout.write(self.style.SUCCESS("\nMigration complete.")) # ── Reset ───────────────────────────────────────────────────────────────── def _reset(self): from apps.rims.scholarships.models import GraduationRecord, ProgressReport, Scholar self.stdout.write(self.style.WARNING(" Resetting legacy-tagged rows...")) gr = GraduationRecord.objects.filter(notes__contains=LEGACY_TAG).count() pr = ProgressReport.objects.filter(notes__contains=LEGACY_TAG).count() sc = Scholar.objects.filter(notes__contains=LEGACY_TAG).count() GraduationRecord.objects.filter(notes__contains=LEGACY_TAG).delete() ProgressReport.objects.filter(notes__contains=LEGACY_TAG).delete() Scholar.objects.filter(notes__contains=LEGACY_TAG).delete() self.stdout.write(f" Deleted: {sc} scholars, {pr} reports, {gr} graduation records.") # ── Phase 1: Scholars ───────────────────────────────────────────────────── def _phase1_scholars(self): from apps.core.authentication.models import CustomUser, Institution from apps.rims.scholarships.models import Scholar, Scholarship self.stdout.write(" Loading legacy students...") self.cur.execute(""" SELECT cs.user_id, cu.business_email AS email, cu.first_name, cu.last_name, cu.is_active, cu.gender, cu.country, cs.university, cs.university_department, cs.university_reg_no, cs.degree_program_level, cs.degree_program_name, cs.intake_year, cs.grad_expected, cs.grad_actual, cs.thesis_title, cs.cohort, cs.supervisor1, cs.supervisor2, cs.supervisor3, cs.research_abstract, cs.grant_type_id, cs.student_type, cs.graduated, cs.student_no, cs.funder, cs.host_university_address, cs.date_of_birth FROM contacts_student cs JOIN contacts_user cu ON cu.id = cs.user_id ORDER BY cs.user_id """) rows = self.cur.fetchall() self.stdout.write(f" Found {len(rows)} student records.") # Ensure a default legacy scholarship programme default_scholarship, _ = Scholarship.objects.get_or_create( name="Legacy RUFORUM Scholars", defaults={ "description": f"{LEGACY_TAG} — Scholar records migrated from ruforum_rims_db contacts_student", "is_active": True, }, ) existing_emails: set[str] = set(CustomUser.objects.values_list("email", flat=True)) existing_scholar_notes: set[str] = set( Scholar.objects.filter(notes__contains="[legacy_student_id:") .values_list("notes", flat=True) ) created_scholars = skipped = errors = users_created = 0 for row in rows: user_id = row["user_id"] email = _str(row.get("email")).lower() if not email: errors += 1 continue # Skip if already migrated tag = f"[legacy_student_id:{user_id}]" if any(tag in n for n in existing_scholar_notes): skipped += 1 continue if self.dry_run: created_scholars += 1 continue try: with transaction.atomic(): # Get or create user user = CustomUser.objects.filter(email=email).first() if not user: user = CustomUser.objects.create( email=email, first_name=_str(row.get("first_name"))[:150], last_name=_str(row.get("last_name"))[:150], is_active=bool(row.get("is_active", True)), password=make_password(None), ) existing_emails.add(email) users_created += 1 # Resolve institution institution = None univ_name = _str(row.get("university")) if univ_name: institution = Institution.objects.filter( name__iexact=univ_name ).first() # Map degree level raw_level = _str(row.get("degree_program_level")).lower() degree_level = DEGREE_MAP.get(raw_level, "other") if raw_level else "" # Map student type raw_type = _str(row.get("student_type")).lower() student_type = STUDENT_TYPE_MAP.get(raw_type, "other") # Determine status graduated = bool(row.get("graduated")) grad_actual = _date(row.get("grad_actual")) if graduated or grad_actual: status = Scholar.Status.COMPLETED else: status = Scholar.Status.ACTIVE # Skip if (user, scholarship) already exists if Scholar.objects.filter(user=user, scholarship=default_scholarship).exists(): skipped += 1 continue Scholar.objects.create( user=user, scholarship=default_scholarship, institution=institution, degree_level=degree_level, degree_name=_str(row.get("degree_program_name"), 255), thesis_title=_str(row.get("thesis_title"), 500), research_abstract=_str(row.get("research_abstract")), supervisor_1=_str(row.get("supervisor1"), 255), supervisor_2=_str(row.get("supervisor2"), 255), supervisor_3=_str(row.get("supervisor3"), 255), university_reg_no=_str(row.get("university_reg_no"), 64), student_number=_str(row.get("student_no"), 64), host_university_address=_str(row.get("host_university_address")), student_type=student_type, cohort_year=max(1, _int(row.get("cohort")) or 0) or None, intake_year=max(1900, _int(row.get("intake_year")) or 0) or None, expected_graduation=_date(row.get("grad_expected")), status=status, status_changed_at=timezone.now(), notes=( f"{LEGACY_TAG}\n" f"{tag}\n" f"university: {_str(row.get('university'))}\n" f"department: {_str(row.get('university_department'))}\n" f"funder: {_str(row.get('funder'))}\n" f"grant_type_id: {row.get('grant_type_id')}\n" ), ) created_scholars += 1 except Exception as exc: logger.warning("Phase 1 error user_id=%s: %s", user_id, exc) errors += 1 self.stdout.write( self.style.SUCCESS( f" Scholars — created: {created_scholars}, users_created: {users_created}, " f"skipped: {skipped}, errors: {errors}" ) ) # ── Phase 2: Progress reports ───────────────────────────────────────────── def _phase2_progress_reports(self): from apps.core.authentication.models import CustomUser from apps.rims.scholarships.models import ( ConferencePaper, Manuscript, Presentation, ProgressReport, Scholar, SkillsImprovement, SupervisorFeedback, ) self.stdout.write(" Loading legacy student reports...") # Load all student reports self.cur.execute(""" SELECT sr.id, sr.student_id, sr.period, sr.semester, sr.year, sr.status, sr.submitted_on, sr.activities_executed, sr.additional_information, sr.areas_university_needs_improvement, sr.critical_issues, sr.farming_community_type, sr.farmrers_or_organisation_number, sr.location, sr.none_farmer_actors, sr.research_timeframe, sr.technology_championed, sr.title_of_research_proposal, sr.skills_required, sr.scholarship_id FROM student_reports_studentreport sr ORDER BY sr.student_id, sr.id """) report_rows = self.cur.fetchall() self.stdout.write(f" Found {len(report_rows)} student report records.") # Build lookup: legacy user_id → Scholar pk # Parse [legacy_student_id:X] tags from notes scholar_map: dict[int, int] = {} for s in Scholar.objects.filter(notes__contains="[legacy_student_id:").values("pk", "notes"): notes = s["notes"] or "" try: legacy_uid = int(notes.split("[legacy_student_id:")[1].split("]")[0]) scholar_map[legacy_uid] = s["pk"] except (ValueError, IndexError): pass if not scholar_map and not self.dry_run: self.stdout.write(self.style.WARNING(" No migrated scholars found — run Phase 1 first.")) return if not scholar_map: self.stdout.write(self.style.WARNING(" DRY RUN: scholar_map empty (Phase 1 not yet run). Counting reportable rows only.")) self.stdout.write(f" Would process {len(report_rows)} reports when scholars exist.") return # Load supervisor details, skills, manuscripts, conference papers, presentations def _load_sub(table, id_field="report_id") -> dict[int, list[dict]]: self.cur.execute(f"SELECT * FROM {table} ORDER BY {id_field}") result: dict[int, list] = {} for row in self.cur.fetchall(): rid = row[id_field] result.setdefault(rid, []).append(row) return result supervisor_map = _load_sub("student_reports_supervisordetails") skills_map: dict[int, list] = {} # from skillsimprovement — check table name # The legacy table is student_reports_skillsimprovement? Let's check try: self.cur.execute("SELECT * FROM student_reports_skillsimprovement LIMIT 1") skills_map = _load_sub("student_reports_skillsimprovement") except Exception: pass # table may not exist under that name manuscript_map = _load_sub("student_reports_manuscript") conference_map = _load_sub("student_reports_conferencepaper") presentation_map = _load_sub("student_reports_presentation") created = skipped = errors = 0 STATUS_MAP = { "draft": ProgressReport.Status.DRAFT, "submitted": ProgressReport.Status.SUBMITTED, "accepted": ProgressReport.Status.ACCEPTED, "": ProgressReport.Status.SUBMITTED, } for row in report_rows: student_id = row["student_id"] report_id = row["id"] scholar_pk = scholar_map.get(student_id) if not scholar_pk: errors += 1 continue period = row.get("period") semester = _int(row.get("semester")) year = _int(row.get("year")) period_label = ( f"Semester {semester} {year}" if semester and year else f"Period {period}" if period else f"Report {report_id}" ) # Skip if already migrated if ProgressReport.objects.filter( scholar_id=scholar_pk, notes__contains=f"[legacy_report_id:{report_id}]", ).exists(): skipped += 1 continue if self.dry_run: created += 1 continue raw_status = _str(row.get("status", "")).lower() pr_status = STATUS_MAP.get(raw_status, ProgressReport.Status.SUBMITTED) submitted_on = _date(row.get("submitted_on")) try: with transaction.atomic(): report = ProgressReport.objects.create( scholar_id=scholar_pk, period_label=period_label, semester=semester, year=year, title_of_research=_str(row.get("title_of_research_proposal"), 500), activities_executed=_str(row.get("activities_executed")), research_timeframe=_str(row.get("research_timeframe")), technology_championed=_str(row.get("technology_championed")), farming_community_type=_str(row.get("farming_community_type")), farmers_or_organisation_count=_int(row.get("farmrers_or_organisation_number")), none_farmer_actors=_str(row.get("none_farmer_actors")), location=_str(row.get("location")), critical_issues=_str(row.get("critical_issues")), skills_required=_str(row.get("skills_required")), additional_information=_str(row.get("additional_information")), areas_university_needs_improvement=_str(row.get("areas_university_needs_improvement")), status=pr_status, submitted_on=submitted_on, submitted_at=timezone.make_aware( datetime.combine(submitted_on, datetime.min.time()) ) if submitted_on else timezone.now(), notes=f"{LEGACY_TAG}\n[legacy_report_id:{report_id}]", ) # Supervisor details for sup in supervisor_map.get(report_id, []): SupervisorFeedback.objects.create( report=report, name=_str(sup.get("name"), 255), title=_str(sup.get("title"), 100), area_of_mentorship=_str(sup.get("area_of_mentorship"), 200), areas_of_achievement=_str(sup.get("areas_of_achievement")), areas_requiring_attention=_str(sup.get("areas_required_more_attention_and_support")), ) # Skills improvement for sk in skills_map.get(report_id, []): SkillsImprovement.objects.create( report=report, skill=_str(sk.get("skill") or sk.get("name") or "Skill", 255), description=_str(sk.get("description") or sk.get("details") or ""), ) # Manuscripts for ms in manuscript_map.get(report_id, []): Manuscript.objects.create( report=report, title=_str(ms.get("title") or "Manuscript", 500), ) # Conference papers for cp in conference_map.get(report_id, []): ConferencePaper.objects.create( report=report, title=_str(cp.get("title") or "Conference paper", 500), ) # Presentations for pres in presentation_map.get(report_id, []): Presentation.objects.create( report=report, title=_str(pres.get("title") or "Presentation", 500), ) created += 1 except Exception as exc: logger.warning("Phase 2 error report_id=%s: %s", report_id, exc) errors += 1 self.stdout.write( self.style.SUCCESS( f" Progress reports — created: {created}, skipped: {skipped}, errors: {errors}" ) ) # ── Phase 3: Graduation records ─────────────────────────────────────────── def _phase3_graduation(self): from apps.rims.scholarships.models import GraduationRecord, Scholar self.stdout.write(" Loading students with graduation dates...") self.cur.execute(""" SELECT cs.user_id, cs.grad_actual, cs.thesis_title FROM contacts_student cs WHERE cs.grad_actual IS NOT NULL ORDER BY cs.user_id """) rows = self.cur.fetchall() self.stdout.write(f" Found {len(rows)} students with grad_actual date.") # Build legacy user_id → scholar_pk map scholar_map: dict[int, int] = {} for s in Scholar.objects.filter(notes__contains="[legacy_student_id:").values("pk", "notes"): notes = s["notes"] or "" try: legacy_uid = int(notes.split("[legacy_student_id:")[1].split("]")[0]) scholar_map[legacy_uid] = s["pk"] except (ValueError, IndexError): pass if not scholar_map and not self.dry_run: self.stdout.write(self.style.WARNING(" No migrated scholars found — run Phase 1 first.")) return if not scholar_map: self.stdout.write(self.style.WARNING(" DRY RUN: Phase 1 not yet run. Would process up to 717 graduation records.")) return created = skipped = errors = 0 for row in rows: user_id = row["user_id"] scholar_pk = scholar_map.get(user_id) if not scholar_pk: errors += 1 continue grad_actual = _date(row.get("grad_actual")) if not grad_actual: errors += 1 continue if GraduationRecord.objects.filter(scholar_id=scholar_pk).exists(): skipped += 1 continue if self.dry_run: created += 1 continue try: with transaction.atomic(): # GraduationRecord.save() fires the signal → AlumniProfile created GraduationRecord.objects.create( scholar_id=scholar_pk, graduated_on=grad_actual, certificate_issued=False, notes=( f"{LEGACY_TAG}\n" f"thesis: {_str(row.get('thesis_title'), 200)}" ), ) created += 1 except Exception as exc: logger.warning("Phase 3 error user_id=%s: %s", user_id, exc) errors += 1 self.stdout.write( self.style.SUCCESS( f" Graduation records — created: {created}, skipped: {skipped}, errors: {errors}" ) ) # ── Phase 4: Re-link scholars to existing Awards ────────────────────────── def _phase4_relink_awards(self): from apps.rims.grants.models import Award from apps.rims.scholarships.models import Scholar self.stdout.write(" Re-linking scholars to awards via grants_studentmembership...") self.cur.execute(""" SELECT gsm.student_id, gsm.grant_id FROM grants_studentmembership gsm ORDER BY gsm.student_id """) rows = self.cur.fetchall() # Build award map: legacy grant_id → Award.pk # Awards store the legacy grant_id in their narrative field (set during migrate_legacy_grants) award_map: dict[int, int] = {} for pk, narrative in Award.objects.values_list("pk", "narrative"): if narrative and "[legacy_grant_id:" in narrative: try: grant_id = int(narrative.split("[legacy_grant_id:")[1].split("]")[0]) award_map[grant_id] = pk except (ValueError, IndexError): pass # Build user_id → scholar scholar_qs = Scholar.objects.filter(notes__contains="[legacy_student_id:") scholar_map: dict[int, Scholar] = {s.user_id: s for s in scholar_qs} updated = skipped = errors = 0 for row in rows: student_id = row["student_id"] grant_id = row["grant_id"] scholar = scholar_map.get(student_id) award_pk = award_map.get(grant_id) if not scholar or not award_pk: errors += 1 continue if scholar.award_id == award_pk: skipped += 1 continue if scholar.award_id is not None: # Already linked to a different award — skip skipped += 1 continue # Check Award not already linked to another Scholar if Scholar.objects.filter(award_id=award_pk).exclude(pk=scholar.pk).exists(): skipped += 1 continue if self.dry_run: updated += 1 continue try: Scholar.objects.filter(pk=scholar.pk).update(award_id=award_pk) scholar.award_id = award_pk updated += 1 except Exception as exc: logger.warning("Phase 4 error student_id=%s: %s", student_id, exc) errors += 1 self.stdout.write( self.style.SUCCESS( f" Award links — updated: {updated}, skipped: {skipped}, errors: {errors}" ) )