"""SME-Hub incubation services — Subprocess SP2. Each function corresponds to one or more PRD functional requirements (FRSME-INC001 → INC037). Side-effects (notifications, audit log, domain-signal emission) live here, not on the view layer. """ from __future__ import annotations import logging import re from collections import defaultdict from datetime import timedelta from decimal import Decimal from typing import Iterable from django.conf import settings from django.contrib.auth import get_user_model from django.core.exceptions import ValidationError from django.db import models, transaction from django.urls import NoReverseMatch, reverse from django.utils import timezone from apps.core.notifications.models import Notification from apps.smehub._shared.audit import AuditLog, _audit from apps.smehub._shared.notifications import bulk_notify_smehub, notify_smehub from apps.smehub.incubation import signals as inc_signals from apps.smehub.incubation.models import ( Application, Certificate, Cohort, CohortMember, JudgeAssignment, JudgeScore, MentorAssignment, MentorSession, ProgressRecord, Programme, ProgrammeCourseLink, RubricSection, ScoringRubric, ) from apps.smehub.onboarding.models import Business, EntrepreneurProfile User = get_user_model() Verb = Notification.Verb logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Exceptions # --------------------------------------------------------------------------- class ProgrammeNotPublishable(Exception): """Raised when a programme is missing required pieces (rubric, deadline).""" class EligibilityError(Exception): """Raised when a business does not meet a programme's eligibility rules.""" class ApplicationStateError(Exception): """Raised on illegal application transitions.""" class IncubationStateError(Exception): """Raised when an SP2 precondition is not yet satisfied (e.g. baseline missing).""" def _absolute(path: str) -> str: base = getattr(settings, "PUBLIC_APP_BASE_URL", "").rstrip("/") return f"{base}{path}" if base else path # --------------------------------------------------------------------------- # Programme creation + publication (FRSME-INC001 → INC007) # --------------------------------------------------------------------------- @transaction.atomic def create_programme(*, officer, **fields) -> Programme: # M2M fields (e.g. the FRSME-INC017 judge roster) can't be passed to # objects.create() — set them after the row exists. judges = fields.pop("judges", None) programme = Programme.objects.create(created_by=officer, **fields) if judges is not None: programme.judges.set(judges) ScoringRubric.objects.create(programme=programme) _audit(officer, AuditLog.Action.CREATE, programme, metadata={"title": programme.title}) return programme @transaction.atomic def add_rubric_section( programme: Programme, *, title: str, weight, max_marks: int = 10, instructions: str = "", order: int = 0, ) -> RubricSection: rubric, _ = ScoringRubric.objects.get_or_create(programme=programme) section = RubricSection.objects.create( rubric=rubric, title=title, weight=Decimal(str(weight)), max_marks=max_marks, instructions=instructions, order=order, ) return section @transaction.atomic def link_programme_course( programme: Programme, *, moodle_course_id: str, course_title: str = "", is_required: bool = True, notes: str = "", ) -> ProgrammeCourseLink: """Link a programme to a Moodle course (by id) whose completion counts toward cohort progress. Actual enrolment is delegated to Moodle.""" link, _ = ProgrammeCourseLink.objects.get_or_create( programme=programme, moodle_course_id=str(moodle_course_id), defaults={"course_title": course_title, "is_required": is_required, "notes": notes}, ) return link @transaction.atomic def publish_programme(programme: Programme, *, by_user) -> Programme: """FRSME-INC004 + INC006 — gates publication on rubric completeness + presence of required fields. """ if programme.status != Programme.Status.DRAFT: raise ProgrammeNotPublishable(f"Programme is already {programme.get_status_display().lower()}.") rubric = getattr(programme, "rubric", None) if rubric is None or not rubric.is_complete: raise ProgrammeNotPublishable( "Scoring rubric must be defined and section weights must sum to 1.0 before publishing.", ) if not programme.application_deadline: raise ProgrammeNotPublishable("Set an application deadline before publishing.") # FRSME-INC002 — a programme cannot be published without at least one # assigned judge (the scoring stage would otherwise have no reviewers). if not programme.judges.filter(is_active=True).exists(): raise ProgrammeNotPublishable( "Assign at least one judge to the programme before publishing.", ) programme.publish() programme.save() _audit(by_user, AuditLog.Action.UPDATE, programme, metadata={"status": "published"}) inc_signals.smehub_programme_published.send( sender=Programme, programme=programme, by_user=by_user, ) # FRSME-INC007 — automatic targeted email + dashboard fan-out to every # eligible entrepreneur. Audit found this task existed but only ran from # a manual admin button. Enqueue via Celery so the publish request stays # snappy under large eligible audiences. try: from apps.smehub.incubation.tasks import dispatch_eligibility_alerts_task dispatch_eligibility_alerts_task.delay(programme.pk) except Exception: # pragma: no cover - Celery broker problem must not break publish logger.exception( "INC007 eligibility-alert dispatch failed to enqueue for programme %s", programme.pk, ) return programme @transaction.atomic def close_programme(programme: Programme, *, by_user) -> Programme: programme.close() programme.save() _audit(by_user, AuditLog.Action.UPDATE, programme, metadata={"status": "closed"}) inc_signals.smehub_programme_closed.send( sender=Programme, programme=programme, by_user=by_user, ) return programme def notify_eligible_entrepreneurs(programme: Programme) -> int: """FRSME-INC007. Returns the number of users notified.""" qs = Business.objects.filter(verification_status=Business.Status.VERIFIED) if programme.target_sectors: qs = qs.filter(sector__in=programme.target_sectors) if programme.target_stages: qs = qs.filter(business_stage__in=programme.target_stages) if programme.eligibility_countries: qs = qs.filter(country__in=programme.eligibility_countries) user_ids = ( qs.select_related("entrepreneur__user") .values_list("entrepreneur__user_id", flat=True) .distinct() ) users = list(User.objects.filter(pk__in=user_ids, is_active=True)) if not users: return 0 try: path = reverse("smehub_incubation:programme_detail", kwargs={"slug": programme.slug}) except NoReverseMatch: path = "" bulk_notify_smehub( users, f"You are eligible to apply to “{programme.title}”. Deadline {programme.application_deadline:%d %b %Y}.", verb=Verb.SMEHUB_PROGRAMME_ELIGIBLE, action_url=_absolute(path) if path else "", ) return len(users) # --------------------------------------------------------------------------- # Application lifecycle (FRSME-INC008 → INC016) # --------------------------------------------------------------------------- def _normalise_aih_name(value: str) -> str: """Normalise an AIH/institution name for tolerant comparison. ``Programme.aih_locations`` is a free-form JSON list typed by an admin ("Makerere AIH"), while ``AIHAffiliation.institution`` points at the Institution directory ("Makerere University"). Requiring an exact string match silently locks every entrepreneur out of the programme, so both sides are casefolded, stripped of punctuation, and common hub suffixes ("AIH", "hub", "innovation hub") are dropped before comparing. """ cleaned = re.sub(r"[^\w\s]", " ", (value or "").casefold()) cleaned = re.sub(r"\s+", " ", cleaned).strip() for suffix in (" innovation hub", " aih", " hub"): if cleaned.endswith(suffix): cleaned = cleaned[: -len(suffix)].strip() break return cleaned def aih_location_matches(location: str, institution_name: str) -> bool: """FRSME-INC008 — tolerant venue ↔ institution match. Conservative on purpose: after normalisation the names must be equal, or one must be a prefix of the other (covers "Makerere AIH" → "makerere" vs "Makerere University" → "makerere university"). Anything fuzzier risks granting eligibility across unrelated institutions. """ a = _normalise_aih_name(location) b = _normalise_aih_name(institution_name) if not a or not b: return False return a == b or a.startswith(b) or b.startswith(a) def check_eligibility(business: Business, programme: Programme) -> tuple[bool, str]: if business.verification_status != Business.Status.VERIFIED: return False, "Business must be verified before applying." if programme.status != Programme.Status.PUBLISHED: return False, "Programme is not open for applications." if programme.application_deadline and programme.application_deadline < timezone.now(): return False, "Application deadline has passed." def _norm(values) -> set[str]: return {str(v).strip().lower() for v in values} # Case-insensitive matching: business fields are free-text (legacy imports # mix "Agriculture"/"agriculture") while programme lists come from forms. if programme.target_sectors and business.sector.strip().lower() not in _norm(programme.target_sectors): return False, "Business sector is outside the programme's eligibility." if programme.target_stages and business.business_stage.strip().lower() not in _norm(programme.target_stages): return False, "Business stage is outside the programme's eligibility." if programme.eligibility_countries and business.country.strip().lower() not in _norm(programme.eligibility_countries): return False, "Business country is outside the programme's eligibility." # FRSME-INC008 — AIH affiliation eligibility gate (audit fix). Programmes # may scope themselves to specific AIH locations; entrepreneurs without a # matching primary affiliation are filtered out before they can apply. aih_locations = list(getattr(programme, "aih_locations", []) or []) if aih_locations: from apps.smehub.onboarding.models import AIHAffiliation affiliation_names = AIHAffiliation.objects.filter( business=business, ).values_list("institution__name", flat=True) has_matching_aih = any( aih_location_matches(location, name) for location in aih_locations for name in affiliation_names ) if not has_matching_aih: return False, "Business is not affiliated with an AIH location for this programme." return True, "" def entrepreneur_has_baseline(entrepreneur: EntrepreneurProfile) -> bool: """FRSME-MEI002 — SP1 baseline initialised (SMEHubTrackingRecord exists).""" from apps.mel.tracking.models import SMEHubTrackingRecord return SMEHubTrackingRecord.objects.filter(entrepreneur=entrepreneur).exists() @transaction.atomic def start_application(*, entrepreneur: EntrepreneurProfile, business: Business, programme: Programme) -> Application: """FRSME-INC008 + INC009. Idempotent — re-opening a returned-for-info application reuses the same row. FRSME-MEI002 — refuses to start an application unless an SMEHubTrackingRecord has been initialised for the entrepreneur in SP1 (i.e. the SP1 baseline has been locked and emitted ``smehub_baseline_initialised``). Without that, M&EL has nowhere to anchor SP2 milestones. """ from apps.mel.tracking.models import SMEHubTrackingRecord if not SMEHubTrackingRecord.objects.filter(entrepreneur=entrepreneur).exists(): raise IncubationStateError( "Baseline must be initialised in SP1 before applying to a programme." ) ok, reason = check_eligibility(business, programme) if not ok: raise EligibilityError(reason) existing = ( Application.objects.filter(programme=programme, business=business) .exclude(status__in=[Application.Status.WITHDRAWN, Application.Status.REJECTED, Application.Status.ACCEPTED]) .first() ) if existing: return existing application = Application.objects.create( programme=programme, entrepreneur=entrepreneur, business=business, ) _audit(entrepreneur.user, AuditLog.Action.CREATE, application) return application @transaction.atomic def auto_save_application(application: Application, *, payload: dict) -> Application: if application.status not in {Application.Status.DRAFT, Application.Status.RETURNED_FOR_INFO}: raise ApplicationStateError("Only drafts can be auto-saved.") application.auto_saved_payload = payload or {} for field in ("business_description", "problem_statement", "traction", "funding_needs", "support_sought"): if field in payload: setattr(application, field, payload[field] or "") application.save( update_fields=[ "auto_saved_payload", "business_description", "problem_statement", "traction", "funding_needs", "support_sought", "updated_at", ], ) return application @transaction.atomic def submit_application(application: Application) -> Application: # FRSME-INC014 — deadline check at submit time, not just at start. The # audit found a draft started before the deadline could still be # submitted afterwards. Programmes without a deadline (admin-curated # invitations) skip this check. deadline = application.programme.application_deadline if deadline and deadline < timezone.now(): raise ApplicationStateError("Application deadline has passed.") # FRSME-INC012 (Slice C) — block submission of empty applications. The # auto-save path lets entrepreneurs persist drafts incrementally, but # submit must enforce mandatory content. Keep the rule small and easy # to extend by adding fields to the required list. required = ("business_description", "problem_statement") missing = [f for f in required if not (getattr(application, f, "") or "").strip()] if missing: # Surface human field labels, not model attribute names. labels = [ Application._meta.get_field(f).verbose_name.capitalize() for f in missing ] raise ApplicationStateError( f"Complete the required section(s) before submitting: {', '.join(labels)}.", ) # FRSME-INC010/012 — enforce every supporting document the programme flags # as mandatory. A requirement is satisfied by an uploaded document whose # label matches (case-insensitively) the required label. required_docs = [str(d).strip() for d in (application.programme.required_document_labels or []) if str(d).strip()] if required_docs: uploaded_labels = { (label or "").strip().lower() for label in application.documents.values_list("label", flat=True) } missing_docs = [d for d in required_docs if d.lower() not in uploaded_labels] if missing_docs: raise ApplicationStateError( f"Attach the required supporting document(s) before submitting: {', '.join(missing_docs)}.", ) if application.status == Application.Status.RETURNED_FOR_INFO: application.resubmit() elif application.status == Application.Status.DRAFT: application.submit() else: raise ApplicationStateError("Application is not in a submittable state.") application.save() _audit(application.entrepreneur.user, AuditLog.Action.UPDATE, application, metadata={"status": application.status}) inc_signals.smehub_application_submitted.send(sender=Application, application=application) notify_smehub( application.entrepreneur.user, f"We have received your application “{application.application_id}” for {application.programme.title}.", verb=Verb.SMEHUB_APPLICATION_RECEIVED, target=application, ) return application @transaction.atomic def pass_first_level_review(application: Application, *, officer) -> Application: """FRSME-INC015/INC017 — programme officer marks the file complete; ready for judging. If the programme has a configured judge roster, the application is auto-assigned to every roster judge (each notified via dashboard + email), so officers no longer have to assign manually. """ if application.status != Application.Status.SUBMITTED: raise ApplicationStateError("Only submitted applications can move to review.") application.begin_review() application.save() _audit(officer, AuditLog.Action.UPDATE, application, metadata={"status": "under_review"}) # FRSME-INC017 — auto-assign the programme's judge roster (idempotent; # assign_judges get_or_creates and only notifies newly-created rows). roster = list(application.programme.judges.filter(is_active=True)) if roster: assign_judges(application, judges=roster, by_user=officer) return application @transaction.atomic def return_for_info(application: Application, *, officer, message: str) -> Application: """FRSME-INC016.""" application.return_for_info(message=message) application.save() _audit(officer, AuditLog.Action.UPDATE, application, metadata={"status": "returned_for_info"}) inc_signals.smehub_application_returned.send( sender=Application, application=application, by_user=officer, message=message, ) notify_smehub( application.entrepreneur.user, f"Application {application.application_id} returned: {message}", verb=Verb.SMEHUB_APPLICATION_RETURNED, target=application, ) return application # --------------------------------------------------------------------------- # Judging (FRSME-INC017 → INC021) # --------------------------------------------------------------------------- @transaction.atomic def assign_judges(application: Application, *, judges: Iterable, by_user=None) -> list[JudgeAssignment]: if application.status not in {Application.Status.SUBMITTED, Application.Status.UNDER_REVIEW}: raise ApplicationStateError("Application must be submitted/under review to assign judges.") assignments: list[JudgeAssignment] = [] for judge in judges: a, created = JudgeAssignment.objects.get_or_create(application=application, judge=judge) if created: a.notified_at = timezone.now() a.save(update_fields=["notified_at"]) try: path = reverse( "smehub_incubation:judge_score_form", kwargs={"pk": application.pk}, ) except NoReverseMatch: path = "" notify_smehub( judge, f"You have been assigned to score application {application.application_id} ({application.business.name}).", verb=Verb.SMEHUB_JUDGE_ASSIGNED, action_url=_absolute(path) if path else "", target=application, ) assignments.append(a) inc_signals.smehub_judges_assigned.send( sender=Application, application=application, assignments=assignments, ) if application.status == Application.Status.SUBMITTED: application.begin_review() application.save() return assignments @transaction.atomic def submit_score( application: Application, *, judge, section_scores: dict, comments: str = "", ) -> list[JudgeScore]: """FRSME-INC018/019. ``section_scores`` is a dict of {section_id: score}. Existing scores are upserted so a judge can revise before the deadline. """ if not JudgeAssignment.objects.filter(application=application, judge=judge).exists(): raise ApplicationStateError("Judge is not assigned to this application.") rubric_section_ids = list(application.programme.rubric.sections.values_list("pk", flat=True)) saved: list[JudgeScore] = [] for section_id, score in section_scores.items(): sid = int(section_id) if sid not in rubric_section_ids: continue section = RubricSection.objects.get(pk=sid) s = Decimal(str(score)) if s < 0 or s > Decimal(str(section.max_marks)): raise ValidationError(f"Score out of range for section {section.title}.") obj, _ = JudgeScore.objects.update_or_create( application=application, judge=judge, section=section, defaults={"score": s, "comments": comments, "submitted_at": timezone.now()}, ) saved.append(obj) # FRSME-INC019 — record the judge's scoring in the audit trail. _audit( judge, AuditLog.Action.UPDATE, application, metadata={ "action": "judge_score_submitted", "sections_scored": len(saved), }, ) inc_signals.smehub_judge_score_submitted.send( sender=Application, application=application, judge=judge, ) return saved def compute_weighted_average(application: Application, *, persist: bool = True) -> Decimal: """FRSME-INC020. Returns 0 when no scores exist yet. Aggregates per section: average across judges, normalised to [0,1] by max_marks, then multiplied by section weight. Sum across sections, expressed as a 0–100 percentage. When ``persist`` is True (default), the result is written back to ``application.computed_score`` so the selection dashboard can read it without re-running the aggregation per render. """ rubric = getattr(application.programme, "rubric", None) if rubric is None: return Decimal("0") sections = list(rubric.sections.all()) if not sections: return Decimal("0") scores_by_section: dict[int, list[Decimal]] = defaultdict(list) for js in JudgeScore.objects.filter(application=application).select_related("section"): scores_by_section[js.section_id].append(js.score) total = Decimal("0") for section in sections: section_scores = scores_by_section.get(section.pk, []) if not section_scores or section.max_marks == 0: continue avg = sum(section_scores) / Decimal(len(section_scores)) normalised = avg / Decimal(section.max_marks) total += normalised * section.weight result = (total * Decimal("100")).quantize(Decimal("0.01")) if persist and application.pk and application.computed_score != result: Application.objects.filter(pk=application.pk).update(computed_score=result) application.computed_score = result return result def rank_applications(programme: Programme): """Returns applications ordered by computed weighted score desc. Prefers ``Application.computed_score`` (cached by the JudgeScore signal); falls back to recomputing only if the cache is missing. """ apps = list( Application.objects.filter( programme=programme, status__in=[ Application.Status.SUBMITTED, Application.Status.UNDER_REVIEW, Application.Status.ACCEPTED, Application.Status.REJECTED, ], ) .select_related("entrepreneur__user", "business") ) ranked = [] for app in apps: if app.computed_score is None: app.computed_score = compute_weighted_average(app) ranked.append(app) ranked.sort(key=lambda a: (a.computed_score or Decimal("0"), a.created_at), reverse=True) return ranked def remind_pending_judges(*, days_pending: int = 3, now=None) -> int: """FRSME-INC021. Returns count of reminders dispatched. Two cadences: * When the programme sets a ``scoring_deadline``, reminders are *deadline-relative* — they start firing ``reminder_days_before`` days ahead of the deadline and repeat at most once per day until the judge submits a score (including while overdue). * Otherwise the legacy days-since-notified cadence applies (fire once the assignment has been pending ``days_pending`` days). """ now = now or timezone.now() legacy_cutoff = now - timedelta(days=days_pending) daily_throttle = now - timedelta(days=1) sent = 0 qs = ( JudgeAssignment.objects.filter( application__status__in=[Application.Status.SUBMITTED, Application.Status.UNDER_REVIEW], ) .select_related("application__programme", "judge") ) for ja in qs: scored = JudgeScore.objects.filter(application=ja.application, judge=ja.judge).exists() if scored: continue deadline = ja.application.programme.scoring_deadline if deadline: # Deadline-relative: too early to remind until inside the window. window_start = deadline - timedelta(days=ja.application.programme.reminder_days_before or 0) if now < window_start: continue # Throttle to once per day so overdue reminders don't spam. if ja.last_reminded_at and ja.last_reminded_at > daily_throttle: continue else: last = ja.last_reminded_at or ja.notified_at or ja.created_at if last and last > legacy_cutoff: continue try: path = reverse("smehub_incubation:judge_score_form", kwargs={"pk": ja.application.pk}) except NoReverseMatch: path = "" notify_smehub( ja.judge, f"Reminder: please score application {ja.application.application_id} ({ja.application.programme.title}).", verb=Verb.SMEHUB_JUDGE_REMINDER, action_url=_absolute(path) if path else "", target=ja.application, ) ja.last_reminded_at = timezone.now() ja.save(update_fields=["last_reminded_at"]) sent += 1 return sent # --------------------------------------------------------------------------- # Cohort confirmation + REP enrolment (FRSME-INC022 → INC025) # --------------------------------------------------------------------------- @transaction.atomic def confirm_cohort( programme: Programme, *, accepted_application_ids: list[int], name: str, start_date, end_date, venue: str = "", rejection_reason: str = "", by_user, ) -> Cohort: """Mark accepted applications, reject the rest, build the Cohort. FRSME-INC020 audit fix: when the programme uses weighted-rubric selection, every accepted application must have a *complete* judge score across every rubric section before it can be confirmed. Selection on a partial average misrepresents standing. Programmes using ``OFFICER_DECISION`` selection bypass this check — admin selection does not require numeric scoring at all. The shared strict-mode helper from ``apps.smehub._shared.calls`` enforces this consistently across SP2/SP4/SP5. """ from apps.smehub._shared.calls import compute_weighted_average as shared_avg rubric = getattr(programme, "rubric", None) needs_full_scoring = ( programme.selection_method == Programme.SelectionMethod.WEIGHTED_RUBRIC and rubric is not None ) if needs_full_scoring: sections = list(rubric.sections.all()) for app_id in accepted_application_ids: scores = JudgeScore.objects.filter(application_id=app_id) if shared_avg(sections, scores, require_all_sections=True) is None: raise ApplicationStateError( f"Cannot confirm cohort: application {app_id} has not been " f"fully scored across all rubric sections." ) cohort = Cohort.objects.create( programme=programme, name=name, start_date=start_date, end_date=end_date, venue=venue, ) members: list[CohortMember] = [] apps = ( Application.objects.filter(programme=programme) .exclude(status__in=[Application.Status.WITHDRAWN]) ) for app in apps: score = compute_weighted_average(app) if app.pk in accepted_application_ids: if app.status not in [Application.Status.SUBMITTED, Application.Status.UNDER_REVIEW]: continue app.accept(by_user=by_user, weighted_score=score) app.save() if app.business and app.business.programme_of_origin_id is None: app.business.programme_of_origin = programme app.business.save(update_fields=["programme_of_origin", "updated_at"]) member = CohortMember.objects.create( cohort=cohort, application=app, entrepreneur=app.entrepreneur, business=app.business, ) ProgressRecord.objects.get_or_create(cohort_member=member) members.append(member) try: path = reverse("smehub_incubation:my_programme") except NoReverseMatch: path = "" notify_smehub( app.entrepreneur.user, f"Congratulations — “{app.business.name}” is in the {cohort.name} cohort for {programme.title}.", verb=Verb.SMEHUB_COHORT_ACCEPTED, action_url=_absolute(path) if path else "", target=member, ) inc_signals.smehub_application_decided.send( sender=Application, application=app, by_user=by_user, accepted=True, score=score, ) elif app.status in [Application.Status.SUBMITTED, Application.Status.UNDER_REVIEW]: app.reject(by_user=by_user, weighted_score=score, reason=rejection_reason) app.save() # FRSME-INC023 — give the applicant the officer's specific reason # when one was supplied, otherwise fall back to the generic notice. if rejection_reason: body = ( f"Your application {app.application_id} for {programme.title} was " f"not selected this round. {rejection_reason}" ) else: body = ( f"Your application {app.application_id} for {programme.title} was " f"not selected this round." ) notify_smehub( app.entrepreneur.user, body, verb=Verb.SMEHUB_COHORT_REJECTED, target=app, ) inc_signals.smehub_application_decided.send( sender=Application, application=app, by_user=by_user, accepted=False, score=score, ) _audit(by_user, AuditLog.Action.CREATE, cohort, metadata={"accepted": len(members)}) inc_signals.smehub_cohort_confirmed.send(sender=Cohort, cohort=cohort, members=members) # FRSME-INC024 — auto-enrol the cohort into all linked REP courses. # Async so a slow REP backend can't block the cohort-confirm transaction. if cohort.programme.course_links.exists(): from apps.smehub.incubation.tasks import enrol_cohort_in_courses_task transaction.on_commit(lambda cid=cohort.pk: enrol_cohort_in_courses_task.delay(cid)) return cohort def enrol_cohort_in_courses(cohort: Cohort, *, by_user=None) -> dict: """FRSME-INC024/025. Returns a summary dict of {moodle_course_id: {...}}. Cohort enrolment used to push members into the REP LMS. With learning on Moodle, actual enrolment is delegated to Moodle (a Moodle Web Services ``enrol_manual_enrol_users`` call is future work); this records the intent per linked course and reports zero programmatic enrolments for now. """ summary: dict = {} active_members = cohort.members.filter(status=CohortMember.Status.ACTIVE).count() for link in cohort.programme.course_links.all(): summary[link.moodle_course_id] = { "course_title": link.course_title or link.moodle_course_id, "enrolled": 0, "failed": 0, "deferred_to_moodle": active_members, } logger.info( "smehub-incubation: cohort %s course enrolment delegated to Moodle (%s members, %s courses)", cohort.pk, active_members, len(summary), ) return summary @transaction.atomic def retry_cohort_enrolment(member: CohortMember, *, by_user) -> dict: """FRSME-INC025 — retry hook for a single cohort member's course enrolment. REP enrolment was retired with the LMS; enrolment is delegated to Moodle, so this clears any stale error and reports zero enrolments. Kept so the retry UI/route stay valid until a Moodle enrol call is wired in. Returns ``{"enrolled": int, "failed": int, "errors": list[str]}``. """ now = timezone.now() member.last_enrol_attempt_at = now member.last_enrol_error = "" member.save(update_fields=["last_enrol_error", "last_enrol_attempt_at", "updated_at"]) _audit( by_user, AuditLog.Action.UPDATE, member, metadata={"action": "retry_cohort_enrolment", "enrolled": 0, "failed": 0, "deferred_to_moodle": True}, ) return {"enrolled": 0, "failed": 0, "errors": []} # --------------------------------------------------------------------------- # Mentorship (FRSME-INC026 → INC030) # --------------------------------------------------------------------------- @transaction.atomic def assign_mentor(cohort_member: CohortMember, *, mentor, by_user=None, notes: str = "") -> MentorAssignment: existing = cohort_member.mentor_assignments.filter(ended_at__isnull=True).first() reassigned = False reassigned_from = None if existing: existing.ended_at = timezone.now() existing.save(update_fields=["ended_at"]) reassigned = True reassigned_from = existing assignment = MentorAssignment.objects.create( cohort_member=cohort_member, mentor=mentor, notes=notes, reassigned_from=reassigned_from, ) _audit(by_user, AuditLog.Action.CREATE, assignment, metadata={"reassigned": reassigned}) notify_smehub( mentor, f"You have been assigned as mentor to {cohort_member.business.name}.", verb=Verb.SMEHUB_MENTOR_ASSIGNED, target=assignment, ) notify_smehub( cohort_member.entrepreneur.user, f"{mentor.get_full_name() or mentor.email} has been assigned as your mentor.", verb=Verb.SMEHUB_MENTOR_ASSIGNED, target=assignment, ) inc_signals.smehub_mentor_assigned.send( sender=MentorAssignment, assignment=assignment, reassigned=reassigned, ) return assignment def reassign_mentor(cohort_member: CohortMember, *, new_mentor, by_user=None, notes: str = "") -> MentorAssignment: return assign_mentor(cohort_member, mentor=new_mentor, by_user=by_user, notes=notes) @transaction.atomic def book_session( assignment: MentorAssignment, *, starts_at, duration_minutes: int = 60, type: str = MentorSession.Type.VIRTUAL, location: str = "", agenda: str = "", booked_by=None, ) -> MentorSession: if not assignment.is_active: raise ApplicationStateError("Mentor assignment is no longer active.") session = MentorSession.objects.create( assignment=assignment, starts_at=starts_at, duration_minutes=duration_minutes, type=type, location=location, agenda=agenda, ) inc_signals.smehub_session_scheduled.send(sender=MentorSession, session=session) body = ( f"Session scheduled with " f"{assignment.mentor.get_full_name() or assignment.mentor.email} on " f"{starts_at:%d %b %Y %H:%M}." ) for recipient in (assignment.mentor, assignment.cohort_member.entrepreneur.user): notify_smehub(recipient, body, verb=Verb.SMEHUB_SESSION_SCHEDULED, target=session) return session @transaction.atomic def complete_session(session: MentorSession, *, notes: str = "", by_user=None) -> MentorSession: if session.status == MentorSession.Status.COMPLETED: return session session.status = MentorSession.Status.COMPLETED session.completed_at = timezone.now() if notes: session.notes = notes session.save(update_fields=["status", "completed_at", "notes", "updated_at"]) member = session.assignment.cohort_member record, _ = ProgressRecord.objects.get_or_create(cohort_member=member) record.sessions_completed = (record.sessions_completed or 0) + 1 record.last_milestone_at = timezone.now() record.save(update_fields=["sessions_completed", "last_milestone_at", "updated_at"]) # FRSME-INC029 — audit the session completion. _audit( by_user or session.assignment.mentor, AuditLog.Action.UPDATE, session, metadata={"action": "session_completed", "cohort_member": str(member.pk)}, ) inc_signals.smehub_session_completed.send(sender=MentorSession, session=session) inc_signals.smehub_progress_milestone.send( sender=ProgressRecord, member=member, milestone="session_completed", payload={"session_id": session.pk}, ) evaluate_completion_criteria(member) return session @transaction.atomic def cancel_session(session: MentorSession, *, reason: str = "") -> MentorSession: session.status = MentorSession.Status.CANCELLED session.cancellation_reason = reason session.save(update_fields=["status", "cancellation_reason", "updated_at"]) inc_signals.smehub_session_cancelled.send(sender=MentorSession, session=session, reason=reason) body = f"Session on {session.starts_at:%d %b %Y %H:%M} cancelled." for recipient in (session.assignment.mentor, session.assignment.cohort_member.entrepreneur.user): notify_smehub(recipient, body, verb=Verb.SMEHUB_SESSION_CANCELLED, target=session) return session # --------------------------------------------------------------------------- # Mentorship Readiness Assessment (PRD §5.X — Mentorship and Industry Advisory) # --------------------------------------------------------------------------- @transaction.atomic def record_readiness_assessment( cohort_member: CohortMember, *, market_positioning_score: int, technical_robustness_score: int, regulatory_compliance_score: int, by_user=None, market_positioning_notes: str = "", technical_robustness_notes: str = "", regulatory_compliance_notes: str = "", overall_notes: str = "", ): """Persist a readiness assessment, audit it, and notify the entrepreneur. Scores are validated to be 1–5; out-of-range values raise ``IncubationStateError`` so callers get a typed failure they can map to HTTP 400 in views and pytest assertions in tests. """ from apps.smehub.incubation.models import ReadinessAssessment for label, value in ( ("market_positioning_score", market_positioning_score), ("technical_robustness_score", technical_robustness_score), ("regulatory_compliance_score", regulatory_compliance_score), ): if not (1 <= int(value) <= 5): raise IncubationStateError(f"{label} must be between 1 and 5.") assessment = ReadinessAssessment.objects.create( cohort_member=cohort_member, conducted_by=by_user, market_positioning_score=market_positioning_score, market_positioning_notes=market_positioning_notes, technical_robustness_score=technical_robustness_score, technical_robustness_notes=technical_robustness_notes, regulatory_compliance_score=regulatory_compliance_score, regulatory_compliance_notes=regulatory_compliance_notes, overall_notes=overall_notes, ) _audit(by_user, AuditLog.Action.CREATE, assessment, metadata={ "average_score": str(assessment.average_score), "gaps": assessment.gaps(), }) notify_smehub( cohort_member.entrepreneur.user, ( f"Readiness assessment recorded for {cohort_member.business.name} — " f"average score {assessment.average_score}/5." ), verb=Verb.SMEHUB_READINESS_RECORDED, target=assessment, ) return assessment # --------------------------------------------------------------------------- # Manual training + completion + certificates (FRSME-INC031 → INC037) # --------------------------------------------------------------------------- @transaction.atomic def record_manual_training_completion( cohort_member: CohortMember, *, title: str, completed_at=None, by_user, ) -> ProgressRecord: record, _ = ProgressRecord.objects.get_or_create(cohort_member=cohort_member) completed_at = completed_at or timezone.now() entries = list(record.manual_trainings or []) entries.append( { "title": title, "completed_at": completed_at.isoformat(), "recorded_by": by_user.email if by_user else "system", }, ) record.manual_trainings = entries record.last_milestone_at = timezone.now() record.save(update_fields=["manual_trainings", "last_milestone_at", "updated_at"]) # FRSME-INC032 — flag manually-recorded training in the audit trail. _audit( by_user, AuditLog.Action.UPDATE, cohort_member, metadata={ "action": "manual_training_recorded", "title": title, "completed_at": completed_at.isoformat(), }, ) inc_signals.smehub_progress_milestone.send( sender=ProgressRecord, member=cohort_member, milestone="manual_training", payload={"title": title}, ) evaluate_completion_criteria(cohort_member) return record def _member_session_attendance(member: CohortMember) -> tuple[int, int]: """Best-effort attendance from mentorship sessions. Attendance is not tracked as a discrete signal, so we derive it from the member's mentorship sessions: of every session that has reached its outcome (completed or cancelled, or a scheduled session whose time has passed), count the completed ones as "attended". Returns ``(attended, held)``; ``held == 0`` means there is no attendance evidence yet. """ sessions = MentorSession.objects.filter(assignment__cohort_member=member) now = timezone.now() attended = 0 held = 0 for s in sessions: if s.status == MentorSession.Status.COMPLETED: attended += 1 held += 1 elif s.status == MentorSession.Status.CANCELLED: held += 1 elif s.starts_at and s.starts_at < now: # A scheduled session whose slot has passed but was never marked # complete counts as a missed session. held += 1 return attended, held def completion_criteria_breakdown(member: CohortMember) -> dict: """FRSME-INC033 — structured view of a member's progress toward completion. Returns every *configured* criterion (required courses, minimum mentorship sessions, attendance threshold) with the member's current standing against it, plus whether they are collectively satisfied. Shared by the auto- completion evaluator and the entrepreneur / officer templates so the same logic drives both the flip and what the UI shows. """ programme = member.cohort.programme record = getattr(member, "progress", None) required_course_ids = set( programme.course_links.filter(is_required=True).values_list("moodle_course_id", flat=True), ) completions = (record.course_completions if record else {}) or {} completed_course_ids = {str(k) for k in completions.keys()} courses_done = len(required_course_ids & completed_course_ids) courses_required = len(required_course_ids) courses_ok = required_course_ids.issubset(completed_course_ids) min_sessions = programme.min_mentorship_sessions or 0 sessions_done = (record.sessions_completed if record else 0) or 0 sessions_ok = sessions_done >= min_sessions threshold = programme.attendance_threshold_pct or 0 attended, held = _member_session_attendance(member) attendance_pct = round(100 * attended / held) if held else None attendance_ok = threshold == 0 or (attendance_pct is not None and attendance_pct >= threshold) criteria: list[dict] = [] if courses_required: criteria.append({ "key": "courses", "label": "Required courses completed", "satisfied": courses_ok, "detail": f"{courses_done} of {courses_required} completed", }) if min_sessions: criteria.append({ "key": "sessions", "label": "Mentorship sessions completed", "satisfied": sessions_ok, "detail": f"{sessions_done} of {min_sessions} sessions", }) if threshold: if attendance_pct is None: detail = f"No sessions recorded yet — needs {threshold}%" else: detail = f"{attendance_pct}% attended — needs {threshold}%" criteria.append({ "key": "attendance", "label": "Session attendance", "satisfied": attendance_ok, "detail": detail, }) configured = bool(criteria) all_satisfied = configured and courses_ok and sessions_ok and attendance_ok return { "configured": configured, "all_satisfied": all_satisfied, "criteria": criteria, # Attendance is derived, not directly tracked — the UI notes this. "attendance_derived": threshold > 0, } def evaluate_completion_criteria(member: CohortMember) -> bool: """FRSME-INC033 — auto-complete when every *configured* criterion is met. A member is flipped to PROGRAMME_COMPLETE only when the programme defines at least one completion criterion (required courses, minimum mentorship sessions, or an attendance threshold) and the member satisfies all of them. Programmes with no configured criteria never auto-complete (they rely on manual completion). Returns True when this call performed the flip. """ if member.status != CohortMember.Status.ACTIVE: return False breakdown = completion_criteria_breakdown(member) if not breakdown["configured"] or not breakdown["all_satisfied"]: return False member.mark_complete(manual=False, notes="Auto-completed on programme completion criteria") member.save() inc_signals.smehub_cohort_member_completed.send( sender=CohortMember, member=member, manual=False, ) notify_smehub( member.entrepreneur.user, f"You've met all completion criteria for {member.cohort.programme.title}. Your certificate is being prepared.", verb=Verb.SMEHUB_PROGRAMME_COMPLETE, target=member, ) generate_certificate(member, issued_by=None) return True @transaction.atomic def manual_complete(member: CohortMember, *, by_user, notes: str = "") -> CohortMember: """FRSME-INC035 — admin-driven completion with audit log.""" if member.status != CohortMember.Status.ACTIVE: raise ApplicationStateError("Cohort member is not active.") # FRSME-INC035 — the override reason is mandatory and recorded in the audit trail. if not (notes or "").strip(): raise ApplicationStateError("A reason is required to manually complete a cohort member.") member.mark_complete(manual=True, notes=notes) member.save() _audit(by_user, AuditLog.Action.UPDATE, member, metadata={"manual_complete": True, "notes": notes}) inc_signals.smehub_cohort_member_completed.send( sender=CohortMember, member=member, manual=True, ) notify_smehub( member.entrepreneur.user, f"You have been marked as having completed {member.cohort.programme.title}.", verb=Verb.SMEHUB_PROGRAMME_COMPLETE, target=member, ) generate_certificate(member, issued_by=by_user) return member @transaction.atomic def withdraw_cohort_member(member: CohortMember, *, by_user, reason: str) -> CohortMember: member.withdraw(reason=reason) member.save() _audit(by_user, AuditLog.Action.UPDATE, member, metadata={"withdrawn": True, "reason": reason}) inc_signals.smehub_cohort_member_withdrawn.send( sender=CohortMember, member=member, reason=reason, ) return member @transaction.atomic def generate_certificate(member: CohortMember, *, issued_by) -> Certificate: """FRSME-INC034. Idempotent — re-running for a member that already has a certificate returns the existing row. """ existing = Certificate.objects.filter(cohort_member=member).first() if existing: return existing cert = Certificate.objects.create(cohort_member=member, issued_by=issued_by) try: from django.core.files.base import ContentFile from django.template.loader import render_to_string from weasyprint import HTML html = render_to_string( "smehub/incubation/certificate_pdf.html", { "member": member, "programme": member.cohort.programme, "cohort": member.cohort, "issued_at": cert.generated_at, "certificate_number": str(cert.certificate_number), }, ) pdf_bytes = HTML(string=html, base_url=getattr(settings, "PUBLIC_APP_BASE_URL", "")).write_pdf() cert.pdf_file.save(f"smehub-cert-{cert.certificate_number}.pdf", ContentFile(pdf_bytes), save=True) except Exception: # Image without cairo/pango — leave pdf_file empty; row still exists. pass inc_signals.smehub_certificate_issued.send(sender=Certificate, certificate=cert, member=member) notify_smehub( member.entrepreneur.user, f"Your certificate for {member.cohort.programme.title} has been issued.", verb=Verb.SMEHUB_CERTIFICATE_ISSUED, target=cert, ) return cert def push_progress_to_mel(member: CohortMember, *, milestone: str, payload: dict | None = None) -> None: """FRSME-INC036/037. Thin wrapper for callers that want to emit the M&EL milestone signal without touching ProgressRecord directly. """ inc_signals.smehub_progress_milestone.send( sender=ProgressRecord, member=member, milestone=milestone, payload=payload or {}, )