aether_ad/cli/main.py

"""`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()