Initial commit

.trellis/scripts/common/tasks.py

@@ -0,0 +1,109 @@
+"""
+Task data access layer.
+
+Single source of truth for loading and iterating task directories.
+Replaces scattered task.json parsing across 9+ files.
+
+Provides:
+    load_task          — Load a single task by directory path
+    iter_active_tasks  — Iterate all non-archived tasks (sorted)
+    get_all_statuses   — Get {dir_name: status} map for children progress
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from pathlib import Path
+
+from .io import read_json
+from .paths import FILE_TASK_JSON
+from .types import TaskInfo
+
+
+def load_task(task_dir: Path) -> TaskInfo | None:
+    """Load task from a directory containing task.json.
+
+    Args:
+        task_dir: Absolute path to the task directory.
+
+    Returns:
+        TaskInfo if task.json exists and is valid, None otherwise.
+    """
+    task_json = task_dir / FILE_TASK_JSON
+    if not task_json.is_file():
+        return None
+
+    data = read_json(task_json)
+    if not data:
+        return None
+
+    return TaskInfo(
+        dir_name=task_dir.name,
+        directory=task_dir,
+        title=data.get("title") or data.get("name") or "unknown",
+        status=data.get("status", "unknown"),
+        assignee=data.get("assignee", ""),
+        priority=data.get("priority", "P2"),
+        children=tuple(data.get("children", [])),
+        parent=data.get("parent"),
+        package=data.get("package"),
+        raw=data,
+    )
+
+
+def iter_active_tasks(tasks_dir: Path) -> Iterator[TaskInfo]:
+    """Iterate all active (non-archived) tasks, sorted by directory name.
+
+    Skips the "archive" directory and directories without valid task.json.
+
+    Args:
+        tasks_dir: Path to the tasks directory.
+
+    Yields:
+        TaskInfo for each valid task.
+    """
+    if not tasks_dir.is_dir():
+        return
+
+    for d in sorted(tasks_dir.iterdir()):
+        if not d.is_dir() or d.name == "archive":
+            continue
+        info = load_task(d)
+        if info is not None:
+            yield info
+
+
+def get_all_statuses(tasks_dir: Path) -> dict[str, str]:
+    """Get a {dir_name: status} mapping for all active tasks.
+
+    Useful for computing children progress without loading full TaskInfo.
+
+    Args:
+        tasks_dir: Path to the tasks directory.
+
+    Returns:
+        Dict mapping directory names to status strings.
+    """
+    return {t.dir_name: t.status for t in iter_active_tasks(tasks_dir)}
+
+
+def children_progress(
+    children: tuple[str, ...] | list[str],
+    all_statuses: dict[str, str],
+) -> str:
+    """Format children progress string like " [2/3 done]".
+
+    Args:
+        children: List of child directory names.
+        all_statuses: Status map from get_all_statuses().
+
+    Returns:
+        Formatted string, or "" if no children.
+    """
+    if not children:
+        return ""
+    done = sum(
+        1 for c in children
+        if all_statuses.get(c) in ("completed", "done")
+    )
+    return f" [{done}/{len(children)} done]"