Add support for Telegram message attachments

homeassistant/components/telegram_bot/bot.py

@@ -7,7 +7,7 @@
 import logging
 from ssl import SSLContext
 from types import MappingProxyType
-from typing import Any
+from typing import Any, cast
 
 import httpx
 from telegram import (
@@ -17,6 +17,7 @@
     InlineKeyboardMarkup,
     InputPollOption,
     Message,
+    PhotoSize,
     ReplyKeyboardMarkup,
     ReplyKeyboardRemove,
     Update,
@@ -50,6 +51,10 @@
     ATTR_DISABLE_NOTIF,
     ATTR_DISABLE_WEB_PREV,
     ATTR_FILE,
+    ATTR_FILE_ID,
+    ATTR_FILE_MIME_TYPE,
+    ATTR_FILE_NAME,
+    ATTR_FILE_SIZE,
     ATTR_FROM_FIRST,
     ATTR_FROM_LAST,
     ATTR_KEYBOARD,
@@ -78,6 +83,7 @@
     CONF_CHAT_ID,
     CONF_PROXY_URL,
     DOMAIN,
+    EVENT_TELEGRAM_ATTACHMENT,
     EVENT_TELEGRAM_CALLBACK,
     EVENT_TELEGRAM_COMMAND,
     EVENT_TELEGRAM_SENT,
@@ -175,6 +181,10 @@ def _get_message_event_data(self, message: Message) -> tuple[str, dict[str, Any]
             # This is a command message - set event type to command and split data into command and args
             event_type = EVENT_TELEGRAM_COMMAND
             event_data.update(self._get_command_event_data(message.text))
+        elif filters.ATTACHMENT.filter(message):
+            event_type = EVENT_TELEGRAM_ATTACHMENT
+            event_data[ATTR_TEXT] = message.caption
+            event_data.update(self._get_file_id_event_data(message))
         else:
             event_type = EVENT_TELEGRAM_TEXT
             event_data[ATTR_TEXT] = message.text
@@ -184,6 +194,26 @@ def _get_message_event_data(self, message: Message) -> tuple[str, dict[str, Any]
 
         return event_type, event_data
 
+    def _get_file_id_event_data(self, message: Message) -> dict[str, Any]:
+        """Extract file_id from a message attachment, if any."""
+        if filters.PHOTO.filter(message):
+            photos = cast(Sequence[PhotoSize], message.effective_attachment)
+            return {
+                ATTR_FILE_ID: photos[-1].file_id,
+                ATTR_FILE_MIME_TYPE: "image/jpeg",  # telegram always uses jpeg for photos
+                ATTR_FILE_SIZE: photos[-1].file_size,
+            }
+        return {
+            k: getattr(message.effective_attachment, v)
+            for k, v in (
+                (ATTR_FILE_ID, "file_id"),
+                (ATTR_FILE_NAME, "file_name"),
+                (ATTR_FILE_MIME_TYPE, "mime_type"),
+                (ATTR_FILE_SIZE, "file_size"),
+            )
+            if hasattr(message.effective_attachment, v)
+        }
+
     def _get_user_event_data(self, user: User) -> dict[str, Any]:
         return {
             ATTR_USER_ID: user.id,

homeassistant/components/telegram_bot/const.py

@@ -53,6 +53,7 @@
 EVENT_TELEGRAM_CALLBACK = "telegram_callback"
 EVENT_TELEGRAM_COMMAND = "telegram_command"
 EVENT_TELEGRAM_TEXT = "telegram_text"
+EVENT_TELEGRAM_ATTACHMENT = "telegram_attachment"
 EVENT_TELEGRAM_SENT = "telegram_sent"
 
 PARSER_HTML = "html"
@@ -89,6 +90,10 @@
 ATTR_DISABLE_WEB_PREV = "disable_web_page_preview"
 ATTR_EDITED_MSG = "edited_message"
 ATTR_FILE = "file"
+ATTR_FILE_ID = "file_id"
+ATTR_FILE_MIME_TYPE = "file_mime_type"
+ATTR_FILE_NAME = "file_name"
+ATTR_FILE_SIZE = "file_size"
 ATTR_FROM_FIRST = "from_first"
 ATTR_FROM_LAST = "from_last"
 ATTR_KEYBOARD = "keyboard"

tests/components/telegram_bot/conftest.py

@@ -242,6 +242,55 @@ def update_callback_query():
     }
 
 
+@pytest.fixture
+def update_message_attachment():
+    """Fixture for mocking an incoming update of type message/attachment."""
+    return {
+        "update_id": 1,
+        "message": {
+            "message_id": 1,
+            "date": 1441645532,
+            "from": {
+                "id": 12345678,
+                "is_bot": False,
+                "last_name": "Test Lastname",
+                "first_name": "Test Firstname",
+                "username": "Testusername",
+            },
+            "chat": {
+                "last_name": "Test Lastname",
+                "id": 1111111,
+                "type": "private",
+                "first_name": "Test Firstname",
+                "username": "Testusername",
+            },
+            "photo": [
+                {
+                    "file_id": "AgACAgUAAxkBAAIBUWJ5aXl6Y3h5bXl6Y3h5bXl6Y3gAAJxvMRtP7nG4Fq7m0m0vBAAMCAAN5AAMjBA",
+                    "file_unique_id": "AQADcbzEbT-5xuBa",
+                    "file_size": 1234,
+                    "width": 90,
+                    "height": 90,
+                },
+                {
+                    "file_id": "AgACAgUAAxkBAAIBUWJ5aXl6Y3h5bXl6Y3h5bXl6Y3gAAJxvMRtP7nG4Fq7m0m0vBAAMCAAN5AAMjBA",
+                    "file_unique_id": "AQADcbzEbT-5xuBa",
+                    "file_size": 12345,
+                    "width": 320,
+                    "height": 320,
+                },
+                {
+                    "file_id": "AgACAgUAAxkBAAIBUWJ5aXl6Y3h5bXl6Y3h5bXl6Y3gAAJxvMRtP7nG4Fq7m0m0vBAAMCAAN5AAMjBA",
+                    "file_unique_id": "AQADcbzEbT-5xuBa",
+                    "file_size": 123456,
+                    "width": 800,
+                    "height": 800,
+                },
+            ],
+        },
+    }
+
+
 @pytest.fixture
 def mock_broadcast_config_entry() -> MockConfigEntry:
     """Return the default mocked config entry."""

tests/components/telegram_bot/test_telegram_bot.py

@@ -476,6 +476,36 @@ async def test_webhook_endpoint_generates_telegram_callback_event(
     assert isinstance(events[0].context, Context)
 
 
+async def test_webhook_endpoint_generates_telegram_attachment_event(
+    hass: HomeAssistant,
+    webhook_platform,
+    hass_client: ClientSessionGenerator,
+    update_message_attachment,
+    mock_generate_secret_token,
+) -> None:
+    """POST to the configured webhook endpoint and assert fired `telegram_attachment` event."""
+    client = await hass_client()
+    events = async_capture_events(hass, "telegram_attachment")
+
+    response = await client.post(
+        f"{TELEGRAM_WEBHOOK_URL}_123456",
+        json=update_message_attachment,
+        headers={"X-Telegram-Bot-Api-Secret-Token": mock_generate_secret_token},
+    )
+    assert response.status == 200
+    assert (await response.read()).decode("utf-8") == ""
+
+    # Make sure event has fired
+    await hass.async_block_till_done()
+
+    assert len(events) == 1
+    assert (
+        events[0].data["file_id"]
+        == update_message_attachment["message"]["photo"][-1]["file_id"]
+    )
+    assert isinstance(events[0].context, Context)
+
+
 async def test_polling_platform_message_text_update(
     hass: HomeAssistant,
     config_polling,