dify-plugin-sdks/python/examples/telegram_trigger/provider/telegram.py

from __future__ import annotations

import secrets
from collections.abc import Mapping
from typing import Any

import requests
from werkzeug import Request, Response

from dify_plugin.entities.provider_config import CredentialType
from dify_plugin.entities.trigger import EventDispatch, Subscription, UnsubscribeResult
from dify_plugin.errors.trigger import (
    SubscriptionError,
    TriggerDispatchError,
    TriggerProviderCredentialValidationError,
    TriggerValidationError,
    UnsubscribeError,
)
from dify_plugin.interfaces.trigger import Trigger, TriggerSubscriptionConstructor


class TelegramTrigger(Trigger):
    """Dispatch Telegram Bot API updates to matching trigger events."""

    def _dispatch_event(self, subscription: Subscription, request: Request) -> EventDispatch:
        secret_token = subscription.properties.get("secret_token")
        if secret_token:
            header_token = request.headers.get("X-Telegram-Bot-Api-Secret-Token")
            if header_token != secret_token:
                raise TriggerValidationError("Invalid Telegram secret token")

        try:
            payload = request.get_json(force=True)
        except Exception as exc:  # pragma: no cover - defensive: werkzeug parsing errors
            raise TriggerDispatchError(f"Failed to parse Telegram payload: {exc}") from exc

        if not payload:
            raise TriggerDispatchError("Empty Telegram webhook payload")

        event = self._resolve_event(payload)
        response = Response(response='{"ok": true}', status=200, mimetype="application/json")
        if not event:
            return EventDispatch(events=[], response=response)
        return EventDispatch(events=[event], response=response)

    def _resolve_event(self, payload: Mapping[str, Any]) -> str:
        """Resolve Telegram update payload to a configured event name."""
        update_event_map = {
            "message": "message_received",
            "edited_message": "message_edited",
            "channel_post": "channel_post_created",
            "edited_channel_post": "channel_post_edited",
            "business_connection": "business_connection_updated",
            "business_message": "business_message_received",
            "edited_business_message": "business_message_edited",
            "deleted_business_messages": "business_messages_deleted",
            "message_reaction": "message_reaction_updated",
            "message_reaction_count": "message_reaction_count_updated",
            "inline_query": "inline_query_received",
            "chosen_inline_result": "inline_result_chosen",
            "callback_query": "callback_query_received",
            "shipping_query": "shipping_query_received",
            "pre_checkout_query": "pre_checkout_query_received",
            "poll": "poll_state_updated",
            "poll_answer": "poll_answer_received",
            "my_chat_member": "my_chat_member_updated",
            "chat_member": "chat_member_updated",
            "chat_join_request": "chat_join_request_received",
            "chat_boost": "chat_boost_updated",
            "removed_chat_boost": "chat_boost_removed",
        }
        for update_key, event_name in update_event_map.items():
            if update_key in payload:
                return event_name
        return ""


class TelegramSubscriptionConstructor(TriggerSubscriptionConstructor):
    """Manage Telegram Bot API webhook subscriptions."""

    _API_BASE = "https://api.telegram.org"

    def _validate_api_key(self, credentials: Mapping[str, Any]) -> None:
        token = credentials.get("bot_token")
        if not token:
            raise TriggerProviderCredentialValidationError("Telegram Bot Token is required")

        url = f"{self._API_BASE}/bot{token}/getMe"
        try:
            response = requests.get(url, timeout=10)
        except requests.RequestException as exc:  # pragma: no cover - network error path
            raise TriggerProviderCredentialValidationError(f"Network error: {exc}") from exc

        try:
            payload = response.json()
        except ValueError as exc:  # pragma: no cover - invalid JSON path
            raise TriggerProviderCredentialValidationError("Invalid response from Telegram API") from exc

        if not payload.get("ok"):
            description = payload.get("description") or "Telegram API rejected the token"
            raise TriggerProviderCredentialValidationError(description)

    def _create_subscription(
        self,
        endpoint: str,
        parameters: Mapping[str, Any],
        credentials: Mapping[str, Any],
        credential_type: CredentialType,
    ) -> Subscription:
        token = credentials.get("bot_token")
        if not token:
            raise SubscriptionError("Telegram Bot Token is required", error_code="MISSING_BOT_TOKEN")

        allowed_updates = parameters.get("allowed_updates") or []
        secret_token = secrets.token_urlsafe(32)
        payload = {"url": endpoint, "secret_token": secret_token}
        if allowed_updates:
            payload["allowed_updates"] = allowed_updates

        response_data = self._telegram_post(token, "setWebhook", payload)
        if not response_data.get("result"):
            description = response_data.get("description") or "Failed to set Telegram webhook"
            raise SubscriptionError(
                description,
                error_code="SET_WEBHOOK_FAILED",
                external_response=response_data,
            )

        bot_profile = self._safe_get_bot_profile(token)
        webhook_info = self._safe_get_webhook_info(token)

        properties: dict[str, Any] = {
            "secret_token": secret_token,
            "allowed_updates": allowed_updates or webhook_info.get("allowed_updates", []),
            "webhook_url": webhook_info.get("url", endpoint),
        }
        if bot_profile:
            properties["bot_id"] = bot_profile.get("id")
            properties["bot_username"] = bot_profile.get("username")
            properties["bot_name"] = bot_profile.get("first_name")
        if webhook_info:
            for key in ("has_custom_certificate", "max_connections", "ip_address", "pending_update_count"):
                if key in webhook_info:
                    properties[key] = webhook_info[key]

        return Subscription(
            expires_at=-1,
            endpoint=endpoint,
            parameters=parameters,
            properties=properties,
        )

    def _delete_subscription(
        self,
        subscription: Subscription,
        credentials: Mapping[str, Any],
        credential_type: CredentialType,
    ) -> UnsubscribeResult:
        token = credentials.get("bot_token")
        if not token:
            raise UnsubscribeError(
                message="Telegram Bot Token is required",
                error_code="MISSING_BOT_TOKEN",
            )

        response_data = self._telegram_post(token, "deleteWebhook", {"drop_pending_updates": True})
        if not response_data.get("ok"):
            description = response_data.get("description") or "Failed to delete Telegram webhook"
            raise UnsubscribeError(
                message=description,
                error_code="DELETE_WEBHOOK_FAILED",
                external_response=response_data,
            )

        message = response_data.get("description") or "Telegram webhook deleted"
        return UnsubscribeResult(success=True, message=message)

    def _refresh_subscription(
        self,
        subscription: Subscription,
        credentials: Mapping[str, Any],
        credential_type: CredentialType,
    ) -> Subscription:
        token = credentials.get("bot_token")
        if not token:
            raise SubscriptionError("Telegram Bot Token is required", error_code="MISSING_BOT_TOKEN")

        webhook_info = self._get_webhook_info(token)
        properties = dict(subscription.properties)
        for key in (
            "url",
            "has_custom_certificate",
            "pending_update_count",
            "ip_address",
            "max_connections",
            "allowed_updates",
        ):
            if key in webhook_info:
                target_key = "webhook_url" if key == "url" else key
                properties[target_key] = webhook_info[key]

        return Subscription(
            expires_at=-1,
            endpoint=subscription.endpoint,
            parameters=subscription.parameters,
            properties=properties,
        )

    # ------------------------------------------------------------------
    # Helpers
    # ------------------------------------------------------------------

    def _telegram_post(self, token: str, method: str, payload: Mapping[str, Any]) -> dict[str, Any]:
        url = f"{self._API_BASE}/bot{token}/{method}"
        try:
            response = requests.post(url, json=payload, timeout=10)
        except requests.RequestException as exc:  # pragma: no cover - network error path
            raise SubscriptionError(
                f"Network error while calling Telegram {method}: {exc}",
                error_code="NETWORK_ERROR",
            ) from exc
        return self._parse_subscription_response(response, method)

    def _telegram_get(self, token: str, method: str) -> dict[str, Any]:
        url = f"{self._API_BASE}/bot{token}/{method}"
        try:
            response = requests.get(url, timeout=10)
        except requests.RequestException as exc:  # pragma: no cover - network error path
            raise SubscriptionError(
                f"Network error while calling Telegram {method}: {exc}",
                error_code="NETWORK_ERROR",
            ) from exc
        return self._parse_subscription_response(response, method)

    def _parse_subscription_response(self, response: requests.Response, method: str) -> dict[str, Any]:
        try:
            data = response.json()
        except ValueError as exc:  # pragma: no cover - invalid JSON path
            raise SubscriptionError(
                "Invalid JSON response from Telegram API",
                error_code="INVALID_RESPONSE",
            ) from exc

        if not data.get("ok"):
            description = data.get("description") or f"Telegram API error during {method}"
            raise SubscriptionError(
                description,
                error_code="TELEGRAM_API_ERROR",
                external_response=data,
            )
        return data

    def _get_webhook_info(self, token: str) -> Mapping[str, Any]:
        response = self._telegram_get(token, "getWebhookInfo")
        return response.get("result") or {}

    def _safe_get_webhook_info(self, token: str) -> Mapping[str, Any]:
        try:
            return self._get_webhook_info(token)
        except SubscriptionError:
            return {}

    def _safe_get_bot_profile(self, token: str) -> Mapping[str, Any]:
        url = f"{self._API_BASE}/bot{token}/getMe"
        try:
            response = requests.get(url, timeout=10)
            data = response.json()
            if data.get("ok"):
                return data.get("result") or {}
        except Exception:  # pragma: no cover - best effort path
            return {}
        return {}