"""Import the SME-Hub (EHC, Laravel) database into ``apps.smehub``. Reads the EHC DB from a **MariaDB/MySQL source** (load the mysqldump into a throwaway container first — runbook below) via PyMySQL and maps it into the live IILMP models: CustomUser / EntrepreneurProfile / Business / MentorProfile / incubation Programme+Rubric+Application+JudgeScore / Faq / NewsletterSubscriber / MeetingRequest, plus the UltimatePOS tenant link (and optional sales). The competitions are a jQuery-formBuilder dynamic-form engine: ``calls.form`` is a field schema, ``applications.data`` is ``{field: answer}``, ``guides.data`` is the schema annotated with ``marks`` (the rubric), and ``results.data`` is ``{field: judge_score}``. We derive RubricSections from the marked guide fields and explode each result into per-section JudgeScores while keeping the raw JSON (``Application.form_responses`` / ``form_answers``, ``JudgeScore.score_breakdown``) so nothing is lost. Idempotent on the ``external_ref`` anchors; FSM statuses are seeded directly (constructor on create, ``.update()`` on existing) — import fires no transitions. Passwords are NOT migrated: accounts are created with an unusable password and imported users reset on first login (matches the Moodle importer). Runbook:: docker run -d --name ehc-src --network ruforum-iilmp_default \ -e MARIADB_ROOT_PASSWORD=ehc -e MARIADB_DATABASE=ehc mariadb:10.11 docker exec -i ehc-src mariadb -uroot -pehc ehc < "smehub_ehc.sql" docker compose exec web python manage.py import_smehub_ehc \ --source-host ehc-src --source-user root --source-password ehc \ --source-db ehc --stage all --report /tmp/ehc_import.csv # optional UltimatePOS sales + products (separate dump): docker run -d --name pos-src --network ruforum-iilmp_default \ -e MARIADB_ROOT_PASSWORD=pos -e MARIADB_DATABASE=pos mariadb:10.11 docker exec -i pos-src mariadb -uroot -ppos pos < "smehub_accounting.sql" … import_smehub_ehc --stage pos --pos-source-host pos-src --pos-source-password pos --pos-source-db pos # file bytes (avatars · business logos · mentor resumes) — needs the EHC # server's Laravel storage/app/public tree (NOT in the SQL dumps; server # access currently blocked). Once obtained: docker cp storage/app/public ruforum-iilmp-web-1:/tmp/ehc-storage … import_smehub_ehc --stage files # idempotent; skips gracefully if absent """ from __future__ import annotations import csv import json from datetime import date, datetime, timezone as dt_timezone from decimal import Decimal, InvalidOperation from django.contrib.auth import get_user_model from django.contrib.contenttypes.models import ContentType from django.core.management.base import BaseCommand, CommandError from django.db import IntegrityError, transaction from django.utils import timezone from django.utils.text import slugify from apps.core.permissions.roles import UserRole User = get_user_model() STAGES = [ "users", "businesses", "mentors", "competitions", "applications", "results", "meetings", "faqs", "subscribers", "community", "pos", "files", ] # imported `businesses.stage` → BusinessStage code STAGE_MAP = { "idea": "idea", "start": "mvp", "break": "early_revenue", "profitable": "growth", } def _ts(value) -> datetime | None: """MySQL DATETIME/TIMESTAMP → aware datetime (PyMySQL returns datetime already).""" if not value: return None if isinstance(value, datetime): return value if value.tzinfo else value.replace(tzinfo=dt_timezone.utc) return None def _date_deadline(value) -> datetime | None: """MySQL DATE deadline → end of that day, UTC. PyMySQL returns DATE columns as datetime.date (not datetime), which _ts drops — calls.expires_at came through as None and historical competitions rendered as open forever. A deadline date means "open through that day". """ if isinstance(value, datetime): return _ts(value) if isinstance(value, date): return datetime(value.year, value.month, value.day, 23, 59, 59, tzinfo=dt_timezone.utc) return None # Literal junk strings the EHC forms stored instead of an empty value # (JS `null`/`undefined` serialised into the column, or user-typed filler). _NULL_SENTINELS = {"null", "undefined", "n/a", "na", "none", "-", "--"} def _txt(value, limit: int | None = None) -> str: """Trim, drop null-sentinel filler, and optionally cap a source string.""" s = (value or "").strip() if s.lower() in _NULL_SENTINELS: s = "" return s[:limit] if limit else s def _url(value, limit: int = 255) -> str: """Like _txt but guarantees an absolute http(s) URL (or empty).""" s = _txt(value, limit) if s and not s.lower().startswith(("http://", "https://")): s = ("https://" + s)[:limit] return s def _split_name(name: str) -> tuple[str, str]: name = (name or "").strip() if not name: return "", "" parts = name.split(" ", 1) return parts[0][:150], (parts[1][:150] if len(parts) > 1 else "") class Command(BaseCommand): help = "Import the SME-Hub (EHC) database into apps.smehub." def add_arguments(self, parser): parser.add_argument("--source-host", default="ehc-src") parser.add_argument("--source-port", type=int, default=3306) parser.add_argument("--source-user", default="root") parser.add_argument("--source-password", default="ehc") parser.add_argument("--source-db", default="ehc") parser.add_argument("--stage", default="all", help=f"all | {' | '.join(STAGES)}") parser.add_argument("--limit", type=int, default=0) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--flush", action="store_true") parser.add_argument("--report", default="") parser.add_argument( "--files-root", default="/tmp/ehc-storage", help="Laravel storage/app/public tree (holds avatars/, business/logos/, " "mentor_resumes/). Stage skips gracefully when absent — the EHC " "server files are not in the SQL dumps.") # Optional UltimatePOS source (for the `pos` stage sales import). parser.add_argument("--pos-source-host", default="") parser.add_argument("--pos-source-port", type=int, default=3306) parser.add_argument("--pos-source-user", default="root") parser.add_argument("--pos-source-password", default="") parser.add_argument("--pos-source-db", default="") # ── entry point ────────────────────────────────────────────────────────── def handle(self, *args, **opts): try: import pymysql from pymysql.cursors import DictCursor except ImportError as exc: # pragma: no cover raise CommandError("PyMySQL is required: pip install PyMySQL") from exc self._DictCursor = DictCursor self._pymysql = pymysql stage = opts["stage"] if stage != "all" and stage not in STAGES: raise CommandError(f"Unknown stage {stage!r}. all | {' | '.join(STAGES)}") self.opts = opts self.limit = opts["limit"] self.report_rows: list[dict] = [] self._country_cache: dict[str, str] = {} self.conn = self._connect( opts["source_host"], opts["source_port"], opts["source_user"], opts["source_password"], opts["source_db"], ) try: with transaction.atomic(): if opts["flush"]: self._flush() self._rebuild_caches() for st in (STAGES if stage == "all" else [stage]): self.stdout.write(self.style.MIGRATE_HEADING(f"▶ stage: {st}")) getattr(self, f"_stage_{st}")() if opts["dry_run"]: self.stdout.write(self.style.WARNING("dry-run — rolling back")) transaction.set_rollback(True) finally: self.conn.close() if opts["report"]: self._write_report(opts["report"]) self.stdout.write(self.style.SUCCESS("EHC import complete.")) # ── source helpers ─────────────────────────────────────────────────────── def _connect(self, host, port, user, password, db): try: return self._pymysql.connect( host=host, port=port, user=user, password=password, database=db, cursorclass=self._DictCursor, charset="utf8mb4", ) except Exception as exc: # pragma: no cover raise CommandError(f"Could not connect to source DB {db}@{host}: {exc}") from exc def _query(self, sql, params=None, conn=None) -> list[dict]: conn = conn or self.conn with conn.cursor() as cur: cur.execute(sql, params or ()) rows = list(cur.fetchall()) return rows[: self.limit] if self.limit else rows def _note(self, stage, **counts): self.report_rows.append({"stage": stage, **counts}) self.stdout.write(" " + " ".join(f"{k}={v}" for k, v in counts.items())) # ── caches ─────────────────────────────────────────────────────────────── def _rebuild_caches(self): from apps.smehub.incubation.models import Programme from apps.smehub.onboarding.models import Business, EntrepreneurProfile self._user_cache = dict( User.objects.exclude(external_ref=None) .values_list("external_ref", "pk") ) self._business_by_ref = dict( Business.objects.exclude(external_ref=None) .values_list("external_ref", "pk") ) # business by owning user (source user id) — for application resolution self._business_by_user = {} self._entrepreneur_by_user = dict( EntrepreneurProfile.objects.exclude(user__external_ref=None) .values_list("user__external_ref", "pk") ) self._programme_by_call = dict( Programme.objects.exclude(external_ref=None) .values_list("external_ref", "pk") ) def _user_pk(self, source_uid): if source_uid in (None, 0): return None return self._user_cache.get(int(source_uid)) def _country_name(self, code) -> str: if not code: return "" code = code.strip().upper() if code in self._country_cache: return self._country_cache[code] name = code try: import pycountry c = pycountry.countries.get(alpha_2=code) if c: name = c.name except Exception: pass self._country_cache[code] = name return name[:100] @staticmethod def _gender(value) -> str: from apps.core.authentication.models import UserProfile v = (value or "").strip().lower() if v == "male": return UserProfile.Gender.MALE if v == "female": return UserProfile.Gender.FEMALE return "" def _ensure_profile(self, user): from apps.core.authentication.models import UserProfile prof, _ = UserProfile.objects.get_or_create(user=user) return prof # ── flush ──────────────────────────────────────────────────────────────── def _flush(self): from apps.smehub.incubation.models import ( Application, JudgeAssignment, JudgeScore, Programme, RubricSection, ScoringRubric, ) from apps.smehub.onboarding.models import ( Business, BusinessBaseline, EntrepreneurProfile, Faq, InvestorProfile, MeetingRequest, MentorProfile, MentorshipCategory, NewsletterSubscriber, ) self.stdout.write(self.style.WARNING("flushing previously-imported EHC records…")) JudgeScore.objects.exclude(external_ref=None).delete() JudgeAssignment.objects.filter(application__external_ref__isnull=False).delete() Application.objects.exclude(external_ref=None).delete() RubricSection.objects.exclude(external_ref=None).delete() ScoringRubric.objects.filter(programme__external_ref__isnull=False).delete() Programme.objects.exclude(external_ref=None).delete() MeetingRequest.objects.exclude(external_ref=None).delete() Faq.objects.exclude(external_ref=None).delete() NewsletterSubscriber.objects.exclude(external_ref=None).delete() BusinessBaseline.objects.filter(business__external_ref__isnull=False).delete() Business.objects.exclude(external_ref=None).delete() MentorProfile.objects.exclude(external_ref=None).delete() MentorshipCategory.objects.exclude(external_ref=None).delete() EntrepreneurProfile.objects.exclude(external_ref=None).delete() # ── STAGE: users ───────────────────────────────────────────────────────── # Source users.role_id → IILMP role for users who registered under a role # but never built the matching profile (the businesses/mentors stages set # the role again for profile-holders — same values, idempotent). Investor # registrants (role_id 4, profile table empty in source) get a bare # InvestorProfile shell and self-complete their focus later; legacy EHC # admins (role_id 1) map to SME_ADMIN (SME-Hub scope only — never Django # staff/superuser; gated behind the password reset). _ROLE_ID_MAP = { 1: UserRole.SME_ADMIN, 2: UserRole.ENTREPRENEUR, 3: UserRole.MENTOR, 4: UserRole.INVESTOR, } def _stage_users(self): from apps.smehub.onboarding.models import InvestorProfile rows = self._query( "SELECT id, name, email, alias, email_verified_at, deleted_at, role_id " "FROM users ORDER BY id" ) created = matched = skipped = roles_set = investor_profiles = 0 for r in rows: email = (r["email"] or "").strip().lower() if not email or "@" not in email: skipped += 1 continue first, last = _split_name(r["name"]) source_role = self._ROLE_ID_MAP.get(r.get("role_id")) user = (User.objects.filter(external_ref=r["id"]).first() or User.objects.filter(email__iexact=email).first()) if user: matched += 1 changed = [] if user.external_ref is None: user.external_ref = r["id"]; changed.append("external_ref") if not user.external_alias and r["alias"]: user.external_alias = r["alias"][:255]; changed.append("external_alias") if source_role and not user.role: user.role = source_role; changed.append("role"); roles_set += 1 if changed: user.save(update_fields=changed) else: user = User( email=email, first_name=first, last_name=last, external_ref=r["id"], external_alias=(r["alias"] or "")[:255], email_verified=bool(r["email_verified_at"]), is_active=r["deleted_at"] is None, role=source_role or "", ) if source_role: roles_set += 1 # Source passwords are NOT migrated — accounts are created with # an unusable password; users reset on first login. user.set_unusable_password() user.save() created += 1 # Investors carry no profile data in the source (investors table is # empty), so seed a bare shell they can complete via the live # RoleProfileCompletionView. Idempotent on the OneToOne(user). if user.role == UserRole.INVESTOR: _, inv_created = InvestorProfile.objects.get_or_create(user=user) if inv_created: investor_profiles += 1 self._user_cache[int(r["id"])] = user.pk self._note("users", created=created, matched=matched, skipped=skipped, roles_set=roles_set, investor_profiles=investor_profiles, total=len(rows)) # ── STAGE: businesses (+ founder EntrepreneurProfile + baseline) ───────── def _stage_businesses(self): from apps.smehub.onboarding.models import ( Business, BusinessBaseline, EntrepreneurProfile, ) rows = self._query("SELECT * FROM businesses ORDER BY id") ent_created = biz_created = updated = skipped = 0 for r in rows: owner_pk = self._user_pk(r["user_id"]) if not owner_pk: skipped += 1 continue owner = User.objects.get(pk=owner_pk) # Business owners need the entrepreneur role to reach their own # SME-Hub pages (EntrepreneurRequired gate). Never overwrite a # role granted elsewhere (e.g. a Moodle-matched learner upgrade). if not owner.role: owner.role = UserRole.ENTREPRENEUR owner.save(update_fields=["role"]) # Founder demographics → core.UserProfile (gender/DOB) prof = self._ensure_profile(owner) dirty = [] g = self._gender(r.get("gender")) if g and not prof.gender: prof.gender = g; dirty.append("gender") if r.get("date_of_birth") and not prof.date_of_birth: prof.date_of_birth = r["date_of_birth"]; dirty.append("date_of_birth") if dirty: prof.save(update_fields=dirty) # EntrepreneurProfile (verified — imported businesses were live) verified = bool(r.get("verified_at")) ent = EntrepreneurProfile.objects.filter(user=owner).first() ent_defaults = dict( is_student=bool(r.get("is_student")), education_institution=_txt(r.get("education_institution"), 255), external_ref=r["id"], country=self._country_name(r.get("country_code")), ) if ent: for k, v in ent_defaults.items(): setattr(ent, k, v) ent.save() else: ent = EntrepreneurProfile( user=owner, verification_status=(EntrepreneurProfile.Status.VERIFIED if verified else EntrepreneurProfile.Status.PENDING_VERIFICATION), verified_at=_ts(r.get("verified_at")), **ent_defaults, ) ent.save() ent_created += 1 self._entrepreneur_by_user[int(r["user_id"])] = ent.pk # Business stage_code = STAGE_MAP.get((r.get("stage") or "").strip().lower(), "idea") biz_defaults = dict( entrepreneur=ent, name=(r.get("name") or "Untitled business")[:255], sector=(r.get("sector") or "Other")[:128], business_stage=stage_code, stage_raw=(r.get("stage") or "")[:64], country=self._country_name(r.get("country_code")), description=_txt(r.get("profile")), pitch=_txt(r.get("pitch")), registration_number=_txt(r.get("reg_no"), 120), business_email=_txt(r.get("email"), 254), website=_url(r.get("website")), phone=_txt(r.get("phone_no"), 40), total_employees=self._pos_int(r.get("no_of_employees")), female_employees=self._pos_int(r.get("female_employees")), accounting_business_id=r.get("accounting_business_id") or None, external_ref=r["id"], ) biz = Business.objects.filter(external_ref=r["id"]).first() if biz: for k, v in biz_defaults.items(): setattr(biz, k, v) biz.save() Business.objects.filter(pk=biz.pk).update( verification_status=(Business.Status.VERIFIED if verified else Business.Status.PENDING_ADMIN_REVIEW)) updated += 1 else: biz = Business( verification_status=(Business.Status.VERIFIED if verified else Business.Status.PENDING_ADMIN_REVIEW), verified_at=_ts(r.get("verified_at")), **biz_defaults, ) biz.save() biz_created += 1 self._business_by_ref[r["id"]] = biz.pk self._business_by_user[int(r["user_id"])] = biz.pk # Baseline snapshot BusinessBaseline.objects.get_or_create( business=biz, defaults=dict( business_stage_at_entry=stage_code, fte_count=self._pos_int(r.get("no_of_employees")) or 0, ), ) self._note("businesses", entrepreneurs=ent_created, businesses=biz_created, updated=updated, skipped=skipped, total=len(rows)) # ── STAGE: mentors (+ categories) ──────────────────────────────────────── def _stage_mentors(self): from apps.smehub.onboarding.models import MentorProfile, MentorshipCategory cats = self._query("SELECT id, name FROM mentorship_categories ORDER BY id") cat_pk_by_ref = {} cat_created = 0 for c in cats: obj, was = MentorshipCategory.objects.update_or_create( external_ref=c["id"], defaults=dict(name=(c["name"] or "")[:255])) cat_pk_by_ref[c["id"]] = obj.pk cat_created += was # mentor ↔ category links links = self._query( "SELECT mentor_id, mentorship_category_id FROM mentors_mentorship_categories") cats_by_mentor: dict[int, list[int]] = {} for l in links: cats_by_mentor.setdefault(l["mentor_id"], []).append(l["mentorship_category_id"]) rows = self._query("SELECT * FROM mentors ORDER BY id") m_created = skipped = 0 for r in rows: user_pk = self._user_pk(r["user_id"]) if not user_pk: skipped += 1 continue user = User.objects.get(pk=user_pk) # Mentor gate parity with live registrations (see businesses stage). if not user.role: user.role = UserRole.MENTOR user.save(update_fields=["role"]) prof = self._ensure_profile(user) dirty = [] g = self._gender(r.get("gender")) if g and not prof.gender: prof.gender = g; dirty.append("gender") if r.get("date_of_birth") and not prof.date_of_birth: prof.date_of_birth = r["date_of_birth"]; dirty.append("date_of_birth") if dirty: prof.save(update_fields=dirty) verified = bool(r.get("approved_at")) langs = [s.strip() for s in (r.get("languages") or "").replace(";", ",").split(",") if s.strip()] defaults = dict( country=_txt(r.get("country"), 100), phone=_txt(r.get("phone_number"), 32), bio=(r.get("bio") or ""), years_experience=r.get("years_of_experience"), languages=langs, current_company=_txt(r.get("current_company"), 255), current_position=_txt(r.get("current_position"), 255), linkedin=_url(r.get("linkedin")), nationality=_txt(r.get("nationality"), 100), awards=(r.get("awards") or ""), publications=(r.get("publications") or ""), external_ref=r["id"], ) mp = MentorProfile.objects.filter(external_ref=r["id"]).first() if mp: for k, v in defaults.items(): setattr(mp, k, v) mp.save() MentorProfile.objects.filter(pk=mp.pk).update( verification_status=(MentorProfile.Status.VERIFIED if verified else MentorProfile.Status.PENDING_VERIFICATION)) else: mp = MentorProfile.objects.filter(user=user).first() if mp: for k, v in defaults.items(): setattr(mp, k, v) mp.save() else: mp = MentorProfile( user=user, verification_status=(MentorProfile.Status.VERIFIED if verified else MentorProfile.Status.PENDING_VERIFICATION), verified_at=_ts(r.get("approved_at")), **defaults, ) mp.save() m_created += 1 cat_pks = [cat_pk_by_ref[c] for c in cats_by_mentor.get(r["id"], []) if c in cat_pk_by_ref] if cat_pks: mp.categories.set(cat_pks) self._note("mentors", categories=cat_created, mentors=m_created, skipped=skipped, total=len(rows)) # ── STAGE: competitions (calls → Programme; guides → Rubric/Sections) ──── def _stage_competitions(self): from apps.smehub.incubation.models import ( Programme, RubricSection, ScoringRubric, ) calls = self._query("SELECT * FROM calls ORDER BY id") c_created = updated = 0 for r in calls: call_type = (Programme.CallType.EXTERNAL if (r.get("call_type") or "").lower() == "external" else Programme.CallType.COMPETITION) form_schema = self._json(r.get("form")) defaults = dict( title=(r.get("title") or "Untitled call")[:255], objectives=(r.get("body") or ""), excerpt=(r.get("excerpt") or ""), call_type=call_type, external_url=(r.get("external_url") or "")[:500], application_form_schema=form_schema, title_fr=(r.get("title_fr") or "")[:255], excerpt_fr=(r.get("excerpt_fr") or ""), description_fr=(r.get("body_fr") or ""), application_deadline=_date_deadline(r.get("expires_at")), ) prog = Programme.objects.filter(external_ref=r["id"]).first() status = (Programme.Status.PUBLISHED if r.get("published_at") else Programme.Status.DRAFT) if prog: for k, v in defaults.items(): setattr(prog, k, v) prog.save() Programme.objects.filter(pk=prog.pk).update(status=status) updated += 1 else: prog = Programme( external_ref=r["id"], status=status, slug=self._unique_slug(Programme, r.get("title"), r["id"]), published_at=_ts(r.get("published_at")), **defaults, ) prog.save() c_created += 1 self._programme_by_call[r["id"]] = prog.pk # guides → rubric + sections (derived from marked fields) guides = self._query("SELECT * FROM guides ORDER BY id") rub_created = sec_created = 0 for g in guides: prog_pk = self._programme_by_call.get(g["call_id"]) if not prog_pk: continue schema = self._json(g.get("data")) or [] rubric, _ = ScoringRubric.objects.update_or_create( programme_id=prog_pk, defaults=dict(pass_mark=g.get("passmark") or None, application_form_schema=schema), ) rub_created += 1 marked = [f for f in schema if isinstance(f, dict) and self._to_int(f.get("marks")) > 0 and f.get("name")] total = sum(self._to_int(f.get("marks")) for f in marked) or 1 for idx, f in enumerate(marked): marks = self._to_int(f.get("marks")) weight = (Decimal(marks) / Decimal(total)).quantize(Decimal("0.0001")) _, was = RubricSection.objects.update_or_create( rubric=rubric, form_field_name=f["name"][:128], defaults=dict( title=(f.get("label") or f["name"])[:255], instructions=(f.get("instructions") or ""), weight=weight, max_marks=min(marks, 32767), order=idx, ), ) sec_created += was self._note("competitions", programmes=c_created, updated=updated, rubrics=rub_created, sections=sec_created, total=len(calls)) # ── STAGE: applications ────────────────────────────────────────────────── def _stage_applications(self): from apps.smehub.incubation.models import Application # label maps per call from Programme.application_form_schema label_by_call = self._label_maps() rows = self._query("SELECT * FROM applications ORDER BY id") created = updated = skipped = dup = 0 for r in rows: prog_pk = self._programme_by_call.get(r["call_id"]) user_pk = self._user_pk(r["user_id"]) if not prog_pk or not user_pk: skipped += 1 continue ent_pk = self._resolve_entrepreneur(user_pk, r["user_id"]) biz_pk = self._business_by_user.get(int(r["user_id"])) data = self._json(r.get("data")) or {} labels = label_by_call.get(r["call_id"], {}) resolved = {labels.get(k, k): v for k, v in data.items()} if isinstance(data, dict) else {} status = (Application.Status.SUBMITTED if str(r.get("status")) == "1" else Application.Status.DRAFT) defaults = dict( programme_id=prog_pk, entrepreneur_id=ent_pk, business_id=biz_pk, form_responses=data, form_answers=resolved, submitted_at=_ts(r.get("created_at")) if status == Application.Status.SUBMITTED else None, **self._structured_from_resolved(resolved), ) existing = Application.objects.filter(external_ref=r["id"]).first() if existing: for k, v in defaults.items(): setattr(existing, k, v) existing.save() Application.objects.filter(pk=existing.pk).update(status=status) updated += 1 continue try: with transaction.atomic(): app = Application(external_ref=r["id"], status=status, **defaults) app.save() created += 1 except IntegrityError: dup += 1 self._note("applications", created=created, updated=updated, duplicates=dup, skipped=skipped, total=len(rows)) def _resolve_entrepreneur(self, user_pk, source_uid) -> int: from apps.smehub.onboarding.models import EntrepreneurProfile cached = self._entrepreneur_by_user.get(int(source_uid)) if cached: return cached ent = EntrepreneurProfile.objects.filter(user_id=user_pk).first() if not ent: ent = EntrepreneurProfile(user_id=user_pk) ent.save() self._entrepreneur_by_user[int(source_uid)] = ent.pk return ent.pk @staticmethod def _structured_from_resolved(resolved: dict) -> dict: """Best-effort fill of typed Application fields from label→answer pairs.""" out = {"business_description": "", "problem_statement": "", "traction": "", "funding_needs": "", "support_sought": ""} if not isinstance(resolved, dict): return out for label, answer in resolved.items(): if not answer: continue low = str(label).lower() if not out["business_description"] and ("describe your business" in low or "about your business" in low): out["business_description"] = str(answer)[:5000] elif not out["problem_statement"] and "problem" in low: out["problem_statement"] = str(answer)[:5000] elif not out["traction"] and ("traction" in low or "customers" in low or "revenue" in low): out["traction"] = str(answer)[:5000] elif not out["funding_needs"] and ("funding" in low or "amount" in low or "budget" in low): out["funding_needs"] = str(answer)[:5000] elif not out["support_sought"] and ("support" in low or "need" in low): out["support_sought"] = str(answer)[:5000] return out # ── STAGE: results (→ JudgeAssignment + JudgeScore) ────────────────────── def _stage_results(self): from apps.smehub.incubation.models import ( Application, JudgeAssignment, JudgeScore, RubricSection, ) # programme_pk → {form_field_name: section_pk} section_by_field: dict[int, dict[str, int]] = {} for prog_pk, fname, sec_pk in RubricSection.objects.exclude( form_field_name="" ).values_list("rubric__programme_id", "form_field_name", "pk"): section_by_field.setdefault(prog_pk, {})[fname] = sec_pk app_by_ref = dict( Application.objects.exclude(external_ref=None) .values_list("external_ref", "pk")) app_programme = dict( Application.objects.exclude(external_ref=None) .values_list("pk", "programme_id")) rows = self._query("SELECT * FROM results ORDER BY id") scores = skipped = assignments = 0 for r in rows: app_pk = app_by_ref.get(r["application_id"]) judge_pk = self._user_pk(r["judge_id"]) if not app_pk or not judge_pk: skipped += 1 continue JudgeAssignment.objects.get_or_create(application_id=app_pk, judge_id=judge_pk) assignments += 1 data = self._json(r.get("data")) or {} field_map = section_by_field.get(app_programme.get(app_pk), {}) first = True for fname, raw in (data.items() if isinstance(data, dict) else []): sec_pk = field_map.get(fname) if not sec_pk: continue val = self._to_decimal(raw) if val is None: continue JudgeScore.objects.update_or_create( application_id=app_pk, judge_id=judge_pk, section_id=sec_pk, defaults=dict( score=val, external_ref=r["id"], score_breakdown=data if first else None, ), ) scores += 1 first = False self._note("results", judge_assignments=assignments, scores=scores, skipped=skipped, total=len(rows)) # ── STAGE: meetings ────────────────────────────────────────────────────── def _stage_meetings(self): from apps.smehub.onboarding.models import ( Business, MeetingRequest, MentorProfile, ) mentor_ct = ContentType.objects.get_for_model(MentorProfile) business_ct = ContentType.objects.get_for_model(Business) # imported mentor.id → MentorProfile pk ; imported business.id → Business pk mentor_by_ref = dict( MentorProfile.objects.exclude(external_ref=None) .values_list("external_ref", "pk")) rows = self._query("SELECT * FROM meeting_requests ORDER BY id") created = skipped = 0 for r in rows: requester_pk = self._user_pk(r["requester_id"]) if not requester_pk: skipped += 1 continue rtype = (r.get("recipient_type") or "") if "Mentor" in rtype: ct, obj_pk = mentor_ct, mentor_by_ref.get(r["recipient_id"]) recipient_type = MeetingRequest.RecipientType.MENTOR elif "Business" in rtype: ct, obj_pk = business_ct, self._business_by_ref.get(r["recipient_id"]) recipient_type = MeetingRequest.RecipientType.BUSINESS else: skipped += 1 continue status_map = {"request": MeetingRequest.Status.REQUEST, "accepted": MeetingRequest.Status.ACCEPTED, "rejected": MeetingRequest.Status.REJECTED, "cancelled": MeetingRequest.Status.CANCELLED} MeetingRequest.objects.update_or_create( external_ref=r["id"], defaults=dict( requester_id=requester_pk, recipient_type=recipient_type, recipient_content_type=ct, recipient_object_id=obj_pk, meeting_time=_ts(r.get("meeting_time")), status=status_map.get((r.get("status") or "").lower(), MeetingRequest.Status.REQUEST), reason=(r.get("reason") or ""), agenda=(r.get("agenda") or ""), meeting_link=(r.get("meeting_link") or "")[:255], ), ) created += 1 self._note("meetings", created=created, skipped=skipped, total=len(rows)) # ── STAGE: faqs ────────────────────────────────────────────────────────── def _stage_faqs(self): from apps.smehub.onboarding.models import Faq rows = self._query("SELECT * FROM faqs ORDER BY id") created = 0 for idx, r in enumerate(rows): Faq.objects.update_or_create( external_ref=r["id"], defaults=dict( question=(r.get("question") or ""), answer=(r.get("answer") or ""), is_published=bool(r.get("display_on_frontend")), order=idx, ), ) created += 1 self._note("faqs", upserted=created, total=len(rows)) # ── STAGE: subscribers ─────────────────────────────────────────────────── def _stage_subscribers(self): from apps.smehub.onboarding.models import NewsletterSubscriber rows = self._query("SELECT * FROM email_subscribers ORDER BY id") created = skipped = 0 for r in rows: email = (r.get("email") or "").strip().lower() if not email or "@" not in email: skipped += 1 continue NewsletterSubscriber.objects.update_or_create( email=email, defaults=dict(external_ref=r["id"], subscribed_at=_ts(r.get("created_at"))), ) created += 1 self._note("subscribers", upserted=created, skipped=skipped, total=len(rows)) # ── STAGE: community (Chatter forum → read-only archive) ──────────────── def _stage_community(self): from apps.smehub.onboarding.models import CommunityDiscussion, CommunityPost categories = { r["id"]: (r.get("name") or "") for r in self._query("SELECT id, name FROM chatter_categories") } d_created = p_created = skipped = 0 disc_pk_by_ref: dict[int, int] = {} for r in self._query("SELECT * FROM chatter_discussion ORDER BY id"): author_pk = self._user_pk(r.get("user_id")) disc, was_created = CommunityDiscussion.objects.update_or_create( external_ref=r["id"], defaults=dict( title=_txt(r.get("title"), 255) or "Untitled discussion", category=_txt(categories.get(r.get("chatter_category_id")), 120), author_id=author_pk, ), ) if was_created: d_created += 1 if r.get("created_at"): # backdate (auto_now_add ignores constructor) CommunityDiscussion.objects.filter(pk=disc.pk).update( created_at=_ts(r["created_at"])) disc_pk_by_ref[r["id"]] = disc.pk for r in self._query("SELECT * FROM chatter_post ORDER BY id"): disc_pk = disc_pk_by_ref.get(r.get("chatter_discussion_id")) if not disc_pk: skipped += 1 continue post, was_created = CommunityPost.objects.update_or_create( external_ref=r["id"], defaults=dict( discussion_id=disc_pk, author_id=self._user_pk(r.get("user_id")), body=(r.get("body") or ""), ), ) if was_created: p_created += 1 if r.get("created_at"): CommunityPost.objects.filter(pk=post.pk).update( created_at=_ts(r["created_at"])) self._note("community", discussions=d_created, posts=p_created, skipped=skipped) # ── STAGE: pos (UltimatePOS sales → SalesRecord; optional) ─────────────── def _stage_pos(self): from apps.smehub.marketplace.models import ProductListing, SalesRecord from apps.smehub.onboarding.models import Business if not self.opts["pos_source_host"] or not self.opts["pos_source_db"]: self._note("pos", note="no --pos-source-host/db given — skipped (link already set on Business)") return pos = self._connect( self.opts["pos_source_host"], self.opts["pos_source_port"], self.opts["pos_source_user"], self.opts["pos_source_password"], self.opts["pos_source_db"], ) try: biz_by_acct = dict( Business.objects.exclude(accounting_business_id=None) .values_list("accounting_business_id", "pk")) contacts = {c["id"]: c for c in self._query( "SELECT id, name FROM contacts", conn=pos)} rows = self._query( "SELECT id, business_id, contact_id, invoice_no, transaction_date, " "final_total FROM transactions WHERE type = 'sell' ORDER BY id", conn=pos) created = skipped = 0 for r in rows: biz_pk = biz_by_acct.get(r["business_id"]) if not biz_pk: skipped += 1 continue biz = Business.objects.select_related("entrepreneur__user").get(pk=biz_pk) entrepreneur_user = biz.entrepreneur.user customer = (contacts.get(r["contact_id"], {}) or {}).get("name") or "" sale_date = r["transaction_date"].date() if r.get("transaction_date") else None SalesRecord.objects.update_or_create( source_system="ultimatepos", source_transaction_id=r["id"], defaults=dict( business_id=biz_pk, entrepreneur=entrepreneur_user, channel=SalesRecord.Channel.OFFLINE, gross_amount=self._to_decimal(r["final_total"]) or Decimal("0"), currency="UGX", sale_date=sale_date, customer_name=customer[:255], source_invoice_no=(r.get("invoice_no") or "")[:191], status=SalesRecord.Status.RECORDED, ), ) created += 1 # ── Products → ProductListing ──────────────────────────────── # Only tenants with real sell activity get their catalogue # imported — the POS install is largely a demo (the "martin # amitu" tenant holds 3 duplicate 150M-UGX "Tractor" test rows). selling_tenants = { r["business_id"] for r in self._query( "SELECT DISTINCT business_id FROM transactions WHERE type = 'sell'", conn=pos) } currency_by_tenant = { r["id"]: r["code"] for r in self._query( "SELECT b.id, c.code FROM business b " "LEFT JOIN currencies c ON c.id = b.currency_id", conn=pos) } qty_by_product: dict[int, Decimal] = {} for r in self._query( "SELECT product_id, SUM(qty_available) AS qty " "FROM variation_location_details GROUP BY product_id", conn=pos): qty_by_product[r["product_id"]] = self._to_decimal(r["qty"]) or Decimal("0") price_by_product = { r["product_id"]: self._to_decimal(r["default_sell_price"]) for r in self._query( "SELECT product_id, MIN(default_sell_price) AS default_sell_price " "FROM variations GROUP BY product_id", conn=pos) } prod_rows = self._query( "SELECT p.id, p.business_id, p.name, p.sku, p.created_at, " "c.name AS category, u.short_name AS unit " "FROM products p " "LEFT JOIN categories c ON c.id = p.category_id " "LEFT JOIN units u ON u.id = p.unit_id ORDER BY p.id", conn=pos) p_created = p_already = p_skipped = 0 for r in prod_rows: biz_pk = biz_by_acct.get(r["business_id"]) if not biz_pk or r["business_id"] not in selling_tenants: p_skipped += 1 continue biz = Business.objects.select_related("entrepreneur").get(pk=biz_pk) price = price_by_product.get(r["id"]) currency = currency_by_tenant.get(r["business_id"]) or "UGX" unit = _txt(r.get("unit"), 32) or "unit" qty = qty_by_product.get(r["id"], Decimal("0")) listing, was_created = ProductListing.objects.update_or_create( external_ref=r["id"], defaults=dict( entrepreneur=biz.entrepreneur, business=biz, name=_txt(r.get("name"), 255) or "Untitled product", sector=biz.sector, sub_sector=_txt(r.get("category"), 128), pricing=(f"{currency} {price:,.0f} / {unit}" if price else ""), availability=(ProductListing.Availability.IN_STOCK if qty > 0 else ProductListing.Availability.OUT_OF_STOCK), status=ProductListing.Status.PUBLISHED, published_at=_ts(r.get("created_at")), ), ) if was_created: p_created += 1 if r.get("created_at"): # backdate (auto_now_add ignores constructor) ProductListing.objects.filter(pk=listing.pk).update( created_at=_ts(r["created_at"])) else: p_already += 1 # ── Link imported sales to their product listing ───────────── listing_by_ref = dict( ProductListing.objects.exclude(external_ref=None) .values_list("external_ref", "pk")) linked = 0 for r in self._query( "SELECT transaction_id, MIN(product_id) AS product_id " "FROM transaction_sell_lines GROUP BY transaction_id", conn=pos): listing_pk = listing_by_ref.get(r["product_id"]) if not listing_pk: continue linked += SalesRecord.objects.filter( source_system="ultimatepos", source_transaction_id=r["transaction_id"], ).exclude(product_listing_id=listing_pk).update(product_listing_id=listing_pk) self._note("pos", sales=created, skipped=skipped, total=len(rows), products=p_created, products_already=p_already, products_skipped=p_skipped, sales_linked=linked) finally: pos.close() # ── STAGE: files (Laravel storage/app/public — avatars · logos · resumes) # The EHC dumps are SQL-only; this stage runs once the server's # storage/app/public tree is copied to --files-root. NOTE: dry-run skips # all writes here — saved media files would not roll back with the txn. def _stage_files(self): import os root = self.opts["files_root"] if not os.path.isdir(root): self._note("files", note=f"files-root {root!r} not found — stage skipped " f"(copy the EHC server's storage/app/public there first)") return self._files_avatars() self._files_business_logos() self._files_mentor_resumes() def _store_file(self, instance, field_name: str, rel_path: str) -> str: """Attach / to instance.. Returns 'created' | 'missing'. Saves the file with save=False and persists via queryset .update() so FSM-protected models never re-save. """ import os from django.core.files import File src = os.path.join(self.opts["files_root"], rel_path.lstrip("/")) if not os.path.isfile(src): return "missing" field = getattr(instance, field_name) with open(src, "rb") as fh: field.save(os.path.basename(rel_path), File(fh), save=False) type(instance).objects.filter(pk=instance.pk).update( **{field_name: field.name}) return "created" def _files_avatars(self): """users.avatar → core.UserProfile.avatar (89 non-default in source).""" rows = self._query( "SELECT id, avatar FROM users " "WHERE avatar IS NOT NULL AND avatar != '' AND avatar NOT LIKE %s " "ORDER BY id", ("%default%",)) created = already = missing = skipped = 0 for r in rows: user_pk = self._user_pk(r["id"]) if not user_pk: skipped += 1 continue prof = self._ensure_profile(User.objects.get(pk=user_pk)) if prof.avatar: already += 1 continue if self.opts["dry_run"]: created += 1 continue result = self._store_file(prof, "avatar", r["avatar"]) if result == "created": created += 1 else: missing += 1 self._note("files:avatars", created=created, already=already, missing_file=missing, skipped=skipped, total=len(rows)) def _files_business_logos(self): """businesses.logo → onboarding.Business.logo (286 in source).""" from apps.smehub.onboarding.models import Business rows = self._query( "SELECT id, logo FROM businesses " "WHERE logo IS NOT NULL AND logo != '' ORDER BY id") by_ref = dict(Business.objects.exclude(external_ref=None) .values_list("external_ref", "pk")) created = already = missing = skipped = 0 for r in rows: biz_pk = by_ref.get(r["id"]) if not biz_pk: skipped += 1 continue biz = Business.objects.get(pk=biz_pk) if biz.logo: already += 1 continue if self.opts["dry_run"]: created += 1 continue result = self._store_file(biz, "logo", r["logo"]) if result == "created": created += 1 else: missing += 1 self._note("files:logos", created=created, already=already, missing_file=missing, skipped=skipped, total=len(rows)) def _files_mentor_resumes(self): """mentors.resume → onboarding.MentorProfile.resume (2 in source).""" from apps.smehub.onboarding.models import MentorProfile rows = self._query( "SELECT id, resume FROM mentors " "WHERE resume IS NOT NULL AND resume != '' ORDER BY id") by_ref = dict(MentorProfile.objects.exclude(external_ref=None) .values_list("external_ref", "pk")) created = already = missing = skipped = 0 for r in rows: mp_pk = by_ref.get(r["id"]) if not mp_pk: skipped += 1 continue mp = MentorProfile.objects.get(pk=mp_pk) if mp.resume: already += 1 continue if self.opts["dry_run"]: created += 1 continue result = self._store_file(mp, "resume", r["resume"]) if result == "created": created += 1 else: missing += 1 self._note("files:resumes", created=created, already=already, missing_file=missing, skipped=skipped, total=len(rows)) # ── small utilities ────────────────────────────────────────────────────── def _json(self, raw): if raw in (None, ""): return None if isinstance(raw, (dict, list)): return raw try: return json.loads(raw) except (json.JSONDecodeError, TypeError): return None @staticmethod def _pos_int(v): """Non-negative int or None (imported employee counts include negatives).""" try: n = int(v) except (TypeError, ValueError): return None return n if n >= 0 else None @staticmethod def _to_int(v) -> int: try: return int(float(v)) except (TypeError, ValueError): return 0 @staticmethod def _to_decimal(v): if v in (None, ""): return None try: return Decimal(str(v)) except (InvalidOperation, TypeError, ValueError): return None def _label_maps(self) -> dict[int, dict[str, str]]: from apps.smehub.incubation.models import Programme out: dict[int, dict[str, str]] = {} for call_ref, schema in Programme.objects.exclude( external_ref=None ).values_list("external_ref", "application_form_schema"): mapping = {} for f in (schema or []): if isinstance(f, dict) and f.get("name"): mapping[f["name"]] = (f.get("label") or f["name"]) out[call_ref] = mapping return out def _unique_slug(self, model, base, ref_id) -> str: slug = (slugify(base or "") or f"call-{ref_id}")[:110] candidate, n = slug, 1 qs = model.objects.exclude(external_ref=ref_id) while qs.filter(slug=candidate).exists(): candidate = f"{slug}-{n}"; n += 1 return candidate[:120] def _write_report(self, path): keys: list[str] = [] for row in self.report_rows: for k in row: if k not in keys: keys.append(k) with open(path, "w", newline="") as fh: w = csv.DictWriter(fh, fieldnames=keys) w.writeheader() for row in self.report_rows: w.writerow(row) self.stdout.write(self.style.SUCCESS(f"report → {path}"))