""" Management command: migrate_legacy_grants ========================================= Migrates grants data from the legacy ruforum_rims_db (MySQL) into the new IILMP system (PostgreSQL) in 5 phases: Phase 1 — Institutions (common_university) Phase 2 — Grant Calls (calls_grantcall → FundingRecord + GrantCall) Phase 3 — Users / Applicants (contacts_user filtered to grant applicants) Phase 4 — Applications + Reviews Phase 5 — Awards (grants_grant where approval_status = approved) Usage: docker compose run --rm web python manage.py migrate_legacy_grants docker compose run --rm web python manage.py migrate_legacy_grants --phase 1 docker compose run --rm web python manage.py migrate_legacy_grants --phase 1 2 3 docker compose run --rm web python manage.py migrate_legacy_grants --dry-run docker compose run --rm web python manage.py migrate_legacy_grants --reset """ import re import uuid from datetime import date, datetime, timezone as dt_timezone from decimal import Decimal import MySQLdb from django.contrib.auth.hashers import make_password from django.contrib.auth.models import Group from django.core.management.base import BaseCommand, CommandError from django.db import transaction from django.utils import timezone from django.utils.text import slugify from apps.core.authentication.models import CustomUser, Institution from apps.core.permissions.roles import UserRole from apps.rims.grants.models import ( Application, Award, FundingRecord, GrantCall, Review, ) # --------------------------------------------------------------------------- # Legacy DB connection # --------------------------------------------------------------------------- from apps.core.legacy_db import LEGACY_DB # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- DEFAULT_CURRENCY = "USD" LEGACY_TAG = "LEGACY_MIGRATED" # Status mapping: old → new Application.Status APPLICATION_STATUS_MAP = { "draft": Application.Status.DRAFT, "submitted": Application.Status.SUBMITTED, "validated": Application.Status.SHORTLISTED, # passed compliance checks in old system "noncompliant": Application.Status.INELIGIBLE, "selected_for_funding": Application.Status.AWARDED, "rejected": Application.Status.REJECTED, } # Review recommendation mapping REVIEW_RECOMMENDATION_MAP = { "approve": Review.RecommendationCode.APPROVE, "approved": Review.RecommendationCode.APPROVE, "revise": Review.RecommendationCode.REVISE, "revision": Review.RecommendationCode.REVISE, "reject": Review.RecommendationCode.REJECT, "rejected": Review.RecommendationCode.REJECT, } # Call type guessing from grant type name CALL_TYPE_HINTS = { "fellowship": GrantCall.CallType.FELLOWSHIP, "post doc": GrantCall.CallType.FELLOWSHIP, "postdoc": GrantCall.CallType.FELLOWSHIP, "fapa": GrantCall.CallType.FELLOWSHIP, "scholarship": GrantCall.CallType.SCHOLARSHIP, "recap": GrantCall.CallType.CHALLENGE, "challenge": GrantCall.CallType.CHALLENGE, } def _guess_call_type(name: str) -> str: name_lower = (name or "").lower() for hint, ctype in CALL_TYPE_HINTS.items(): if hint in name_lower: return ctype return GrantCall.CallType.GRANT def _safe_date(val) -> date | None: if val is None: return None if isinstance(val, date): return val try: return datetime.strptime(str(val), "%Y-%m-%d").date() except (ValueError, TypeError): return None def _safe_decimal(val, default=Decimal("0.00")) -> Decimal: try: return Decimal(str(val)).quantize(Decimal("0.01")) except Exception: return default def _unique_slug(base: str, existing: set) -> str: slug = slugify(base)[:100] or "call" candidate = slug i = 2 while candidate in existing: candidate = f"{slug[:95]}-{i}" i += 1 existing.add(candidate) return candidate def _unique_email(email: str, existing: set) -> str: """Ensure no duplicate email — suffix with counter if needed.""" email = (email or "").strip().lower() if not email: email = f"legacy-{uuid.uuid4().hex[:8]}@ruforum.org" candidate = email i = 2 while candidate in existing: parts = candidate.split("@") candidate = f"{parts[0]}-{i}@{parts[1]}" i += 1 existing.add(candidate) return candidate class Command(BaseCommand): help = "Migrate grants data from legacy ruforum_rims_db (MySQL) to IILMP (PostgreSQL)" def add_arguments(self, parser): parser.add_argument( "--phase", nargs="+", type=int, choices=[1, 2, 3, 4, 5, 6], default=[1, 2, 3, 4, 5, 6], help="Phases to run (default: all). Phase 6 = assign roles. e.g. --phase 6", ) parser.add_argument( "--dry-run", action="store_true", help="Print what would be migrated without writing anything.", ) parser.add_argument( "--reset", action="store_true", help="Delete all previously migrated records before running (uses legacy_id tag).", ) # ------------------------------------------------------------------ # Entry point # ------------------------------------------------------------------ def handle(self, *args, **options): self.dry_run = options["dry_run"] self.verbosity = options["verbosity"] phases = sorted(options["phase"]) if self.dry_run: self.stdout.write(self.style.WARNING("DRY RUN — no data will be written.\n")) try: self.conn = MySQLdb.connect(**LEGACY_DB) except Exception as exc: raise CommandError(f"Cannot connect to legacy MySQL: {exc}") from exc self.cursor = self.conn.cursor(MySQLdb.cursors.DictCursor) if options["reset"]: self._reset() # In-memory id maps (legacy_id → new Django id) self.inst_map: dict[int, int] = {} # legacy university_id → Institution.pk self.call_map: dict[int, int] = {} # legacy grantcall_id → GrantCall.pk self.user_map: dict[int, int] = {} # legacy user_id → CustomUser.pk self.app_map: dict[int, int] = {} # legacy application_id → Application.pk self.reviewer_map: dict[int, int] = {} # legacy reviewer_id → CustomUser.pk # Pre-load existing slugs to avoid conflicts self.existing_slugs: set[str] = set(GrantCall.objects.values_list("slug", flat=True)) self.existing_emails: set[str] = set(CustomUser.objects.values_list("email", flat=True)) runners = { 1: self._phase1_institutions, 2: self._phase2_calls, 3: self._phase3_users, 4: self._phase4_applications, 5: self._phase5_awards, 6: self._phase6_roles, } for phase in phases: self.stdout.write(self.style.MIGRATE_HEADING(f"\n=== Phase {phase} ===")) runners[phase]() self.cursor.close() self.conn.close() self.stdout.write(self.style.SUCCESS("\nMigration complete.")) # ------------------------------------------------------------------ # Reset # ------------------------------------------------------------------ def _reset(self): self.stdout.write(self.style.WARNING("Resetting previously migrated records...")) if not self.dry_run: # Delete in reverse FK order Award.objects.filter(narrative__contains=LEGACY_TAG).delete() Application.objects.filter(eligibility_notes__contains=LEGACY_TAG).delete() CustomUser.objects.filter(legacy_notes__contains=LEGACY_TAG).delete() if hasattr(CustomUser, "legacy_notes") else None GrantCall.objects.filter(description__contains=LEGACY_TAG).delete() FundingRecord.objects.filter(strategic_objectives__contains=LEGACY_TAG).delete() Institution.objects.filter(address__contains=LEGACY_TAG).delete() self.stdout.write(" Reset done.") # ------------------------------------------------------------------ # Phase 1 — Institutions # ------------------------------------------------------------------ def _phase1_institutions(self): self.cursor.execute( "SELECT u.id, u.name, u.acronym, c.name as country_name " "FROM common_university u " "LEFT JOIN common_country c ON c.id = u.country_id" ) rows = self.cursor.fetchall() created = skipped = 0 for row in rows: name = (row["name"] or "").strip() if not name: continue existing = Institution.objects.filter(name=name).first() if existing: self.inst_map[row["id"]] = existing.pk skipped += 1 continue if self.dry_run: self.stdout.write(f" [DRY] Institution: {name}") created += 1 continue inst = Institution.objects.create( name=name, country=row["country_name"] or "", institution_type=Institution.InstitutionType.UNIVERSITY, address=LEGACY_TAG, region="", website="", is_active=True, is_member=False, ) self.inst_map[row["id"]] = inst.pk created += 1 self.stdout.write( self.style.SUCCESS(f" Institutions — created: {created}, skipped (already exist): {skipped}") ) # ------------------------------------------------------------------ # Phase 2 — Grant Calls → FundingRecord + GrantCall # ------------------------------------------------------------------ def _phase2_calls(self): self.cursor.execute( "SELECT gc.*, gt.name as grant_type_name " "FROM calls_grantcall gc " "LEFT JOIN grant_types_granttype gt ON gt.id = gc.grant_type_id" ) rows = self.cursor.fetchall() created = skipped = 0 # Pre-load admin user as created_by (fallback) admin = CustomUser.objects.filter(is_superuser=True).first() for row in rows: title = (row["title"] or "").strip() or f"Legacy Call #{row['id']}" call_ref = (row["call_id"] or "").strip() grant_type_name = (row["grant_type_name"] or "").strip() # Check if already migrated (match by call_id in description) if GrantCall.objects.filter(description__contains=f"[legacy_call_id:{row['id']}]").exists(): # Load into map gc = GrantCall.objects.get(description__contains=f"[legacy_call_id:{row['id']}]") self.call_map[row["id"]] = gc.pk skipped += 1 continue start = _safe_date(row.get("start_date")) or date.today() end = _safe_date(row.get("end_date")) or date.today() deadline = _safe_date(row.get("submission_deadline")) or start call_type = _guess_call_type(grant_type_name or title) if self.dry_run: self.stdout.write(f" [DRY] Call: {title} ({call_ref}) [{call_type}]") created += 1 continue with transaction.atomic(): # One FundingRecord per legacy call # Budget unknown — set to 0; staff can update # Make public_grant_id unique by always appending legacy id raw_ref = call_ref[:32] if call_ref else "LEG" pub_id = f"{raw_ref}-{row['id']}"[:40] fr = FundingRecord.objects.create( public_grant_id=pub_id, title=title, amount=Decimal("0.00"), currency=DEFAULT_CURRENCY, start_date=start, end_date=end, status=FundingRecord.Status.APPROVED, strategic_objectives=f"{LEGACY_TAG} grant_type:{grant_type_name}", created_by=admin, ) slug = _unique_slug(title, self.existing_slugs) # closes_at must be a datetime closes_at = datetime.combine(deadline, datetime.min.time()).replace(tzinfo=dt_timezone.utc) gc = GrantCall( title=title, slug=slug, call_type=call_type, description=( f"{LEGACY_TAG}\n" f"[legacy_call_id:{row['id']}]\n" f"[legacy_ref:{call_ref}]\n" f"grant_type: {grant_type_name}\n" f"call_year: {row.get('call_year', '')}" ), opens_at=datetime.combine(start, datetime.min.time()).replace(tzinfo=dt_timezone.utc), closes_at=closes_at, status=GrantCall.Status.CLOSED, funding_record=fr, created_by=admin, blind_review=False, ) # Save bypassing FSM (already closed historical calls) GrantCall.objects.bulk_create([gc]) gc = GrantCall.objects.get(slug=slug) self.call_map[row["id"]] = gc.pk created += 1 self.stdout.write( self.style.SUCCESS(f" Grant Calls — created: {created}, skipped: {skipped}") ) # ------------------------------------------------------------------ # Phase 3 — Users (only those with grant applications) # ------------------------------------------------------------------ def _phase3_users(self): self.cursor.execute( "SELECT DISTINCT u.id, u.business_email as email, u.first_name, u.last_name, " "u.is_active, u.date_joined, " "a.institution, a.country " "FROM contacts_user u " "JOIN grants_applications_grantapplication a ON a.user_id = u.id" ) rows = self.cursor.fetchall() # Also load reviewer users (may not have applications) self.cursor.execute( "SELECT DISTINCT u.id, u.business_email as email, u.first_name, u.last_name, u.is_active, u.date_joined " "FROM contacts_user u " "JOIN grants_applications_grantappreview r ON r.reviewer_id = u.id" ) reviewer_rows = self.cursor.fetchall() all_rows = {r["id"]: r for r in rows} for r in reviewer_rows: if r["id"] not in all_rows: all_rows[r["id"]] = r created = skipped = 0 for uid, row in all_rows.items(): email = (row.get("email") or "").strip().lower() if not email: email = f"legacy-user-{row['id']}@ruforum.org" existing = CustomUser.objects.filter(email=email).first() if existing: self.user_map[row["id"]] = existing.pk skipped += 1 continue if self.dry_run: self.stdout.write(f" [DRY] User: {email}") created += 1 continue email = _unique_email(email, self.existing_emails) user = CustomUser.objects.create( email=email, first_name=(row.get("first_name") or "").strip()[:150], last_name=(row.get("last_name") or "").strip()[:150], is_active=bool(row.get("is_active", True)), password=make_password(None), # unusable — reset on first login date_joined=row.get("date_joined") or timezone.now(), ) self.user_map[row["id"]] = user.pk created += 1 self.stdout.write( self.style.SUCCESS(f" Users — created: {created}, skipped: {skipped}") ) # ------------------------------------------------------------------ # Phase 4 — Applications + Reviews # ------------------------------------------------------------------ def _phase4_applications(self): # Ensure user & call maps are populated if running phase 4 standalone if not self.user_map: self._load_user_map() if not self.call_map: self._load_call_map() self.cursor.execute( "SELECT * FROM grants_applications_grantapplication ORDER BY id" ) rows = self.cursor.fetchall() created = skipped = errors = 0 # Track (call_id, user_id) pairs to enforce uniqueness existing_pairs: set[tuple] = set( Application.objects.values_list("call_id", "applicant_id") ) for row in rows: legacy_id = row["id"] # Skip if already migrated if Application.objects.filter( eligibility_notes__contains=f"[legacy_app_id:{legacy_id}]" ).exists(): app = Application.objects.get( eligibility_notes__contains=f"[legacy_app_id:{legacy_id}]" ) self.app_map[legacy_id] = app.pk skipped += 1 continue call_new_id = self.call_map.get(row.get("call_id")) user_new_id = self.user_map.get(row.get("user_id")) if not call_new_id: self.stdout.write( self.style.WARNING(f" SKIP app {legacy_id}: call {row.get('call_id')} not in map") ) errors += 1 continue if not user_new_id: self.stdout.write( self.style.WARNING(f" SKIP app {legacy_id}: user {row.get('user_id')} not in map") ) errors += 1 continue # Enforce unique (call, applicant) pair = (call_new_id, user_new_id) if pair in existing_pairs: self.stdout.write( self.style.WARNING( f" SKIP app {legacy_id}: duplicate (call={call_new_id}, user={user_new_id})" ) ) errors += 1 continue old_status = (row.get("status") or "draft").lower() new_status = APPLICATION_STATUS_MAP.get(old_status, Application.Status.DRAFT) submitted_at = None if old_status not in ("draft",): # Use last_modified as a proxy for submitted_at lm = row.get("last_modified") if lm: submitted_at = datetime.combine(lm, datetime.min.time()).replace(tzinfo=dt_timezone.utc) proposal_text = "\n\n".join( filter(None, [ row.get("proposal_title") or "", row.get("project_objectives") or "", ]) ) notes = ( f"{LEGACY_TAG}\n" f"[legacy_app_id:{legacy_id}]\n" f"[legacy_ref:{row.get('ref_number', '')}]\n" f"institution: {row.get('institution', '')}\n" f"country: {row.get('country', '')}\n" f"budget: {row.get('total_budget', '')}\n" f"duration_months: {row.get('duration_months', '')}\n" f"compliance: {row.get('compliance_comments', '')}" ) if self.dry_run: self.stdout.write( f" [DRY] Application {legacy_id}: user={user_new_id} call={call_new_id} status={new_status}" ) created += 1 existing_pairs.add(pair) continue with transaction.atomic(): app = Application( call_id=call_new_id, applicant_id=user_new_id, proposal=proposal_text, status=new_status, eligibility_notes=notes, eligibility_passed=old_status not in ("noncompliant", "draft"), submitted_at=submitted_at, anonymized_code=f"LEG-{legacy_id}", review_conflict_acknowledged=False, interview_completed=False, shortlist_recommended=old_status in ("selected_for_funding",), ) # Try institution lookup from application text inst_name = (row.get("institution") or "").strip() if inst_name: inst = Institution.objects.filter(name__icontains=inst_name[:50]).first() if inst: app.institution_id = inst.pk # Save bypassing FSM protected field Application.objects.bulk_create([app]) saved_app = Application.objects.filter( eligibility_notes__contains=f"[legacy_app_id:{legacy_id}]" ).first() if saved_app: self.app_map[legacy_id] = saved_app.pk existing_pairs.add(pair) created += 1 self.stdout.write( self.style.SUCCESS( f" Applications — created: {created}, skipped: {skipped}, errors: {errors}" ) ) # Reviews self._phase4b_reviews() def _phase4b_reviews(self): if not self.user_map: self._load_user_map() self.cursor.execute( "SELECT * FROM grants_applications_grantappreview ORDER BY id" ) rows = self.cursor.fetchall() created = skipped = errors = 0 existing_pairs: set[tuple] = set( Review.objects.values_list("application_id", "reviewer_id") ) for row in rows: app_new_id = self.app_map.get(row.get("application_id")) reviewer_new_id = self.user_map.get(row.get("reviewer_id")) if not app_new_id or not reviewer_new_id: errors += 1 continue pair = (app_new_id, reviewer_new_id) if pair in existing_pairs: skipped += 1 continue rec_raw = (row.get("recommendation") or "").lower().strip() rec_code = REVIEW_RECOMMENDATION_MAP.get(rec_raw, "") score = _safe_decimal(row.get("score"), Decimal("0")) if self.dry_run: created += 1 existing_pairs.add(pair) continue with transaction.atomic(): review = Review( application_id=app_new_id, reviewer_id=reviewer_new_id, scores={"legacy_score": float(score)}, total_score=score, recommendation_code=rec_code, recommendation=(row.get("comments") or ""), submitted_at=timezone.now(), ) Review.objects.bulk_create([review], ignore_conflicts=True) existing_pairs.add(pair) created += 1 self.stdout.write( self.style.SUCCESS( f" Reviews — created: {created}, skipped: {skipped}, errors: {errors}" ) ) # ------------------------------------------------------------------ # Phase 5 — Awards # ------------------------------------------------------------------ def _phase5_awards(self): if not self.app_map: self._load_app_map() self.cursor.execute( "SELECT * FROM grants_grant WHERE approval_status = 'approved' ORDER BY id" ) rows = self.cursor.fetchall() created = skipped = errors = 0 for row in rows: legacy_id = row["id"] if Award.objects.filter(narrative__contains=f"[legacy_grant_id:{legacy_id}]").exists(): skipped += 1 continue # Prefer link via grant_application_id legacy_app_id = row.get("grant_application_id") app_new_id = self.app_map.get(legacy_app_id) if legacy_app_id else None if not app_new_id: # Try to find by title match in applications title = (row.get("title") or "").strip() app = Application.objects.filter( proposal__icontains=title[:80] ).first() if title else None if app: app_new_id = app.pk else: self.stdout.write( self.style.WARNING( f" SKIP grant {legacy_id} '{row.get('grant_id')}': no application found" ) ) errors += 1 continue # Skip if this application already has an award if Award.objects.filter(application_id=app_new_id).exists(): skipped += 1 continue amount = _safe_decimal(row.get("budget"), Decimal("0.00")) start = _safe_date(row.get("start_date")) or date.today() end = _safe_date(row.get("new_end_date")) or _safe_date(row.get("end_date")) or date.today() # Determine award status based on end date if end < date.today(): award_status = Award.Status.CLOSED else: award_status = Award.Status.ACTIVE narrative = ( f"{LEGACY_TAG}\n" f"[legacy_grant_id:{legacy_id}]\n" f"[legacy_grant_ref:{row.get('grant_id', '')}]\n" f"title: {row.get('title', '')}\n" f"grant_year: {row.get('grant_year', '')}\n" f"duration: {row.get('duration', '')} months\n" f"theme: {row.get('theme', '')}\n" f"value_chain: {row.get('value_chain', '')}" ) if self.dry_run: self.stdout.write( f" [DRY] Award for grant {legacy_id}: app={app_new_id} " f"amount={amount} status={award_status}" ) created += 1 continue with transaction.atomic(): Award.objects.create( application_id=app_new_id, amount=amount, currency=DEFAULT_CURRENCY, project_end_date=end, activation_date=start, status=award_status, narrative=narrative, budget_deferred=False, ) created += 1 self.stdout.write( self.style.SUCCESS( f" Awards — created: {created}, skipped: {skipped}, errors: {errors}" ) ) # ------------------------------------------------------------------ # Phase 6 — Assign roles to migrated users # ------------------------------------------------------------------ def _phase6_roles(self): """ Assign roles based on how each user appears in the legacy DB: - Has a grant application → applicant - Has reviewed an application → reviewer - Both applicant AND reviewer → both groups - Has an approved grant (is PI) → scholar (awarded applicant) - grant_manager_id on applications → grants_manager Users already in a group are skipped (idempotent). """ # Load role groups applicant_group = Group.objects.filter(name=UserRole.APPLICANT.value).first() reviewer_group = Group.objects.filter(name=UserRole.REVIEWER.value).first() scholar_group = Group.objects.filter(name=UserRole.SCHOLAR.value).first() gm_group = Group.objects.filter(name=UserRole.GRANTS_MANAGER.value).first() missing = [name for name, g in [ ("applicant", applicant_group), ("reviewer", reviewer_group), ("scholar", scholar_group), ("grants_manager", gm_group) ] if g is None] if missing: self.stdout.write( self.style.WARNING( f" Missing groups: {missing}. Run `python manage.py seed_roles` first." ) ) return # --- Collect legacy user sets --- self.cursor.execute( "SELECT DISTINCT user_id FROM grants_applications_grantapplication WHERE user_id IS NOT NULL" ) applicant_legacy_ids = {r["user_id"] for r in self.cursor.fetchall()} self.cursor.execute( "SELECT DISTINCT reviewer_id FROM grants_applications_grantappreview WHERE reviewer_id IS NOT NULL" ) reviewer_legacy_ids = {r["reviewer_id"] for r in self.cursor.fetchall()} self.cursor.execute( "SELECT DISTINCT grant_manager_id FROM grants_applications_grantapplication " "WHERE grant_manager_id IS NOT NULL" ) gm_legacy_ids = {r["grant_manager_id"] for r in self.cursor.fetchall()} # PI users (user linked to an approved grant via application) self.cursor.execute( "SELECT DISTINCT a.user_id " "FROM grants_applications_grantapplication a " "JOIN grants_grant g ON g.grant_application_id = a.id " "WHERE g.approval_status = 'approved' AND a.user_id IS NOT NULL" ) scholar_legacy_ids = {r["user_id"] for r in self.cursor.fetchall()} # Reload user map if needed if not self.user_map: self._load_user_map() # --- Assign roles --- applicant_assigned = reviewer_assigned = scholar_assigned = gm_assigned = 0 for legacy_id, new_id in self.user_map.items(): try: user = CustomUser.objects.get(pk=new_id) except CustomUser.DoesNotExist: continue if self.dry_run: roles = [] if legacy_id in applicant_legacy_ids: roles.append("applicant") if legacy_id in reviewer_legacy_ids: roles.append("reviewer") if legacy_id in scholar_legacy_ids: roles.append("scholar") if legacy_id in gm_legacy_ids: roles.append("grants_manager") if roles: self.stdout.write(f" [DRY] {user.email} → {', '.join(roles)}") continue existing_groups = set(user.groups.values_list("name", flat=True)) if legacy_id in applicant_legacy_ids and UserRole.APPLICANT.value not in existing_groups: user.groups.add(applicant_group) # Also set role field on CustomUser if it exists if hasattr(user, "role") and not user.role: user.role = UserRole.APPLICANT.value user.save(update_fields=["role"]) applicant_assigned += 1 if legacy_id in reviewer_legacy_ids and UserRole.REVIEWER.value not in existing_groups: user.groups.add(reviewer_group) reviewer_assigned += 1 if legacy_id in scholar_legacy_ids and UserRole.SCHOLAR.value not in existing_groups: user.groups.add(scholar_group) # Promote role to scholar (higher than applicant) if hasattr(user, "role"): user.role = UserRole.SCHOLAR.value user.save(update_fields=["role"]) scholar_assigned += 1 if legacy_id in gm_legacy_ids and UserRole.GRANTS_MANAGER.value not in existing_groups: user.groups.add(gm_group) if hasattr(user, "role"): user.role = UserRole.GRANTS_MANAGER.value user.save(update_fields=["role"]) gm_assigned += 1 self.stdout.write( self.style.SUCCESS( f" Roles assigned — applicant: {applicant_assigned}, " f"reviewer: {reviewer_assigned}, scholar: {scholar_assigned}, " f"grants_manager: {gm_assigned}" ) ) # ------------------------------------------------------------------ # Helpers: reload maps when running individual phases # ------------------------------------------------------------------ def _load_user_map(self): for user in CustomUser.objects.all(): # Try to map via email stored in legacy DB pass self.cursor.execute("SELECT id, business_email as email FROM contacts_user") for row in self.cursor.fetchall(): email = (row["email"] or "").strip().lower() user = CustomUser.objects.filter(email=email).first() if user: self.user_map[row["id"]] = user.pk def _load_call_map(self): for gc in GrantCall.objects.filter(description__contains="[legacy_call_id:"): match = re.search(r"\[legacy_call_id:(\d+)\]", gc.description) if match: self.call_map[int(match.group(1))] = gc.pk def _load_app_map(self): for app in Application.objects.filter(eligibility_notes__contains="[legacy_app_id:"): match = re.search(r"\[legacy_app_id:(\d+)\]", app.eligibility_notes) if match: self.app_map[int(match.group(1))] = app.pk