# (c) cavaliba.com - data - views_task.py import json from app_data.models import DataTask from app_data.task_manager import abort_task from app_home.log import DEBUG, log from app_user.aaa import start_view from django.http import HttpResponse from django.shortcuts import redirect, render from django.utils import timezone def _json_items(data): """Return list of (key, value_str) for a JSON dict, or None if not a dict.""" if not isinstance(data, dict): return None items = [] for k, v in data.items(): if k == "aaa" and isinstance(v, dict): v = {ik: iv for ik, iv in v.items() if ik != "perms"} if isinstance(v, (dict, list)): items.append((k, json.dumps(v, indent=2, ensure_ascii=False))) else: items.append((k, str(v) if v is not None else "-")) return items NAVBAR_ACTIVE_STATES = ["QUEUED", "RUNNING"] NAVBAR_TERMINAL_STATES = ["DONE", "FAILED", "ABORTED"] NAVBAR_RECENT_HOURS = 24 NAVBAR_RECENT_MAX = 5 # ------------------------------------------------------------------ # HTMX partial — navbar dropdown fragment # GET /data/private/tasks/navbar/ # ------------------------------------------------------------------ def task_navbar_partial(request): context = start_view(request, app="data", view="task_navbar", noauth="app_sirene:index", perm="p_task_view", noauthz="app_home:private") if context["redirect"]: return HttpResponse("") aaa = context["aaa"] username = aaa.get("username", "") if 'p_task_all' in aaa['perms']: active_tasks = list( DataTask.objects.filter(state__in=NAVBAR_ACTIVE_STATES) .order_by("-created_at") ) else: active_tasks = list( DataTask.objects.filter(owner_id=username, state__in=NAVBAR_ACTIVE_STATES) .order_by("-created_at") ) cutoff = timezone.now() - timezone.timedelta(hours=NAVBAR_RECENT_HOURS) if 'p_task_all' in aaa['perms']: recent_tasks = list( DataTask.objects.filter( state__in=NAVBAR_TERMINAL_STATES, finished_at__gte=cutoff, ) .order_by("-finished_at")[:NAVBAR_RECENT_MAX] ) else: recent_tasks = list( DataTask.objects.filter( owner_id=username, state__in=NAVBAR_TERMINAL_STATES, finished_at__gte=cutoff, ) .order_by("-finished_at")[:NAVBAR_RECENT_MAX] ) # always display for p_task_all users if 'p_task_view' in aaa['perms']: context['ui_perm_task'] = True context['active_tasks'] = active_tasks context['recent_tasks'] = recent_tasks return render(request, "app_data/tasks/navbar_partial.html", context) # ------------------------------------------------------------------ # list task # perm p_task_view : needed # perm p_task_all : show all tasks # GET /data/private/tasks/ # ------------------------------------------------------------------ def task_list(request): context = start_view(request, app="data", view="task_list", noauth="app_sirene:index", perm="p_task_view", noauthz="app_home:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] username = aaa.get("username", "") state_filter = request.GET.get("state", "").strip().upper() search = request.GET.get("search", "").strip() if 'p_task_all' in aaa['perms']: qs = DataTask.objects.all().order_by("-created_at") else: qs = DataTask.objects.filter(owner_id=username).order_by("-created_at") if state_filter: qs = qs.filter(state=state_filter) if search: from django.db.models import Q qs = qs.filter( Q(name__icontains=search) | Q(owner_id__icontains=search) | Q(owner_type__icontains=search) ) tasks = qs[:200] log(DEBUG, aaa=aaa, app="data", view="task_list", action="GET", status="OK", data=f"state={state_filter} search={search}") context['tasks'] = tasks context['state_filter'] = state_filter context['search'] = search context['states'] = ["", "QUEUED", "RUNNING", "DONE", "FAILED", "ABORTED"] return render(request, "app_data/tasks/list.html", context) # ------------------------------------------------------------------ # Stop task # POST /data/private/tasks//stop/ (stop button) # ------------------------------------------------------------------ def task_stop(request, handle): context = start_view(request, app="data", view="task_stop", noauth="app_sirene:index", perm="p_task_manage", noauthz="app_home:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] if request.method == "POST": abort_task(handle, aaa=aaa) log(DEBUG, aaa=aaa, app="data", view="task_stop", action="POST", status="OK", data=f"handle={handle}") next_url = request.POST.get("next") or request.GET.get("next") or "" if next_url: return redirect(next_url) return redirect("app_data:task_list") # ------------------------------------------------------------------ # Task detail # GET /data/private/tasks// # perm p_task_view : required # perm p_task_all : required to view another user's task # ------------------------------------------------------------------ def task_detail(request, handle): context = start_view(request, app="data", view="task_detail", noauth="app_sirene:index", perm="p_task_view", noauthz="app_home:private") if context["redirect"]: return redirect(context["redirect"]) aaa = context["aaa"] username = aaa.get("username", "") try: task = DataTask.objects.get(handle=handle) except DataTask.DoesNotExist: from django.http import Http404 raise Http404 if task.owner_id != username and "p_task_all" not in aaa.get("perms", []): return redirect("app_home:private") log(DEBUG, aaa=aaa, app="data", view="task_detail", action="GET", status="OK", data=f"handle={handle}") estimated = None if task.state == "RUNNING" and task.duration: progress = task.progress or {} percent = progress.get("percent", 0) if isinstance(progress, dict) else 0 if percent and percent > 0: estimated = round(task.duration * (100 - percent) / percent, 1) context["task"] = task context["estimated"] = estimated context["output_items"] = _json_items(task.output) context["params_items"] = _json_items(task.params) return render(request, "app_data/tasks/detail.html", context)