claude-code/automation/workflow-info.py

#!/usr/bin/env python3
"""
List and describe available workflows.
Usage: python3 workflow-info.py [--category CAT] [name]
"""

import argparse
import json
import sys
from pathlib import Path
from typing import Dict, List, Optional
import yaml

CLAUDE_DIR = Path.home() / ".claude"
WORKFLOWS_DIR = CLAUDE_DIR / "workflows"
REGISTRY_PATH = CLAUDE_DIR / "state" / "component-registry.json"


def load_registry() -> Dict:
    """Load component registry."""
    try:
        with open(REGISTRY_PATH) as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return {}


def find_workflow_files() -> List[Path]:
    """Find all workflow YAML files."""
    if not WORKFLOWS_DIR.exists():
        return []

    files = []
    for pattern in ["*.yaml", "*.yml", "**/*.yaml", "**/*.yml"]:
        files.extend(WORKFLOWS_DIR.glob(pattern))

    # Filter out README and other non-workflow files
    return [f for f in files if f.name not in ["README.md"]]


def parse_workflow(path: Path) -> Optional[Dict]:
    """Parse a workflow YAML file."""
    try:
        with open(path) as f:
            content = f.read()
            # Handle YAML front matter or full YAML
            if content.startswith("---"):
                parts = content.split("---", 2)
                if len(parts) >= 2:
                    return yaml.safe_load(parts[1])
            return yaml.safe_load(content)
    except Exception:
        return None


def get_workflow_category(path: Path) -> str:
    """Get workflow category from path."""
    rel_path = path.relative_to(WORKFLOWS_DIR)
    if len(rel_path.parts) > 1:
        return rel_path.parts[0]
    return "general"


def list_workflows(category: Optional[str] = None):
    """List all available workflows."""
    registry = load_registry()
    workflows = registry.get("workflows", {})

    # Group by category
    categories: Dict[str, List] = {}

    for name, info in workflows.items():
        cat = name.split("/")[0] if "/" in name else "general"
        if category and cat != category:
            continue

        if cat not in categories:
            categories[cat] = []

        categories[cat].append({
            "name": name,
            "description": info.get("description", "No description"),
            "triggers": info.get("triggers", [])
        })

    if not categories:
        print("No workflows found.")
        return

    print(f"\n📋 Available Workflows\n")

    for cat in sorted(categories.keys()):
        print(f"=== {cat.title()} ===")
        for wf in categories[cat]:
            print(f"  {wf['name']}")
            print(f"    {wf['description']}")
            if wf['triggers']:
                print(f"    Triggers: {', '.join(wf['triggers'][:3])}")
        print("")


def show_workflow(name: str):
    """Show details for a specific workflow."""
    registry = load_registry()
    workflows = registry.get("workflows", {})

    # Find matching workflow
    matches = [n for n in workflows.keys() if name in n]

    if not matches:
        print(f"Workflow '{name}' not found.")
        print("\nAvailable workflows:")
        for n in sorted(workflows.keys()):
            print(f"  - {n}")
        return

    if len(matches) > 1 and name not in matches:
        print(f"Multiple matches for '{name}':")
        for m in matches:
            print(f"  - {m}")
        return

    wf_name = name if name in matches else matches[0]
    wf_info = workflows[wf_name]

    print(f"\n📋 Workflow: {wf_name}\n")
    print(f"Description: {wf_info.get('description', 'No description')}")

    triggers = wf_info.get("triggers", [])
    if triggers:
        print(f"\nTriggers:")
        for t in triggers:
            print(f"  - {t}")

    # Try to find and show the actual workflow file
    possible_paths = [
        WORKFLOWS_DIR / f"{wf_name}.yaml",
        WORKFLOWS_DIR / f"{wf_name}.yml",
        WORKFLOWS_DIR / wf_name / "workflow.yaml",
    ]

    for path in possible_paths:
        if path.exists():
            wf_data = parse_workflow(path)
            if wf_data:
                print(f"\nFile: {path.relative_to(CLAUDE_DIR)}")

                if "steps" in wf_data:
                    print(f"\nSteps:")
                    for i, step in enumerate(wf_data["steps"], 1):
                        step_name = step.get("name", f"Step {i}")
                        agent = step.get("agent", "unknown")
                        print(f"  {i}. {step_name} (agent: {agent})")

                if "trigger" in wf_data:
                    trigger = wf_data["trigger"]
                    if isinstance(trigger, dict):
                        if trigger.get("schedule"):
                            print(f"\nSchedule: {trigger['schedule']}")
                        if trigger.get("manual"):
                            print("Manual trigger: Yes")
            break

    print("")


def main():
    parser = argparse.ArgumentParser(description="List and describe workflows")
    parser.add_argument("name", nargs="?", help="Workflow name to show details")
    parser.add_argument("--category", "-c", type=str, help="Filter by category")
    parser.add_argument("--list", "-l", action="store_true", help="List all workflows")

    args = parser.parse_args()

    if args.name and not args.list:
        show_workflow(args.name)
    else:
        list_workflows(args.category)


if __name__ == "__main__":
    main()