feat: add click-based colorized output with progress events and build timer

- Add click as explicit dependency (already bundled with typer)
- Replace typer.echo calls with click.echo + click.style for colorized output
- Add BuildEvent enum and ProgressCallback to builder for decoupled progress reporting
- Remove direct typer dependency from builder module
- Show per-target status with colored labels (skip/ok/fail/...)
- Display elapsed build time in summary
- Colorize graph and clean command output
- Update CLI tests to match new output format
This commit is contained in:
Konstantin Fickel 2026-02-14 21:25:38 +01:00
parent 6a9d7efd5d
commit ee6c411f3c
Signed by: kfickel
GPG key ID: A793722F9933C1A5
5 changed files with 141 additions and 26 deletions

View file

@ -3,12 +3,12 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import enum
import os import os
from collections.abc import Callable
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
import typer
from bulkgen.config import ( from bulkgen.config import (
ProjectConfig, ProjectConfig,
TargetType, TargetType,
@ -28,6 +28,25 @@ from bulkgen.state import (
) )
class BuildEvent(enum.Enum):
"""Events emitted during a build for progress reporting."""
TARGET_SKIPPED = "skipped"
TARGET_BUILDING = "building"
TARGET_OK = "ok"
TARGET_FAILED = "failed"
TARGET_DEP_FAILED = "dep_failed"
TARGET_NO_PROVIDER = "no_provider"
ProgressCallback = Callable[[BuildEvent, str, str], None]
"""Signature: (event, target_name, detail_message)."""
def _noop_callback(_event: BuildEvent, _name: str, _detail: str) -> None:
"""Default no-op progress callback."""
@dataclass @dataclass
class BuildResult: class BuildResult:
"""Summary of a build run.""" """Summary of a build run."""
@ -35,6 +54,7 @@ class BuildResult:
built: list[str] = field(default_factory=list) built: list[str] = field(default_factory=list)
skipped: list[str] = field(default_factory=list) skipped: list[str] = field(default_factory=list)
failed: dict[str, str] = field(default_factory=dict) failed: dict[str, str] = field(default_factory=dict)
total_targets: int = 0
def _resolve_prompt(prompt_value: str, project_dir: Path) -> str: def _resolve_prompt(prompt_value: str, project_dir: Path) -> str:
@ -118,6 +138,7 @@ async def run_build(
config: ProjectConfig, config: ProjectConfig,
project_dir: Path, project_dir: Path,
target: str | None = None, target: str | None = None,
on_progress: ProgressCallback = _noop_callback,
) -> BuildResult: ) -> BuildResult:
"""Execute the build. """Execute the build.
@ -143,6 +164,11 @@ async def run_build(
generations = get_build_order(graph) generations = get_build_order(graph)
target_names = set(config.targets) target_names = set(config.targets)
# Count total buildable targets for progress reporting.
result.total_targets = sum(
1 for gen in generations for n in gen if n in target_names
)
for generation in generations: for generation in generations:
targets_in_gen = [n for n in generation if n in target_names] targets_in_gen = [n for n in generation if n in target_names]
@ -150,23 +176,28 @@ async def run_build(
for name in targets_in_gen: for name in targets_in_gen:
if _should_skip_failed_dep(name, config, result): if _should_skip_failed_dep(name, config, result):
result.failed[name] = "Dependency failed" result.failed[name] = "Dependency failed"
on_progress(BuildEvent.TARGET_DEP_FAILED, name, "Dependency failed")
continue continue
if _is_dirty(name, config, project_dir, state): if _is_dirty(name, config, project_dir, state):
if not _has_provider(name, providers, result): if not _has_provider(name, providers, result, on_progress):
continue continue
dirty_targets.append(name) dirty_targets.append(name)
else: else:
result.skipped.append(name) result.skipped.append(name)
on_progress(BuildEvent.TARGET_SKIPPED, name, "up to date")
if not dirty_targets: if not dirty_targets:
continue continue
for name in dirty_targets:
on_progress(BuildEvent.TARGET_BUILDING, name, "")
outcomes = await _build_generation( outcomes = await _build_generation(
dirty_targets, config, project_dir, providers dirty_targets, config, project_dir, providers
) )
_process_outcomes(outcomes, config, project_dir, state, result) _process_outcomes(outcomes, config, project_dir, state, result, on_progress)
save_state(state, project_dir) save_state(state, project_dir)
return result return result
@ -207,6 +238,7 @@ def _has_provider(
target_name: str, target_name: str,
providers: dict[TargetType, Provider], providers: dict[TargetType, Provider],
result: BuildResult, result: BuildResult,
on_progress: ProgressCallback = _noop_callback,
) -> bool: ) -> bool:
"""Check that the required provider is available; record failure if not.""" """Check that the required provider is available; record failure if not."""
target_type = infer_target_type(target_name) target_type = infer_target_type(target_name)
@ -214,7 +246,9 @@ def _has_provider(
env_var = ( env_var = (
"BFL_API_KEY" if target_type is TargetType.IMAGE else "MISTRAL_API_KEY" "BFL_API_KEY" if target_type is TargetType.IMAGE else "MISTRAL_API_KEY"
) )
result.failed[target_name] = f"Missing {env_var} environment variable" msg = f"Missing {env_var} environment variable"
result.failed[target_name] = msg
on_progress(BuildEvent.TARGET_NO_PROVIDER, target_name, msg)
return False return False
return True return True
@ -243,12 +277,13 @@ def _process_outcomes(
project_dir: Path, project_dir: Path,
state: BuildState, state: BuildState,
result: BuildResult, result: BuildResult,
on_progress: ProgressCallback = _noop_callback,
) -> None: ) -> None:
"""Process build outcomes: record state for successes, log failures.""" """Process build outcomes: record state for successes, log failures."""
for name, error in outcomes: for name, error in outcomes:
if error is not None: if error is not None:
result.failed[name] = str(error) result.failed[name] = str(error)
typer.echo(f"FAIL: {name} -- {error}", err=True) on_progress(BuildEvent.TARGET_FAILED, name, str(error))
else: else:
target_cfg = config.targets[name] target_cfg = config.targets[name]
model = resolve_model(name, target_cfg, config.defaults) model = resolve_model(name, target_cfg, config.defaults)
@ -266,4 +301,4 @@ def _process_outcomes(
project_dir=project_dir, project_dir=project_dir,
) )
result.built.append(name) result.built.append(name)
typer.echo(f" OK: {name}") on_progress(BuildEvent.TARGET_OK, name, "")

View file

@ -3,13 +3,15 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import time
from pathlib import Path from pathlib import Path
from typing import Annotated from typing import Annotated
import click
import typer import typer
from bulkgen.builder import run_build from bulkgen.builder import BuildEvent, BuildResult, run_build
from bulkgen.config import load_config from bulkgen.config import ProjectConfig, load_config
from bulkgen.graph import build_graph, get_build_order from bulkgen.graph import build_graph, get_build_order
app = typer.Typer(name="bulkgen", help="AI artifact build tool.") app = typer.Typer(name="bulkgen", help="AI artifact build tool.")
@ -19,15 +21,76 @@ def _find_config(directory: Path) -> Path:
"""Find the single ``*.bulkgen.yaml`` file in *directory*.""" """Find the single ``*.bulkgen.yaml`` file in *directory*."""
candidates = list(directory.glob("*.bulkgen.yaml")) candidates = list(directory.glob("*.bulkgen.yaml"))
if len(candidates) == 0: if len(candidates) == 0:
typer.echo("Error: No .bulkgen.yaml file found in current directory", err=True) click.echo(
click.style("Error: ", fg="red", bold=True)
+ "No .bulkgen.yaml file found in current directory",
err=True,
)
raise typer.Exit(code=1) raise typer.Exit(code=1)
if len(candidates) > 1: if len(candidates) > 1:
names = ", ".join(str(c.name) for c in candidates) names = ", ".join(str(c.name) for c in candidates)
typer.echo(f"Error: Multiple .bulkgen.yaml files found: {names}", err=True) click.echo(
click.style("Error: ", fg="red", bold=True)
+ f"Multiple .bulkgen.yaml files found: {names}",
err=True,
)
raise typer.Exit(code=1) raise typer.Exit(code=1)
return candidates[0] 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() @app.command()
def build( def build(
target: Annotated[ target: Annotated[
@ -39,16 +102,25 @@ def build(
config_path = _find_config(project_dir) config_path = _find_config(project_dir)
config = load_config(config_path) config = load_config(config_path)
result = asyncio.run(run_build(config, project_dir, target)) 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: if result.built:
typer.echo(f"\nBuilt {len(result.built)} target(s)") parts.append(click.style(f"{len(result.built)} built", fg="green", bold=True))
if result.skipped: if result.skipped:
typer.echo(f"Skipped {len(result.skipped)} target(s) (up to date)") 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: if result.failed:
typer.echo(f"Failed {len(result.failed)} target(s):", err=True)
for name, err in result.failed.items():
typer.echo(f" {name}: {err}", err=True)
raise typer.Exit(code=1) raise typer.Exit(code=1)
@ -64,15 +136,15 @@ def clean() -> None:
target_path = project_dir / target_name target_path = project_dir / target_name
if target_path.exists(): if target_path.exists():
target_path.unlink() target_path.unlink()
typer.echo(f"Removed: {target_name}") click.echo(click.style(" rm ", fg="red") + target_name)
removed += 1 removed += 1
state_path = project_dir / ".bulkgen.state.yaml" state_path = project_dir / ".bulkgen.state.yaml"
if state_path.exists(): if state_path.exists():
state_path.unlink() state_path.unlink()
typer.echo("Removed: .bulkgen.state.yaml") click.echo(click.style(" rm ", fg="red") + ".bulkgen.state.yaml")
typer.echo(f"Cleaned {removed} artifact(s)") click.echo(click.style(f"\nCleaned {removed} artifact(s)", bold=True))
@app.command() @app.command()
@ -91,11 +163,16 @@ def graph() -> None:
externals_in_gen = [n for n in gen if n not in target_names] externals_in_gen = [n for n in gen if n not in target_names]
if externals_in_gen: if externals_in_gen:
typer.echo(f"Stage {i} (inputs): {', '.join(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: if targets_in_gen:
typer.echo(f"Stage {i} (targets): {', '.join(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: for node in gen:
preds: list[str] = list(dep_graph.predecessors(node)) preds: list[str] = list(dep_graph.predecessors(node))
if preds: if preds:
typer.echo(f" {node} <- {', '.join(preds)}") arrow = click.style(" <- ", dim=True)
click.echo(f" {node}{arrow}{', '.join(preds)}")

View file

@ -5,6 +5,7 @@ description = "Bulk-Generate Images with Generative AI"
readme = "README.md" readme = "README.md"
requires-python = ">=3.13" requires-python = ">=3.13"
dependencies = [ dependencies = [
"click>=8.0.0",
"httpx>=0.27.0", "httpx>=0.27.0",
"mistralai>=1.0.0", "mistralai>=1.0.0",
"networkx>=3.6.1", "networkx>=3.6.1",

View file

@ -77,7 +77,7 @@ class TestBuildCommand:
result = runner.invoke(app, ["build"]) result = runner.invoke(app, ["build"])
assert result.exit_code == 0 assert result.exit_code == 0
assert "Built 2 target(s)" in result.output assert "2 built" in result.output
def test_build_with_skipped(self, cli_project: Path) -> None: def test_build_with_skipped(self, cli_project: Path) -> None:
build_result = BuildResult( build_result = BuildResult(
@ -95,7 +95,7 @@ class TestBuildCommand:
result = runner.invoke(app, ["build"]) result = runner.invoke(app, ["build"])
assert result.exit_code == 0 assert result.exit_code == 0
assert "Skipped 2 target(s) (up to date)" in result.output assert "2 skipped" in result.output
def test_build_with_failures(self, cli_project: Path) -> None: def test_build_with_failures(self, cli_project: Path) -> None:
build_result = BuildResult( build_result = BuildResult(
@ -115,7 +115,7 @@ class TestBuildCommand:
result = runner.invoke(app, ["build"]) result = runner.invoke(app, ["build"])
assert result.exit_code == 1 assert result.exit_code == 1
assert "Failed 1 target(s)" in result.output assert "1 failed" in result.output
def test_build_specific_target(self, cli_project: Path) -> None: def test_build_specific_target(self, cli_project: Path) -> None:
build_result = BuildResult(built=["output.txt"], skipped=[], failed={}) build_result = BuildResult(built=["output.txt"], skipped=[], failed={})

2
uv.lock generated
View file

@ -49,6 +49,7 @@ name = "bulkgen"
version = "0.1.0" version = "0.1.0"
source = { editable = "." } source = { editable = "." }
dependencies = [ dependencies = [
{ name = "click" },
{ name = "httpx" }, { name = "httpx" },
{ name = "mistralai" }, { name = "mistralai" },
{ name = "networkx" }, { name = "networkx" },
@ -67,6 +68,7 @@ dev = [
[package.metadata] [package.metadata]
requires-dist = [ requires-dist = [
{ name = "click", specifier = ">=8.0.0" },
{ name = "httpx", specifier = ">=0.27.0" }, { name = "httpx", specifier = ">=0.27.0" },
{ name = "mistralai", specifier = ">=1.0.0" }, { name = "mistralai", specifier = ">=1.0.0" },
{ name = "networkx", specifier = ">=3.6.1" }, { name = "networkx", specifier = ">=3.6.1" },