"""SME-Hub marketplace models — Subprocess SP3b. Implements PRD Table 39 (FRSME-MPL014 → FRSME-MPL016) plus buyer registry: • ProductListing (FRSME-MPL014) — entrepreneur catalogue entries with media • ProductMedia — supporting images / pitch decks per listing • BuyerRegistryEntry — buyer organisation profile with verification FSM • MarketDemandListing (FRSME-MPL015) — buyer-posted procurement need • DemandResponse (FRSME-MPL016) — entrepreneur response to a demand listing """ from __future__ import annotations import uuid from decimal import Decimal from django.conf import settings from django.core.exceptions import ValidationError from django.core.validators import MaxValueValidator, MinValueValidator from django.db import models from django.utils import timezone from django_fsm import FSMField, transition from apps.core.money.currencies import DEFAULT_CURRENCY, ISO4217_CHOICES from apps.core.storage.paths import upload_to_smehub_marketplace from apps.smehub.onboarding.models import Business, EntrepreneurProfile # --------------------------------------------------------------------------- # Buyer registry (Buyer self-onboards via SP1 with role=BUYER; richer # org-level metadata captured here for the marketplace flows.) # --------------------------------------------------------------------------- class BuyerRegistryEntry(models.Model): class OrgType(models.TextChoices): WHOLESALE = "wholesale", "Wholesaler / aggregator" RETAIL = "retail", "Retailer" EXPORTER = "exporter", "Exporter" PROCESSOR = "processor", "Processor" INSTITUTIONAL = "institutional", "Institutional buyer" GOVERNMENT = "government", "Government / parastatal" OTHER = "other", "Other" class Status(models.TextChoices): PENDING = "pending", "Pending verification" VERIFIED = "verified", "Verified" REJECTED = "rejected", "Rejected" DEACTIVATED = "deactivated", "Deactivated" user = models.OneToOneField( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="smehub_buyer_registry", ) organisation_name = models.CharField(max_length=255, db_index=True) org_type = models.CharField(max_length=24, choices=OrgType.choices, db_index=True) sectors_sourced = models.JSONField(default=list, blank=True) geographic_coverage = models.JSONField(default=list, blank=True) contact_email = models.EmailField(blank=True) contact_phone = models.CharField(max_length=64, blank=True) website = models.URLField(blank=True) description = models.TextField(blank=True) typical_volume = models.CharField( max_length=255, blank=True, help_text="Free-form: e.g. '5–10 tonnes/month maize'.", ) verification_status = FSMField( max_length=16, choices=Status.choices, default=Status.PENDING, protected=True, db_index=True, ) verified_by = models.ForeignKey( settings.AUTH_USER_MODEL, null=True, blank=True, on_delete=models.SET_NULL, related_name="+", ) verified_at = models.DateTimeField(null=True, blank=True) rejection_reason = models.TextField(blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["organisation_name"] verbose_name_plural = "Buyer registry entries" def __str__(self) -> str: return self.organisation_name @transition( field=verification_status, source=[Status.PENDING, Status.REJECTED], target=Status.VERIFIED, ) def verify(self, *, by_user) -> None: self.verified_by = by_user self.verified_at = timezone.now() self.rejection_reason = "" @transition(field=verification_status, source=Status.PENDING, target=Status.REJECTED) def reject(self, *, reason: str) -> None: self.rejection_reason = reason @transition( field=verification_status, source=[Status.VERIFIED, Status.PENDING, Status.REJECTED], target=Status.DEACTIVATED, ) def deactivate(self, *, reason: str = "") -> None: """FRSME-MPL001 — admin can deactivate a buyer registry entry.""" if reason: self.rejection_reason = reason @transition(field=verification_status, source=Status.DEACTIVATED, target=Status.VERIFIED) def reactivate(self, *, by_user) -> None: self.verified_by = by_user self.verified_at = timezone.now() self.rejection_reason = "" # --------------------------------------------------------------------------- # Product listings (FRSME-MPL014) # --------------------------------------------------------------------------- class ProductListing(models.Model): class Availability(models.TextChoices): IN_STOCK = "in_stock", "In stock" TO_ORDER = "to_order", "Available to order" SEASONAL = "seasonal", "Seasonal" OUT_OF_STOCK = "out_of_stock", "Out of stock" class Status(models.TextChoices): DRAFT = "draft", "Draft" PUBLISHED = "published", "Published" UNPUBLISHED = "unpublished", "Unpublished" entrepreneur = models.ForeignKey( EntrepreneurProfile, on_delete=models.CASCADE, related_name="product_listings", ) business = models.ForeignKey( Business, on_delete=models.CASCADE, related_name="product_listings", ) listing_id = models.CharField( max_length=20, unique=True, editable=False, help_text="Auto-generated public identifier.", ) name = models.CharField(max_length=255, db_index=True) description = models.TextField(blank=True) sector = models.CharField(max_length=128, db_index=True) sub_sector = models.CharField(max_length=128, blank=True) pricing = models.CharField( max_length=255, blank=True, help_text="Free-form pricing string (e.g. 'USD 12.5 / kg', 'POA').", ) minimum_order = models.CharField(max_length=128, blank=True) availability = models.CharField( max_length=24, choices=Availability.choices, default=Availability.IN_STOCK, ) geographic_reach = models.JSONField( default=list, blank=True, help_text="ISO-2 codes the listing serves.", ) cover_image = models.FileField( upload_to=upload_to_smehub_marketplace, null=True, blank=True, ) status = models.CharField( max_length=16, choices=Status.choices, default=Status.DRAFT, db_index=True, ) published_at = models.DateTimeField(null=True, blank=True) unpublished_at = models.DateTimeField(null=True, blank=True) external_ref = models.PositiveBigIntegerField( null=True, blank=True, unique=True, db_index=True, help_text="Source id from the UltimatePOS `products` table.", ) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["-created_at"] indexes = [ models.Index(fields=["sector", "status"]), models.Index(fields=["entrepreneur", "status"]), ] def __str__(self) -> str: return f"{self.listing_id} — {self.name}" def save(self, *args, **kwargs): if not self.listing_id: self.listing_id = f"PRD-{uuid.uuid4().hex[:8].upper()}" super().save(*args, **kwargs) class ProductMedia(models.Model): """Additional pitch / spec media attached to a product listing.""" class Kind(models.TextChoices): IMAGE = "image", "Image" DOCUMENT = "document", "Document" VIDEO_LINK = "video_link", "Video link" listing = models.ForeignKey( ProductListing, on_delete=models.CASCADE, related_name="media", ) kind = models.CharField(max_length=16, choices=Kind.choices, default=Kind.IMAGE) title = models.CharField(max_length=255, blank=True) file = models.FileField( upload_to=upload_to_smehub_marketplace, null=True, blank=True, ) external_url = models.URLField(blank=True) sort_order = models.PositiveSmallIntegerField(default=0) created_at = models.DateTimeField(auto_now_add=True) class Meta: ordering = ["sort_order", "id"] def __str__(self) -> str: return f"{self.kind} for {self.listing.listing_id}" # --------------------------------------------------------------------------- # Market demand + responses (FRSME-MPL015 / MPL016) # --------------------------------------------------------------------------- class MarketDemandListing(models.Model): class Kind(models.TextChoices): PROBLEM = "problem", "Problem statement" TENDER = "tender", "Tender" PROCUREMENT = "procurement", "Procurement need" OFFTAKE = "offtake", "Off-take opportunity" class Status(models.TextChoices): OPEN = "open", "Open" CLOSED = "closed", "Closed" AWARDED = "awarded", "Awarded" buyer = models.ForeignKey( BuyerRegistryEntry, on_delete=models.CASCADE, related_name="demand_listings", ) title = models.CharField(max_length=255, db_index=True) listing_id = models.CharField( max_length=20, unique=True, editable=False, ) kind = models.CharField(max_length=16, choices=Kind.choices, default=Kind.PROBLEM) description = models.TextField() sector_targeting = models.JSONField(default=list, blank=True) geographic_targeting = models.JSONField(default=list, blank=True) quantity = models.CharField( max_length=255, blank=True, help_text="Free-form quantity / volume requirement.", ) deadline = models.DateTimeField(null=True, blank=True) status = models.CharField( max_length=16, choices=Status.choices, default=Status.OPEN, db_index=True, ) closed_at = models.DateTimeField(null=True, blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["-created_at"] indexes = [ models.Index(fields=["status", "deadline"]), ] def __str__(self) -> str: return f"{self.listing_id} — {self.title}" def save(self, *args, **kwargs): if not self.listing_id: self.listing_id = f"DMD-{uuid.uuid4().hex[:8].upper()}" super().save(*args, **kwargs) class DemandResponse(models.Model): class Status(models.TextChoices): SUBMITTED = "submitted", "Submitted" SHORTLISTED = "shortlisted", "Shortlisted" AWARDED = "awarded", "Awarded" DECLINED = "declined", "Declined" WITHDRAWN = "withdrawn", "Withdrawn" demand_listing = models.ForeignKey( MarketDemandListing, on_delete=models.CASCADE, related_name="responses", ) entrepreneur = models.ForeignKey( EntrepreneurProfile, on_delete=models.CASCADE, related_name="demand_responses", ) business = models.ForeignKey( Business, on_delete=models.CASCADE, related_name="demand_responses", ) product_listing = models.ForeignKey( ProductListing, on_delete=models.SET_NULL, null=True, blank=True, related_name="demand_responses", help_text="Optional reference to a published product listing.", ) message = models.TextField() pricing_offer = models.CharField(max_length=255, blank=True) attachment = models.FileField( upload_to=upload_to_smehub_marketplace, null=True, blank=True, ) status = models.CharField( max_length=16, choices=Status.choices, default=Status.SUBMITTED, db_index=True, ) decided_at = models.DateTimeField(null=True, blank=True) decided_by = models.ForeignKey( settings.AUTH_USER_MODEL, null=True, blank=True, on_delete=models.SET_NULL, related_name="+", ) decision_note = models.TextField(blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["-created_at"] constraints = [ # Prevent a business from posting two active responses to the same listing. models.UniqueConstraint( fields=["demand_listing", "business"], condition=models.Q(status__in=["submitted", "shortlisted"]), name="uniq_smehub_active_demand_response", ), ] indexes = [ models.Index(fields=["demand_listing", "status"]), models.Index(fields=["entrepreneur", "status"]), ] def __str__(self) -> str: return f"Response by {self.business.name} → {self.demand_listing.listing_id}" # --------------------------------------------------------------------------- # Market validation (Phase 6.3 — customer discovery evidence) # --------------------------------------------------------------------------- class MarketValidation(models.Model): """One row per customer-discovery touchpoint for a ProductListing. Captures the evidence an entrepreneur uses to argue product/market fit. Aggregations (count, average score, WTP distribution) are computed at view time; this model stays event-grained so trends can be charted. """ class Method(models.TextChoices): INTERVIEW = "interview", "Customer interview" SURVEY = "survey", "Survey" PILOT = "pilot", "Pilot / trial" LOI = "loi", "Letter of intent" FOCUS_GROUP = "focus_group", "Focus group" DEMO_FEEDBACK = "demo_feedback", "Demo feedback" product_listing = models.ForeignKey( ProductListing, on_delete=models.PROTECT, related_name="validations", ) business = models.ForeignKey( Business, on_delete=models.PROTECT, related_name="market_validations", help_text="Denormalised from product_listing.business; auto-set in save().", ) recorded_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="smehub_market_validations_recorded", ) validator_name = models.CharField( max_length=255, blank=True, help_text="Free-form: 'Acme Foods Procurement Lead' or 'Anonymous interview #3'.", ) validator_email = models.EmailField(blank=True) validator_user = models.ForeignKey( settings.AUTH_USER_MODEL, null=True, blank=True, on_delete=models.SET_NULL, related_name="smehub_market_validations_given", ) validator_buyer_entry = models.ForeignKey( BuyerRegistryEntry, null=True, blank=True, on_delete=models.SET_NULL, related_name="market_validations", ) method = models.CharField(max_length=24, choices=Method.choices, db_index=True) conducted_at = models.DateTimeField(default=timezone.now) problem_fit_score = models.PositiveSmallIntegerField( validators=[MinValueValidator(1), MaxValueValidator(5)], help_text="1 (no fit) – 5 (strong fit).", ) solution_fit_score = models.PositiveSmallIntegerField( validators=[MinValueValidator(1), MaxValueValidator(5)], help_text="1 (poor solution) – 5 (great solution).", ) willingness_to_pay_score = models.PositiveSmallIntegerField( validators=[MinValueValidator(1), MaxValueValidator(5)], help_text="1 (won't pay) – 5 (will pay readily).", ) willingness_to_pay_amount = models.DecimalField( max_digits=14, decimal_places=2, null=True, blank=True, help_text="Optional indicative amount the validator would pay.", ) currency = models.CharField( max_length=8, choices=ISO4217_CHOICES, blank=True, help_text="Required iff willingness_to_pay_amount is set.", ) notes = models.TextField(blank=True) evidence_file = models.FileField( upload_to=upload_to_smehub_marketplace, null=True, blank=True, help_text="Optional supporting artefact (transcript, signed LOI, survey export).", ) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["-conducted_at"] indexes = [ models.Index(fields=["product_listing", "-conducted_at"]), models.Index(fields=["business", "method"]), ] def __str__(self) -> str: return f"{self.method} for {self.product_listing.listing_id}" def clean(self): super().clean() if self.willingness_to_pay_amount is not None and not self.currency: raise ValidationError( {"currency": "Currency is required when willingness_to_pay_amount is set."} ) def save(self, *args, **kwargs): if self.product_listing_id and not self.business_id: self.business = self.product_listing.business super().save(*args, **kwargs) @property def average_score(self) -> Decimal: total = self.problem_fit_score + self.solution_fit_score + self.willingness_to_pay_score return (Decimal(total) / Decimal(3)).quantize(Decimal("0.01")) @property def has_strong_signal(self) -> bool: if self.method in (self.Method.LOI, self.Method.PILOT): return True return min( self.problem_fit_score, self.solution_fit_score, self.willingness_to_pay_score, ) >= 4 # --------------------------------------------------------------------------- # Sales record (Phase 6.3 — post-marketplace transaction tracking) # --------------------------------------------------------------------------- class SalesRecord(models.Model): """One row per booked sale. May be auto-stubbed (status=PENDING_DETAILS) when a DemandResponse is awarded, or manually recorded for off-platform / direct-listing sales. Status is a plain CharField — the state space is small and there is no need for FSM protection. """ class Channel(models.TextChoices): MARKETPLACE_AWARD = "marketplace_award", "Marketplace award" DIRECT_LISTING = "direct_listing", "Direct listing" OFFLINE = "offline", "Offline / walk-in" REPEAT_CUSTOMER = "repeat_customer", "Repeat customer" REFERRAL = "referral", "Referral" EXPORT = "export", "Export" class Status(models.TextChoices): PENDING_DETAILS = "pending_details", "Pending details" RECORDED = "recorded", "Recorded" VERIFIED = "verified", "Verified" DISPUTED = "disputed", "Disputed" CANCELLED = "cancelled", "Cancelled" product_listing = models.ForeignKey( ProductListing, on_delete=models.PROTECT, null=True, blank=True, related_name="sales", help_text="Nullable so off-catalogue sales can still be logged.", ) business = models.ForeignKey( Business, on_delete=models.PROTECT, related_name="sales", ) entrepreneur = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="smehub_sales_recorded", ) demand_response = models.ForeignKey( DemandResponse, null=True, blank=True, on_delete=models.SET_NULL, related_name="sales", help_text="Set by the auto-stub when a DemandResponse is awarded.", ) buyer_entry = models.ForeignKey( BuyerRegistryEntry, null=True, blank=True, on_delete=models.SET_NULL, related_name="sales_purchased", ) customer_name = models.CharField( max_length=255, blank=True, help_text="Required when buyer_entry is null (validated in clean()).", ) channel = models.CharField(max_length=24, choices=Channel.choices, db_index=True) quantity = models.DecimalField( max_digits=12, decimal_places=2, default=Decimal("1.00"), ) unit = models.CharField(max_length=32, blank=True, default="units") unit_price = models.DecimalField( max_digits=14, decimal_places=2, null=True, blank=True, help_text="Auto-derived from gross_amount / quantity in save() if omitted.", ) gross_amount = models.DecimalField( max_digits=14, decimal_places=2, default=Decimal("0.00"), ) currency = models.CharField( max_length=8, choices=ISO4217_CHOICES, default=DEFAULT_CURRENCY, ) sale_date = models.DateField(null=True, blank=True) status = models.CharField( max_length=24, choices=Status.choices, default=Status.RECORDED, db_index=True, ) notes = models.TextField(blank=True) evidence_file = models.FileField( upload_to=upload_to_smehub_marketplace, null=True, blank=True, help_text="Invoice / receipt / contract.", ) verified_by = models.ForeignKey( settings.AUTH_USER_MODEL, null=True, blank=True, on_delete=models.SET_NULL, related_name="smehub_sales_verified", ) verified_at = models.DateTimeField(null=True, blank=True) dispute_reason = models.TextField(blank=True) cancel_reason = models.TextField(blank=True) # Legacy external-system provenance (UltimatePOS sales import). Idempotent # on (source_system, source_transaction_id). source_system = models.CharField( max_length=24, blank=True, help_text='e.g. "ultimatepos".' ) source_transaction_id = models.PositiveIntegerField(null=True, blank=True, db_index=True) source_invoice_no = models.CharField(max_length=191, blank=True) recorded_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["-sale_date", "-recorded_at"] indexes = [ models.Index(fields=["business", "-sale_date"]), models.Index(fields=["product_listing", "-sale_date"]), models.Index(fields=["channel", "status"]), ] constraints = [ models.UniqueConstraint( fields=["demand_response"], condition=models.Q(demand_response__isnull=False), name="uniq_smehub_sale_per_demand_response", ), models.UniqueConstraint( fields=["source_system", "source_transaction_id"], condition=models.Q(source_transaction_id__isnull=False), name="uniq_smehub_sale_per_source_txn", ), ] def __str__(self) -> str: return ( f"Sale {self.business.name} {self.gross_amount} {self.currency} " f"({self.get_status_display()})" ) def clean(self): super().clean() if not self.buyer_entry_id and not self.customer_name: raise ValidationError( {"customer_name": "Provide a customer name when no buyer is linked."} ) if self.status == self.Status.RECORDED and self.gross_amount and self.gross_amount <= 0: raise ValidationError( {"gross_amount": "Recorded sales must have a positive gross amount."} ) def save(self, *args, **kwargs): if ( self.unit_price is None and self.gross_amount and self.quantity and self.quantity > 0 ): self.unit_price = ( Decimal(self.gross_amount) / Decimal(self.quantity) ).quantize(Decimal("0.01")) super().save(*args, **kwargs) @property def is_complete(self) -> bool: return bool( self.gross_amount and self.gross_amount > 0 and self.sale_date is not None )