108 lines
3.0 KiB
Python
108 lines
3.0 KiB
Python
"""Safe, non-executing task discovery."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
|
|
def display_name(value: str) -> str:
|
|
return value.replace("_", " ")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Task:
|
|
id: str
|
|
group_id: str
|
|
filename: str
|
|
path: Path
|
|
display_name: str
|
|
description: str | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Group:
|
|
id: str
|
|
path: Path
|
|
display_name: str
|
|
tasks: tuple[Task, ...]
|
|
|
|
|
|
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
|
|
tasks = tuple(self._tasks_in_group(group_path))
|
|
groups.append(
|
|
Group(
|
|
id=group_path.name,
|
|
path=group_path,
|
|
display_name=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 _tasks_in_group(self, group_path: Path) -> list[Task]:
|
|
tasks = []
|
|
for path in sorted(group_path.iterdir(), key=lambda candidate: candidate.name):
|
|
if (
|
|
path.name.startswith((".", "_"))
|
|
or path.is_symlink()
|
|
or not path.is_file()
|
|
or path.suffix != ".py"
|
|
):
|
|
continue
|
|
tasks.append(
|
|
Task(
|
|
id=f"{group_path.name}/{path.name}",
|
|
group_id=group_path.name,
|
|
filename=path.name,
|
|
path=path,
|
|
display_name=display_name(path.stem),
|
|
description=read_description(path),
|
|
)
|
|
)
|
|
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,
|
|
)
|