Ledgerly/ledgerly/cli.py

"""Command line interface: parsing, dispatch, and text presentation.

Handlers stay pure with respect to I/O: each returns the text to print (or
None). ``main`` is the only place that talks to stdout/stderr (D7), which
keeps the CLI trivially testable.
"""

from __future__ import annotations

import argparse
import sys
from datetime import date
from pathlib import Path
from typing import Callable, Sequence

from . import __version__
from .errors import LedgerlyError, ValidationError
from .money import InvalidMoney, Money
from .models import EXPENSE, INCOME, Month, Transaction, parse_date
from .repository import SqliteRepository
from .services import BudgetService, LedgerService, ReportingService

__all__ = ["build_parser", "default_db_path", "main"]


def default_db_path() -> Path:
    return Path.home() / ".ledgerly" / "ledgerly.db"


# -- argument types ----------------------------------------------------------


def money_arg(text: str) -> Money:
    try:
        return Money.parse(text)
    except InvalidMoney as exc:
        raise argparse.ArgumentTypeError(str(exc)) from exc


def date_arg(text: str) -> date:
    try:
        return parse_date(text)
    except ValidationError as exc:
        raise argparse.ArgumentTypeError(str(exc)) from exc


def month_arg(text: str) -> Month:
    try:
        return Month.from_str(text)
    except ValidationError as exc:
        raise argparse.ArgumentTypeError(str(exc)) from exc


# -- parser -------------------------------------------------------------------


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="ledgerly",
        description=(
            "A tiny personal finance tracker: transactions, budgets, "
            "monthly reports."
        ),
    )
    parser.add_argument(
        "--version", action="version", version=f"ledgerly {__version__}"
    )
    parser.add_argument(
        "--db",
        default=None,
        metavar="PATH",
        help=f"database file (default: {default_db_path()})",
    )
    sub = parser.add_subparsers(dest="command", required=True)

    add = sub.add_parser("add", help="record a transaction")
    add.add_argument("amount", type=money_arg)
    add.add_argument("--category", required=True)
    add.add_argument("--date", type=date_arg, default=None, help="YYYY-MM-DD (default: today)")
    add.add_argument("--note", default=None)

    lst = sub.add_parser("list", help="list transactions")
    lst.add_argument("--month", type=month_arg, default=None, help="YYYY-MM (default: current)")
    lst.add_argument("--category", default=None)
    lst.add_argument("--all", action="store_true", help="ignore the month filter")

    delete = sub.add_parser("delete", help="delete a transaction by id")
    delete.add_argument("id", type=int)

    sub.add_parser("categories", help="list categories")

    cat_add = sub.add_parser("category-add", help="add a category")
    cat_add.add_argument("name")
    cat_add.add_argument("--type", dest="kind", required=True, choices=[INCOME, EXPENSE])

    cat_del = sub.add_parser("category-del", help="delete an unused category")
    cat_del.add_argument("name")

    budget_set = sub.add_parser("budget-set", help="set the monthly budget for an expense category")
    budget_set.add_argument("category")
    budget_set.add_argument("amount", type=money_arg)

    budget_del = sub.add_parser("budget-del", help="remove a budget")
    budget_del.add_argument("category")

    budget_show = sub.add_parser("budget-show", help="budgets vs. spending for a month")
    budget_show.add_argument("--month", type=month_arg, default=None, help="YYYY-MM (default: current)")

    report = sub.add_parser("report", help="monthly summary and category breakdown")
    report.add_argument("--month", type=month_arg, default=None, help="YYYY-MM (default: current)")

    return parser


# -- command handlers (return the text to print) ------------------------------


def _cmd_add(args: argparse.Namespace, repo: SqliteRepository) -> str:
    service = LedgerService(repo)
    day = args.date or date.today()
    tx = service.add_transaction(args.amount, args.category, day, args.note)
    note = f"  ({tx.note})" if tx.note else ""
    return (
        f"recorded #{tx.id}: {tx.amount} in {tx.category} "
        f"on {tx.date.isoformat()}{note}"
    )


def _cmd_list(args: argparse.Namespace, repo: SqliteRepository) -> str:
    service = LedgerService(repo)
    if args.all:
        start = end = None
    else:
        start, end = (args.month or Month.now()).range
    rows = service.list_transactions(start, end, args.category)
    return _render_transactions(rows)


def _render_transactions(rows: list[Transaction]) -> str:
    if not rows:
        return "no transactions"
    out = []
    for tx in rows:
        line = f"#{tx.id:<4} {tx.date.isoformat()}  {tx.category:<16} {tx.kind:<8} {tx.amount}"
        if tx.note:
            line += f"  [{tx.note}]"
        out.append(line)
    return "\n".join(out)


def _cmd_delete(args: argparse.Namespace, repo: SqliteRepository) -> str:
    LedgerService(repo).delete_transaction(args.id)
    return f"deleted transaction #{args.id}"


def _cmd_categories(args: argparse.Namespace, repo: SqliteRepository) -> str:
    cats = LedgerService(repo).list_categories()
    if not cats:
        return "no categories"
    return "\n".join(f"{cat.kind:<8} {cat.name}" for cat in cats)


def _cmd_category_add(args: argparse.Namespace, repo: SqliteRepository) -> str:
    cat = LedgerService(repo).add_category(args.name, args.kind)
    return f"added {cat.kind} category {cat.name!r}"


def _cmd_category_del(args: argparse.Namespace, repo: SqliteRepository) -> str:
    LedgerService(repo).delete_category(args.name)
    return f"deleted category {args.name!r}"


def _cmd_budget_set(args: argparse.Namespace, repo: SqliteRepository) -> str:
    budget = BudgetService(repo).set_budget(args.category, args.amount)
    return f"budget for {budget.category}: {budget.amount} per month"


def _cmd_budget_del(args: argparse.Namespace, repo: SqliteRepository) -> str:
    budget = BudgetService(repo).delete_budget(args.category)
    return f"removed budget for {budget.category}: {budget.amount}"


def _cmd_budget_show(args: argparse.Namespace, repo: SqliteRepository) -> str:
    month = args.month or Month.now()
    lines = ReportingService(repo).budget_status(month)
    out = [f"budgets for {month.label}"]
    if not lines:
        out.append("no budgets set, no unbudgeted spending")
    for line in lines:
        marker = "  OVER" if line.is_over else ""
        out.append(
            f"{line.category:<16} budget {line.budget:<12} "
            f"spent {line.spent:<12} left {line.remaining}{marker}"
        )
    return "\n".join(out)


def _cmd_report(args: argparse.Namespace, repo: SqliteRepository) -> str:
    month = args.month or Month.now()
    service = ReportingService(repo)
    summary = service.monthly_summary(month)
    out = [
        f"report for {month.label}",
        f"  income     {summary.income}",
        f"  expenses   {summary.expenses}",
        f"  net        {summary.net}",
    ]
    totals = service.category_totals(month)
    if not totals:
        out.append("  no transactions this month")
    else:
        current_kind = None
        for total in totals:
            if total.kind != current_kind:
                out.append(f"  {total.kind}")
                current_kind = total.kind
            out.append(f"    {total.category:<16} {total.total}")
    return "\n".join(out)


HANDLERS: dict[str, Callable[[argparse.Namespace, SqliteRepository], str]] = {
    "add": _cmd_add,
    "list": _cmd_list,
    "delete": _cmd_delete,
    "categories": _cmd_categories,
    "category-add": _cmd_category_add,
    "category-del": _cmd_category_del,
    "budget-set": _cmd_budget_set,
    "budget-del": _cmd_budget_del,
    "budget-show": _cmd_budget_show,
    "report": _cmd_report,
}


# -- entry point ---------------------------------------------------------------


def main(argv: Sequence[str] | None = None, *, db_path: str | Path | None = None) -> int:
    """Run the CLI. Returns a process exit code (SPECS.md §7)."""
    args = build_parser().parse_args(argv)
    path = str(db_path) if db_path is not None else str(args.db or default_db_path())
    repo = SqliteRepository(path)
    try:
        output = HANDLERS[args.command](args, repo)
    except LedgerlyError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1
    finally:
        repo.close()
    if output:
        print(output)
    return 0