"""`aether-ad` CLI — generate, export-storyboard, evaluate.""" from __future__ import annotations import json import logging import os from pathlib import Path from typing import Optional import typer from dotenv import load_dotenv from rich.console import Console from rich.table import Table from aether_ad.cli.commands import export_storyboard_markdown, hit_rate_report from aether_ad.core.context import load_context from aether_ad.core.engine import GenesisEngine from aether_ad.core.genome import load_genome from aether_ad.llm.fireworks_kimi import FireworksKimiBackend from aether_ad.llm.hf_inference import HFInferenceBackend load_dotenv() logging.basicConfig( level=os.getenv("AETHER_LOG_LEVEL", "INFO"), format="%(asctime)s %(name)s %(levelname)s %(message)s", ) app = typer.Typer(help="AETHER-Ad Genesis CLI") console = Console() def _build_backend(name: str): name = (name or os.getenv("AETHER_DEFAULT_BACKEND", "hf")).lower() if name == "fireworks": return FireworksKimiBackend() if name == "hf": return HFInferenceBackend() if name == "none": return None raise typer.BadParameter(f"Unknown backend '{name}'") def _default_context_path() -> Path: return Path(__file__).resolve().parent.parent / "data" / "tensions" / "default_tensions.json" @app.command("generate") def generate( product: Path = typer.Option(..., help="Path to product genome JSON"), tension: str = typer.Option("T01_power_asymmetry", help="Tension archetype id"), persona: str = typer.Option("일반 소비자", help="Persona free-text"), duration: int = typer.Option(15, help="Spot duration (15 or 30)"), n_seeds: int = typer.Option(30, help="Raw seeds per run"), top_k: int = typer.Option(5, help="Top-k after wow filter"), backend: str = typer.Option("hf", help="LLM backend: hf | fireworks | none"), output: Path = typer.Option(Path("outputs/seeds.json"), help="Output JSON path"), tensions_path: Optional[Path] = typer.Option(None, help="Override context tensions JSON"), rng_seed: Optional[int] = typer.Option(None, help="Random seed for reproducibility"), ): """Generate scored ad-seeds for a product + tension + persona.""" if duration not in (15, 30): raise typer.BadParameter("duration must be 15 or 30") genome = load_genome(product) context = load_context(tensions_path or _default_context_path()) llm = _build_backend(backend) engine = GenesisEngine( llm=llm, context=context, seed_count_per_run=n_seeds, keep_top_k=top_k, rng_seed=rng_seed, ) with console.status("[bold cyan]Running 5-stage pipeline...[/]"): scored = engine.generate( product=genome, tension_id=tension, persona_text=persona, duration=duration, # type: ignore[arg-type] ) table = Table(title=f"Top-{top_k} Ad Seeds — {genome.brand} / {tension}") table.add_column("rank", justify="right") table.add_column("seed_id", overflow="fold") table.add_column("final", justify="right") table.add_column("rules") table.add_column("concept", overflow="fold") for i, s in enumerate(scored, 1): table.add_row( str(i), s.seed.seed_id, f"{s.score.final:.3f}", ",".join(s.seed.rules_applied), s.seed.concept[:180], ) console.print(table) output.parent.mkdir(parents=True, exist_ok=True) output.write_text( json.dumps([s.model_dump() for s in scored], ensure_ascii=False, indent=2), encoding="utf-8", ) console.print(f"[green]Wrote[/] {output}") @app.command("export-storyboard") def export_storyboard( input: Path = typer.Option(..., "--input", help="Seeds JSON from `generate`"), output: Path = typer.Option(Path("outputs/storyboards.md"), help="Markdown output"), ): """Convert a seeds JSON into a markdown storyboard.""" data = json.loads(input.read_text(encoding="utf-8")) md = export_storyboard_markdown(data) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(md, encoding="utf-8") console.print(f"[green]Wrote[/] {output}") @app.command("evaluate") def evaluate( input: Path = typer.Option(..., "--input", help="Seeds JSON"), report: Path = typer.Option(Path("outputs/hit_rate_report.md")), threshold: float = typer.Option(0.5, help="final_score threshold to count as 'hit'"), ): """Compute wow_filter hit-rate for a seeds JSON.""" data = json.loads(input.read_text(encoding="utf-8")) md = hit_rate_report(data, threshold=threshold) report.parent.mkdir(parents=True, exist_ok=True) report.write_text(md, encoding="utf-8") console.print(f"[green]Wrote[/] {report}") if __name__ == "__main__": # pragma: no cover app()