"""Typer CLI for bulkgen: build, clean, graph commands.""" from __future__ import annotations import asyncio import time from pathlib import Path from typing import Annotated import click import typer from bulkgen.builder import BuildEvent, BuildResult, run_build from bulkgen.config import ProjectConfig, load_config from bulkgen.graph import build_graph, get_build_order from bulkgen.models import ALL_MODELS app = typer.Typer(name="bulkgen", help="AI artifact build tool.") def _find_config(directory: Path) -> Path: """Find the single ``*.bulkgen.yaml`` file in *directory*.""" candidates = list(directory.glob("*.bulkgen.yaml")) if len(candidates) == 0: click.echo( click.style("Error: ", fg="red", bold=True) + "No .bulkgen.yaml file found in current directory", err=True, ) raise typer.Exit(code=1) if len(candidates) > 1: names = ", ".join(str(c.name) for c in candidates) click.echo( click.style("Error: ", fg="red", bold=True) + f"Multiple .bulkgen.yaml files found: {names}", err=True, ) raise typer.Exit(code=1) return candidates[0] def _format_elapsed(seconds: float) -> str: """Format elapsed time as a human-readable string.""" if seconds < 60: return f"{seconds:.1f}s" minutes = int(seconds // 60) secs = seconds % 60 return f"{minutes}m{secs:.1f}s" def _run_build( config: ProjectConfig, project_dir: Path, target: str | None ) -> tuple[BuildResult, float]: """Run the async build with click-styled progress output and timing.""" def on_progress(event: BuildEvent, name: str, detail: str) -> None: if event is BuildEvent.TARGET_SKIPPED: click.echo( click.style(" skip ", fg="yellow") + click.style(name, bold=True) + click.style(f" ({detail})", dim=True) ) elif event is BuildEvent.TARGET_BUILDING: click.echo(click.style(" ... ", fg="cyan") + click.style(name, bold=True)) elif event is BuildEvent.TARGET_OK: click.echo(click.style(" ok ", fg="green") + click.style(name, bold=True)) elif event is BuildEvent.TARGET_FAILED: click.echo( click.style(" fail ", fg="red", bold=True) + click.style(name, bold=True) + f" {detail}" ) elif event is BuildEvent.TARGET_DEP_FAILED: click.echo( click.style(" fail ", fg="red") + click.style(name, bold=True) + click.style(f" ({detail})", dim=True) ) elif event is BuildEvent.TARGET_NO_PROVIDER: click.echo( click.style(" fail ", fg="red", bold=True) + click.style(name, bold=True) + f" {detail}" ) start = time.monotonic() result = asyncio.run( run_build(config, project_dir, target, on_progress=on_progress) ) elapsed = time.monotonic() - start return result, elapsed @app.command() def build( target: Annotated[ str | None, typer.Argument(help="Specific target to build.") ] = None, ) -> None: """Build all targets (or a specific target) in dependency order.""" project_dir = Path.cwd() config_path = _find_config(project_dir) config = load_config(config_path) click.echo(click.style("bulkgen", fg="cyan", bold=True) + " building targets...\n") result, elapsed = _run_build(config, project_dir, target) # Summary click.echo("") parts: list[str] = [] if result.built: parts.append(click.style(f"{len(result.built)} built", fg="green", bold=True)) if result.skipped: parts.append(click.style(f"{len(result.skipped)} skipped", fg="yellow")) if result.failed: parts.append(click.style(f"{len(result.failed)} failed", fg="red", bold=True)) summary = ", ".join(parts) if parts else click.style("nothing to do", dim=True) timer = click.style(f" in {_format_elapsed(elapsed)}", dim=True) click.echo(summary + timer) if result.failed: raise typer.Exit(code=1) @app.command() def clean() -> None: """Remove generated artifacts (targets only, not input files).""" project_dir = Path.cwd() config_path = _find_config(project_dir) config = load_config(config_path) removed = 0 for target_name in config.targets: target_path = project_dir / target_name if target_path.exists(): target_path.unlink() click.echo(click.style(" rm ", fg="red") + target_name) removed += 1 state_path = project_dir / ".bulkgen.state.yaml" if state_path.exists(): state_path.unlink() click.echo(click.style(" rm ", fg="red") + ".bulkgen.state.yaml") click.echo(click.style(f"\nCleaned {removed} artifact(s)", bold=True)) @app.command() def graph() -> None: """Print the dependency graph with build stages.""" project_dir = Path.cwd() config_path = _find_config(project_dir) config = load_config(config_path) dep_graph = build_graph(config, project_dir) generations = get_build_order(dep_graph) target_names = set(config.targets) for i, gen in enumerate(generations): targets_in_gen = [n for n in gen if n in target_names] externals_in_gen = [n for n in gen if n not in target_names] if externals_in_gen: label = click.style(f"Stage {i}", fg="cyan", bold=True) kind = click.style(" (inputs)", dim=True) click.echo(f"{label}{kind}: {', '.join(externals_in_gen)}") if targets_in_gen: label = click.style(f"Stage {i}", fg="green", bold=True) kind = click.style(" (targets)", dim=True) click.echo(f"{label}{kind}: {', '.join(targets_in_gen)}") for node in gen: preds: list[str] = list(dep_graph.predecessors(node)) if preds: arrow = click.style(" <- ", dim=True) click.echo(f" {node}{arrow}{', '.join(preds)}") @app.command() def models() -> None: """List available models and their capabilities.""" name_width = max(len(m.name) for m in ALL_MODELS) provider_width = max(len(m.provider) for m in ALL_MODELS) type_width = max(len(m.type) for m in ALL_MODELS) header_name = "Model".ljust(name_width) header_provider = "Provider".ljust(provider_width) header_type = "Type".ljust(type_width) header_caps = "Capabilities" click.echo( click.style(header_name, bold=True) + " " + click.style(header_provider, bold=True) + " " + click.style(header_type, bold=True) + " " + click.style(header_caps, bold=True) ) click.echo( "─" * name_width + " " + "─" * provider_width + " " + "─" * type_width + " " + "─" * len(header_caps) ) for model in ALL_MODELS: name_col = model.name.ljust(name_width) provider_col = model.provider.ljust(provider_width) type_col = model.type.ljust(type_width) caps_col = ", ".join(model.capabilities) type_color = "green" if model.type == "image" else "cyan" click.echo( click.style(name_col, fg="white", bold=True) + " " + provider_col + " " + click.style(type_col, fg=type_color) + " " + click.style(caps_col, dim=True) )