"""Application-level use cases for the task automation API. The automation engine owns rule evaluation and mutations. This module owns the orchestration required by HTTP and other adapters: access-scoped task lookup, eligible button selection and interpretation of manual run outcomes. """ from __future__ import annotations from dataclasses import dataclass from django.db.models import Q, QuerySet from tasks.access import task_board_access_q from tasks.automation.engine import conditions_match, execute_automation from tasks.models import ( Task, TaskAutomation, TaskAutomationKind, TaskAutomationRun, TaskAutomationRunStatus, TaskBoard, ) class TaskAutomationApplicationError(Exception): """Base error raised by task automation use cases.""" class AccessibleTaskNotFound(TaskAutomationApplicationError): """The task does not exist or is outside the actor's accessible boards.""" class ManualAutomationUnavailable(TaskAutomationApplicationError): """The selected rule cannot be launched as a manual button.""" @dataclass(frozen=True) class ManualAutomationOutcome: run: TaskAutomationRun | None conflict_message: str | None = None def get_accessible_automation_task(*, actor, task_id: int) -> Task: task = ( Task.objects.select_related("board", "column", "created_by", "assignee") .filter( board__in=TaskBoard.objects.filter(task_board_access_q(actor)), id=task_id, ) .first() ) if task is None: raise AccessibleTaskNotFound return task def eligible_button_automations( queryset: QuerySet[TaskAutomation], *, task: Task, ) -> list[TaskAutomation]: candidates = ( queryset.filter(kind=TaskAutomationKind.BUTTON, is_active=True) .filter(Q(applies_to_all_boards=True) | Q(boards=task.board_id)) .distinct() ) return [ automation for automation in candidates if conditions_match(task, automation.conditions) ] def run_manual_automation( automation: TaskAutomation, *, task: Task, actor, ) -> ManualAutomationOutcome: if automation.kind != TaskAutomationKind.BUTTON or not automation.is_active: raise ManualAutomationUnavailable run = execute_automation( automation, task, source="manual", actor=actor, ) if run is None: return ManualAutomationOutcome( run=None, conflict_message="Автоматизация не была запущена.", ) if run.status == TaskAutomationRunStatus.SKIPPED: return ManualAutomationOutcome( run=run, conflict_message=(run.error or "Условия автоматизации не выполнены."), ) return ManualAutomationOutcome(run=run)