"""SP3 marketplace views — products, buyer registry, market demand.""" from __future__ import annotations from django.contrib import messages from django.contrib.auth.mixins import LoginRequiredMixin from django.core.exceptions import PermissionDenied from django.db.models import Q from django.http import Http404 from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse, reverse_lazy from django.views import View from django.views.decorators.http import require_POST from django.views.generic import ( CreateView, DetailView, ListView, TemplateView, UpdateView, ) from apps.core.lists import paginate_qs from apps.core.permissions.roles import UserRole from apps.smehub._shared.role_gates import ( BuyerRequired, EntrepreneurRequired, SMEAdminRequired, ) from apps.smehub.marketplace import services from apps.smehub.marketplace.forms import ( BuyerRegistryForm, DemandResponseDecisionForm, DemandResponseForm, MarketDemandForm, MarketValidationForm, ProductListingForm, ProductMediaFormSet, SalesCancelForm, SalesDisputeForm, SalesRecordForm, ) from apps.smehub.marketplace.models import ( BuyerRegistryEntry, DemandResponse, MarketDemandListing, MarketValidation, ProductListing, SalesRecord, ) from apps.smehub.onboarding.models import Business, EntrepreneurProfile SME_ADMIN_ROLES = { UserRole.SYSTEM_ADMIN, UserRole.ADMIN, UserRole.SME_ADMIN, } def _is_sme_admin(user) -> bool: if not user.is_authenticated: return False if user.is_superuser: return True return user.role in SME_ADMIN_ROLES def _get_entrepreneur_profile(user) -> EntrepreneurProfile | None: if not user.is_authenticated: return None return EntrepreneurProfile.objects.filter(user=user).first() def _get_buyer_entry(user) -> BuyerRegistryEntry | None: if not user.is_authenticated: return None return BuyerRegistryEntry.objects.filter(user=user).first() # --------------------------------------------------------------------------- # Product catalogue (public) # --------------------------------------------------------------------------- class MarketplaceHubView(LoginRequiredMixin, TemplateView): """Hub landing — product catalogue, demand, sales, responses (Slice B.2). Bridges across the marketplace and linkage modules so users see a single cohesive page rather than navigating two columns of the mega-menu. """ template_name = "smehub/marketplace/hub.html" def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) user = self.request.user ctx["is_admin"] = _is_sme_admin(user) ctx["entrepreneur"] = _get_entrepreneur_profile(user) if user.is_authenticated else None ctx["buyer_entry"] = _get_buyer_entry(user) if user.is_authenticated else None ctx["product_count"] = ProductListing.objects.filter( status=ProductListing.Status.PUBLISHED, ).count() ctx["demand_count"] = MarketDemandListing.objects.filter( status=MarketDemandListing.Status.OPEN, ).count() ctx["recent_products"] = ( ProductListing.objects.filter(status=ProductListing.Status.PUBLISHED) .select_related("entrepreneur__user", "business") .order_by("-published_at")[:6] ) ctx["recent_demand"] = ( MarketDemandListing.objects.filter(status=MarketDemandListing.Status.OPEN) .select_related("buyer") .order_by("-created_at")[:6] ) ctx["my_product_count"] = ( ProductListing.objects.filter(entrepreneur=ctx["entrepreneur"]).count() if ctx["entrepreneur"] else 0 ) ctx["my_demand_count"] = ( MarketDemandListing.objects.filter(buyer=ctx["buyer_entry"]).count() if ctx["buyer_entry"] else 0 ) return ctx class ProductCatalogueView(ListView): model = ProductListing context_object_name = "listings" paginate_by = 24 def get_template_names(self): if getattr(self.request, "htmx", False): return ["smehub/marketplace/partials/product_catalogue_results.html"] if not self.request.user.is_authenticated: return ["smehub/marketplace/product_catalogue_public.html"] return ["smehub/marketplace/product_catalogue.html"] def get_queryset(self): qs = ProductListing.objects.filter(status=ProductListing.Status.PUBLISHED).select_related( "business", "entrepreneur__user", ) q = self.request.GET.get("q") if q: qs = qs.filter(Q(name__icontains=q) | Q(description__icontains=q) | Q(business__name__icontains=q)) # Slice C — sector + country facets, mirroring the buyer-registry pattern. sector = self.request.GET.get("sector") if sector: qs = qs.filter(sector__iexact=sector) country = self.request.GET.get("country") if country: qs = qs.filter(business__country__iexact=country) business = self.request.GET.get("business") if business: qs = qs.filter(business_id=business) return qs def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["q_filter"] = self.request.GET.get("q", "") ctx["sector_filter"] = self.request.GET.get("sector", "") ctx["country_filter"] = self.request.GET.get("country", "") # Distinct values from the existing published catalogue — keeps # the dropdowns relevant without a static config list. published = ProductListing.objects.filter(status=ProductListing.Status.PUBLISHED) ctx["sectors"] = sorted({s for s in published.values_list("sector", flat=True) if s}) ctx["countries"] = sorted({c for c in published.values_list("business__country", flat=True) if c}) return ctx class ProductDetailView(DetailView): model = ProductListing context_object_name = "listing" def get_template_names(self): if not self.request.user.is_authenticated: return ["smehub/marketplace/product_detail_public.html"] return ["smehub/marketplace/product_detail.html"] def get_queryset(self): qs = super().get_queryset().select_related("business", "entrepreneur__user") if not self.request.user.is_authenticated: return qs.filter(status=ProductListing.Status.PUBLISHED) if not _is_sme_admin(self.request.user): qs = qs.filter( Q(status=ProductListing.Status.PUBLISHED) | Q(entrepreneur__user=self.request.user), ) return qs def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) listing: ProductListing = self.object # type: ignore[assignment] seller_listings = ( ProductListing.objects.filter( business=listing.business, status=ProductListing.Status.PUBLISHED, ) .exclude(pk=listing.pk) .order_by("-created_at") ) ctx["seller_listing_count"] = seller_listings.count() ctx["other_listings"] = list(seller_listings[:6]) ctx["is_seller"] = listing.entrepreneur.user_id == self.request.user.pk return ctx # --------------------------------------------------------------------------- # Entrepreneur — manage own products # --------------------------------------------------------------------------- class MyProductsView(EntrepreneurRequired, ListView): model = ProductListing template_name = "smehub/marketplace/my_products.html" context_object_name = "listings" paginate_by = 30 def get_queryset(self): profile = _get_entrepreneur_profile(self.request.user) if profile is None: return ProductListing.objects.none() return ProductListing.objects.filter(entrepreneur=profile).select_related("business") def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) profile = _get_entrepreneur_profile(self.request.user) qs = ProductListing.objects.filter(entrepreneur=profile) if profile else ProductListing.objects.none() ctx["stats"] = { "total": qs.count(), "published": qs.filter(status=ProductListing.Status.PUBLISHED).count(), "draft": qs.filter(status=ProductListing.Status.DRAFT).count(), } return ctx class _BusinessChoiceMixin: """Provides the entrepreneur's verified businesses to product forms.""" def _verified_businesses(self): profile = _get_entrepreneur_profile(self.request.user) if profile is None: return Business.objects.none() return Business.objects.filter(entrepreneur=profile, verification_status=Business.Status.VERIFIED) class ProductCreateView(EntrepreneurRequired, _BusinessChoiceMixin, View): template_name = "smehub/marketplace/product_list_form.html" def get(self, request): return render(request, self.template_name, { "form": ProductListingForm(), "media_formset": ProductMediaFormSet(), "businesses": self._verified_businesses(), }) def post(self, request): form = ProductListingForm(request.POST, request.FILES) businesses = self._verified_businesses() business_id = request.POST.get("business") if not form.is_valid() or not business_id: return render(request, self.template_name, { "form": form, "media_formset": ProductMediaFormSet(request.POST, request.FILES), "businesses": businesses, }) try: business = businesses.get(pk=business_id) except Business.DoesNotExist: messages.error(request, "Pick a verified business.") return render(request, self.template_name, { "form": form, "media_formset": ProductMediaFormSet(request.POST, request.FILES), "businesses": businesses, }) profile = _get_entrepreneur_profile(request.user) listing = services.create_product_listing( entrepreneur=profile, business=business, **form.cleaned_data, ) media_formset = ProductMediaFormSet(request.POST, request.FILES, instance=listing) if media_formset.is_valid(): media_formset.save() messages.success(request, "Product listing created. Publish when ready.") return redirect("smehub_marketplace:product_update", pk=listing.pk) class ProductUpdateView(EntrepreneurRequired, _BusinessChoiceMixin, UpdateView): model = ProductListing form_class = ProductListingForm template_name = "smehub/marketplace/product_list_form.html" def get_object(self, queryset=None): obj = super().get_object(queryset) profile = _get_entrepreneur_profile(self.request.user) if profile is None or obj.entrepreneur_id != profile.pk: raise PermissionDenied() return obj def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["businesses"] = self._verified_businesses() if "media_formset" not in ctx: ctx["media_formset"] = ProductMediaFormSet(instance=self.object) return ctx def post(self, request, *args, **kwargs): self.object = self.get_object() form = self.get_form() media_formset = ProductMediaFormSet(request.POST, request.FILES, instance=self.object) if form.is_valid() and media_formset.is_valid(): self.object = form.save() media_formset.save() messages.success(request, "Listing updated.") return redirect(self.get_success_url()) return self.render_to_response(self.get_context_data(form=form, media_formset=media_formset)) def get_success_url(self): return reverse("smehub_marketplace:my_products") @require_POST def product_publish_view(request, pk: int): listing = get_object_or_404(ProductListing, pk=pk) profile = _get_entrepreneur_profile(request.user) if profile is None or listing.entrepreneur_id != profile.pk: raise PermissionDenied() services.publish_product(listing, by_user=request.user) messages.success(request, "Product published.") return redirect("smehub_marketplace:my_products") @require_POST def product_express_interest_view(request, pk: int): """In-app fallback for "Express interest" — works without a mail client.""" if not request.user.is_authenticated: raise PermissionDenied() listing = get_object_or_404( ProductListing, pk=pk, status=ProductListing.Status.PUBLISHED, ) profile = _get_entrepreneur_profile(request.user) if profile is not None and listing.entrepreneur_id == profile.pk: messages.info(request, "This is your own product listing.") return redirect("smehub_marketplace:product_detail", pk=listing.pk) services.express_product_interest( listing, from_user=request.user, message=request.POST.get("message", ""), ) messages.success( request, "Your interest has been sent to the seller — they'll receive it in-app and by email.", ) return redirect("smehub_marketplace:product_detail", pk=listing.pk) @require_POST def product_unpublish_view(request, pk: int): listing = get_object_or_404(ProductListing, pk=pk) profile = _get_entrepreneur_profile(request.user) if profile is None or listing.entrepreneur_id != profile.pk: raise PermissionDenied() services.unpublish_product(listing, by_user=request.user) messages.success(request, "Product unpublished.") return redirect("smehub_marketplace:my_products") # --------------------------------------------------------------------------- # Buyer registry # --------------------------------------------------------------------------- class BuyerRegistryView(LoginRequiredMixin, ListView): model = BuyerRegistryEntry template_name = "smehub/marketplace/buyer_registry.html" context_object_name = "buyers" paginate_by = 30 def get_queryset(self): """Audit fix — keyword + sector + geographic coverage filters. The legacy view supported only ``?q=`` keyword filtering; PRD FRSME-MPL004 calls for sector and geography facets across every directory. ``sectors_sourced`` and ``geographic_coverage`` are JSONField lists, so we use ``__contains=[value]`` to match any entry whose list includes the requested filter. """ qs = BuyerRegistryEntry.objects.filter(verification_status=BuyerRegistryEntry.Status.VERIFIED) q = self.request.GET.get("q") if q: qs = qs.filter(Q(organisation_name__icontains=q) | Q(description__icontains=q)) sector = self.request.GET.get("sector") if sector: qs = qs.filter(sectors_sourced__contains=[sector]) country = self.request.GET.get("country") if country: qs = qs.filter(geographic_coverage__contains=[country]) org_type = self.request.GET.get("org_type") if org_type: qs = qs.filter(org_type=org_type) return qs def get_context_data(self, **kwargs): # FRSME-MPL004 — expose the facets the queryset supports as filter # controls, mirroring the partner / service-provider directories. from apps.smehub._shared.choices import COUNTRIES, SECTORS ctx = super().get_context_data(**kwargs) ctx["sector_choices"] = SECTORS ctx["country_choices"] = COUNTRIES ctx["org_type_choices"] = BuyerRegistryEntry.OrgType.choices ctx["sector_selected"] = self.request.GET.get("sector", "") ctx["country_selected"] = self.request.GET.get("country", "") ctx["org_type_selected"] = self.request.GET.get("org_type", "") return ctx class BuyerDetailView(LoginRequiredMixin, DetailView): model = BuyerRegistryEntry template_name = "smehub/marketplace/buyer_detail.html" context_object_name = "buyer" _BUYER_PROFILE_WIZARD_STEPS = [ {"n": 1, "title": "About"}, {"n": 2, "title": "Sourcing"}, ] class BuyerRegistryFormView(BuyerRequired, View): template_name = "smehub/marketplace/buyer_profile_form.html" def _instance(self, user): return BuyerRegistryEntry.objects.filter(user=user).first() def _ctx(self, form, instance): return { "form": form, "buyer": instance, "wizard_steps": _BUYER_PROFILE_WIZARD_STEPS, "cancel_url": reverse("smehub_marketplace:buyer_registry"), "submit_label": "Save changes" if instance else "Submit for verification", } def get(self, request): instance = self._instance(request.user) form = BuyerRegistryForm(instance=instance) return render(request, self.template_name, self._ctx(form, instance)) def post(self, request): instance = self._instance(request.user) form = BuyerRegistryForm(request.POST, instance=instance) if not form.is_valid(): return render(request, self.template_name, self._ctx(form, instance)) services.register_buyer(user=request.user, **form.cleaned_data) messages.success(request, "Buyer profile saved.") return redirect("smehub_marketplace:buyer_profile_form") # --------------------------------------------------------------------------- # Buyer admin verification # --------------------------------------------------------------------------- class AdminBuyerListView(SMEAdminRequired, TemplateView): template_name = "smehub/marketplace/admin_buyer_list.html" def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["pending"] = paginate_qs( self.request, BuyerRegistryEntry.objects.filter( verification_status=BuyerRegistryEntry.Status.PENDING, ), param="pending_page", ) ctx["verified"] = paginate_qs( self.request, BuyerRegistryEntry.objects.filter( verification_status=BuyerRegistryEntry.Status.VERIFIED, ), param="verified_page", ) ctx["deactivated"] = paginate_qs( self.request, BuyerRegistryEntry.objects.filter( verification_status=BuyerRegistryEntry.Status.DEACTIVATED, ), param="deactivated_page", ) return ctx @require_POST def buyer_verify_view(request, pk: int): if not _is_sme_admin(request.user): raise PermissionDenied() buyer = get_object_or_404(BuyerRegistryEntry, pk=pk) services.verify_buyer(buyer, by_user=request.user) messages.success(request, f"{buyer.organisation_name} verified.") return redirect("smehub_marketplace:admin_buyer_list") @require_POST def buyer_reject_view(request, pk: int): if not _is_sme_admin(request.user): raise PermissionDenied() buyer = get_object_or_404(BuyerRegistryEntry, pk=pk) reason = request.POST.get("reason", "") if not reason.strip(): messages.error(request, "Rejection reason required.") return redirect("smehub_marketplace:admin_buyer_list") services.reject_buyer(buyer, by_user=request.user, reason=reason) messages.success(request, f"{buyer.organisation_name} rejected.") return redirect("smehub_marketplace:admin_buyer_list") @require_POST def buyer_deactivate_view(request, pk: int): if not _is_sme_admin(request.user): raise PermissionDenied() buyer = get_object_or_404(BuyerRegistryEntry, pk=pk) reason = request.POST.get("reason", "") try: services.deactivate_buyer(buyer, by_user=request.user, reason=reason) except Exception as exc: messages.error(request, f"Cannot deactivate: {exc}") else: messages.success(request, f"{buyer.organisation_name} deactivated.") return redirect("smehub_marketplace:admin_buyer_list") @require_POST def buyer_reactivate_view(request, pk: int): if not _is_sme_admin(request.user): raise PermissionDenied() buyer = get_object_or_404(BuyerRegistryEntry, pk=pk) try: services.reactivate_buyer(buyer, by_user=request.user) except Exception as exc: messages.error(request, f"Cannot reactivate: {exc}") else: messages.success(request, f"{buyer.organisation_name} reactivated.") return redirect("smehub_marketplace:admin_buyer_list") # --------------------------------------------------------------------------- # Market demand # --------------------------------------------------------------------------- class MarketDemandListView(LoginRequiredMixin, ListView): model = MarketDemandListing template_name = "smehub/marketplace/market_demand_list.html" context_object_name = "listings" paginate_by = 24 def _viewer_sectors(self) -> set[str]: """Verified-business sectors of the viewing entrepreneur (FRSME-MPL015).""" from apps.smehub.onboarding.models import Business return set( Business.objects.filter( entrepreneur__user=self.request.user, verification_status=Business.Status.VERIFIED, ) .exclude(sector="") .values_list("sector", flat=True) ) def get_queryset(self): qs = MarketDemandListing.objects.filter(status=MarketDemandListing.Status.OPEN).select_related("buyer") q = self.request.GET.get("q") if q: qs = qs.filter(Q(title__icontains=q) | Q(description__icontains=q)) return qs def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) # FRSME-MPL015 — flag listings whose sector targeting matches one of the # viewer's verified businesses so entrepreneurs can spot relevant demand. sectors = self._viewer_sectors() if sectors: for listing in ctx["listings"]: listing.matches_me = bool(set(listing.sector_targeting or []) & sectors) ctx["viewer_has_sectors"] = bool(sectors) return ctx class MarketDemandDetailView(LoginRequiredMixin, DetailView): model = MarketDemandListing template_name = "smehub/marketplace/demand_detail.html" context_object_name = "listing" def get_queryset(self): return super().get_queryset().select_related("buyer__user") def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) listing: MarketDemandListing = self.object # type: ignore[assignment] user = self.request.user buyer = _get_buyer_entry(user) is_owner_buyer = buyer is not None and listing.buyer_id == buyer.pk is_admin = _is_sme_admin(user) ctx["is_owner_buyer"] = is_owner_buyer ctx["is_sme_admin"] = is_admin ctx["can_manage"] = is_owner_buyer or is_admin # Responses table — only for buyer/admin to avoid leaking competitive intel. if is_owner_buyer or is_admin: ctx["responses"] = paginate_qs( self.request, listing.responses .select_related("business", "entrepreneur__user", "product_listing") .order_by("status", "-created_at"), param="resp_page", ) # If the viewer is an entrepreneur, surface their existing response (if any). entrepreneur = _get_entrepreneur_profile(user) if entrepreneur is not None: ctx["my_response"] = ( listing.responses .filter(entrepreneur=entrepreneur) .order_by("-created_at") .first() ) return ctx class MyDemandListingsView(BuyerRequired, ListView): model = MarketDemandListing template_name = "smehub/marketplace/my_demand_listings.html" context_object_name = "listings" paginate_by = 30 def get_queryset(self): buyer = _get_buyer_entry(self.request.user) if buyer is None: return MarketDemandListing.objects.none() return MarketDemandListing.objects.filter(buyer=buyer) def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) buyer = _get_buyer_entry(self.request.user) qs = MarketDemandListing.objects.filter(buyer=buyer) if buyer else MarketDemandListing.objects.none() responses_qs = DemandResponse.objects.filter(demand_listing__buyer=buyer) if buyer else DemandResponse.objects.none() ctx["stats"] = { "total": qs.count(), "open": qs.filter(status=MarketDemandListing.Status.OPEN).count(), "responses": responses_qs.count(), } return ctx class MarketDemandCreateView(BuyerRequired, View): template_name = "smehub/marketplace/demand_listing_form.html" def _verified_buyer_or_redirect(self, request): """FRSME-MPL015 — only verified buyers may post demand listings. Gate GET as well as POST: rendering the form to an unverified buyer and 403-ing on submit loses their input. Redirect to the buyer profile form instead. """ buyer = _get_buyer_entry(request.user) if buyer is not None and buyer.verification_status == BuyerRegistryEntry.Status.VERIFIED: return buyer, None messages.warning( request, "Complete and get your buyer profile verified before posting demand listings.", ) return None, redirect("smehub_marketplace:buyer_profile_form") def get(self, request): _buyer, deny = self._verified_buyer_or_redirect(request) if deny: return deny return render(request, self.template_name, {"form": MarketDemandForm()}) def post(self, request): buyer, deny = self._verified_buyer_or_redirect(request) if deny: return deny form = MarketDemandForm(request.POST) if not form.is_valid(): return render(request, self.template_name, {"form": form}) try: listing = services.post_market_demand(buyer=buyer, **form.cleaned_data) except services.MarketplaceStateError as exc: messages.error(request, str(exc)) return render(request, self.template_name, {"form": form}) messages.success(request, "Demand listing posted.") return redirect("smehub_marketplace:demand_detail", pk=listing.pk) @require_POST def demand_close_view(request, pk: int): listing = get_object_or_404(MarketDemandListing, pk=pk) buyer = _get_buyer_entry(request.user) if (buyer is None or listing.buyer_id != buyer.pk) and not _is_sme_admin(request.user): raise PermissionDenied() services.close_market_demand(listing, by_user=request.user) messages.success(request, "Demand listing closed.") return redirect("smehub_marketplace:my_demand_listings") class DemandResponsesView(BuyerRequired, DetailView): model = MarketDemandListing template_name = "smehub/marketplace/demand_responses.html" context_object_name = "listing" def get_object(self, queryset=None): obj = super().get_object(queryset) buyer = _get_buyer_entry(self.request.user) if (buyer is None or obj.buyer_id != buyer.pk) and not _is_sme_admin(self.request.user): raise PermissionDenied() return obj def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["responses"] = paginate_qs( self.request, self.object.responses.select_related("business", "entrepreneur__user"), param="resp_page", ) ctx["decision_form"] = DemandResponseDecisionForm() return ctx class DemandResponseCreateView(EntrepreneurRequired, View): template_name = "smehub/marketplace/demand_response_form.html" def get(self, request, pk: int): listing = get_object_or_404(MarketDemandListing, pk=pk) profile = _get_entrepreneur_profile(request.user) return render(request, self.template_name, { "listing": listing, "form": DemandResponseForm(entrepreneur=profile), }) def post(self, request, pk: int): listing = get_object_or_404(MarketDemandListing, pk=pk) profile = _get_entrepreneur_profile(request.user) form = DemandResponseForm(request.POST, request.FILES, entrepreneur=profile) if not form.is_valid(): return render(request, self.template_name, {"listing": listing, "form": form}) try: services.respond_to_demand( demand_listing=listing, entrepreneur=profile, business=form.cleaned_data["business"], message=form.cleaned_data["message"], pricing_offer=form.cleaned_data.get("pricing_offer", ""), attachment=form.cleaned_data.get("attachment"), product_listing=form.cleaned_data.get("product_listing"), ) except services.MarketplaceStateError as exc: messages.error(request, str(exc)) return render(request, self.template_name, {"listing": listing, "form": form}) messages.success(request, "Response submitted.") return redirect("smehub_marketplace:my_demand_responses") class DemandResponseDecideView(LoginRequiredMixin, View): def post(self, request, pk: int): response = get_object_or_404(DemandResponse, pk=pk) buyer = _get_buyer_entry(request.user) if (buyer is None or response.demand_listing.buyer_id != buyer.pk) and not _is_sme_admin(request.user): raise PermissionDenied() form = DemandResponseDecisionForm(request.POST) if not form.is_valid(): messages.error(request, "Pick an outcome.") return redirect("smehub_marketplace:demand_responses", pk=response.demand_listing.pk) try: services.decide_demand_response( response, by_user=request.user, outcome=form.cleaned_data["outcome"], note=form.cleaned_data.get("note", ""), ) except services.MarketplaceStateError as exc: messages.error(request, str(exc)) else: messages.success(request, "Response decision recorded.") return redirect("smehub_marketplace:demand_responses", pk=response.demand_listing.pk) class MyDemandResponsesView(EntrepreneurRequired, ListView): model = DemandResponse template_name = "smehub/marketplace/my_demand_responses.html" context_object_name = "responses" paginate_by = 30 def get_queryset(self): profile = _get_entrepreneur_profile(self.request.user) if profile is None: return DemandResponse.objects.none() return DemandResponse.objects.filter(entrepreneur=profile).select_related( "demand_listing__buyer", "business", ) def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) profile = _get_entrepreneur_profile(self.request.user) qs = DemandResponse.objects.filter(entrepreneur=profile) if profile else DemandResponse.objects.none() ctx["stats"] = { "total": qs.count(), "shortlisted": qs.filter(status=DemandResponse.Status.SHORTLISTED).count(), "awarded": qs.filter(status=DemandResponse.Status.AWARDED).count(), } return ctx # --------------------------------------------------------------------------- # Buyer workspace — landing dashboard, self-register, responses inbox # --------------------------------------------------------------------------- class BuyerSelfRegisterView(LoginRequiredMixin, View): """Public-facing buyer onboarding. Any logged-in user can submit; the BUYER role is assigned on success (via services.register_buyer_self) and the entry lands in PENDING for admin review.""" template_name = "smehub/marketplace/buyer_self_register.html" def get(self, request): existing = _get_buyer_entry(request.user) if existing: messages.info(request, "You already have a buyer profile.") return redirect("smehub_marketplace:buyer_dashboard") return render(request, self.template_name, {"form": BuyerRegistryForm()}) def post(self, request): existing = _get_buyer_entry(request.user) if existing: return redirect("smehub_marketplace:buyer_dashboard") form = BuyerRegistryForm(request.POST) if not form.is_valid(): return render(request, self.template_name, {"form": form}) services.register_buyer_self(user=request.user, **form.cleaned_data) messages.success( request, "Buyer profile submitted. An admin will verify it shortly — typical turnaround is 24–48 hours.", ) return redirect("smehub_marketplace:buyer_dashboard") class BuyerDashboardView(BuyerRequired, LoginRequiredMixin, TemplateView): template_name = "smehub/marketplace/buyer_dashboard.html" def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) user = self.request.user buyer = _get_buyer_entry(user) ctx["buyer"] = buyer if buyer is None: return ctx listings = MarketDemandListing.objects.filter(buyer=buyer) responses = DemandResponse.objects.filter(demand_listing__buyer=buyer) ctx["stats"] = { "open_listings": listings.filter(status=MarketDemandListing.Status.OPEN).count(), "closed_listings": listings.exclude(status=MarketDemandListing.Status.OPEN).count(), "total_responses": responses.count(), "pending_decisions": responses.filter(status=DemandResponse.Status.SUBMITTED).count(), "awarded": responses.filter(status=DemandResponse.Status.AWARDED).count(), } ctx["pending_responses"] = ( responses.filter(status=DemandResponse.Status.SUBMITTED) .select_related("entrepreneur__user", "business", "demand_listing") .order_by("-created_at")[:5] ) ctx["recent_listings"] = ( listings.order_by("-created_at")[:5] ) # Suggested products: published listings whose sector overlaps with # the buyer's sourced sectors. JSONField intersection — best-effort. sectors = buyer.sectors_sourced or [] suggested = ProductListing.objects.filter( status=ProductListing.Status.PUBLISHED, ).select_related("business", "entrepreneur__user") if sectors: suggested = suggested.filter(sector__in=sectors) ctx["suggested_products"] = suggested.order_by("-created_at")[:6] return ctx class BuyerResponsesInboxView(BuyerRequired, LoginRequiredMixin, ListView): template_name = "smehub/marketplace/buyer_responses_inbox.html" context_object_name = "responses" paginate_by = 30 def get_queryset(self): buyer = _get_buyer_entry(self.request.user) if buyer is None: return DemandResponse.objects.none() qs = ( DemandResponse.objects.filter(demand_listing__buyer=buyer) .select_related("entrepreneur__user", "business", "demand_listing") .order_by("status", "-created_at") ) status = self.request.GET.get("status") if status: qs = qs.filter(status=status) return qs def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["selected_status"] = self.request.GET.get("status", "") ctx["status_choices"] = DemandResponse.Status.choices return ctx # --------------------------------------------------------------------------- # Market validation (Phase 6.3) # --------------------------------------------------------------------------- def _user_owned_product(request, pk: int) -> ProductListing: listing = get_object_or_404( ProductListing.objects.select_related("business", "entrepreneur__user"), pk=pk, ) is_admin = _is_sme_admin(request.user) is_owner = ( request.user.is_authenticated and getattr(listing.entrepreneur, "user_id", None) == request.user.pk ) if not (is_owner or is_admin): raise PermissionDenied("Not your product.") return listing class MarketValidationListView(LoginRequiredMixin, ListView): template_name = "smehub/marketplace/market_validation_list.html" context_object_name = "validations" paginate_by = 20 def get_queryset(self): listing = _user_owned_product(self.request, self.kwargs["pk"]) self.listing = listing return listing.validations.select_related( "recorded_by", "validator_user", "validator_buyer_entry", ).order_by("-conducted_at") def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["product"] = self.listing from decimal import Decimal validations = list(self.object_list) if validations: avg = sum((v.average_score for v in validations), Decimal("0")) / len(validations) ctx["aggregate_average"] = avg.quantize(Decimal("0.01")) else: ctx["aggregate_average"] = None ctx["strong_count"] = sum(1 for v in validations if v.has_strong_signal) return ctx class MarketValidationCreateView(LoginRequiredMixin, View): template_name = "smehub/marketplace/market_validation_form.html" def get(self, request, pk): listing = _user_owned_product(request, pk) return render( request, self.template_name, {"product": listing, "form": MarketValidationForm()}, ) def post(self, request, pk): listing = _user_owned_product(request, pk) form = MarketValidationForm(request.POST, request.FILES) if not form.is_valid(): return render( request, self.template_name, {"product": listing, "form": form}, status=400, ) try: services.validate_market_fit( product_listing=listing, recorded_by=request.user, method=form.cleaned_data["method"], problem_fit_score=form.cleaned_data["problem_fit_score"], solution_fit_score=form.cleaned_data["solution_fit_score"], willingness_to_pay_score=form.cleaned_data["willingness_to_pay_score"], conducted_at=form.cleaned_data.get("conducted_at"), validator_name=form.cleaned_data.get("validator_name", ""), validator_email=form.cleaned_data.get("validator_email", ""), willingness_to_pay_amount=form.cleaned_data.get("willingness_to_pay_amount"), currency=form.cleaned_data.get("currency", ""), notes=form.cleaned_data.get("notes", ""), evidence_file=form.cleaned_data.get("evidence_file"), ) except services.MarketplaceStateError as exc: messages.error(request, str(exc)) return render( request, self.template_name, {"product": listing, "form": form}, status=400, ) messages.success(request, "Market validation recorded.") return redirect("smehub_marketplace:market_validation_list", pk=listing.pk) class MarketValidationDetailView(LoginRequiredMixin, DetailView): template_name = "smehub/marketplace/market_validation_detail.html" context_object_name = "validation" pk_url_kwarg = "vid" def get_queryset(self): listing = _user_owned_product(self.request, self.kwargs["pk"]) self.listing = listing return listing.validations.all() def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["product"] = self.listing return ctx # --------------------------------------------------------------------------- # Sales records (Phase 6.3) # --------------------------------------------------------------------------- def _sales_for_user(user): """All sales the user can see (their own as entrepreneur, or all if admin).""" if _is_sme_admin(user): return SalesRecord.objects.all() return SalesRecord.objects.filter(entrepreneur=user) class SalesRecordListView(LoginRequiredMixin, ListView): template_name = "smehub/marketplace/sales_record_list.html" context_object_name = "sales" paginate_by = 25 def get_queryset(self): qs = _sales_for_user(self.request.user).select_related( "business", "product_listing", "buyer_entry", "verified_by", ) status = self.request.GET.get("status", "").strip() if status: qs = qs.filter(status=status) channel = self.request.GET.get("channel", "").strip() if channel: qs = qs.filter(channel=channel) product_id = self.request.GET.get("product") if product_id and product_id.isdigit(): qs = qs.filter(product_listing_id=int(product_id)) business_id = self.request.GET.get("business") if business_id and business_id.isdigit(): qs = qs.filter(business_id=int(business_id)) return qs.order_by("-sale_date", "-recorded_at") def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) all_sales = list(_sales_for_user(self.request.user)) from collections import Counter from decimal import Decimal totals: dict[str, Decimal] = {} for s in all_sales: if s.status == SalesRecord.Status.CANCELLED: continue totals[s.currency] = totals.get(s.currency, Decimal("0")) + Decimal(s.gross_amount) ctx["totals_by_currency"] = totals ctx["count_total"] = len(all_sales) ctx["count_recorded"] = sum(1 for s in all_sales if s.status == SalesRecord.Status.RECORDED) ctx["count_verified"] = sum(1 for s in all_sales if s.status == SalesRecord.Status.VERIFIED) ctx["count_pending"] = sum(1 for s in all_sales if s.status == SalesRecord.Status.PENDING_DETAILS) ctx["count_disputed"] = sum(1 for s in all_sales if s.status == SalesRecord.Status.DISPUTED) ctx["channel_counts"] = Counter(s.channel for s in all_sales) ctx["status_filter"] = self.request.GET.get("status", "") ctx["channel_filter"] = self.request.GET.get("channel", "") return ctx class SalesRecordCreateView(LoginRequiredMixin, View): """Record a sale. Two paths: * **Entrepreneur path** — defaults business to the entrepreneur's most recent verified business; the entrepreneur is the recording user. * **Admin path** (PRD A5 manual-entry pattern) — admins must pick a business via ``?business=``; the entrepreneur on the resulting record is the business's owning entrepreneur user, not the admin. """ template_name = "smehub/marketplace/sales_record_form.html" def _resolve_business(self, request): if _is_sme_admin(request.user): biz_id = request.GET.get("business") or request.POST.get("business") if not biz_id: raise PermissionDenied( "Admin must specify ?business= when recording on behalf.", ) try: biz = ( Business.objects.select_related("entrepreneur__user") .get(pk=int(biz_id), verification_status=Business.Status.VERIFIED) ) except (Business.DoesNotExist, ValueError): raise PermissionDenied("Business not found or not verified.") return biz, biz.entrepreneur.user # Entrepreneur path. ent = _get_entrepreneur_profile(request.user) if ent is None: raise PermissionDenied("Entrepreneur profile required.") biz = ( Business.objects.filter( entrepreneur=ent, verification_status=Business.Status.VERIFIED, ) .order_by("-created_at") .first() ) if biz is None: raise PermissionDenied("Verified business required.") return biz, request.user def get(self, request): biz, _ = self._resolve_business(request) return render( request, self.template_name, {"form": SalesRecordForm(business=biz), "sale": None, "business": biz}, ) def post(self, request): biz, recording_user = self._resolve_business(request) form = SalesRecordForm(request.POST, request.FILES, business=biz) if not form.is_valid(): return render( request, self.template_name, {"form": form, "sale": None, "business": biz}, status=400, ) try: sale = services.record_sale( business=biz, entrepreneur=recording_user, channel=form.cleaned_data["channel"], quantity=form.cleaned_data["quantity"], gross_amount=form.cleaned_data["gross_amount"], currency=form.cleaned_data["currency"], sale_date=form.cleaned_data["sale_date"], product_listing=form.cleaned_data.get("product_listing"), buyer_entry=form.cleaned_data.get("buyer_entry"), customer_name=form.cleaned_data.get("customer_name", ""), unit=form.cleaned_data.get("unit") or "units", unit_price=form.cleaned_data.get("unit_price"), notes=form.cleaned_data.get("notes", ""), evidence_file=form.cleaned_data.get("evidence_file"), ) except services.MarketplaceStateError as exc: messages.error(request, str(exc)) return render( request, self.template_name, {"form": form, "sale": None, "business": biz}, status=400, ) messages.success(request, "Sale recorded.") return redirect("smehub_marketplace:sale_detail", pk=sale.pk) def _user_owned_sale(request, pk: int) -> SalesRecord: sale = get_object_or_404( SalesRecord.objects.select_related( "business", "product_listing", "buyer_entry", "demand_response", ), pk=pk, ) if not ( _is_sme_admin(request.user) or sale.entrepreneur_id == request.user.pk ): raise PermissionDenied("Not your sale.") return sale class SalesRecordDetailView(LoginRequiredMixin, DetailView): template_name = "smehub/marketplace/sales_record_detail.html" context_object_name = "sale" def get_object(self, queryset=None): return _user_owned_sale(self.request, self.kwargs["pk"]) def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["dispute_form"] = SalesDisputeForm() ctx["cancel_form"] = SalesCancelForm() return ctx class SalesRecordUpdateView(LoginRequiredMixin, View): template_name = "smehub/marketplace/sales_record_form.html" def get(self, request, pk): sale = _user_owned_sale(request, pk) form = SalesRecordForm(instance=sale, business=sale.business) return render( request, self.template_name, {"form": form, "sale": sale, "business": sale.business}, ) def post(self, request, pk): sale = _user_owned_sale(request, pk) form = SalesRecordForm(request.POST, request.FILES, instance=sale, business=sale.business) if not form.is_valid(): return render( request, self.template_name, {"form": form, "sale": sale, "business": sale.business}, status=400, ) try: services.update_sale_record( sale, by_user=request.user, **{k: v for k, v in form.cleaned_data.items() if v is not None}, ) except services.MarketplaceStateError as exc: messages.error(request, str(exc)) return render( request, self.template_name, {"form": form, "sale": sale, "business": sale.business}, status=400, ) messages.success(request, "Sale updated.") return redirect("smehub_marketplace:sale_detail", pk=sale.pk) @require_POST def sale_verify_view(request, pk): if not _is_sme_admin(request.user): raise PermissionDenied("Admin only.") sale = get_object_or_404(SalesRecord, pk=pk) try: services.verify_sale(sale, by_user=request.user) except services.MarketplaceStateError as exc: messages.error(request, str(exc)) else: messages.success(request, "Sale verified.") return redirect("smehub_marketplace:sale_detail", pk=pk) @require_POST def sale_dispute_view(request, pk): sale = _user_owned_sale(request, pk) if not _is_sme_admin(request.user) else get_object_or_404(SalesRecord, pk=pk) form = SalesDisputeForm(request.POST) if not form.is_valid(): messages.error(request, "Dispute reason is required.") return redirect("smehub_marketplace:sale_detail", pk=pk) try: services.dispute_sale(sale, by_user=request.user, reason=form.cleaned_data["reason"]) except services.MarketplaceStateError as exc: messages.error(request, str(exc)) else: messages.success(request, "Sale disputed.") return redirect("smehub_marketplace:sale_detail", pk=pk) @require_POST def sale_cancel_view(request, pk): sale = _user_owned_sale(request, pk) form = SalesCancelForm(request.POST) form.is_valid() services.cancel_sale( sale, by_user=request.user, reason=form.cleaned_data.get("reason", "") if form.is_valid() else "", ) messages.success(request, "Sale cancelled.") return redirect("smehub_marketplace:sale_list")