"""Seed SME-Hub showcasing (SP4) demo data. Idempotent. Builds on: • SP1 onboarding (verified entrepreneurs + businesses) • SP3 linkage (partner / service provider directory entries) • SP3b marketplace (buyer registry entries) Produces approximately: • 4 ShowcaseEvents (mix of formats / statuses) • 3 InnovationChallenges (each with the ESG 6-category / 20-criterion rubric) • 25 ShowcaseApplications (mixed statuses, mix event/challenge) • 18 InnovationCatalogueEntries • 6 DealRooms (4 open / 1 expired / 1 agreement-reached) with realistic message threads and one signed agreement. """ from __future__ import annotations import logging import random from datetime import timedelta from decimal import Decimal from django.contrib.auth import get_user_model from django.contrib.auth.hashers import make_password from django.contrib.contenttypes.models import ContentType from django.core.files.base import ContentFile from django.db import transaction from django.utils import timezone from apps.core.permissions.roles import UserRole from apps.smehub.linkage.models import ( PartnerDirectoryEntry, ServiceProviderEntry, ) from apps.smehub.marketplace.models import BuyerRegistryEntry from apps.smehub.onboarding.models import Business, EntrepreneurProfile from apps.smehub.showcasing.models import ( ChallengeScoringRubric, DealAgreement, DealRoom, DealRoomDocument, DealRoomMessage, EventAttendance, InnovationCatalogueEntry, InnovationCatalogueVersion, InnovationChallenge, PitchScore, ShowcaseApplication, ShowcaseEvent, ) from apps.smehub.showcasing.rubric_defaults import apply_esg_rubric logger = logging.getLogger(__name__) User = get_user_model() DEMO_PASSWORD = "smehub-demo-2026" # noqa: S105 - demo only def _ensure_user(email: str, *, first: str, last: str, role: str) -> User: user, created = User.objects.get_or_create( email=email, defaults={ "first_name": first, "last_name": last, "is_active": True, "email_verified": True, "role": role, }, ) if created: user.password = make_password(DEMO_PASSWORD) user.save(update_fields=["password"]) return user def _ensure_admin() -> User: return _ensure_user( "smehub-admin@ruforum.org", first="Sara", last="Admin", role=UserRole.SME_ADMIN, ) def _ensure_judges(n: int = 6) -> list[User]: judges = [] for i in range(1, n + 1): judges.append( _ensure_user( f"judge{i}@ruforum.org", first=f"Judge{i}", last="Panel", role=UserRole.JUDGE, ) ) return judges # --------------------------------------------------------------------------- # Showcase events # --------------------------------------------------------------------------- _EVENTS = [ { "name": "RUFORUM AgriTech Demo Day 2026", "format": ShowcaseEvent.Format.HYBRID, "summary": "Top entrepreneurs from the Spring AgriTech cohort pitch live to investors and partners.", "venue": "Speke Resort, Kampala", "video_link": "https://www.youtube.com/embed/dQw4w9WgXcQ", "sectors": ["Agriculture", "Agribusiness", "Tech"], "days_from_now": 7, "publish": True, }, { "name": "Continental Innovation Showcase", "format": ShowcaseEvent.Format.VIRTUAL, "summary": "Pan-African virtual stage for emerging climate, energy, and water innovations.", "venue": "https://www.ruforum.org/showcase", "video_link": "https://www.youtube.com/embed/dQw4w9WgXcQ", "sectors": ["Energy", "Water", "Agriculture"], "days_from_now": -10, # past — set to LIVE "publish": True, "go_live": True, }, { "name": "FinTech Investor Roundtable", "format": ShowcaseEvent.Format.PHYSICAL, "summary": "Curated showcase for FinTech ventures targeting cross-border MSME credit.", "venue": "Radisson Blu, Nairobi", "video_link": "", "sectors": ["FinTech", "Trade"], "days_from_now": -45, # past — concluded "publish": True, "go_live": True, "conclude": True, }, { "name": "Health & Wellness Innovation Day (Draft)", "format": ShowcaseEvent.Format.HYBRID, "summary": "Coming soon — a curated showcase for digital health and assistive-tech ventures.", "venue": "TBD", "video_link": "", "sectors": ["Health", "Tech"], "days_from_now": 60, "publish": False, }, ] def _seed_events(admin: User, judges: list[User]) -> list[ShowcaseEvent]: out = [] now = timezone.now() for idx, payload in enumerate(_EVENTS): starts_at = now + timedelta(days=payload["days_from_now"]) ends_at = starts_at + timedelta(hours=6) deadline = starts_at - timedelta(days=2) event, _ = ShowcaseEvent.objects.get_or_create( name=payload["name"], defaults={ "summary": payload["summary"], "description": payload["summary"] + " " * 0 + " More details will be shared with confirmed participants.", "format": payload["format"], "starts_at": starts_at, "ends_at": ends_at, "application_deadline": deadline, "target_sectors": payload["sectors"], "venue": payload["venue"], "video_link": payload["video_link"], "organiser": admin, }, ) if event.evaluation_panel.count() == 0: event.evaluation_panel.set(judges[: 3 + (idx % 2)]) # Refetch to dodge FSM cached-value gotchas event = ShowcaseEvent.objects.get(pk=event.pk) if event.status == ShowcaseEvent.Status.DRAFT and payload.get("publish"): event.publish() event.save() event = ShowcaseEvent.objects.get(pk=event.pk) if event.status == ShowcaseEvent.Status.PUBLISHED and payload.get("go_live"): event.go_live() event.save() event = ShowcaseEvent.objects.get(pk=event.pk) if event.status == ShowcaseEvent.Status.LIVE and payload.get("conclude"): event.conclude() event.save() event = ShowcaseEvent.objects.get(pk=event.pk) out.append(event) return out # --------------------------------------------------------------------------- # Innovation challenges # --------------------------------------------------------------------------- _CHALLENGES = [ { "title": "Climate-Smart Smallholder Innovations 2026", "summary": "Solutions that help smallholder farmers adapt to climate variability.", "theme": "Climate adaptation", "sectors": ["Agriculture", "Energy", "Water"], "prize": "$15,000 grant + 6 months of mentorship.", "open": True, }, { "title": "FinTech for Rural Africa", "summary": "Affordable credit, insurance, and payment products for MSMEs and rural households.", "theme": "Financial inclusion", "sectors": ["FinTech", "Trade"], "prize": "$10,000 grant + investor pitch slot.", "open": True, }, { "title": "Women-Led AgriBusiness Awards", "summary": "Recognising women entrepreneurs scaling agribusiness ventures across the network.", "theme": "Gender & enterprise", "sectors": ["Agribusiness", "Manufacturing"], "prize": "Cohort sponsorship for 5 winners.", "open": False, }, ] def _seed_challenges(admin: User) -> list[InnovationChallenge]: out = [] now = timezone.now() for idx, payload in enumerate(_CHALLENGES): deadline = now + timedelta(days=30 if payload["open"] else -10) challenge, _ = InnovationChallenge.objects.get_or_create( title=payload["title"], defaults={ "summary": payload["summary"], "description": payload["summary"] + " Open to all RUFORUM-network entrepreneurs with a verified business.", "theme": payload["theme"], "target_sectors": payload["sectors"], "prize_summary": payload["prize"], "application_deadline": deadline, "judging_deadline": deadline + timedelta(days=14), "organiser": admin, }, ) rubric, _ = ChallengeScoringRubric.objects.get_or_create(challenge=challenge) if rubric.sections.count() == 0: apply_esg_rubric(rubric) if challenge.status == InnovationChallenge.Status.DRAFT and payload["open"]: challenge.status = InnovationChallenge.Status.OPEN challenge.published_at = timezone.now() challenge.save(update_fields=["status", "published_at", "updated_at"]) elif not payload["open"] and challenge.status == InnovationChallenge.Status.DRAFT: challenge.status = InnovationChallenge.Status.ANNOUNCED challenge.published_at = timezone.now() - timedelta(days=60) challenge.announced_at = timezone.now() - timedelta(days=5) challenge.save(update_fields=["status", "published_at", "announced_at", "updated_at"]) out.append(challenge) return out # --------------------------------------------------------------------------- # Applications + scoring # --------------------------------------------------------------------------- def _verified_businesses(limit: int = 12) -> list[Business]: return list( Business.objects.filter(verification_status=Business.Status.VERIFIED) .select_related("entrepreneur__user")[:limit] ) def _seed_applications( events: list[ShowcaseEvent], challenges: list[InnovationChallenge], judges: list[User], admin: User, ) -> dict: businesses = _verified_businesses() if not businesses or not events or not challenges: return {"applications": 0, "scores": 0} rng = random.Random(73) rng.shuffle(businesses) event_ct = ContentType.objects.get_for_model(ShowcaseEvent) challenge_ct = ContentType.objects.get_for_model(InnovationChallenge) apps_created = 0 scores_created = 0 plan = [] # (business, target, kind) seen: set[tuple[int, int, str]] = set() # (entrepreneur_id, target_pk, kind) # ~15 event applications across the 3 published events pub_events = [e for e in events if e.status != ShowcaseEvent.Status.DRAFT] if pub_events: for i, biz in enumerate(businesses): event = pub_events[i % len(pub_events)] key = (biz.entrepreneur_id, event.pk, "event") if key in seen: continue seen.add(key) plan.append((biz, event, "event")) # ~10 challenge applications open_challenges = [c for c in challenges if c.status == InnovationChallenge.Status.OPEN] pool = open_challenges or challenges for i, biz in enumerate(businesses[:10]): challenge = pool[i % len(pool)] key = (biz.entrepreneur_id, challenge.pk, "challenge") if key in seen: continue seen.add(key) plan.append((biz, challenge, "challenge")) plan = plan[:25] for index, (biz, target, kind) in enumerate(plan): ct = event_ct if kind == "event" else challenge_ct innovation_title = f"{biz.name} — {target.target_sectors[0] if target.target_sectors else 'Innovation'} platform" application, was_created = ShowcaseApplication.objects.get_or_create( target_content_type=ct, target_object_id=str(target.pk), entrepreneur=biz.entrepreneur, business=biz, defaults={ "innovation_title": innovation_title, "innovation_description": f"{biz.name} is building solutions in {biz.sector or 'their sector'}.", "problem_statement": "Our beneficiaries struggle with reliable access to inputs and markets.", "solution_summary": "A digital platform that connects entrepreneurs to verified suppliers and buyers.", "impact_potential": "Targets 5,000 households in year 1; 25,000 by year 3.", }, ) if was_created: apps_created += 1 # Distribute statuses outcome = index % 5 application = ShowcaseApplication.objects.get(pk=application.pk) if outcome in (0, 1) and application.status in ( ShowcaseApplication.Status.SUBMITTED, ShowcaseApplication.Status.UNDER_REVIEW, ): application.confirm(by_user=admin, note="Welcome to the lineup.") application.save() elif outcome == 2 and application.status in ( ShowcaseApplication.Status.SUBMITTED, ShowcaseApplication.Status.UNDER_REVIEW, ): application.reject(by_user=admin, note="Strong entry but missed the shortlist.") application.save() # Add 2 judge scores per application (challenges) or 1 per event app if kind == "challenge": challenge = target rubric = getattr(challenge, "rubric", None) if rubric is not None: for j_idx, judge in enumerate(judges[:2]): for sec in rubric.sections.all(): # Keep demo scores within each criterion's ceiling # (ESG criteria are out of 5, legacy rubrics out of 10). hi = max(1.0, sec.max_marks * 0.95) lo = max(1.0, sec.max_marks * 0.5) score_value = Decimal(str(rng.uniform(lo, hi))).quantize(Decimal("0.5")) _, sc_created = PitchScore.objects.get_or_create( application=application, judge=judge, section=sec, defaults={"score": score_value, "comments": "Solid signal."}, ) if sc_created: scores_created += 1 else: for judge in judges[: 1 + (index % 2)]: score_value = Decimal(str(rng.uniform(6.0, 9.0))).quantize(Decimal("0.5")) _, sc_created = PitchScore.objects.get_or_create( application=application, judge=judge, section=None, defaults={"score": score_value, "comments": "Compelling pitch overall."}, ) if sc_created: scores_created += 1 return {"applications": apps_created, "scores": scores_created} # --------------------------------------------------------------------------- # Catalogue # --------------------------------------------------------------------------- def _seed_catalogue(admin: User) -> int: confirmed = list( ShowcaseApplication.objects.filter(status=ShowcaseApplication.Status.CONFIRMED) .select_related("entrepreneur__user", "business")[:18] ) created = 0 for app in confirmed: entry, was_created = InnovationCatalogueEntry.objects.get_or_create( application=app, defaults={ "entrepreneur": app.entrepreneur, "business": app.business, "title": app.innovation_title, "summary": app.solution_summary or app.innovation_description[:200], "description": app.innovation_description, "sectors": [app.business.sector] if getattr(app.business, "sector", None) else [], "impact_metrics": { "beneficiaries_year_1": "5,000", "geographic_reach": "Uganda, Kenya, Tanzania", }, "visibility": InnovationCatalogueEntry.Visibility.STAKEHOLDER, "last_updated_by": admin, }, ) if was_created: created += 1 InnovationCatalogueVersion.objects.create( entry=entry, version=1, title=entry.title, summary=entry.summary, description=entry.description, sectors=entry.sectors, impact_metrics=entry.impact_metrics, video_link=entry.video_link, website=entry.website, notes="Seeded initial publication.", captured_by=admin, ) return created # --------------------------------------------------------------------------- # Deal rooms # --------------------------------------------------------------------------- _MESSAGES = [ "Thanks for the introduction — happy to explore a partnership here.", "We'd like to see audited financials before going further.", "Sharing the v2 proposal — let me know if anything is unclear.", "Could we set up a call next week to align on scope?", "Term sheet uploaded; let's plan a working session next Tuesday.", "Looks great — we'll loop in our compliance team and circle back.", ] def _seed_deal_rooms(admin: User) -> dict: profiles = list( EntrepreneurProfile.objects.filter(verification_status=EntrepreneurProfile.Status.VERIFIED) .select_related("user")[:6] ) if not profiles: return {"rooms": 0, "messages": 0, "documents": 0, "agreements": 0} partners = list( PartnerDirectoryEntry.objects.filter(verification_status=PartnerDirectoryEntry.Status.VERIFIED)[:3] ) providers = list( ServiceProviderEntry.objects.filter(verification_status=ServiceProviderEntry.Status.VERIFIED)[:3] ) buyers = list(BuyerRegistryEntry.objects.filter(verification_status=BuyerRegistryEntry.Status.VERIFIED)[:3]) counterparties = [] counterparties.extend([(p, "partner") for p in partners]) counterparties.extend([(p, "service_provider") for p in providers]) counterparties.extend([(b, "buyer") for b in buyers]) if not counterparties: return {"rooms": 0, "messages": 0, "documents": 0, "agreements": 0} pairs = [] for profile in profiles: biz = profile.businesses.filter(verification_status=Business.Status.VERIFIED).first() if not biz: continue # Pair this entrepreneur with up to 2 different counterparties so we # exercise PRD A4 (concurrent rooms across different counterparties). used = 0 for cp_idx, (cp, kind) in enumerate(counterparties): if used >= 2: break pairs.append((profile, biz, cp, kind)) used += 1 if cp_idx > 0: # rotate offset so different entrepreneurs hit different cps counterparties.append(counterparties.pop(0)) break pairs = pairs[:6] now = timezone.now() rooms_created = 0 messages_created = 0 documents_created = 0 agreements_created = 0 n = len(pairs) expired_idx = max(n - 2, 1) if n >= 3 else None agreement_idx = max(n - 1, 2) if n >= 2 else None for index, (profile, biz, counterparty, kind) in enumerate(pairs): cp_ct = ContentType.objects.get_for_model(counterparty.__class__) # Distribute outcomes so we always have at least one EXPIRED + one # AGREEMENT_REACHED when we have ≥3 pairs. if index == expired_idx: expiry = now - timedelta(days=2) target_status = DealRoom.Status.EXPIRED elif index == agreement_idx: expiry = now + timedelta(days=5) target_status = DealRoom.Status.AGREEMENT_REACHED else: expiry = now + timedelta(days=30 + index * 5) target_status = DealRoom.Status.OPEN room, was_created = DealRoom.objects.get_or_create( entrepreneur=profile, business=biz, counterparty_content_type=cp_ct, counterparty_object_id=str(counterparty.pk), defaults={ "title": f"{biz.name} ↔ {counterparty}", "summary": "Initial commercial conversation seeded for demo.", "expiry_date": expiry, "created_by": admin, }, ) if was_created: rooms_created += 1 # Move to target status if needed room = DealRoom.objects.get(pk=room.pk) if target_status == DealRoom.Status.EXPIRED and room.status == DealRoom.Status.OPEN: room.expire() room.save() room = DealRoom.objects.get(pk=room.pk) elif target_status == DealRoom.Status.AGREEMENT_REACHED and room.status == DealRoom.Status.OPEN: # We add the agreement below; flip status here. room.reach_agreement() room.save() room = DealRoom.objects.get(pk=room.pk) # Seed messages — 2-4 per room, alternating author cp_owner = ( getattr(counterparty, "managed_by", None) or getattr(counterparty, "user", None) ) thread_authors = [profile.user, cp_owner or admin] for j in range(2 + (index % 3)): body = _MESSAGES[j % len(_MESSAGES)] existing = DealRoomMessage.objects.filter( deal_room=room, author=thread_authors[j % 2], body=body, ).exists() if existing: continue DealRoomMessage.objects.create( deal_room=room, author=thread_authors[j % 2], body=body, ) messages_created += 1 # Seed one proposal document on every room if not DealRoomDocument.objects.filter(deal_room=room, label="Proposal v1").exists(): DealRoomDocument.objects.create( deal_room=room, label="Proposal v1", kind=DealRoomDocument.Kind.PROPOSAL, file=ContentFile(b"DEMO PROPOSAL", name=f"proposal-{room.pk}.pdf"), version=1, notes="Seeded baseline proposal.", uploaded_by=profile.user, ) documents_created += 1 # Add a signed agreement for the agreement-reached room if target_status == DealRoom.Status.AGREEMENT_REACHED and not hasattr(room, "agreement"): DealAgreement.objects.create( deal_room=room, title=f"Memorandum of Understanding — {biz.name}", summary="High-level MoU covering distribution and capacity-building.", signed_file=ContentFile(b"SIGNED PDF", name=f"agreement-{room.pk}.pdf"), formalised_by_admin=admin, notes="Seeded demo agreement.", ) agreements_created += 1 return { "rooms": rooms_created, "messages": messages_created, "documents": documents_created, "agreements": agreements_created, } # --------------------------------------------------------------------------- # Attendance # --------------------------------------------------------------------------- def _seed_attendance(admin: User) -> int: concluded_or_live = ShowcaseEvent.objects.filter( status__in=[ShowcaseEvent.Status.LIVE, ShowcaseEvent.Status.CONCLUDED], ) created = 0 for event in concluded_or_live: ct = ContentType.objects.get_for_model(ShowcaseEvent) confirmed = ShowcaseApplication.objects.filter( target_content_type=ct, target_object_id=str(event.pk), status=ShowcaseApplication.Status.CONFIRMED, ).select_related("entrepreneur__user") for idx, app in enumerate(confirmed): user = app.entrepreneur.user attended = idx % 4 != 0 # ~75% attendance _, was_created = EventAttendance.objects.get_or_create( event=event, participant=user, defaults={ "application": app, "attended": attended, "checked_in_at": timezone.now() if attended else None, "checked_in_by": admin, }, ) if was_created: created += 1 return created # --------------------------------------------------------------------------- # Top-level seed # --------------------------------------------------------------------------- @transaction.atomic def seed() -> dict: admin = _ensure_admin() judges = _ensure_judges() events = _seed_events(admin, judges) challenges = _seed_challenges(admin) apps_summary = _seed_applications(events, challenges, judges, admin) catalogue_count = _seed_catalogue(admin) deal_summary = _seed_deal_rooms(admin) attendance_count = _seed_attendance(admin) return { "events": len(events), "challenges": len(challenges), "applications": apps_summary["applications"], "scores": apps_summary["scores"], "catalogue_entries": catalogue_count, "deal_rooms": deal_summary["rooms"], "deal_room_messages": deal_summary["messages"], "deal_room_documents": deal_summary["documents"], "deal_agreements": deal_summary["agreements"], "attendance_records": attendance_count, }