from django.shortcuts import render, redirect, get_object_or_404 from django.contrib.admin.views.decorators import staff_member_required from django.views.decorators.http import require_POST from django.contrib import messages from django.urls import reverse from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger import django_filters as filters from admin_frontend.templatetags.public_urls import public_route from admin_frontend.nav import _nav_items from cms.views import breadcrumbs from orgs.models import Organization, UserProfile from .models import ( MaterialCategory, Material, ProvidedService, Customer, CustomerSite, PickupOrder, WeighTicket, WeighLine, ScrapListing, ScrapBid, Invoice, Payment, Document, AuditLog, Payout, ) from .forms import ( MaterialForm, MaterialCategoryForm, CustomerForm, PickupAssignForm, PickupStatusForm, PaymentForm, DocumentForm, ProvidedServiceForm, ) from .services.billing import generate_invoice_for_pickup from django.contrib.contenttypes.models import ContentType from django.contrib.auth import get_user_model from django.http import HttpResponse import csv from django.db.models import Sum, F, ExpressionWrapper, DecimalField, Value from django.db.models.functions import Coalesce from decimal import Decimal from functools import wraps def owner_required(view_func): @wraps(view_func) def _wrapped(request, *args, **kwargs): if not request.user.is_authenticated: from django.contrib.auth.views import redirect_to_login return redirect_to_login(request.get_full_path(), login_url=reverse("admin_frontend:login")) profile = getattr(request.user, "recycle_profile", None) if not profile or profile.role != UserProfile.ROLE_OWNER: messages.error(request, "Only organization owners can manage users.") return redirect("admin_frontend:dashboard") if getattr(request, "org", None) is not None and profile.organization_id != request.org.id: messages.error(request, "You are not permitted for this organization.") return redirect("admin_frontend:dashboard") return view_func(request, *args, **kwargs) return _wrapped @staff_member_required @public_route(label="Materials", order=10, icon="beaker") @breadcrumbs(label="Materials", name="re_materials") def materials_list(request): # Create forms mat_form = MaterialForm(request.POST or None) cat_form = MaterialCategoryForm(request.POST or None) # Restrict organization choices in forms to current org if getattr(request, "org", None) is not None: try: mat_form.fields["organization"].queryset = Organization.objects.filter(pk=request.org.pk) cat_form.fields["organization"].queryset = Organization.objects.filter(pk=request.org.pk) except Exception: pass if request.method == "POST": # Distinguish which form posted via a hidden field or field presence if "default_unit" in request.POST: if mat_form.is_valid(): obj = mat_form.save(commit=False) if getattr(request, "org", None) is not None: obj.organization = request.org obj.save() messages.success(request, "Material created.") return redirect("recycle_core:materials_list") else: messages.error(request, "Please correct the errors in material form.") else: if cat_form.is_valid(): obj = cat_form.save(commit=False) if getattr(request, "org", None) is not None: obj.organization = request.org obj.save() messages.success(request, "Category created.") return redirect("recycle_core:materials_list") else: messages.error(request, "Please correct the errors in category form.") # Filters via django-filter to match list pattern class MaterialFilter(filters.FilterSet): organization = filters.ModelChoiceFilter(queryset=Organization.objects.all()) category = filters.ModelChoiceFilter(queryset=MaterialCategory.objects.all()) name = filters.CharFilter(field_name="name", lookup_expr="icontains") class Meta: model = Material fields = ["organization", "category", "name"] base_mats = Material.objects.select_related("organization", "category").order_by("organization_id", "name") mat_filter = MaterialFilter(request.GET, queryset=base_mats) mats = mat_filter.qs # Scope to current organization if present org = getattr(request, "org", None) if org is not None: mats = mats.filter(organization=org) cats = MaterialCategory.objects.select_related("organization").order_by("organization_id", "name") try: sel_org = mat_filter.form.cleaned_data.get("organization") if sel_org: cats = cats.filter(organization=sel_org) except Exception: pass if org is not None: cats = cats.filter(organization=org) context = { "nav": _nav_items(), "materials": mats, "categories": cats, "mat_form": mat_form, "cat_form": cat_form, "filter": mat_filter, } return render(request, "recycle_core/materials_list.html", context) # Organization Users (owner-only) from .forms import UserCreateForm, UserEditForm @owner_required @public_route(label="Users", order=5, icon="user-group") @breadcrumbs(label="Users", name="re_users") def org_users_list(request): User = get_user_model() org = getattr(request, "org", None) qs = User.objects.order_by("id") if org is not None: qs = qs.filter(recycle_profile__organization=org) else: qs = qs.none() paginator = Paginator(qs, 15) page = request.GET.get("page") try: page_obj = paginator.page(page) except PageNotAnInteger: page_obj = paginator.page(1) except EmptyPage: page_obj = paginator.page(paginator.num_pages) context = {"nav": _nav_items(), "users": page_obj.object_list, "page_obj": page_obj} return render(request, "recycle_core/users_list.html", context) @owner_required @breadcrumbs(label="Create User", parent="re_users") def org_user_create(request): form = UserCreateForm(request.POST or None) if request.method == "POST": if form.is_valid(): User = get_user_model() user = User.objects.create(username=form.cleaned_data["username"], email=form.cleaned_data.get("email") or "") user.set_password(form.cleaned_data["password1"]) user.save() org = getattr(request, "org", None) UserProfile.objects.create(user=user, organization=org, role=form.cleaned_data["role"]) # type: ignore[arg-type] messages.success(request, f"User '{user.username}' created.") return redirect("recycle_core:org_users_list") messages.error(request, "Please correct the errors below.") return render(request, "recycle_core/user_form.html", {"nav": _nav_items(), "form": form, "creating": True, "cancel_url": reverse("recycle_core:org_users_list")}) @owner_required @breadcrumbs(label="Edit User", parent="re_users") def org_user_edit(request, pk: int): User = get_user_model() org = getattr(request, "org", None) base_qs = User.objects.all() if org is not None: base_qs = base_qs.filter(recycle_profile__organization=org) user = get_object_or_404(base_qs, pk=pk) profile = getattr(user, "recycle_profile", None) form = UserEditForm(request.POST or None, initial={ "email": user.email, "role": getattr(profile, "role", UserProfile.ROLE_MANAGER), }) if request.method == "POST": if form.is_valid(): user.email = form.cleaned_data.get("email") or "" p1 = form.cleaned_data.get("password1") if p1: user.set_password(p1) user.save() if profile is None: profile = UserProfile(user=user, organization=org) # type: ignore[arg-type] profile.role = form.cleaned_data["role"] if org is not None: profile.organization = org profile.save() messages.success(request, f"User '{user.username}' updated.") return redirect("recycle_core:org_users_list") messages.error(request, "Please correct the errors below.") return render(request, "recycle_core/user_form.html", {"nav": _nav_items(), "form": form, "item": user, "creating": False, "cancel_url": reverse("recycle_core:org_users_list")}) @owner_required @require_POST def org_user_delete(request, pk: int): User = get_user_model() org = getattr(request, "org", None) base_qs = User.objects.all() if org is not None: base_qs = base_qs.filter(recycle_profile__organization=org) user = get_object_or_404(base_qs, pk=pk) if user.id == request.user.id: messages.error(request, "You cannot delete your own account.") return redirect("recycle_core:org_users_list") user.delete() messages.success(request, "User deleted.") return redirect("recycle_core:org_users_list") @staff_member_required @breadcrumbs(label="Edit Material", parent="re_materials") def material_edit(request, pk: int): item = get_object_or_404(Material, pk=pk) if request.method == "POST": form = MaterialForm(request.POST, instance=item) if form.is_valid(): form.save() messages.success(request, "Material updated.") return redirect("recycle_core:material_edit", pk=item.pk) else: form = MaterialForm(instance=item) return render( request, "recycle_core/material_form.html", {"nav": _nav_items(), "item": item, "form": form, "cancel_url": reverse("recycle_core:materials_list")}, ) @staff_member_required @require_POST def material_delete(request, pk: int): item = get_object_or_404(Material, pk=pk) item.delete() messages.success(request, "Material deleted.") return redirect("recycle_core:materials_list") @staff_member_required @public_route(label="Customers", order=20, icon="building-storefront") @breadcrumbs(label="Customers", name="re_customers") def customers_list(request): form = CustomerForm(request.POST or None) # Restrict organization field to current org if present if getattr(request, "org", None) is not None: try: form.fields["organization"].queryset = Organization.objects.filter(pk=request.org.pk) except Exception: pass if request.method == "POST": if form.is_valid(): obj = form.save(commit=False) if getattr(request, "org", None) is not None: obj.organization = request.org obj.save() messages.success(request, "Customer created.") return redirect("recycle_core:customers_list") messages.error(request, "Please correct the errors below.") # FilterSet similar to opportunities list class CustomerFilter(filters.FilterSet): organization = filters.ModelChoiceFilter(queryset=Organization.objects.all()) name = filters.CharFilter(field_name="name", lookup_expr="icontains") email = filters.CharFilter(field_name="email", lookup_expr="icontains") phone = filters.CharFilter(field_name="phone", lookup_expr="icontains") class Meta: model = Customer fields = ["organization", "name", "email", "phone"] base_qs = Customer.objects.select_related("organization", "price_list").order_by("organization_id", "name") cust_filter = CustomerFilter(request.GET, queryset=base_qs) qs = cust_filter.qs org = getattr(request, "org", None) if org is not None: qs = qs.filter(organization=org) paginator = Paginator(qs, 10) page = request.GET.get("page") try: page_obj = paginator.page(page) except PageNotAnInteger: page_obj = paginator.page(1) except EmptyPage: page_obj = paginator.page(paginator.num_pages) context = { "nav": _nav_items(), "customers": page_obj.object_list, "page_obj": page_obj, "form": form, "filter": cust_filter, } return render(request, "recycle_core/customers_list.html", context) @staff_member_required @breadcrumbs(label="Edit Customer", parent="re_customers") def customer_edit(request, pk: int): item = get_object_or_404(Customer, pk=pk) if request.method == "POST": form = CustomerForm(request.POST, instance=item) if form.is_valid(): form.save() messages.success(request, "Customer updated.") return redirect("recycle_core:customer_edit", pk=item.pk) else: form = CustomerForm(instance=item) return render( request, "recycle_core/customer_form.html", {"nav": _nav_items(), "item": item, "form": form, "cancel_url": reverse("recycle_core:customers_list")}, ) @staff_member_required @require_POST def customer_delete(request, pk: int): item = get_object_or_404(Customer, pk=pk) item.delete() messages.success(request, "Customer deleted.") return redirect("recycle_core:customers_list") # Pickups UI ----------------------------------------------------------------- @staff_member_required @public_route(label="Pickups", order=30, icon="truck") @breadcrumbs(label="Pickups", name="re_pickups") def pickups_list(request): # FilterSet for pickups class PickupFilter(filters.FilterSet): organization = filters.ModelChoiceFilter(queryset=Organization.objects.all()) customer = filters.ModelChoiceFilter(queryset=Customer.objects.all()) site = filters.ModelChoiceFilter(queryset=CustomerSite.objects.all()) assigned_driver = filters.CharFilter(field_name="assigned_driver__username", lookup_expr="icontains", label="Driver") status = filters.ChoiceFilter(field_name="status", choices=PickupOrder.STATUS_CHOICES) scheduled_at = filters.DateFromToRangeFilter(field_name="scheduled_at", label="Scheduled between") class Meta: model = PickupOrder fields = ["organization", "customer", "site", "assigned_driver", "status", "scheduled_at"] base_qs = ( PickupOrder.objects.select_related("organization", "customer", "site", "assigned_driver") .order_by("-created_at") ) p_filter = PickupFilter(request.GET, queryset=base_qs) qs = p_filter.qs org = getattr(request, "org", None) if org is not None: qs = qs.filter(organization=org) paginator = Paginator(qs, 10) page = request.GET.get("page") try: page_obj = paginator.page(page) except PageNotAnInteger: page_obj = paginator.page(1) except EmptyPage: page_obj = paginator.page(paginator.num_pages) # empty forms used in row actions assign_form = PickupAssignForm() status_form = PickupStatusForm() context = { "nav": _nav_items(), "pickups": page_obj.object_list, "page_obj": page_obj, "filter": p_filter, "assign_form": assign_form, "status_form": status_form, } return render(request, "recycle_core/pickups_list.html", context) @staff_member_required @require_POST def pickup_assign(request, pk: int): pickup = get_object_or_404(PickupOrder, pk=pk) form = PickupAssignForm(request.POST) if form.is_valid(): pickup.assigned_driver = form.cleaned_data["driver"] pickup.status = PickupOrder.STATUS_SCHEDULED pickup.save(update_fields=["assigned_driver", "status"]) messages.success(request, f"Assigned driver to pickup #{pickup.id}.") else: messages.error(request, "Invalid driver selection.") return redirect("recycle_core:pickups_list") @staff_member_required @require_POST def pickup_set_status(request, pk: int): pickup = get_object_or_404(PickupOrder, pk=pk) form = PickupStatusForm(request.POST) if form.is_valid(): pickup.status = form.cleaned_data["status"] pickup.save(update_fields=["status"]) messages.success(request, f"Updated status for pickup #{pickup.id}.") else: messages.error(request, "Invalid status selection.") return redirect("recycle_core:pickups_list") @staff_member_required @require_POST def pickup_generate_invoice(request, pk: int): pickup = get_object_or_404(PickupOrder, pk=pk) try: if not hasattr(pickup, "weigh_ticket"): messages.error(request, "Pickup has no weigh ticket.") return redirect("recycle_core:pickups_list") invoice = generate_invoice_for_pickup(pickup) messages.success(request, f"Invoice #{invoice.id} generated.") except Exception as e: messages.error(request, f"Failed to generate invoice: {e}") return redirect("recycle_core:pickups_list") # Scrap Listings UI ---------------------------------------------------------- @staff_member_required @public_route(label="Scrap Listings", order=40, icon="banknotes") @breadcrumbs(label="Scrap Listings", name="re_listings") def scrap_listings_list(request): class ListingFilter(filters.FilterSet): organization = filters.ModelChoiceFilter(queryset=Organization.objects.all()) customer = filters.ModelChoiceFilter(queryset=Customer.objects.all()) status = filters.ChoiceFilter(field_name="status", choices=ScrapListing.STATUS_CHOICES) is_public = filters.BooleanFilter(field_name="is_public") starts_at = filters.DateFromToRangeFilter(field_name="starts_at", label="Starts between") ends_at = filters.DateFromToRangeFilter(field_name="ends_at", label="Ends between") class Meta: model = ScrapListing fields = ["organization", "customer", "status", "is_public", "starts_at", "ends_at"] base_qs = ( ScrapListing.objects.select_related("organization", "customer", "site", "created_by") .order_by("-created_at") ) l_filter = ListingFilter(request.GET, queryset=base_qs) qs = l_filter.qs org = getattr(request, "org", None) if org is not None: qs = qs.filter(organization=org) paginator = Paginator(qs, 10) page = request.GET.get("page") try: page_obj = paginator.page(page) except PageNotAnInteger: page_obj = paginator.page(1) except EmptyPage: page_obj = paginator.page(paginator.num_pages) context = { "nav": _nav_items(), "listings": page_obj.object_list, "page_obj": page_obj, "filter": l_filter, } return render(request, "recycle_core/scrap_listings_list.html", context) @staff_member_required @require_POST def scrap_listing_open(request, pk: int): listing = get_object_or_404(ScrapListing, pk=pk) listing.status = ScrapListing.STATUS_OPEN if not listing.starts_at: from django.utils import timezone listing.starts_at = timezone.now() listing.save(update_fields=["status", "starts_at"]) messages.success(request, f"Listing #{listing.id} opened.") return redirect("recycle_core:scrap_listings_list") @staff_member_required @require_POST def scrap_listing_close(request, pk: int): listing = get_object_or_404(ScrapListing, pk=pk) listing.status = ScrapListing.STATUS_CLOSED if not listing.ends_at: from django.utils import timezone listing.ends_at = timezone.now() listing.save(update_fields=["status", "ends_at"]) messages.success(request, f"Listing #{listing.id} closed.") return redirect("recycle_core:scrap_listings_list") @staff_member_required @require_POST def scrap_listing_award(request, pk: int): listing = get_object_or_404(ScrapListing, pk=pk) top_bid = ( ScrapBid.objects.filter(listing=listing, status=ScrapBid.STATUS_ACTIVE) .order_by("-price_total") .first() ) if not top_bid: messages.error(request, "No bids to award.") return redirect("recycle_core:scrap_listings_list") if listing.reserve_price and top_bid.price_total < listing.reserve_price: messages.error(request, "Top bid below reserve price.") return redirect("recycle_core:scrap_listings_list") # Update bid statuses and listing ScrapBid.objects.filter(listing=listing, status=ScrapBid.STATUS_ACTIVE).exclude(id=top_bid.id).update( status=ScrapBid.STATUS_REJECTED ) top_bid.status = ScrapBid.STATUS_ACCEPTED top_bid.save(update_fields=["status"]) listing.status = ScrapListing.STATUS_AWARDED listing.save(update_fields=["status"]) messages.success(request, f"Listing #{listing.id} awarded to bid #{top_bid.id}.") return redirect("recycle_core:scrap_listings_list") # Weigh Tickets UI ----------------------------------------------------------- @staff_member_required @public_route(label="Weigh Tickets", order=45, icon="scale") @breadcrumbs(label="Weigh Tickets", name="re_tickets") def weigh_tickets_list(request): class TicketFilter(filters.FilterSet): organization = filters.ModelChoiceFilter(queryset=Organization.objects.all(), field_name="pickup__organization") customer = filters.ModelChoiceFilter(queryset=Customer.objects.all(), field_name="pickup__customer") recorded_at = filters.DateFromToRangeFilter(field_name="recorded_at", label="Recorded between") ticket_number = filters.CharFilter(field_name="ticket_number", lookup_expr="icontains") class Meta: model = WeighTicket fields = ["organization", "customer", "recorded_at", "ticket_number"] base_qs = WeighTicket.objects.select_related("pickup", "pickup__customer", "pickup__organization").order_by("-recorded_at") t_filter = TicketFilter(request.GET, queryset=base_qs) qs = t_filter.qs org = getattr(request, "org", None) if org is not None: qs = qs.filter(pickup__organization=org) paginator = Paginator(qs, 10) page = request.GET.get("page") try: page_obj = paginator.page(page) except PageNotAnInteger: page_obj = paginator.page(1) except EmptyPage: page_obj = paginator.page(paginator.num_pages) context = { "nav": _nav_items(), "tickets": page_obj.object_list, "page_obj": page_obj, "filter": t_filter, } return render(request, "recycle_core/weigh_tickets_list.html", context) @staff_member_required @breadcrumbs(label="Ticket Detail", parent="re_tickets") def weigh_ticket_detail(request, pk: int): ticket = get_object_or_404(WeighTicket.objects.select_related("pickup", "pickup__customer"), pk=pk) lines = WeighLine.objects.select_related("material").filter(ticket=ticket) ct = ContentType.objects.get_for_model(WeighTicket) docs = Document.objects.filter(content_type=ct, object_id=ticket.id) context = { "nav": _nav_items(), "ticket": ticket, "lines": lines, "generate_url": reverse("recycle_core:weigh_ticket_generate_invoice", args=[ticket.id]), "documents": docs, "ct_id": ct.id, } return render(request, "recycle_core/weigh_ticket_detail.html", context) @staff_member_required @require_POST def weigh_ticket_generate_invoice(request, pk: int): ticket = get_object_or_404(WeighTicket, pk=pk) try: invoice = generate_invoice_for_pickup(ticket.pickup) messages.success(request, f"Invoice #{invoice.id} generated from ticket {ticket.ticket_number or ticket.id}.") except Exception as e: messages.error(request, f"Failed to generate invoice: {e}") return redirect("recycle_core:weigh_tickets_list") # Invoices UI --------------------------------------------------------------- @staff_member_required @public_route(label="Invoices", order=50, icon="receipt-percent") @breadcrumbs(label="Invoices", name="re_invoices") def invoices_list(request): class InvoiceFilter(filters.FilterSet): organization = filters.ModelChoiceFilter(queryset=Organization.objects.all()) customer = filters.ModelChoiceFilter(queryset=Customer.objects.all()) status = filters.ChoiceFilter(field_name="status", choices=Invoice.STATUS_CHOICES) issued_at = filters.DateFromToRangeFilter(field_name="issued_at", label="Issued between") class Meta: model = Invoice fields = ["organization", "customer", "status", "issued_at"] base_qs = ( Invoice.objects.select_related("organization", "customer") .annotate( payments_total=Coalesce( Sum("payments__amount"), Value(Decimal("0.00")), output_field=DecimalField(max_digits=14, decimal_places=2), ), ) .annotate( balance=ExpressionWrapper( F("total_amount") - F("payments_total"), output_field=DecimalField(max_digits=14, decimal_places=2), ) ) .order_by("-issued_at", "-id") ) i_filter = InvoiceFilter(request.GET, queryset=base_qs) qs = i_filter.qs org = getattr(request, "org", None) if org is not None: qs = qs.filter(organization=org) paginator = Paginator(qs, 10) page = request.GET.get("page") try: page_obj = paginator.page(page) except PageNotAnInteger: page_obj = paginator.page(1) except EmptyPage: page_obj = paginator.page(paginator.num_pages) pay_form = PaymentForm() context = { "nav": _nav_items(), "invoices": page_obj.object_list, "page_obj": page_obj, "filter": i_filter, "pay_form": pay_form, } return render(request, "recycle_core/invoices_list.html", context) @staff_member_required @require_POST def invoice_add_payment(request, pk: int): invoice = get_object_or_404(Invoice, pk=pk) form = PaymentForm(request.POST) if form.is_valid(): Payment.objects.create( invoice=invoice, amount=form.cleaned_data["amount"], currency_code=invoice.currency_code, received_at=form.cleaned_data.get("received_at") or None, reference=form.cleaned_data.get("reference", ""), ) messages.success(request, f"Payment recorded for invoice #{invoice.id}.") else: messages.error(request, "Invalid payment data.") return redirect("recycle_core:invoices_list") # Payouts UI + CSV ---------------------------------------------------------- @staff_member_required @public_route(label="Payouts", order=52, icon="banknotes") @breadcrumbs(label="Payouts", name="re_payouts") def payouts_list(request): class PayoutFilter(filters.FilterSet): organization = filters.ModelChoiceFilter(queryset=Organization.objects.all()) customer = filters.ModelChoiceFilter(queryset=Customer.objects.all()) paid_at = filters.DateFromToRangeFilter(field_name="paid_at", label="Paid between") class Meta: model = Payout fields = ["organization", "customer", "paid_at"] base_qs = Payout.objects.select_related("organization", "customer").order_by("-paid_at", "-id") p_filter = PayoutFilter(request.GET, queryset=base_qs) qs = p_filter.qs org = getattr(request, "org", None) if org is not None: qs = qs.filter(organization=org) paginator = Paginator(qs, 10) page = request.GET.get("page") try: page_obj = paginator.page(page) except PageNotAnInteger: page_obj = paginator.page(1) except EmptyPage: page_obj = paginator.page(paginator.num_pages) context = { "nav": _nav_items(), "payouts": page_obj.object_list, "page_obj": page_obj, "filter": p_filter, } return render(request, "recycle_core/payouts_list.html", context) @staff_member_required def payouts_export_csv(request): qs = Payout.objects.select_related("organization", "customer").all() org_obj = getattr(request, "org", None) if org_obj is not None: qs = qs.filter(organization=org_obj) org = request.GET.get("organization") cust = request.GET.get("customer") start = request.GET.get("paid_at_after") or request.GET.get("paid_at_min") end = request.GET.get("paid_at_before") or request.GET.get("paid_at_max") from django.utils.dateparse import parse_datetime if org: qs = qs.filter(organization_id=org) if cust: qs = qs.filter(customer_id=cust) if start: dt = parse_datetime(start) if dt: qs = qs.filter(paid_at__gte=dt) if end: dt = parse_datetime(end) if dt: qs = qs.filter(paid_at__lte=dt) response = HttpResponse(content_type="text/csv") response["Content-Disposition"] = "attachment; filename=payouts.csv" writer = csv.writer(response) writer.writerow(["organization", "customer", "amount", "currency", "paid_at", "reference", "pickup_id"]) for p in qs.iterator(): writer.writerow([ p.organization.code, p.customer.name, p.amount, p.currency_code, p.paid_at.isoformat(), p.reference, p.pickup_id or "", ]) return response # Documents UI -------------------------------------------------------------- @staff_member_required @public_route(label="Documents", order=55, icon="paper-clip") @breadcrumbs(label="Documents", name="re_documents") def documents_list(request): class DocumentFilter(filters.FilterSet): organization = filters.ModelChoiceFilter(queryset=Organization.objects.all()) kind = filters.CharFilter(field_name="kind", lookup_expr="icontains") content_type = filters.ModelChoiceFilter(queryset=ContentType.objects.all()) object_id = filters.NumberFilter(field_name="object_id") uploaded_by = filters.ModelChoiceFilter(queryset=get_user_model().objects.all()) created_at = filters.DateFromToRangeFilter(field_name="created_at", label="Uploaded between") class Meta: model = Document fields = ["organization", "kind", "content_type", "object_id", "uploaded_by", "created_at"] form = DocumentForm(request.POST or None, request.FILES or None) if getattr(request, "org", None) is not None: try: form.fields["organization"].queryset = Organization.objects.filter(pk=request.org.pk) except Exception: pass if request.method == "POST": if form.is_valid(): doc = Document( organization=(getattr(request, "org", None) or form.cleaned_data["organization"]), file=form.cleaned_data["file"], kind=form.cleaned_data.get("kind") or "", content_type=form.cleaned_data["content_type"], object_id=form.cleaned_data["object_id"], uploaded_by=request.user, ) doc.save() messages.success(request, "Document uploaded.") return redirect("recycle_core:documents_list") else: messages.error(request, "Please correct the document form errors.") base_qs = Document.objects.select_related("organization", "uploaded_by", "content_type").order_by("-created_at") d_filter = DocumentFilter(request.GET, queryset=base_qs) qs = d_filter.qs org = getattr(request, "org", None) if org is not None: qs = qs.filter(organization=org) paginator = Paginator(qs, 10) page = request.GET.get("page") try: page_obj = paginator.page(page) except PageNotAnInteger: page_obj = paginator.page(1) except EmptyPage: page_obj = paginator.page(paginator.num_pages) context = { "nav": _nav_items(), "documents": page_obj.object_list, "page_obj": page_obj, "filter": d_filter, "form": form, } return render(request, "recycle_core/documents_list.html", context) # Audit Logs UI ------------------------------------------------------------- @staff_member_required @public_route(label="Audit Logs", order=60, icon="clock") @breadcrumbs(label="Audit Logs", name="re_audit_logs") def audit_logs_list(request): class AuditFilter(filters.FilterSet): organization = filters.ModelChoiceFilter(queryset=Organization.objects.all()) user = filters.ModelChoiceFilter(queryset=get_user_model().objects.all()) action = filters.CharFilter(field_name="action", lookup_expr="icontains") content_type = filters.ModelChoiceFilter(queryset=ContentType.objects.all()) object_id = filters.NumberFilter(field_name="object_id") created_at = filters.DateFromToRangeFilter(field_name="created_at", label="Between") class Meta: model = AuditLog fields = ["organization", "user", "action", "content_type", "object_id", "created_at"] base_qs = AuditLog.objects.select_related("organization", "user", "content_type").order_by("-created_at") a_filter = AuditFilter(request.GET, queryset=base_qs) qs = a_filter.qs org = getattr(request, "org", None) if org is not None: qs = qs.filter(organization=org) paginator = Paginator(qs, 10) page = request.GET.get("page") try: page_obj = paginator.page(page) except PageNotAnInteger: page_obj = paginator.page(1) except EmptyPage: page_obj = paginator.page(paginator.num_pages) context = { "nav": _nav_items(), "logs": page_obj.object_list, "page_obj": page_obj, "filter": a_filter, } return render(request, "recycle_core/audit_logs_list.html", context) @staff_member_required def audit_logs_export_csv(request): qs = AuditLog.objects.select_related("organization", "user", "content_type").all() org_obj = getattr(request, "org", None) if org_obj is not None: qs = qs.filter(organization=org_obj) org = request.GET.get("organization") user = request.GET.get("user") action = request.GET.get("action") ctype = request.GET.get("content_type") obj_id = request.GET.get("object_id") start = request.GET.get("created_at_after") or request.GET.get("created_at_min") end = request.GET.get("created_at_before") or request.GET.get("created_at_max") from django.utils.dateparse import parse_datetime if org: qs = qs.filter(organization_id=org) if user: qs = qs.filter(user_id=user) if action: qs = qs.filter(action__icontains=action) if ctype: qs = qs.filter(content_type_id=ctype) if obj_id: qs = qs.filter(object_id=obj_id) if start: dt = parse_datetime(start) if dt: qs = qs.filter(created_at__gte=dt) if end: dt = parse_datetime(end) if dt: qs = qs.filter(created_at__lte=dt) response = HttpResponse(content_type="text/csv") response["Content-Disposition"] = "attachment; filename=audit_logs.csv" writer = csv.writer(response) writer.writerow(["created_at", "organization", "user", "action", "content_type", "object_id", "metadata"]) for log in qs.iterator(): writer.writerow([ log.created_at.isoformat(), log.organization.code, getattr(log.user, "username", ""), log.action, log.content_type.model, log.object_id, log.metadata, ]) return response @staff_member_required @public_route(label="Provided Services", order=15, icon="sparkles") @breadcrumbs(label="Provided Services", name="rc_services") def services_list(request): if not request.user.is_authenticated or not getattr(request.user, "is_staff", False): from django.contrib.auth.views import redirect_to_login return redirect_to_login(request.get_full_path(), login_url=reverse("admin_frontend:login")) org = getattr(request, "org", None) services = ProvidedService.objects.filter(organization=org).order_by("display_order", "id") return render(request, "recycle_core/services_list.html", {"nav": _nav_items(), "services": services}) @breadcrumbs(label="New Service", parent="rc_services") @owner_required def service_create(request): if request.method == "POST": form = ProvidedServiceForm(request.POST) if form.is_valid(): obj = form.save(commit=False) obj.organization = getattr(request, "org", None) obj.save() messages.success(request, "Service created.") return redirect("recycle_core:services_list") messages.error(request, "Please correct the errors below.") else: form = ProvidedServiceForm() return render(request, "recycle_core/service_form.html", {"nav": _nav_items(), "form": form, "cancel_url": reverse("recycle_core:services_list")}) @breadcrumbs(label="Edit Service", parent="rc_services") @owner_required def service_edit(request, pk: int): org = getattr(request, "org", None) item = ProvidedService.objects.filter(organization=org).filter(pk=pk).first() if not item: messages.error(request, "Service not found.") return redirect("recycle_core:services_list") if request.method == "POST": form = ProvidedServiceForm(request.POST, instance=item) if form.is_valid(): obj = form.save(commit=False) obj.organization = org obj.save() messages.success(request, "Service updated.") return redirect("recycle_core:service_edit", pk=item.pk) messages.error(request, "Please correct the errors below.") else: form = ProvidedServiceForm(instance=item) return render(request, "recycle_core/service_form.html", {"nav": _nav_items(), "item": item, "form": form, "cancel_url": reverse("recycle_core:services_list")}) @require_POST @owner_required def service_delete(request, pk: int): org = getattr(request, "org", None) item = ProvidedService.objects.filter(organization=org).filter(pk=pk).first() if not item: messages.error(request, "Service not found.") else: item.delete() messages.success(request, "Service deleted.") return redirect("recycle_core:services_list") @require_POST @owner_required def service_toggle_enabled(request, pk: int): org = getattr(request, "org", None) item = ProvidedService.objects.filter(organization=org).filter(pk=pk).first() if not item: messages.error(request, "Service not found.") else: item.is_enabled = not item.is_enabled item.save(update_fields=["is_enabled"]) messages.success(request, f"Service '{item.title}' {'enabled' if item.is_enabled else 'disabled'}.") return redirect("recycle_core:services_list")