notify-bridge/packages/server/src/notify_bridge_server/commands/planka_handler.py

"""Planka-specific bot command handler."""

from __future__ import annotations

import logging
from collections.abc import Callable, Coroutine
from typing import Any

from ..database.models import (
    CommandConfig, CommandTracker, ServiceProvider, TelegramBot,
)
from ..services import make_planka_provider
from ..services.http_session import get_http_session
from .base import CommandResponse, ProviderCommandHandler
from .command_utils import get_last_event_str, get_tracked_collection_ids, get_trackers_for_provider
from .handler import _render_cmd_template

_LOGGER = logging.getLogger(__name__)

_PLANKA_COMMANDS = {"status", "boards", "cards", "lists"}


def _get_tracked_board_ids(
    provider: ServiceProvider,
    trackers: list,
) -> list[str]:
    """Get board IDs from tracked collection_ids for this provider."""
    if not provider.config.get("api_key"):
        return []
    return get_tracked_collection_ids(provider, trackers)


# ---------------------------------------------------------------------------
# Command dispatch table
# ---------------------------------------------------------------------------

_TEXT_COMMANDS: dict[str, Callable[..., Coroutine[Any, Any, dict[str, Any]]]] = {}


def _text_cmd(fn: Callable[..., Coroutine[Any, Any, dict[str, Any]]]) -> Callable[..., Coroutine[Any, Any, dict[str, Any]]]:
    """Register a function in the text command dispatch table."""
    name = fn.__name__.removeprefix("_cmd_")
    _TEXT_COMMANDS[name] = fn
    return fn


class PlankaCommandHandler(ProviderCommandHandler):
    """Handles Planka-specific bot commands."""

    provider_type = "planka"

    def get_provider_commands(self) -> set[str]:
        return _PLANKA_COMMANDS

    def get_rate_categories(self) -> dict[str, str]:
        return {
            "boards": "api", "cards": "api", "lists": "api",
        }

    async def handle(
        self,
        cmd: str,
        args: str,
        count: int,
        locale: str,
        response_mode: str,
        provider: ServiceProvider,
        cmd_templates: dict[str, dict[str, str]],
        bot: TelegramBot,
        tracker: CommandTracker,
        config: CommandConfig,
        *,
        listener: Any = None,
        allowed_album_ids: set[str] | None = None,  # noqa: ARG002 — unused (Planka has no album model)
        page: int = 1,
    ) -> CommandResponse | None:
        fn = _TEXT_COMMANDS.get(cmd)
        if fn is None:
            return None
        ctx = await fn(provider, count)
        return CommandResponse(text=_render_cmd_template(cmd_templates, cmd, locale, ctx))


@_text_cmd
async def _cmd_status(provider: ServiceProvider, count: int) -> dict[str, Any]:
    trackers = await get_trackers_for_provider(provider.id)
    tracked_boards = _get_tracked_board_ids(provider, trackers)

    tracker_ids = [t.id for t in trackers]
    last_str = await get_last_event_str(tracker_ids)

    return {
        "boards_count": len(tracked_boards),
        "last_event": last_str,
    }


@_text_cmd
async def _cmd_boards(provider: ServiceProvider, count: int) -> dict[str, Any]:
    trackers = await get_trackers_for_provider(provider.id)
    tracked_boards = _get_tracked_board_ids(provider, trackers)

    boards_data: list[dict[str, Any]] = []
    http = await get_http_session()
    planka = make_planka_provider(http, provider)
    all_boards = await planka.client.get_boards()
    board_names = {str(b.get("id", "")): b.get("name", "") for b in all_boards}
    for board_id in tracked_boards:
        boards_data.append({"name": board_names.get(board_id, board_id)})

    return {"boards": boards_data}


@_text_cmd
async def _cmd_cards(provider: ServiceProvider, count: int) -> dict[str, Any]:
    trackers = await get_trackers_for_provider(provider.id)
    tracked_boards = _get_tracked_board_ids(provider, trackers)

    all_cards: list[dict[str, Any]] = []
    http = await get_http_session()
    planka = make_planka_provider(http, provider)
    boards = await planka.client.get_boards()
    board_names = {str(b.get("id", "")): b.get("name", "") for b in boards}

    for board_id in tracked_boards:
        cards = await planka.client.get_board_cards(board_id, limit=count)
        lists = await planka.client.get_board_lists(board_id)
        lists_by_id = {str(lst.get("id", "")): lst.get("name", "") for lst in lists}
        board_name = board_names.get(board_id, board_id)

        for card in cards:
            list_id = str(card.get("listId", ""))
            all_cards.append({
                "name": card.get("name", ""),
                "list_name": lists_by_id.get(list_id, ""),
                "board_name": board_name,
            })

    return {"cards": all_cards[:count]}


@_text_cmd
async def _cmd_lists(provider: ServiceProvider, count: int) -> dict[str, Any]:
    trackers = await get_trackers_for_provider(provider.id)
    tracked_boards = _get_tracked_board_ids(provider, trackers)

    all_lists: list[dict[str, Any]] = []
    http = await get_http_session()
    planka = make_planka_provider(http, provider)
    boards = await planka.client.get_boards()
    board_names = {str(b.get("id", "")): b.get("name", "") for b in boards}

    for board_id in tracked_boards:
        lists = await planka.client.get_board_lists(board_id)
        board_name = board_names.get(board_id, board_id)

        for lst in lists:
            all_lists.append({
                "name": lst.get("name", ""),
                "board_name": board_name,
            })

    return {"lists": all_lists}