docker-images/python-tools/app/discovery.py

"""Safe, non-executing task discovery."""

from __future__ import annotations

import ast
import json
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any


FIELD_NAME = re.compile(r"[a-z][a-z0-9_]*\Z")
FIELD_TYPES = frozenset({"text", "integer", "date", "datetime", "choice", "multi_choice", "file"})
DISPLAY_ACRONYMS = {
    "api": "API", "csv": "CSV", "iata": "IATA", "icao": "ICAO", "id": "ID",
    "ids": "IDs", "url": "URL", "urls": "URLs", "uuid": "UUID",
}


def display_name(value: str) -> str:
    titled = value.replace("_", " ").title()
    return re.sub(
        r"\b[A-Za-z]+\b",
        lambda match: DISPLAY_ACRONYMS.get(match.group().casefold(), match.group()),
        titled,
    )


@dataclass(frozen=True)
class Task:
    id: str
    group_id: str
    filename: str
    path: Path
    display_name: str
    description: str | None
    inputs: tuple["InputField", ...]
    wait_for_result: bool = False
    download_artifacts: bool = False
    execution: "ExecutionMode | None" = None


@dataclass(frozen=True)
class BackgroundTask:
    """A declared internal task that is never exposed through the web UI."""

    id: str
    group_id: str
    path: Path
    interval_seconds: int


@dataclass(frozen=True)
class InputOption:
    value: str
    label: str


@dataclass(frozen=True)
class InputField:
    name: str
    label: str
    type: str
    required: bool
    default: str | int | tuple[str, ...] | None
    minimum: int | None
    maximum: int | None
    step: int | None
    pattern: str | None
    options: tuple[InputOption, ...]
    accept: tuple[str, ...]
    maximum_bytes: int | None
    sensitive: bool = False


@dataclass(frozen=True)
class ExecutionMode:
    """A task-declared report/apply mode exposed as a generated choice field."""

    field: InputField
    dry_run_value: str
    execute_value: str


@dataclass(frozen=True)
class Group:
    id: str
    path: Path
    display_name: str
    tasks: tuple[Task, ...]

    @property
    def can_run_as_group(self) -> bool:
        return all(not task.inputs for task in self.tasks)


class TaskCatalog:
    """Discovers only direct Python children of non-hidden group directories."""

    def __init__(self, task_root: Path) -> None:
        self.task_root = task_root

    def groups(self) -> tuple[Group, ...]:
        if not self.task_root.is_dir():
            return ()

        groups = []
        for group_path in sorted(self.task_root.iterdir(), key=lambda path: path.name):
            if (
                group_path.name.startswith((".", "_"))
                or group_path.is_symlink()
                or not group_path.is_dir()
            ):
                continue
            manifest = load_input_manifest(
                group_path,
                {
                    path.name for path in group_path.iterdir()
                    if (
                        path.is_file()
                        and not path.is_symlink()
                        and path.suffix == ".py"
                        and not path.name.startswith((".", "_"))
                    )
                },
            )
            tasks = tuple(self._tasks_in_group(group_path, manifest.declarations))
            groups.append(
                Group(
                    id=group_path.name,
                    path=group_path,
                    display_name=manifest.title or display_name(group_path.name),
                    tasks=tasks,
                )
            )
        return tuple(groups)

    def group(self, group_id: str) -> Group | None:
        return next((group for group in self.groups() if group.id == group_id), None)

    def task(self, task_id: str) -> Task | None:
        for group in self.groups():
            for task in group.tasks:
                if task.id == task_id:
                    return task
        return None

    def background_tasks(self) -> tuple[BackgroundTask, ...]:
        if not self.task_root.is_dir():
            return ()

        tasks = []
        for group_path in sorted(self.task_root.iterdir(), key=lambda path: path.name):
            if (
                group_path.name.startswith((".", "_"))
                or group_path.is_symlink()
                or not group_path.is_dir()
            ):
                continue
            tasks.extend(load_background_tasks(group_path))
        return tuple(tasks)

    def _tasks_in_group(
        self,
        group_path: Path,
        declarations_by_filename: dict[str, "TaskDeclaration"] | None = None,
    ) -> list[Task]:
        task_paths = [
            path
            for path in sorted(group_path.iterdir(), key=lambda candidate: candidate.name)
            if not (
                path.name.startswith((".", "_"))
                or path.is_symlink()
                or not path.is_file()
                or path.suffix != ".py"
            )
        ]
        if declarations_by_filename is None:
            declarations_by_filename = load_input_manifest(
                group_path,
                {path.name for path in task_paths},
            ).declarations
        paths_by_filename = {path.name: path for path in task_paths}
        ordered_filenames = [
            *declarations_by_filename,
            *(path.name for path in task_paths if path.name not in declarations_by_filename),
        ]
        tasks = []
        for filename in ordered_filenames:
            path = paths_by_filename[filename]
            declaration = declarations_by_filename.get(path.name)
            tasks.append(
                Task(
                    id=f"{group_path.name}/{path.name}",
                    group_id=group_path.name,
                    filename=path.name,
                    path=path,
                    display_name=declaration.title if declaration and declaration.title else display_name(path.stem),
                    description=(
                        declaration.description
                        if declaration and declaration.description is not None
                        else read_description(path)
                    ),
                    inputs=(
                        (*declaration.inputs, declaration.execution.field)
                        if declaration and declaration.execution
                        else declaration.inputs if declaration else ()
                    ),
                    wait_for_result=declaration.wait_for_result if declaration else False,
                    download_artifacts=declaration.download_artifacts if declaration else False,
                    execution=declaration.execution if declaration else None,
                )
            )
        return tasks


def load_background_tasks(group_path: Path) -> list[BackgroundTask]:
    """Read optional internal recurring-task declarations for one group."""

    manifest_path = group_path / "background-tasks.json"
    if not manifest_path.exists():
        return []
    if manifest_path.is_symlink():
        raise ValueError(f"Background task manifest cannot be a symlink: {manifest_path}")
    try:
        data = json.loads(manifest_path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error:
        raise ValueError(f"Could not read background task manifest {manifest_path}: {error}") from error
    if not isinstance(data, dict) or set(data) != {"version", "tasks"}:
        raise ValueError(
            f"Background task manifest {manifest_path} must contain only version and tasks."
        )
    if data["version"] != 1 or not isinstance(data["tasks"], list):
        raise ValueError(f"Background task manifest {manifest_path} has an unsupported structure.")

    tasks = []
    seen_ids = set()
    for declaration in data["tasks"]:
        if not isinstance(declaration, dict) or set(declaration) != {"id", "script", "interval_seconds"}:
            raise ValueError(f"Background task manifest {manifest_path} has an invalid task declaration.")
        task_id = declaration["id"]
        script = declaration["script"]
        interval_seconds = declaration["interval_seconds"]
        if not isinstance(task_id, str) or not FIELD_NAME.fullmatch(task_id) or task_id in seen_ids:
            raise ValueError(f"Background task manifest {manifest_path} has an invalid or duplicate task id.")
        if (
            not isinstance(script, str)
            or not script.startswith("internal/")
            or Path(script).suffix != ".py"
            or Path(script).is_absolute()
            or ".." in Path(script).parts
        ):
            raise ValueError(f"Background task manifest {manifest_path} task {task_id} has an invalid script.")
        if not isinstance(interval_seconds, int) or isinstance(interval_seconds, bool) or interval_seconds < 1:
            raise ValueError(f"Background task manifest {manifest_path} task {task_id} has an invalid interval.")
        path = group_path / script
        try:
            path.relative_to(group_path)
        except ValueError as error:
            raise ValueError(f"Background task manifest {manifest_path} task {task_id} escapes its group.") from error
        if path.is_symlink() or not path.is_file():
            raise ValueError(f"Background task manifest {manifest_path} task {task_id} script is unavailable.")
        parent = path.parent
        while parent != group_path:
            if parent.is_symlink():
                raise ValueError(f"Background task manifest {manifest_path} task {task_id} script uses a symlink.")
            parent = parent.parent
        tasks.append(
            BackgroundTask(
                id=f"{group_path.name}/{task_id}",
                group_id=group_path.name,
                path=path,
                interval_seconds=interval_seconds,
            )
        )
        seen_ids.add(task_id)
    return tasks


def read_description(path: Path) -> str | None:
    """Return the first docstring line without importing or executing a task."""

    try:
        module = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
    except (OSError, SyntaxError, UnicodeDecodeError):
        return None
    docstring = ast.get_docstring(module, clean=True)
    if not docstring:
        return None
    return next(
        (line.strip() for line in docstring.splitlines() if line.strip()),
        None,
    )


@dataclass(frozen=True)
class GroupManifest:
    title: str | None
    declarations: dict[str, "TaskDeclaration"]


def load_input_manifest(
    group_path: Path,
    task_filenames: set[str],
) -> GroupManifest:
    """Read the optional data-only input declaration for one task group."""

    manifest_path = group_path / "task-inputs.json"
    if not manifest_path.exists():
        return GroupManifest(title=None, declarations={})
    if manifest_path.is_symlink():
        raise ValueError(f"Input manifest cannot be a symlink: {manifest_path}")
    try:
        data = json.loads(manifest_path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error:
        raise ValueError(f"Could not read input manifest {manifest_path}: {error}") from error
    if not isinstance(data, dict) or set(data) - {"version", "title", "tasks"}:
        raise ValueError(
            f"Input manifest {manifest_path} has an invalid structure."
        )
    if data.get("version") != 1 or not isinstance(data.get("tasks"), dict):
        raise ValueError(f"Input manifest {manifest_path} has an unsupported structure.")
    title = data.get("title")
    if title is not None and (not isinstance(title, str) or not title.strip()):
        raise ValueError(f"Input manifest {manifest_path} title must be non-empty text.")

    unknown_tasks = set(data["tasks"]) - task_filenames
    if unknown_tasks:
        names = ", ".join(sorted(unknown_tasks))
        raise ValueError(f"Input manifest {manifest_path} names unknown tasks: {names}.")

    return GroupManifest(
        title=title,
        declarations={
            filename: parse_task_declaration(manifest_path, filename, declaration)
            for filename, declaration in data["tasks"].items()
        },
    )


@dataclass(frozen=True)
class TaskDeclaration:
    title: str | None
    description: str | None
    inputs: tuple[InputField, ...]
    wait_for_result: bool
    download_artifacts: bool
    execution: ExecutionMode | None


def parse_task_declaration(
    manifest_path: Path,
    filename: str,
    declaration: Any,
) -> TaskDeclaration:
    if (
        not isinstance(declaration, dict)
        or set(declaration) - {"title", "description", "inputs", "wait_for_result", "download_artifacts", "execution"}
    ):
        raise ValueError(f"{manifest_path} task {filename} has an invalid declaration.")
    raw_inputs = declaration.get("inputs", [])
    if not isinstance(raw_inputs, list):
        raise ValueError(f"{manifest_path} task {filename} inputs must be a list.")
    fields = tuple(parse_input_field(manifest_path, filename, raw) for raw in raw_inputs)
    title = declaration.get("title")
    if title is not None and (not isinstance(title, str) or not title.strip()):
        raise ValueError(f"{manifest_path} task {filename} title must be non-empty text.")
    description = declaration.get("description")
    if description is not None and (not isinstance(description, str) or not description.strip()):
        raise ValueError(f"{manifest_path} task {filename} description must be non-empty text.")
    names = [field.name for field in fields]
    if len(names) != len(set(names)):
        raise ValueError(f"{manifest_path} task {filename} has duplicate input names.")
    execution = parse_execution_mode(manifest_path, filename, declaration.get("execution"), set(names))
    wait_for_result = declaration.get("wait_for_result", False)
    if not isinstance(wait_for_result, bool):
        raise ValueError(f"{manifest_path} task {filename} wait_for_result must be true or false.")
    download_artifacts = declaration.get("download_artifacts", False)
    if not isinstance(download_artifacts, bool):
        raise ValueError(f"{manifest_path} task {filename} download_artifacts must be true or false.")
    if download_artifacts and not wait_for_result:
        raise ValueError(f"{manifest_path} task {filename} download_artifacts requires wait_for_result.")
    return TaskDeclaration(
        title=title,
        description=description,
        inputs=fields,
        wait_for_result=wait_for_result,
        download_artifacts=download_artifacts,
        execution=execution,
    )


def parse_execution_mode(
    manifest_path: Path,
    filename: str,
    raw: Any,
    input_names: set[str],
) -> ExecutionMode | None:
    if raw is None:
        return None
    required_keys = {
        "field", "dry_run_value", "dry_run_label", "execute_value", "execute_label",
    }
    if not isinstance(raw, dict) or set(raw) != required_keys:
        raise ValueError(f"{manifest_path} task {filename} has an invalid execution declaration.")
    values = {key: raw[key] for key in required_keys}
    if not all(isinstance(value, str) and value.strip() for value in values.values()):
        raise ValueError(f"{manifest_path} task {filename} execution values must be non-empty text.")
    field_name = values["field"]
    if not FIELD_NAME.fullmatch(field_name) or field_name in input_names:
        raise ValueError(f"{manifest_path} task {filename} has an invalid execution field.")
    if values["dry_run_value"] == values["execute_value"]:
        raise ValueError(f"{manifest_path} task {filename} execution values must differ.")
    field = InputField(
        name=field_name,
        label="Run mode",
        type="choice",
        required=True,
        default=values["dry_run_value"],
        minimum=None,
        maximum=None,
        step=None,
        pattern=None,
        options=(
            InputOption(values["dry_run_value"], values["dry_run_label"]),
            InputOption(values["execute_value"], values["execute_label"]),
        ),
        accept=(),
        maximum_bytes=None,
    )
    return ExecutionMode(
        field=field,
        dry_run_value=values["dry_run_value"],
        execute_value=values["execute_value"],
    )


def parse_input_field(manifest_path: Path, filename: str, raw: Any) -> InputField:
    if not isinstance(raw, dict):
        raise ValueError(f"{manifest_path} task {filename} has a non-object input field.")
    allowed = {
        "name", "label", "type", "required", "default", "minimum", "maximum",
        "step", "pattern", "options", "accept", "maximum_bytes", "sensitive",
    }
    unknown = set(raw) - allowed
    if unknown:
        raise ValueError(
            f"{manifest_path} task {filename} input has unknown keys: {', '.join(sorted(unknown))}."
        )
    name = raw.get("name")
    label = raw.get("label")
    field_type = raw.get("type")
    if not isinstance(name, str) or not FIELD_NAME.fullmatch(name):
        raise ValueError(f"{manifest_path} task {filename} has an invalid input name.")
    if not isinstance(label, str) or not label.strip():
        raise ValueError(f"{manifest_path} task {filename} input {name} needs a label.")
    if field_type not in FIELD_TYPES:
        raise ValueError(f"{manifest_path} task {filename} input {name} has an invalid type.")
    required = raw.get("required", False)
    if not isinstance(required, bool):
        raise ValueError(f"{manifest_path} task {filename} input {name} required must be true or false.")
    sensitive = raw.get("sensitive", False)
    if not isinstance(sensitive, bool):
        raise ValueError(f"{manifest_path} task {filename} input {name} sensitive must be true or false.")

    numeric_keys = ("minimum", "maximum", "step")
    numeric_values = {key: raw.get(key) for key in numeric_keys}
    if field_type == "integer":
        for key, value in numeric_values.items():
            if value is not None and (not isinstance(value, int) or isinstance(value, bool)):
                raise ValueError(f"{manifest_path} task {filename} input {name} {key} must be an integer.")
        if numeric_values["minimum"] is not None and numeric_values["maximum"] is not None and numeric_values["minimum"] > numeric_values["maximum"]:
            raise ValueError(f"{manifest_path} task {filename} input {name} minimum exceeds maximum.")
        if numeric_values["step"] is not None and numeric_values["step"] <= 0:
            raise ValueError(f"{manifest_path} task {filename} input {name} step must be positive.")
    elif any(value is not None for value in numeric_values.values()):
        raise ValueError(f"{manifest_path} task {filename} input {name} only integer fields accept numeric limits.")

    pattern = raw.get("pattern")
    if pattern is not None:
        if field_type != "text" or not isinstance(pattern, str):
            raise ValueError(f"{manifest_path} task {filename} input {name} has an invalid pattern.")
        try:
            re.compile(pattern)
        except re.error as error:
            raise ValueError(f"{manifest_path} task {filename} input {name} has an invalid pattern: {error}.") from error

    accept = parse_file_accept(manifest_path, filename, name, field_type, raw.get("accept"))
    maximum_bytes = raw.get("maximum_bytes")
    if field_type == "file":
        if pattern is not None or raw.get("default") is not None:
            raise ValueError(f"{manifest_path} task {filename} file input {name} cannot define a pattern or default.")
        if maximum_bytes is not None and (
            not isinstance(maximum_bytes, int)
            or isinstance(maximum_bytes, bool)
            or maximum_bytes < 1
        ):
            raise ValueError(f"{manifest_path} task {filename} file input {name} maximum_bytes must be positive.")
    elif maximum_bytes is not None:
        raise ValueError(f"{manifest_path} task {filename} input {name} only file fields accept maximum_bytes.")

    options = parse_options(manifest_path, filename, name, field_type, raw.get("options"))
    default = parse_default(manifest_path, filename, name, field_type, options, raw.get("default"))
    validate_default(
        manifest_path,
        filename,
        name,
        field_type,
        default,
        numeric_values["minimum"],
        numeric_values["maximum"],
        numeric_values["step"],
        pattern,
    )
    return InputField(
        name=name,
        label=label,
        type=field_type,
        required=required,
        default=default,
        minimum=numeric_values["minimum"],
        maximum=numeric_values["maximum"],
        step=numeric_values["step"],
        pattern=pattern,
        options=options,
        accept=accept,
        maximum_bytes=maximum_bytes,
        sensitive=sensitive,
    )


def parse_file_accept(
    manifest_path: Path,
    filename: str,
    name: str,
    field_type: str,
    raw_accept: Any,
) -> tuple[str, ...]:
    if field_type != "file":
        if raw_accept is not None:
            raise ValueError(f"{manifest_path} task {filename} input {name} only file fields accept accept.")
        return ()
    if raw_accept is None:
        return ()
    if (
        not isinstance(raw_accept, list)
        or not raw_accept
        or not all(isinstance(value, str) and re.fullmatch(r"\.[a-z0-9]{1,10}", value) for value in raw_accept)
        or len(set(raw_accept)) != len(raw_accept)
    ):
        raise ValueError(f"{manifest_path} task {filename} file input {name} has invalid accept values.")
    return tuple(raw_accept)


def parse_options(
    manifest_path: Path,
    filename: str,
    name: str,
    field_type: str,
    raw_options: Any,
) -> tuple[InputOption, ...]:
    if field_type not in {"choice", "multi_choice"}:
        if raw_options is not None:
            raise ValueError(f"{manifest_path} task {filename} input {name} only choice fields accept options.")
        return ()
    if not isinstance(raw_options, list) or not raw_options:
        raise ValueError(f"{manifest_path} task {filename} input {name} needs non-empty options.")
    options = []
    for raw_option in raw_options:
        if isinstance(raw_option, str) and raw_option:
            options.append(InputOption(value=raw_option, label=raw_option))
        elif (
            isinstance(raw_option, dict)
            and set(raw_option) == {"value", "label"}
            and isinstance(raw_option["value"], str)
            and raw_option["value"]
            and isinstance(raw_option["label"], str)
            and raw_option["label"].strip()
        ):
            options.append(InputOption(value=raw_option["value"], label=raw_option["label"]))
        else:
            raise ValueError(f"{manifest_path} task {filename} input {name} has an invalid option.")
    if len({option.value for option in options}) != len(options):
        raise ValueError(f"{manifest_path} task {filename} input {name} has duplicate option values.")
    return tuple(options)


def parse_default(
    manifest_path: Path,
    filename: str,
    name: str,
    field_type: str,
    options: tuple[InputOption, ...],
    default: Any,
) -> str | int | tuple[str, ...] | None:
    if default is None:
        return None
    if field_type == "integer":
        if not isinstance(default, int) or isinstance(default, bool):
            raise ValueError(f"{manifest_path} task {filename} input {name} default must be an integer.")
        return default
    if field_type == "multi_choice":
        if not isinstance(default, list) or not all(isinstance(value, str) for value in default):
            raise ValueError(f"{manifest_path} task {filename} input {name} default must be a list of options.")
        values = tuple(default)
        if len(values) != len(set(values)) or not set(values) <= {option.value for option in options}:
            raise ValueError(f"{manifest_path} task {filename} input {name} default contains an unknown option.")
        return values
    if not isinstance(default, str):
        raise ValueError(f"{manifest_path} task {filename} input {name} default must be text.")
    if field_type == "choice" and default not in {option.value for option in options}:
        raise ValueError(f"{manifest_path} task {filename} input {name} default is not an option.")
    return default


def validate_default(
    manifest_path: Path,
    filename: str,
    name: str,
    field_type: str,
    default: str | int | tuple[str, ...] | None,
    minimum: int | None,
    maximum: int | None,
    step: int | None,
    pattern: str | None,
) -> None:
    if default is None or field_type not in {"integer", "text"}:
        return
    if field_type == "text":
        if pattern and not re.fullmatch(pattern, default):
            raise ValueError(f"{manifest_path} task {filename} input {name} default does not match its pattern.")
        return
    assert isinstance(default, int)
    if minimum is not None and default < minimum:
        raise ValueError(f"{manifest_path} task {filename} input {name} default is below minimum.")
    if maximum is not None and default > maximum:
        raise ValueError(f"{manifest_path} task {filename} input {name} default is above maximum.")
    if step is not None and (default - (minimum or 0)) % step:
        raise ValueError(f"{manifest_path} task {filename} input {name} default does not match step.")