from urllib.parse import urlencode from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin from django.db.models import Count from django.shortcuts import get_object_or_404, redirect from django.urls import reverse from django.views import View from django.views.generic import TemplateView, ListView, DetailView, FormView from apps.batches.models import ProcessingBatch from apps.shipments.models import Shipment from apps.documents.downloads import file_response from apps.documents.models import CertificateOfAnalysis, Document from apps.esg.models import ESGFieldReport from apps.subscriptions.models import Subscription from apps.guides.views import GuideDetailView as BaseGuideDetailView from apps.guides.views import GuideIndexView as BaseGuideIndexView from apps.users.keycloak import ( DuplicateEmailError, KeycloakError, create_keycloak_user, ) from apps.users.models import User from .forms import DUPLICATE_EMAIL_MESSAGE, BuyerSignupForm class BuyerRequiredMixin(UserPassesTestMixin): def test_func(self): user = self.request.user return user.is_authenticated and (user.role in ("buyer", "admin") or user.is_superuser) class SignupView(FormView): """Public buyer self-registration. Creates the Keycloak user first — the Django row stores its subject id — then the matching ``users`` row as a buyer, and sends the new buyer through the normal sign-in flow. There is one sign-in mechanism platform-wide; the sign-up screen only creates the account. A duplicate is refused with the same sentence whether Django or Keycloak spots it, and an unreachable Keycloak is reported, never a 500. """ template_name = "portal/signup.html" form_class = BuyerSignupForm def dispatch(self, request, *args, **kwargs): if request.user.is_authenticated: return redirect("home") return super().dispatch(request, *args, **kwargs) def form_valid(self, form): data = form.cleaned_data try: keycloak_id = create_keycloak_user( email=data["email"], full_name=data["full_name"], password=data["password"], ) except DuplicateEmailError: form.add_error("email", DUPLICATE_EMAIL_MESSAGE) return self.form_invalid(form) except KeycloakError: form.add_error( None, "We could not create your account just now. Please try again shortly.", ) return self.form_invalid(form) User.objects.create_user( email=data["email"], full_name=data["full_name"], company_name=data["company_name"], role=User.Roles.BUYER, is_active=True, is_staff=False, keycloak_id=keycloak_id, ) sign_in = reverse("oidc_authentication_init") return redirect(f"{sign_in}?{urlencode({'next': '/portal/?welcome=1'})}") class DashboardView(BuyerRequiredMixin, TemplateView): template_name = "portal/dashboard.html" def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["welcome"] = self.request.GET.get("welcome") == "1" ctx["total_batches"] = ProcessingBatch.objects.count() ctx["recent_batches"] = ProcessingBatch.objects.order_by("-created_at")[:5] ctx["documents"] = Document.objects.filter(visibility="buyer").order_by("-uploaded_at")[:5] ctx["active_shipments"] = Shipment.objects.exclude(status="delivered").count() ctx["subscription"] = Subscription.objects.filter(user=self.request.user).first() return ctx class BatchListView(BuyerRequiredMixin, ListView): model = ProcessingBatch template_name = "portal/batch_list.html" context_object_name = "batches" paginate_by = 25 def get_queryset(self): qs = ProcessingBatch.objects.all() status = self.request.GET.get("status") q = self.request.GET.get("q") if status: qs = qs.filter(status=status) if q: qs = qs.filter(batch_reference__icontains=q) return qs def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["status_choices"] = ProcessingBatch.Status.choices ctx["current_status"] = self.request.GET.get("status", "") ctx["current_q"] = self.request.GET.get("q", "") return ctx class BatchDetailView(BuyerRequiredMixin, DetailView): model = ProcessingBatch template_name = "portal/batch_detail.html" context_object_name = "batch" def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) b = self.object ctx["shipments"] = Shipment.objects.filter(batch=b) ctx["documents"] = Document.objects.filter(batch=b, visibility="buyer") ctx["esg_reports"] = ESGFieldReport.objects.filter(batch=b) return ctx class DocumentListView(BuyerRequiredMixin, ListView): model = Document template_name = "portal/document_list.html" context_object_name = "documents" paginate_by = 25 def get_queryset(self): qs = Document.objects.filter(visibility="buyer").select_related("batch", "uploaded_by") doc_type = self.request.GET.get("doc_type") if doc_type: qs = qs.filter(doc_type=doc_type) return qs def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["doc_type_choices"] = Document.DocType.choices ctx["current_doc_type"] = self.request.GET.get("doc_type", "") return ctx class DocumentDownloadView(BuyerRequiredMixin, View): """Serve the stored file behind a buyer-visible document. A buyer may only fetch files from documents marked buyer-visible, so a guessed document id belonging to an operations-only row is a 404 rather than a leak. """ def get(self, request, pk): doc = get_object_or_404(Document, pk=pk, visibility=Document.Visibility.BUYER) return file_response(doc.file_url, doc.file_name) class CoADownloadView(BuyerRequiredMixin, View): """Serve the summary PDF of a published certificate of analysis. Only a published certificate is downloadable, matching the compliance page that links to it: a certificate still awaiting release stays internal. """ def get(self, request, pk): coa = get_object_or_404( CertificateOfAnalysis, pk=pk, status=CertificateOfAnalysis.Status.PUBLISHED, ) return file_response(coa.file_url, coa.file_name) class SubscriptionView(BuyerRequiredMixin, DetailView): model = Subscription template_name = "portal/subscription.html" context_object_name = "sub" def get_object(self): return Subscription.objects.filter(user=self.request.user).first() class ChatView(BuyerRequiredMixin, TemplateView): def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) from apps.ai.models import AgentConfig user = self.request.user if user.role in ("admin", "ops") or user.is_superuser: ctx["agents"] = AgentConfig.objects.filter(is_active=True) else: ctx["agents"] = AgentConfig.objects.filter(is_active=True, user__isnull=True) return ctx template_name = "portal/chat.html" class ChatEmbedView(BuyerRequiredMixin, TemplateView): template_name = "portal/chat_embed.html" def dispatch(self, request, *args, **kwargs): response = super().dispatch(request, *args, **kwargs) response["X-Frame-Options"] = "ALLOWALL" return response class ComplianceView(BuyerRequiredMixin, TemplateView): template_name = "portal/compliance.html" def get_context_data(self, **kwargs): from apps.compliance.models import DueDiligenceCase from apps.documents.models import CertificateOfAnalysis ctx = super().get_context_data(**kwargs) ctx["approved_coas"] = CertificateOfAnalysis.objects.filter( status=CertificateOfAnalysis.Status.PUBLISHED ).select_related("batch", "quality_approved_by").order_by("-published_at") ctx["eudr_cases"] = DueDiligenceCase.objects.select_related("batch", "approved_by").order_by("-updated_at") return ctx class GuideIndexView(BuyerRequiredMixin, BaseGuideIndexView): template_name = "guides/index_portal.html" class GuideDetailView(BuyerRequiredMixin, BaseGuideDetailView): template_name = "guides/detail_portal.html"