src/densefeed/cli.py

"""CLI — typer app for DenseFeed."""

from __future__ import annotations

import typer
from rich import print as rprint

from densefeed import __version__

app = typer.Typer(
    name="densefeed", help="Daily tech news podcast pipeline", no_args_is_help=True
)
sources_app = typer.Typer(help="Manage sources")
app.add_typer(sources_app, name="sources")


def _version(value: bool) -> None:
    if value:
        rprint(f"densefeed {__version__}")
        raise typer.Exit()


@app.callback()
def main(
    version: bool = typer.Option(
        None, "--version", "-v", callback=_version, is_eager=True
    ),
) -> None:
    """DenseFeed — daily tech news podcast pipeline."""


@app.command()
def run(
    date: str | None = typer.Option(
        None, "--date", "-d", help="YYYY-MM-DD (default: today)"
    ),
    stage: int | None = typer.Option(
        None, "--stage", "-s", help="Run single stage 1-6"
    ),
    single_stage: bool = typer.Option(
        False, "--single-stage", help="Stop after the first requested stage"
    ),
    config: str | None = typer.Option(None, "--config", "-c", help="Config file path"),
    verbose: bool = typer.Option(False, "--verbose", "-v", help="Show debug logs"),
) -> None:
    """Run the DenseFeed pipeline."""
    import asyncio
    import logging
    from datetime import datetime
    from rich.table import Table
    from densefeed.auth import hf_token
    from densefeed.config import load_config
    from densefeed.pipeline import Pipeline
    from densefeed.storage.duckdb_backend import DuckDBBackend
    from densefeed.storage import sync as storage_sync

    level = logging.DEBUG if verbose else logging.INFO
    logging.basicConfig(level=level, format="%(message)s")

    cfg = load_config(config)
    target_date = date or datetime.now().strftime("%Y-%m-%d")
    start = stage or 1

    rprint(f"[bold]DenseFeed[/bold] — [cyan]{target_date}[/cyan]")
    rprint(f"LLM Backend: {cfg.llm.backend.upper()}")
    if cfg.llm.backend == "baml":
        from densefeed.backends.baml_backend import _resolve_client_name
        baml_client = _resolve_client_name(cfg) or "MlxPrecise/MlxCreative"
        rprint(f"LLM Client: {baml_client}")
    rprint(f"Run dir: {cfg.pipeline.run_dir}\n")

    async def _run() -> None:
        import os

        storage = None
        if cfg.storage.backend == "duckdb":
            if token := os.environ.get("HF_TOKEN"):
                hf_token.set(token)
            storage_sync.download_db(cfg.storage.hf_dataset_repo, cfg.storage.duckdb_path)
            storage = DuckDBBackend(
                cfg.storage.duckdb_path, hf_audio_repo=cfg.storage.hf_audio_repo
            )

        user_id = os.environ.get("DENSEFEED_RUN_USER_ID", "")

        pipeline = Pipeline(cfg, storage=storage)
        result = await pipeline.run(
            target_date, start_from=start, single_stage=single_stage, user_id=user_id
        )

        table = Table(title=f"Pipeline — {target_date}")
        for col in ["Stage", "Name", "Status", "Time", "Items"]:
            table.add_column(col, style="bold" if col == "Stage" else None)

        for s in result.stages:
            status = (
                "[green]✓[/green]" if s.success else f"[red]✗ {s.error or ''}[/red]"
            )
            items = f"{s.items_in}→{s.items_out}" if s.items_in else str(s.items_out)
            table.add_row(str(s.stage), s.name, status, f"{s.duration_s:.1f}s", items)

        rprint(table)
        if result.stages and result.stages[-1].success:
            rprint(
                f"\n[bold green]Output:[/bold green] {result.stages[-1].output_path}"
            )

    asyncio.run(_run())


@app.command()
def config_show(
    config: str | None = typer.Option(None, "--config", "-c", help="Config file path"),
) -> None:
    """Display resolved configuration."""
    from densefeed.config import load_config
    import yaml as _yaml

    rprint(_yaml.dump(load_config(config).model_dump(), default_flow_style=False))


@app.command()
def ui(
    host: str = typer.Option("0.0.0.0", "--host", help="Host to bind the UI to"),
    port: int = typer.Option(7860, "--port", help="Port to run the UI on"),
    config: str | None = typer.Option(None, "--config", "-c", help="Config file path"),
) -> None:
    """Launch the Gradio web interface."""
    from densefeed.app import launch_app

    launch_app(host=host, port=port, config_path=config)


@sources_app.command("list")
def sources_list(
    config: str | None = typer.Option(None, "--config", "-c", help="Config file path"),
) -> None:
    """List configured sources."""
    from densefeed.config import load_config

    src = load_config(config).sources
    for name, s in [
        ("RSS", src.rss),
        ("Hacker News", src.hn),
        ("arXiv", src.arxiv),
        ("GitHub", src.github),
    ]:
        if s.enabled:
            info = (
                f"{len(s.feeds)} feeds"
                if name == "RSS"
                else f"{len(getattr(s, 'terms', []))} terms"
                if name == "Hacker News"
                else f"{getattr(s, 'categories', '') or getattr(s, 'trending_topic', '')}"
            )
            rprint(
                f"[bold]{name}[/bold] — cap {s.cap}, {info}"
                if info
                else f"[bold]{name}[/bold] — cap {s.cap}"
            )
            if name == "RSS":
                for f in s.feeds:
                    rprint(f"  {f.label}: {f.url}")


@sources_app.command("test")
def sources_test(
    source: str | None = typer.Option(
        None, "--source", "-s", help="Test specific source: rss|hn|arxiv|github"
    ),
    config: str | None = typer.Option(None, "--config", "-c", help="Config file path"),
) -> None:
    """Test source fetching (requires network)."""
    import asyncio
    import time
    from densefeed.config import load_config
    from densefeed.fetchers import fetch_all

    async def _test() -> None:
        items = await fetch_all(
            load_config(config),
            time.time() - 86400,
            enabled={source} if source else None,
        )
        rprint(f"\n[bold]Fetched {len(items)} items[/bold]")
        for item in items[:10]:
            rprint(f"  [{item.source}] {item.title[:80]}")
        if len(items) > 10:
            rprint(f"  ... and {len(items) - 10} more")

    asyncio.run(_test())