"""The extension point. A *feature* is a self-contained bundle of bot behaviour: some commands, and optionally an interest in plain messages or uploaded files. Adding one means subclassing :class:`Feature` and registering it -- no edits to the router, the webhook, or any other feature. Expense tracking is simply the first feature registered. Anything else a bot needs to do (reminders, notes, standups, ops alerts) drops in beside it on equal terms. """ from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import date from typing import TYPE_CHECKING, Optional, Sequence if TYPE_CHECKING: # pragma: no cover - typing only from ..advice import Advisor from ..config import Settings from ..ingest import Ingestor from ..llm import LLMClient from ..models import PendingImport from ..store import Store from ..registry import FeatureRegistry @dataclass(frozen=True) class CommandSpec: """One slash command, and enough metadata to build /help and setMyCommands.""" name: str help: str args: str = "" aliases: tuple[str, ...] = () #: Hidden commands still work but stay out of the menu (e.g. /confirm). hidden: bool = False @property def usage(self) -> str: return f"/{self.name} {self.args}".strip() @dataclass class Attachment: """A file the user sent, already downloaded.""" data: bytes filename: str = "" mime_type: str = "" kind: str = "document" # "photo" | "document" caption: str = "" @property def is_image(self) -> bool: return self.kind == "photo" or self.mime_type.startswith("image/") @dataclass class Services: """Shared dependencies handed to every feature.""" settings: "Settings" store: "Store" llm: "LLMClient" ingestor: "Ingestor" advisor: "Advisor" registry: "FeatureRegistry" #: Imports awaiting /confirm, keyed by user id. Held in memory only -- a #: Space restart drops them, which is acceptable because the user simply #: re-sends the file, and it keeps unconfirmed model output out of storage. pending: dict[int, "PendingImport"] = field(default_factory=dict) #: Recent coach conversation per user, so follow-ups make sense. In memory #: only and shared across surfaces, so a thread started on Telegram #: continues in the web chat. Deliberately not persisted: it is chat #: scratch, not ledger data, and keeping it out of the mirrored database #: avoids storing conversation content alongside the finances. conversations: dict[int, list[dict]] = field(default_factory=dict) @dataclass class BotContext: """Everything a feature needs to answer one update.""" user_id: int chat_id: int services: Services today: date #: Current @handle without the @, empty if unset. Display and access only -- #: everything stored is keyed on the immutable user_id. username: str = "" text: str = "" command: str = "" args: str = "" attachment: Optional[Attachment] = None @property def settings(self) -> "Settings": return self.services.settings @property def store(self) -> "Store": return self.services.store @dataclass class Reply: """What to send back. HTML parse mode -- see finbot.render for why.""" text: str parse_mode: Optional[str] = "HTML" disable_preview: bool = True #: Optional file to send alongside the text, as (filename, bytes). document: Optional[tuple[str, bytes]] = None class Feature(ABC): #: Stable identifier, used in logs and /help grouping. name: str = "feature" #: One line shown in /help. description: str = "" def commands(self) -> Sequence[CommandSpec]: """Slash commands this feature owns.""" return () def handle_command(self, ctx: BotContext) -> Optional[Reply]: """Handle one of this feature's commands. Return None to decline.""" return None def handle_message(self, ctx: BotContext) -> Optional[Reply]: """Handle plain (non-command) text. Return None to let others try.""" return None def handle_attachment(self, ctx: BotContext) -> Optional[Reply]: """Handle an uploaded file. Return None to let others try.""" return None #: Lower runs first when several features want the same plain message. priority: int = 100 class SimpleFeature(Feature): """Convenience base that routes commands to ``cmd_`` methods.""" @abstractmethod def commands(self) -> Sequence[CommandSpec]: # pragma: no cover - abstract ... def handle_command(self, ctx: BotContext) -> Optional[Reply]: for spec in self.commands(): if ctx.command == spec.name or ctx.command in spec.aliases: handler = getattr(self, f"cmd_{spec.name}", None) if handler is None: return None return handler(ctx) return None