Spaces:
Running on Zero
Running on Zero
File size: 5,038 Bytes
bfd6783 260fe4e bfd6783 50094a0 bfd6783 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | """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_<name>`` 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
|