"""Books ledger reads — the two thin list endpoints behind the Books UI
(P4). Receivables (`invoices`) and payables (`expenses`) presented as
filtered, paginated, tenant-scoped tables with full-set sums.

The settlement math lives in ``services/money/settlement.py`` (FROZEN) —
this module never recomputes it; it only lists the underlying ledger
rows the sheet used to hold, plus a tiny tenant-meta read (``parallel_run``)
the UI banner needs. Money values are emitted as 2dp decimal STRINGS to
match the settlement engine's wire convention (the frontend renders, it
never does money arithmetic).

Everything here is gated on ``can_see_income`` — the whole Books section
is owner-level money (same 403 shape as the settlement engine).
"""

from __future__ import annotations

from datetime import date
from decimal import Decimal

from sqlalchemy import Select, and_, func, or_, select
from sqlalchemy.orm import Session, aliased

from app.models import Expense, Invoice, Party, Project, Tenant
from app.services.errors import ServiceError
from app.services.reporting.money import quantize2

DEFAULT_PAGE_SIZE = 50
MAX_PAGE_SIZE = 200


# --------------------------------------------------------------- helpers


def _require_income(actor) -> None:
    """Owner-only gate — mirrors settlement.py's INCOME_FORBIDDEN shape."""
    if not actor.capabilities.get("can_see_income"):
        raise ServiceError(
            403,
            "INCOME_FORBIDDEN",
            "Only an owner can view the books.",
            "The books are owner-level money data.",
        )


def _money_str(value) -> str | None:
    """A ledger amount as a 2dp decimal string, or None (never priced)."""
    if value is None:
        return None
    return str(quantize2(Decimal(str(value))))


def _sum_str(value) -> str:
    """A COALESCE(SUM(...), 0) fold to a 2dp decimal string (never None)."""
    return str(quantize2(Decimal(str(value if value is not None else 0))))


def _month_str(value: date | None) -> str | None:
    """First-of-month settlement tag -> 'YYYY-MM' (or None when untagged)."""
    if value is None:
        return None
    return f"{value.year:04d}-{value.month:02d}"


def _escape_like(term: str) -> str:
    """Escape LIKE metacharacters so user text matches literally (registry
    parity). Order matters — escape the escape char (``\\``) FIRST."""
    return term.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def _parse_settlement_month(raw: str) -> date:
    """'YYYY-MM' -> the first-of-month Date the ledgers are tagged with.
    Same contract (and error code) as settlement.parse_month, kept local
    so books.py does not reach into the frozen engine's internals."""
    try:
        year_s, month_s = raw.strip().split("-", 1)
        return date(int(year_s), int(month_s), 1)
    except (ValueError, AttributeError) as exc:
        raise ServiceError(
            422,
            "BAD_MONTH",
            "month must look like YYYY-MM",
            f"Got {raw!r}. Use a settlement month, 'unsettled', or omit it.",
        ) from exc


def _clean(raw: str | None) -> str | None:
    """Trim; treat ''/'all' as filter-off (legacy URL vocabulary)."""
    if raw is None:
        return None
    trimmed = raw.strip()
    if trimmed == "" or trimmed == "all":
        return None
    return trimmed


def _paging(page: int | None, page_size: int | None) -> tuple[int, int]:
    size = DEFAULT_PAGE_SIZE if page_size is None else int(page_size)
    size = max(1, min(size, MAX_PAGE_SIZE))
    current = max(1, int(page or 1))
    return current, size


def _is_unsettled(month: str | None) -> bool:
    return month is not None and month.strip().lower() == "unsettled"


# --------------------------------------------------------- invoices ledger


def list_invoices(
    session: Session,
    actor,
    *,
    month: str | None = None,
    status: str | None = None,
    q: str | None = None,
    page: int | None = None,
    page_size: int | None = None,
) -> dict:
    """Receivables ledger: the tenant's invoices, filtered + paginated,
    with sums (amount + fees) over the FULL filtered set."""
    _require_income(actor)
    current_page, size = _paging(page, page_size)

    to_party = aliased(Party, name="to_party")

    def base(*columns) -> Select:
        return (
            select(*columns)
            .select_from(Invoice)
            .outerjoin(to_party, to_party.id == Invoice.to_party_id)
            .where(Invoice.tenant_id == actor.tenant_id)
        )

    conditions = []
    unsettled = _is_unsettled(month)
    month_clean = _clean(month)
    if unsettled:
        conditions.append(Invoice.settlement_month.is_(None))
    elif month_clean is not None:
        conditions.append(Invoice.settlement_month == _parse_settlement_month(month_clean))

    status_clean = _clean(status)
    if status_clean is not None:
        conditions.append(Invoice.status == status_clean)

    q_clean = _clean(q)
    if q_clean is not None:
        pattern = f"%{_escape_like(q_clean)}%"
        conditions.append(
            or_(
                Invoice.name.ilike(pattern, escape="\\"),
                Invoice.external_id.ilike(pattern, escape="\\"),
                to_party.name.ilike(pattern, escape="\\"),
            )
        )

    where = and_(*conditions) if conditions else None

    totals_stmt = base(
        func.count(Invoice.id),
        func.coalesce(func.sum(Invoice.manual_total), 0),
        func.coalesce(func.sum(Invoice.fee), 0),
    )
    if where is not None:
        totals_stmt = totals_stmt.where(where)
    total_count, sum_amount, sum_fees = session.execute(totals_stmt).one()

    rows_stmt = base(
        Invoice.id,
        Invoice.external_id,
        Invoice.name,
        to_party.name.label("client"),
        Invoice.status,
        Invoice.invoice_date,
        Invoice.due_date,
        Invoice.settlement_month,
        Invoice.currency,
        Invoice.original_amount,
        Invoice.original_currency,
        Invoice.manual_total,
        Invoice.fee,
        Invoice.commission_amount,
        Invoice.billing_type,
        Invoice.relevant_period,
        Invoice.notes,
        Invoice.collected_amount,
        Invoice.collected_currency,
    )
    if where is not None:
        rows_stmt = rows_stmt.where(where)

    # settlement_month desc (nulls first only when the unsettled bucket is
    # the selected view — where every row is null anyway), then newest
    # invoice date, then id for a stable page boundary.
    month_order = (
        Invoice.settlement_month.desc().nulls_first()
        if unsettled
        else Invoice.settlement_month.desc().nulls_last()
    )
    rows_stmt = rows_stmt.order_by(
        month_order, Invoice.invoice_date.desc().nulls_last(), Invoice.id
    ).limit(size).offset((current_page - 1) * size)

    rows = [
        {
            "id": str(r.id),
            "external_id": r.external_id,
            "name": r.name,
            "client": r.client,
            "status": r.status,
            "invoice_date": r.invoice_date.isoformat() if r.invoice_date else None,
            "due_date": r.due_date.isoformat() if r.due_date else None,
            "settlement_month": _month_str(r.settlement_month),
            "currency": r.currency,
            "original_amount": _money_str(r.original_amount),
            "original_currency": r.original_currency,
            "amount": _money_str(r.manual_total),
            "fee": _money_str(r.fee),
            "commission_amount": _money_str(r.commission_amount),
            "billing_type": r.billing_type,
            "relevant_period": r.relevant_period,
            "notes": r.notes,
            # realized-FX pair recorded at Paid for a USD invoice (slice 2b)
            "collected_amount": _money_str(r.collected_amount),
            "collected_currency": r.collected_currency,
        }
        for r in session.execute(rows_stmt)
    ]

    return {
        "rows": rows,
        "total_count": int(total_count),
        "page": current_page,
        "page_size": size,
        "sum_amount": _sum_str(sum_amount),
        "sum_fees": _sum_str(sum_fees),
    }


# --------------------------------------------------------- expenses ledger


def list_expenses(
    session: Session,
    actor,
    *,
    month: str | None = None,
    category: str | None = None,
    paid_by: str | None = None,
    q: str | None = None,
    page: int | None = None,
    page_size: int | None = None,
) -> dict:
    """Payables ledger: the tenant's expenses, filtered + paginated, with
    a full-set sum, plus the distinct category + paid-by lists the filter
    dropdowns need (returned in the same response — one round trip)."""
    _require_income(actor)
    current_page, size = _paging(page, page_size)

    tenant = session.get(Tenant, actor.tenant_id)
    tenant_party_id = tenant.party_id if tenant else None

    vendor = aliased(Party, name="vendor")
    payer = aliased(Party, name="payer")

    def base(*columns) -> Select:
        return (
            select(*columns)
            .select_from(Expense)
            .outerjoin(vendor, vendor.id == Expense.vendor_party_id)
            .outerjoin(payer, payer.id == Expense.paid_by_party_id)
            .outerjoin(Project, Project.id == Expense.project_id)
            .where(Expense.tenant_id == actor.tenant_id)
        )

    conditions = []
    unsettled = _is_unsettled(month)
    month_clean = _clean(month)
    if unsettled:
        conditions.append(Expense.settlement_month.is_(None))
    elif month_clean is not None:
        conditions.append(Expense.settlement_month == _parse_settlement_month(month_clean))

    category_clean = _clean(category)
    if category_clean is not None:
        conditions.append(Expense.category == category_clean)

    paid_by_clean = _clean(paid_by)
    if paid_by_clean is not None:
        conditions.append(payer.name == paid_by_clean)

    q_clean = _clean(q)
    if q_clean is not None:
        pattern = f"%{_escape_like(q_clean)}%"
        conditions.append(
            or_(
                vendor.name.ilike(pattern, escape="\\"),
                Expense.external_invoice_id.ilike(pattern, escape="\\"),
                Project.name.ilike(pattern, escape="\\"),
            )
        )

    where = and_(*conditions) if conditions else None

    totals_stmt = base(
        func.count(Expense.id),
        func.coalesce(func.sum(Expense.amount), 0),
    )
    if where is not None:
        totals_stmt = totals_stmt.where(where)
    total_count, sum_amount = session.execute(totals_stmt).one()

    rows_stmt = base(
        Expense.id,
        Expense.expense_date,
        vendor.name.label("vendor"),
        Project.name.label("project"),
        Expense.category,
        Expense.type,
        Expense.amount,
        Expense.currency,
        payer.name.label("paid_by"),
        Expense.paid_by_party_id,
        Expense.settlement_month,
        Expense.external_invoice_id,
        Expense.notes,
    )
    if where is not None:
        rows_stmt = rows_stmt.where(where)

    month_order = (
        Expense.settlement_month.desc().nulls_first()
        if unsettled
        else Expense.settlement_month.desc().nulls_last()
    )
    rows_stmt = rows_stmt.order_by(
        month_order, Expense.expense_date.desc().nulls_last(), Expense.id
    ).limit(size).offset((current_page - 1) * size)

    rows = [
        {
            "id": str(r.id),
            "expense_date": r.expense_date.isoformat() if r.expense_date else None,
            "vendor": r.vendor,
            "project": r.project,
            "category": r.category,
            "type": r.type,
            "amount": _money_str(r.amount),
            "currency": r.currency,
            "paid_by": r.paid_by,
            # the settlement engine reimburses expenses fronted by a party
            # OTHER than the tenant itself — the UI accents those rows.
            "paid_by_is_tenant": (
                r.paid_by_party_id is not None
                and tenant_party_id is not None
                and r.paid_by_party_id == tenant_party_id
            ),
            "settlement_month": _month_str(r.settlement_month),
            "external_invoice_id": r.external_invoice_id,
            "notes": r.notes,
        }
        for r in session.execute(rows_stmt)
    ]

    categories = [
        c
        for (c,) in session.execute(
            select(Expense.category)
            .where(Expense.tenant_id == actor.tenant_id, Expense.category.is_not(None))
            .distinct()
            .order_by(Expense.category)
        ).all()
    ]
    paid_by_options = [
        name
        for (name,) in session.execute(
            select(payer.name)
            .select_from(Expense)
            .join(payer, payer.id == Expense.paid_by_party_id)
            .where(Expense.tenant_id == actor.tenant_id, payer.name.is_not(None))
            .distinct()
            .order_by(payer.name)
        ).all()
    ]

    return {
        "rows": rows,
        "total_count": int(total_count),
        "page": current_page,
        "page_size": size,
        "sum_amount": _sum_str(sum_amount),
        "categories": categories,
        "paid_by_options": paid_by_options,
    }


# ------------------------------------------------------------------- meta


def books_meta(session: Session, actor) -> dict:
    """Tenant flags the Books UI needs. ``parallel_run`` drives the banner
    warning that these books rebuild nightly from the sheet (so app edits
    would be overwritten). Owner-only, like the rest of Books. Since
    ADR #021 the flag is COMPUTED from ``books_owner`` (sheet-owned →
    true; app-owned or legacy-mirrored → false; the legacy boolean
    ``parallel_run: true`` still reads as sheet-owned) — the API shape is
    unchanged, so the banner lifts by itself when the tenant flips."""
    from app.services.tenancy import BOOKS_OWNER_SHEET, effective_books_owner

    _require_income(actor)
    tenant = session.get(Tenant, actor.tenant_id)
    settings = tenant.settings if tenant else None
    return {"parallel_run": effective_books_owner(settings) == BOOKS_OWNER_SHEET}


__all__ = ["list_invoices", "list_expenses", "books_meta"]