portfolio-hub/app/cli/core/utils.py

"""Formatting and execution helpers for the CLI."""

from __future__ import annotations

import json
from dataclasses import asdict, is_dataclass
from typing import Any, Callable, TypeVar

import typer
from pydantic import BaseModel, ValidationError
from rich.console import Console
from rich.panel import Panel
from rich.table import Table

from app.core.errors import FeatureNotReadyError
from app.domain.models import Daemon, Tag
from app.schemas.auth import AuthTokenResponse
from app.schemas.portfolio import CommentResponse, PortfolioDetailResponse, PortfolioListItem
from app.schemas.user import UserResponse

T = TypeVar("T")


def format_currency(value: float | None) -> str:
    if value is None:
        return "N/A"
    return f"{value:,.0f}"


def to_plain_data(value: Any) -> Any:
    if isinstance(value, BaseModel):
        return value.model_dump(mode="json")
    if is_dataclass(value):
        return asdict(value)
    if isinstance(value, list):
        return [to_plain_data(item) for item in value]
    if isinstance(value, dict):
        return {key: to_plain_data(item) for key, item in value.items()}
    return value


def print_portfolio_list(console: Console, portfolios: list[PortfolioListItem]) -> None:
    for i, item in enumerate(portfolios):
        if i:
            console.print()
        tags_str = f"  [{', '.join(item.tags)}]" if item.tags else ""
        console.print(f"[bold]{item.name}[/bold]  {item.symbol}  [dim]{item.status}[/dim]  {item.broker}{tags_str}")
        console.print(
            f"  cash {format_currency(item.cash_balance)}"
            f"  qty {item.asset_quantity:g}"
            f"  price {format_currency(item.price)}"
            f"  buffer {format_currency(item.profit_buffer)}"
            f"  total {format_currency(item.total_value)}"
        )
        console.print(f"  [dim]{item.portfolio_id}[/dim]")
        if item.managed_by_daemon_id:
            console.print(f"  daemon {item.managed_by_daemon_id}")


def print_portfolio_detail(console: Console, detail: PortfolioDetailResponse) -> None:
    portfolio = detail.portfolio
    position_value = float(portfolio.asset_quantity) * float(portfolio.price)
    bot_total = float(portfolio.cash_balance) + position_value
    user_total = bot_total + float(portfolio.profit_buffer)
    panel = Panel.fit(
        "\n".join(
            [
                f"id: {portfolio.portfolio_id}",
                f"broker: {portfolio.broker}",
                f"status: {portfolio.status}",
                f"symbol: {portfolio.symbol}",
                f"active_strategy_id: {portfolio.active_strategy_id or 'N/A'}",
                f"avabile_strategies: {', '.join(portfolio.avabile_strategies) if portfolio.avabile_strategies else 'N/A'}",
                f"cash_balance: {format_currency(portfolio.cash_balance)}",
                f"asset_quantity: {portfolio.asset_quantity:g}",
                f"price: {format_currency(portfolio.price)}",
                f"profit_buffer: {format_currency(portfolio.profit_buffer)}",
                f"position_value: {format_currency(position_value)}",
                f"bot_total: {format_currency(bot_total)}",
                f"user_total: {format_currency(user_total)}",
                f"live_quantity: {detail.live_quantity if detail.live_quantity is not None else 'N/A'}",
            ]
        ),
        title="Portfolio Status",
    )
    console.print(panel)

    if detail.strategy_set is not None:
        console.print(
            Panel.fit(
                "\n".join(
                    [
                        f"id: {detail.strategy_set.id}",
                        f"name: {detail.strategy_set.name}",
                        f"config: {detail.strategy_set.config}",
                        f"nick: {detail.strategy_set.nick}",
                    ]
                ),
                title="Strategy Set",
            )
        )

    ledger_table = Table(title="Recent Ledger Entries", expand=True)
    ledger_table.add_column("Recorded At", style="cyan", no_wrap=True)
    ledger_table.add_column("Action")
    ledger_table.add_column("Qty", justify="right")
    ledger_table.add_column("Price", justify="right")
    ledger_table.add_column("Cash After", justify="right")
    ledger_table.add_column("Qty After", justify="right")
    ledger_table.add_column("Buffer After", justify="right")
    for entry in detail.recent_entries:
        ledger_table.add_row(
            entry.recorded_at.isoformat(timespec="seconds"),
            entry.action,
            f"{entry.quantity:g}",
            format_currency(entry.price),
            format_currency(entry.cash_balance_after),
            f"{entry.asset_quantity_after:g}",
            format_currency(entry.profit_buffer_after),
        )
    console.print(ledger_table)


def print_user_list(console: Console, users: list[UserResponse]) -> None:
    for i, user in enumerate(users):
        if i:
            console.print()
        console.print(f"[bold]{user.username}[/bold]  [dim]{user.id}[/dim]")
        console.print(f"  email {user.email or 'N/A'}")
        console.print(
            f"  created {user.created_at.isoformat(timespec='seconds') if user.created_at else 'N/A'}"
            f"  updated {user.updated_at.isoformat(timespec='seconds') if user.updated_at else 'N/A'}"
        )


def print_user_detail(console: Console, user: UserResponse) -> None:
    console.print(
        Panel.fit(
            "\n".join(
                [
                    f"id: {user.id}",
                    f"username: {user.username}",
                    f"email: {user.email or 'N/A'}",
                    f"created_at: {user.created_at.isoformat(timespec='seconds') if user.created_at else 'N/A'}",
                    f"updated_at: {user.updated_at.isoformat(timespec='seconds') if user.updated_at else 'N/A'}",
                ]
            ),
            title="User",
        )
    )


def print_daemon_table(console: Console, daemons: list[Daemon]) -> None:
    table = Table(title="Daemons", expand=True)
    table.add_column("ID", style="cyan", no_wrap=True)
    table.add_column("Slug", no_wrap=True)
    table.add_column("Display Name")
    table.add_column("Active", no_wrap=True)
    table.add_column("Created At", no_wrap=True)
    table.add_column("Updated At", no_wrap=True)
    for daemon in daemons:
        table.add_row(
            daemon.id,
            daemon.daemon_slug,
            daemon.display_name or "-",
            "yes" if daemon.is_active else "no",
            daemon.created_at.isoformat(timespec="seconds") if daemon.created_at else "N/A",
            daemon.updated_at.isoformat(timespec="seconds") if daemon.updated_at else "N/A",
        )
    console.print(table)


def print_daemon_detail(console: Console, daemon: Daemon) -> None:
    console.print(
        Panel.fit(
            "\n".join(
                [
                    f"id: {daemon.id}",
                    f"owner_user_id: {daemon.owner_user_id}",
                    f"daemon_slug: {daemon.daemon_slug}",
                    f"display_name: {daemon.display_name or '-'}",
                    f"is_active: {daemon.is_active}",
                    f"created_at: {daemon.created_at.isoformat(timespec='seconds') if daemon.created_at else 'N/A'}",
                    f"updated_at: {daemon.updated_at.isoformat(timespec='seconds') if daemon.updated_at else 'N/A'}",
                ]
            ),
            title="Daemon",
        )
    )


def print_service_token_table(console: Console, tokens: list[AuthTokenResponse]) -> None:
    table = Table(title="Service Tokens", expand=True)
    table.add_column("Name", style="cyan", no_wrap=True)
    table.add_column("Permission", no_wrap=True)
    table.add_column("User ID", no_wrap=True)
    table.add_column("Daemon ID", no_wrap=True)
    table.add_column("Prefix", no_wrap=True)
    table.add_column("Expires At", no_wrap=True)
    table.add_column("Revoked At", no_wrap=True)
    table.add_column("Created At", no_wrap=True)
    for token in tokens:
        table.add_row(
            token.name,
            token.permission,
            token.user_id,
            token.daemon_id or "-",
            token.token_prefix,
            token.expires_at.isoformat(timespec="seconds") if token.expires_at else "never",
            token.revoked_at.isoformat(timespec="seconds") if token.revoked_at else "-",
            token.created_at.isoformat(timespec="seconds") if token.created_at else "N/A",
        )
    console.print(table)


def print_comment_table(console: Console, comments: list[CommentResponse]) -> None:
    table = Table(title="Portfolio Comments", expand=True)
    table.add_column("ID", style="cyan", no_wrap=True)
    table.add_column("Portfolio ID", no_wrap=True)
    table.add_column("Content")
    table.add_column("Created At", no_wrap=True)
    table.add_column("Updated At", no_wrap=True)
    for comment in comments:
        table.add_row(
            comment.id,
            comment.portfolio_id,
            comment.content,
            comment.created_at.isoformat(timespec="seconds") if comment.created_at else "N/A",
            comment.updated_at.isoformat(timespec="seconds") if comment.updated_at else "N/A",
        )
    console.print(table)


def print_tag_table(console: Console, tags: list[Tag]) -> None:
    table = Table(title="Tags", expand=True)
    table.add_column("ID", style="cyan", no_wrap=True)
    table.add_column("Name", no_wrap=True)
    table.add_column("Admin", no_wrap=True)
    table.add_column("Active", no_wrap=True)
    table.add_column("Created By", no_wrap=True)
    table.add_column("Created At", no_wrap=True)
    for tag in tags:
        table.add_row(
            tag.tag_id,
            tag.name,
            "yes" if tag.is_admin_tag else "no",
            "yes" if tag.is_active else "no",
            tag.created_by_user_id or "-",
            tag.created_at.isoformat(timespec="seconds") if tag.created_at else "N/A",
        )
    console.print(table)


def print_comment_detail(console: Console, comment: CommentResponse) -> None:
    console.print(
        Panel.fit(
            "\n".join(
                [
                    f"id: {comment.id}",
                    f"portfolio_id: {comment.portfolio_id}",
                    f"content: {comment.content}",
                    f"created_at: {comment.created_at.isoformat(timespec='seconds') if comment.created_at else 'N/A'}",
                    f"updated_at: {comment.updated_at.isoformat(timespec='seconds') if comment.updated_at else 'N/A'}",
                ]
            ),
            title="Comment",
        )
    )


def print_json(console: Console, payload: Any) -> None:
    console.print_json(data=to_plain_data(payload))


def render_output(console: Console, output_format: str, payload: Any, *, empty_message: str | None = None) -> None:
    if output_format == "json":
        print_json(console, payload)
        return

    if isinstance(payload, list):
        if not payload and empty_message:
            console.print(empty_message)
            return
        if all(isinstance(item, Daemon) for item in payload):
            print_daemon_table(console, payload)
            return
        if all(isinstance(item, PortfolioListItem) for item in payload):
            print_portfolio_list(console, payload)
            return
        if all(isinstance(item, UserResponse) for item in payload):
            print_user_list(console, payload)
            return
        if all(isinstance(item, CommentResponse) for item in payload):
            print_comment_table(console, payload)
            return
        if all(isinstance(item, AuthTokenResponse) for item in payload):
            print_service_token_table(console, payload)
            return
        if all(isinstance(item, Tag) for item in payload):
            print_tag_table(console, payload)
            return
        return

    if isinstance(payload, PortfolioDetailResponse):
        print_portfolio_detail(console, payload)
        return

    if isinstance(payload, UserResponse):
        print_user_detail(console, payload)
        return

    if isinstance(payload, Daemon):
        print_daemon_detail(console, payload)
        return

    if isinstance(payload, CommentResponse):
        print_comment_detail(console, payload)
        return

    console.print(payload)


def execute_command(console: Console, callback: Callable[[], T]) -> T:
    try:
        return callback()
    except ValidationError as exc:
        console.print(f"Input validation failed: {exc}")
        raise typer.Exit(code=1) from exc
    except FeatureNotReadyError as exc:
        console.print(str(exc))
        raise typer.Exit(code=1) from exc
    except ValueError as exc:
        console.print(str(exc))
        raise typer.Exit(code=1) from exc