taskrun.py

#!/usr/bin/env python3
"""
taskrun -- Simple task runner with dependencies. Like just/make, zero deps.

Define tasks in a Taskfile, run them by name. Handles dependencies,
environment variables, working directories. No Makefile syntax to learn.

Taskfile format:
    # Comments start with #

    build:
      desc: Build the project
      run: python setup.py build

    test:
      desc: Run tests
      deps: build
      run: pytest tests/ -v

    deploy:
      desc: Deploy to production
      deps: test
      env: NODE_ENV=production
      run: python deploy.py --prod

    lint:
      desc: Run linters
      run: flake8 src/
      run: black --check src/

    clean:
      desc: Clean build artifacts
      dir: .
      run: rm -rf dist/ build/

Usage:
    py taskrun.py                     # List available tasks
    py taskrun.py build               # Run 'build' task
    py taskrun.py test deploy         # Run multiple tasks
    py taskrun.py -f custom.tasks build    # Custom taskfile
    py taskrun.py --dry-run deploy    # Show what would run
    py taskrun.py --list              # List tasks with descriptions
"""

import argparse
import os
import subprocess
import sys
import time
from pathlib import Path

# Colors
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
CYAN = "\033[36m"
RED = "\033[31m"

DEFAULT_FILES = ["Taskfile", "taskfile", "Taskfile.txt", "tasks.txt", "tasks"]


def color_supported() -> bool:
    if os.environ.get("NO_COLOR"):
        return False
    if sys.platform == "win32":
        return bool(os.environ.get("TERM") or os.environ.get("WT_SESSION"))
    return hasattr(sys.stdout, "isatty") and sys.stdout.isatty()


USE_COLOR = color_supported()


def c(code: str, text: str) -> str:
    return f"{code}{text}{RESET}" if USE_COLOR else text


def find_taskfile() -> str | None:
    """Find a taskfile in the current directory."""
    for name in DEFAULT_FILES:
        if Path(name).is_file():
            return name
    return None


def parse_taskfile(path: str) -> dict[str, dict]:
    """Parse a taskfile into task definitions."""
    tasks = {}
    current_task = None

    try:
        lines = Path(path).read_text(encoding="utf-8").splitlines()
    except FileNotFoundError:
        print(f"Error: taskfile not found: {path}", file=sys.stderr)
        sys.exit(1)

    for lineno, line in enumerate(lines, 1):
        stripped = line.strip()

        # Skip empty lines and comments
        if not stripped or stripped.startswith("#"):
            continue

        # New task definition: "name:" at column 0
        if not line[0].isspace() and stripped.endswith(":") and " " not in stripped.rstrip(":"):
            task_name = stripped[:-1].strip()
            current_task = task_name
            tasks[task_name] = {
                "desc": "",
                "run": [],
                "deps": [],
                "env": {},
                "dir": None,
                "silent": False,
            }
            continue

        # Task property (indented)
        if current_task and line[0].isspace() and ":" in stripped:
            key, _, value = stripped.partition(":")
            key = key.strip().lower()
            value = value.strip()

            if key == "desc":
                tasks[current_task]["desc"] = value
            elif key == "run":
                tasks[current_task]["run"].append(value)
            elif key == "deps":
                deps = [d.strip() for d in value.split(",") if d.strip()]
                tasks[current_task]["deps"].extend(deps)
            elif key == "env":
                if "=" in value:
                    env_key, _, env_val = value.partition("=")
                    tasks[current_task]["env"][env_key.strip()] = env_val.strip()
            elif key == "dir":
                tasks[current_task]["dir"] = value
            elif key == "silent":
                tasks[current_task]["silent"] = value.lower() in ("true", "yes", "1")
            else:
                print(f"Warning: unknown property '{key}' at line {lineno}", file=sys.stderr)
            continue

        # Continuation of a task (indented line without key:value)
        if current_task and line[0].isspace():
            # Treat as additional run command
            if stripped:
                tasks[current_task]["run"].append(stripped)

    return tasks


def resolve_deps(tasks: dict, target: str, resolved: list, seen: set):
    """Topological sort of task dependencies."""
    if target in seen:
        return
    if target not in tasks:
        print(f"Error: unknown task '{target}'", file=sys.stderr)
        sys.exit(1)

    seen.add(target)
    for dep in tasks[target]["deps"]:
        if dep not in tasks:
            print(f"Error: task '{target}' depends on unknown task '{dep}'", file=sys.stderr)
            sys.exit(1)
        if dep in seen and dep not in resolved:
            print(f"Error: circular dependency: {target} -> {dep}", file=sys.stderr)
            sys.exit(1)
        resolve_deps(tasks, dep, resolved, seen)

    if target not in resolved:
        resolved.append(target)


def run_task(task_name: str, task: dict, dry_run: bool = False) -> bool:
    """Run a single task. Returns True on success."""
    desc = f" -- {task['desc']}" if task["desc"] else ""
    print(c(BOLD + CYAN, f"  [{task_name}]") + c(DIM, desc))

    if not task["run"]:
        print(c(DIM, "  (no commands)"))
        return True

    # Prepare environment
    env = os.environ.copy()
    env.update(task["env"])

    # Working directory
    cwd = task["dir"] if task["dir"] else None

    for cmd in task["run"]:
        if not task["silent"]:
            print(c(DIM, f"  $ {cmd}"))

        if dry_run:
            continue

        start = time.perf_counter()
        try:
            if sys.platform == "win32":
                result = subprocess.run(cmd, shell=True, env=env, cwd=cwd)
            else:
                result = subprocess.run(cmd, shell=True, executable="/bin/sh",
                                        env=env, cwd=cwd)

            elapsed = time.perf_counter() - start

            if result.returncode != 0:
                print(c(RED, f"  FAILED (exit {result.returncode}, {elapsed:.1f}s)"))
                return False

        except Exception as e:
            print(c(RED, f"  ERROR: {e}"))
            return False

    if not dry_run:
        elapsed = time.perf_counter() - start if task["run"] else 0
        print(c(GREEN, f"  OK") + c(DIM, f" ({elapsed:.1f}s)") if task["run"] else "")

    return True


def list_tasks(tasks: dict):
    """List all available tasks."""
    if not tasks:
        print("  No tasks defined.")
        return

    print()
    print(c(BOLD, "  Available tasks:"))
    print()

    max_name = max(len(n) for n in tasks) if tasks else 0
    for name, task in tasks.items():
        desc = task["desc"] or c(DIM, "(no description)")
        deps_str = ""
        if task["deps"]:
            deps_str = c(DIM, f" [deps: {', '.join(task['deps'])}]")
        print(f"  {c(CYAN, name.ljust(max_name))}  {desc}{deps_str}")

    print()


def main():
    parser = argparse.ArgumentParser(
        description="taskrun -- simple task runner with dependencies",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("tasks", nargs="*", help="Task(s) to run")
    parser.add_argument("-f", "--file", default=None, help="Path to taskfile")
    parser.add_argument("-l", "--list", action="store_true", help="List available tasks")
    parser.add_argument("--dry-run", action="store_true", help="Show commands without running")

    args = parser.parse_args()

    # Find taskfile
    taskfile = args.file or find_taskfile()
    if not taskfile:
        print("Error: no taskfile found. Create a 'Taskfile' or use -f.", file=sys.stderr)
        print(f"Looked for: {', '.join(DEFAULT_FILES)}", file=sys.stderr)
        sys.exit(1)

    tasks = parse_taskfile(taskfile)

    # List mode
    if args.list or not args.tasks:
        list_tasks(tasks)
        if not args.tasks:
            print(c(DIM, f"  Run: py taskrun.py <task-name>"))
            print()
        return

    # Resolve dependencies and build execution order
    execution_order = []
    seen = set()
    for target in args.tasks:
        resolve_deps(tasks, target, execution_order, seen)

    # Dry-run header
    if args.dry_run:
        print()
        print(c(YELLOW, "  DRY RUN -- commands will not be executed"))

    print()
    print(c(BOLD, f"  taskrun") + c(DIM, f" ({taskfile})"))
    print(c(DIM, f"  Execution order: {' -> '.join(execution_order)}"))
    print()

    # Execute tasks
    start_total = time.perf_counter()
    failed = False

    for task_name in execution_order:
        success = run_task(task_name, tasks[task_name], dry_run=args.dry_run)
        print()
        if not success:
            failed = True
            print(c(RED, f"  Task '{task_name}' failed. Stopping."))
            break

    total_time = time.perf_counter() - start_total

    if not args.dry_run:
        if failed:
            print(c(RED, f"  FAILED") + c(DIM, f" in {total_time:.1f}s"))
        else:
            print(c(GREEN, f"  All tasks completed") + c(DIM, f" in {total_time:.1f}s"))
        print()

    if failed:
        sys.exit(1)


if __name__ == "__main__":
    main()