from __future__ import annotations from django.conf import settings from django.contrib import messages from django.contrib.auth.decorators import login_required from django.contrib.auth.mixins import LoginRequiredMixin from django.http import HttpRequest from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse from django.views.decorators.cache import never_cache from django.views.decorators.http import require_POST from django.views.generic import ListView from apps.core.rate_limit import clear_rate_limit, is_rate_limited, register_rate_limit_failure from .forms import ManualImportForm from .models import Source, SourcePolicyReview from .services.manual_import import ManualImportError, ManualImportSummary, import_manual_source from .services.policy import create_policy_review from .tasks import fetch_source def _source_list_queryset(): return Source.objects.prefetch_related("policy_reviews").all().order_by("name") def _manual_import_context(request: HttpRequest, *, form: ManualImportForm, manual_result=None): base_queryset = _source_list_queryset() return { "sources": base_queryset, "active_source_count": Source.objects.filter(status=Source.Status.ACTIVE).count(), "source_total_count": base_queryset.count(), "review_pending_count": Source.objects.filter(policy=Source.Policy.REVIEW).count(), "manual_import_form": form, "manual_import_bookmarklet_endpoint": request.build_absolute_uri( reverse("sources:manual_import") ), "manual_import_paste_max_bytes": settings.MANUAL_IMPORT_PASTE_MAX_BYTES, "manual_import_result": manual_result, } class SourceListView(LoginRequiredMixin, ListView): model = Source template_name = "sources/list.html" context_object_name = "sources" def get_queryset(self): return _source_list_queryset() def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) context.update( _manual_import_context( self.request, form=ManualImportForm( initial={"source_url": self.request.GET.get("source_url", "")} ), ) ) return context def _bulk_summary(action: str) -> str: mapping = { "allow": "toestaan", "approve_to_trial": "proefrun", "pause": "pauzeren", "dismiss": "afwijzen", } return mapping.get(action, "onbekend") def _action_to_review( action: str, ) -> tuple[SourcePolicyReview.Decision, str, str, str]: if action == "allow": return ( SourcePolicyReview.Decision.ALLOW, Source.Status.ACTIVE, Source.Policy.ALLOW, "Bron actief gemaakt door expliciete goedkeuring", ) if action == "approve_to_trial": return ( SourcePolicyReview.Decision.TRIAL, Source.Status.TRIAL, Source.Policy.REVIEW, "Bron naar proefrun voor technische controle", ) if action == "pause": return ( SourcePolicyReview.Decision.PAUSE, Source.Status.PAUSED, Source.Policy.REVIEW, "Bron tijdelijk gepauzeerd", ) return ( SourcePolicyReview.Decision.DENY, Source.Status.DISABLED, Source.Policy.DENY, "Bron afgewezen", ) @login_required @require_POST def bulk_candidate_action(request): action = request.POST.get("action") source_ids = [int(raw_id) for raw_id in request.POST.getlist("source_ids") if raw_id.isdigit()] sources = Source.objects.filter(id__in=source_ids).exclude(policy=Source.Policy.DENY) if not sources.exists(): messages.info(request, "Geen bruikbare bronregels geselecteerd.") return redirect("sources:list") if action not in {"allow", "approve_to_trial", "pause", "dismiss"}: messages.error(request, "Onbekende actie.") return redirect("sources:list") decision, status, policy, reason = _action_to_review(action) for source in sources: create_policy_review( source, actor=request.user, decision=decision, scope=SourcePolicyReview.Scope.SOURCE, reason=reason, notes="Bulkactie", ) source.status = status source.policy = policy source.policy_reason = reason source.save(update_fields=["status", "policy", "policy_reason", "updated_at"]) messages.success(request, f"{sources.count()} bron(nen) naar {_bulk_summary(action)}.") return redirect("sources:list") @login_required def retry_source(request, pk: int): if request.method == "POST": source = get_object_or_404(Source, pk=pk) if source.policy == Source.Policy.DENY: messages.error(request, "Deze bron is door het bronbeleid geblokkeerd.") else: fetch_source.delay(source.pk, force=True) messages.success(request, "Broncontrole ingepland.") return redirect("sources:list") @login_required @never_cache def manual_import_view(request): rate_limit = is_rate_limited( request, namespace="manual_import", max_attempts=settings.MANUAL_IMPORT_RATE_LIMIT_MAX_ATTEMPTS, window_seconds=settings.MANUAL_IMPORT_RATE_LIMIT_WINDOW_SECONDS, block_seconds=settings.MANUAL_IMPORT_RATE_LIMIT_BLOCK_SECONDS, ) if rate_limit.is_blocked: messages.error( request, "Te veel handmatige importverzoeken. Wacht 5 minuten en probeer daarna opnieuw.", ) return render( request, "sources/list.html", _manual_import_context(request, form=ManualImportForm(), manual_result=None), ) if request.method == "POST": form = ManualImportForm(request.POST) result_payload = None if not form.is_valid(): messages.error(request, "Controleer de handmatige importvelden.") return render( request, "sources/list.html", _manual_import_context(request, form=form, manual_result=result_payload), ) cleaned = form.cleaned_data try: summary: ManualImportSummary = import_manual_source( actor=request.user, source_url=cleaned.get("source_url"), pasted_text=cleaned.get("pasted_text"), ) result_payload = summary.to_session_payload() messages.success( request, "Handmatige import uitgevoerd. Bekijk het resultaat hieronder.", ) clear_rate_limit(request, namespace="manual_import") form = ManualImportForm() except ManualImportError as exc: messages.error(request, str(exc)) register_rate_limit_failure( request, namespace="manual_import", max_attempts=settings.MANUAL_IMPORT_RATE_LIMIT_MAX_ATTEMPTS, window_seconds=settings.MANUAL_IMPORT_RATE_LIMIT_WINDOW_SECONDS, block_seconds=settings.MANUAL_IMPORT_RATE_LIMIT_BLOCK_SECONDS, ) return render( request, "sources/list.html", _manual_import_context(request, form=form, manual_result=result_payload), ) return render( request, "sources/list.html", _manual_import_context( request, form=ManualImportForm(initial={"source_url": request.GET.get("source_url", "")}), ), )