setup_flow.py

Go to the documentation of this file.

"""Corrected operator-initiated Telegram setup flow.
 
Dependencies
------------
- oc_connect.connectors.telegram.destinations (DestinationStore, TelegramDestination - records the pinned operator destination)
- oc_connect.connectors.telegram.transport (TelegramTransport - the default transport for update discovery and the test send)
- oc_connect.credentials (CredentialBroker - stores the bot token inside a rollback transaction)
- oc_connect.egress (require_safe - the default egress check on the test message)
- oc_diagnostics (_dbg structured logger)
"""
from __future__ import annotations
 
import getpass
from collections.abc import Callable
from typing import Any, Protocol
 
from oc_diagnostics import _dbg
 
from ...credentials import CredentialBroker
from ...egress import require_safe
from .destinations import DestinationStore, TelegramDestination
from .transport import TelegramTransport
 
TELEGRAM_CREDENTIAL_NAME = "telegram-operator-bot"
TEST_MESSAGE = "OC Connect is live."
 
 
class SetupCancelled(RuntimeError):
    """The operator cancelled setup; all local writes were rolled back."""
 
 
class TelegramSetupError(RuntimeError):
    """Setup could not identify a valid operator-initiated update."""
 
 
class SetupTransport(Protocol):
    """The two Bot API methods needed during setup."""
 
    def get_updates(self) -> list[dict[str, Any]]: ...
 
    def send_message(self, chat_id: str, text: str) -> dict[str, Any]: ...
 
    def close(self) -> None: ...
 
 
def _default_transport(raw_value: str) -> TelegramTransport:
    return TelegramTransport(raw_value)
 
 
class TelegramSetup:
    """Capture, pin, test, and commit one operator destination atomically."""
 
    def __init__(
        self,
        *,
        broker: CredentialBroker,
        destinations: DestinationStore,
        transport_factory: Callable[[str], SetupTransport] = _default_transport,
        token_reader: Callable[[str], str] = getpass.getpass,
        input_reader: Callable[[str], str] = input,
        output: Callable[[str], None] = print,
        egress_check: Callable[[str], None] = require_safe,
    ) -> None:
        self.broker = broker
        self.destinations = destinations
        self.transport_factory = transport_factory
        self.token_reader = token_reader
        self.input_reader = input_reader
        self.output = output
        self.egress_check = egress_check
 
    def _read_token(self) -> str:
        try:
            raw_value = self.token_reader(
                "Telegram bot token (input hidden, Enter to cancel): "
            )
        except (EOFError, KeyboardInterrupt):
            raise SetupCancelled("Telegram setup cancelled at credential capture") from None
        if raw_value is None or not raw_value.strip():
            raise SetupCancelled("Telegram setup cancelled at credential capture")
        return raw_value.strip()
 
    def _answer(self, prompt: str) -> str:
        try:
            return self.input_reader(prompt).strip()
        except (EOFError, KeyboardInterrupt):
            raise SetupCancelled("Telegram setup cancelled") from None
 
    @staticmethod
    def _operator_update(updates: list[dict[str, Any]]) -> tuple[str, str]:
        for update in reversed(updates):
            message = update.get("message")
            if not isinstance(message, dict):
                continue
            account = message.get("from")
            chat = message.get("chat")
            if not isinstance(account, dict) or not isinstance(chat, dict):
                continue
            if account.get("is_bot") is True:
                continue
            account_id = account.get("id")
            chat_id = chat.get("id")
            if account_id is not None and chat_id is not None:
                return str(account_id), str(chat_id)
        raise TelegramSetupError(
            "no operator-authored Telegram message was found; message the bot and retry"
        )
 
    def run(self) -> TelegramDestination:
        """Run setup, committing local state only after the test send succeeds."""
        self.output(
            "Create the bot with BotFather, then provide its token. The bot cannot "
            "start a conversation or create a group."
        )
        raw_value = self._read_token()
        with self.broker.transaction():
            self.broker.store(TELEGRAM_CREDENTIAL_NAME, raw_value)
            transport = self.transport_factory(raw_value)
            try:
                self.output(
                    "Now message the bot from your operator account. A private direct "
                    "chat is the Phase 1 default."
                )
                answer = self._answer(
                    "Press Enter after sending the message, or type cancel: "
                )
                if answer.lower() == "cancel":
                    raise SetupCancelled("Telegram setup cancelled before discovery")
                account_id, chat_id = self._operator_update(transport.get_updates())
                self.output(
                    f"Found operator account {account_id} in chat {chat_id}."
                )
                confirmed = self._answer("Pin this account and chat? [y/N]: ")
                if confirmed.lower() not in ("y", "yes"):
                    raise SetupCancelled("Telegram setup cancelled before pinning")
                self.egress_check(TEST_MESSAGE)
                transport.send_message(chat_id, TEST_MESSAGE)
                destination = self.destinations.record(
                    name="operator",
                    chat_id=chat_id,
                    operator_account_id=account_id,
                    credential_name=TELEGRAM_CREDENTIAL_NAME,
                )
                _dbg(
                    "OC_CONNECT.TELEGRAM",
                    "run",
                    f"setup complete destination={destination.name} "
                    f"account={account_id} chat={chat_id}",
                    2,
                )
                return destination
            finally:
                transport.close()
 
 
__all__ = [
    "SetupCancelled",
    "TELEGRAM_CREDENTIAL_NAME",
    "TEST_MESSAGE",
    "TelegramSetup",
    "TelegramSetupError",
]