Add download file service to Telegram bot integration

homeassistant/components/telegram_bot/__init__.py

@@ -44,9 +44,12 @@
     ATTR_CAPTION,
     ATTR_CHAT_ACTION,
     ATTR_CHAT_ID,
+    ATTR_DIRECTORY_PATH,
     ATTR_DISABLE_NOTIF,
     ATTR_DISABLE_WEB_PREV,
     ATTR_FILE,
+    ATTR_FILE_ID,
+    ATTR_FILE_NAME,
     ATTR_IS_ANONYMOUS,
     ATTR_IS_BIG,
     ATTR_KEYBOARD,
@@ -90,6 +93,7 @@
     PLATFORM_WEBHOOKS,
     SERVICE_ANSWER_CALLBACK_QUERY,
     SERVICE_DELETE_MESSAGE,
+    SERVICE_DOWNLOAD_FILE,
     SERVICE_EDIT_CAPTION,
     SERVICE_EDIT_MESSAGE,
     SERVICE_EDIT_MESSAGE_MEDIA,
@@ -312,6 +316,15 @@
     }
 )
 
+SERVICE_SCHEMA_DOWNLOAD_FILE = vol.Schema(
+    {
+        vol.Optional(CONF_CONFIG_ENTRY_ID): cv.string,
+        vol.Required(ATTR_FILE_ID): cv.string,
+        vol.Optional(ATTR_DIRECTORY_PATH): cv.string,
+        vol.Optional(ATTR_FILE_NAME): cv.string,
+    }
+)
+
 SERVICE_MAP: dict[str, VolSchemaType] = {
     SERVICE_SEND_MESSAGE: SERVICE_SCHEMA_SEND_MESSAGE,
     SERVICE_SEND_CHAT_ACTION: SERVICE_SCHEMA_SEND_CHAT_ACTION,
@@ -331,6 +344,7 @@
     SERVICE_DELETE_MESSAGE: SERVICE_SCHEMA_DELETE_MESSAGE,
     SERVICE_LEAVE_CHAT: SERVICE_SCHEMA_LEAVE_CHAT,
     SERVICE_SET_MESSAGE_REACTION: SERVICE_SCHEMA_SET_MESSAGE_REACTION,
+    SERVICE_DOWNLOAD_FILE: SERVICE_SCHEMA_DOWNLOAD_FILE,
 }
 
 
@@ -426,6 +440,8 @@ async def async_send_telegram_message(service: ServiceCall) -> ServiceResponse:
             await notify_service.set_message_reaction(context=service.context, **kwargs)
         elif msgtype == SERVICE_EDIT_MESSAGE_MEDIA:
             await notify_service.edit_message_media(context=service.context, **kwargs)
+        elif msgtype == SERVICE_DOWNLOAD_FILE:
+            return await notify_service.download_file(context=service.context, **kwargs)
         else:
             await notify_service.edit_message(
                 msgtype, context=service.context, **kwargs
@@ -471,6 +487,7 @@ async def async_send_telegram_message(service: ServiceCall) -> ServiceResponse:
             SERVICE_SEND_STICKER,
             SERVICE_SEND_LOCATION,
             SERVICE_SEND_POLL,
+            SERVICE_DOWNLOAD_FILE,
         ]:
             supports_response = SupportsResponse.OPTIONAL
 

homeassistant/components/telegram_bot/bot.py

@@ -5,6 +5,8 @@
 from collections.abc import Callable, Sequence
 import io
 import logging
+import os
+from pathlib import Path
 from ssl import SSLContext
 from types import MappingProxyType
 from typing import Any, cast
@@ -13,6 +15,7 @@
 from telegram import (
     Bot,
     CallbackQuery,
+    File,
     InlineKeyboardButton,
     InlineKeyboardMarkup,
     InputMedia,
@@ -45,6 +48,7 @@
 from homeassistant.core import Context, HomeAssistant
 from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
 from homeassistant.helpers.dispatcher import async_dispatcher_send
+from homeassistant.util.json import JsonValueType
 from homeassistant.util.ssl import get_default_context, get_default_no_verify_context
 
 from .const import (
@@ -61,6 +65,7 @@
     ATTR_FILE_ID,
     ATTR_FILE_MIME_TYPE,
     ATTR_FILE_NAME,
+    ATTR_FILE_PATH,
     ATTR_FILE_SIZE,
     ATTR_FROM_FIRST,
     ATTR_FROM_LAST,
@@ -1036,6 +1041,68 @@ async def set_message_reaction(
             context=context,
         )
 
+    async def download_file(
+        self,
+        file_id: str,
+        directory_path: str | None = None,
+        file_name: str | None = None,
+        context: Context | None = None,
+        **kwargs: dict[str, Any],
+    ) -> dict[str, JsonValueType]:
+        """Download a file from Telegram."""
+        if not directory_path:
+            directory_path = self.hass.config.path(DOMAIN)
+        if not await self.hass.async_add_executor_job(
+            self.hass.config.is_allowed_path, directory_path
+        ):
+            raise ServiceValidationError(
+                "File path has not been configured in allowlist_external_dirs.",
+                translation_domain=DOMAIN,
+                translation_key="allowlist_external_dirs_error",
+            )
+        file: File = await self._send_msg(
+            self.bot.get_file,
+            "Error getting file",
+            None,
+            file_id=file_id,
+            context=context,
+        )
+        if not file.file_path:
+            raise HomeAssistantError(
+                translation_domain=DOMAIN,
+                translation_key="action_failed",
+                translation_placeholders={
+                    "error": "No file path returned from Telegram"
+                },
+            )
+        if not file_name:
+            file_name = os.path.basename(file.file_path)
+
+        custom_path = os.path.join(directory_path, file_name)
+        await self.hass.async_add_executor_job(
+            self._prepare_download_directory, directory_path
+        )
+        _LOGGER.debug("Download file %s to %s", file_id, custom_path)
+        try:
+            file_content = await file.download_as_bytearray()
+            await self.hass.async_add_executor_job(
+                Path(custom_path).write_bytes, file_content
+            )
+        except (RuntimeError, OSError, TelegramError) as exc:
+            raise HomeAssistantError(
+                translation_domain=DOMAIN,
+                translation_key="action_failed",
+                translation_placeholders={"error": str(exc)},
+            ) from exc
+        return {ATTR_FILE_PATH: custom_path}
+
+    @staticmethod
+    def _prepare_download_directory(directory_path: str) -> None:
+        """Create download directory if it does not exist."""
+        if not os.path.exists(directory_path):
+            _LOGGER.debug("directory %s does not exist, creating it", directory_path)
+            os.makedirs(directory_path, exist_ok=True)
+
 
 def initialize_bot(hass: HomeAssistant, p_config: MappingProxyType[str, Any]) -> Bot:
     """Initialize telegram bot with proxy support."""

homeassistant/components/telegram_bot/const.py

@@ -43,6 +43,7 @@
 SERVICE_ANSWER_CALLBACK_QUERY = "answer_callback_query"
 SERVICE_DELETE_MESSAGE = "delete_message"
 SERVICE_LEAVE_CHAT = "leave_chat"
+SERVICE_DOWNLOAD_FILE = "download_file"
 
 SIGNAL_UPDATE_EVENT = "telegram_bot_update_event"
 EVENT_TELEGRAM_CALLBACK = "telegram_callback"
@@ -83,9 +84,11 @@
 ATTR_DATE = "date"
 ATTR_DISABLE_NOTIF = "disable_notification"
 ATTR_DISABLE_WEB_PREV = "disable_web_page_preview"
+ATTR_DIRECTORY_PATH = "directory_path"
 ATTR_EDITED_MSG = "edited_message"
 ATTR_FILE = "file"
 ATTR_FILE_ID = "file_id"
+ATTR_FILE_PATH = "file_path"
 ATTR_FILE_MIME_TYPE = "file_mime_type"
 ATTR_FILE_NAME = "file_name"
 ATTR_FILE_SIZE = "file_size"

homeassistant/components/telegram_bot/icons.json

@@ -13,6 +13,9 @@
     "delete_message": {
       "service": "mdi:delete"
     },
+    "download_file": {
+      "service": "mdi:paperclip"
+    },
     "edit_caption": {
       "service": "mdi:pencil"
     },

homeassistant/components/telegram_bot/services.yaml

@@ -911,3 +911,25 @@ set_message_reaction:
       required: false
       selector:
         boolean:
+
+download_file:
+  fields:
+    config_entry_id:
+      selector:
+        config_entry:
+          integration: telegram_bot
+    file_id:
+      required: true
+      example: "ABCD1234Efgh5678Ijkl90mnopQRStuvwx"
+      selector:
+        text:
+    directory_path:
+      required: false
+      default: "/config/telegram_bot"
+      selector:
+        text:
+    file_name:
+      required: false
+      example: "my_downloaded_file"
+      selector:
+        text:

homeassistant/components/telegram_bot/strings.json

@@ -324,6 +324,29 @@
       },
       "name": "Delete message"
     },
+    "download_file": {
+      "description": "Download the file to a local path.",
+      "fields": {
+        "config_entry_id": {
+          "description": "The config entry representing the Telegram bot to get the file.",
+          "name": "[%key:component::telegram_bot::services::send_message::fields::config_entry_id::name%]"
+        },
+        "directory_path": {
+          "description": "Local directory path to save the file to. Defaults to the 'telegram_bot' directory within your Home Assistant configuration directory.",
+          "example": "/config/telegram_bot",
+          "name": "Directory path"
+        },
+        "file_id": {
+          "description": "ID of the file to get.",
+          "name": "File ID"
+        },
+        "file_name": {
+          "description": "Name to save the file as. If not provided, the original file name will be used.",
+          "name": "File name"
+        }
+      },
+      "name": "Download file"
+    },
     "edit_caption": {
       "description": "Edits the caption of a previously sent message.",
       "fields": {