Repository: mandarons/icloud-drive-docker (python)

Issue:
[FEATURE] Telegram Notifications
**Use case**
I would love to see the option to receive the messages that are sent via mail alternatively via telegram.

**Describe the solution you'd like**
Just put two env vars into the config files. ChatID and token
https://github.com/python-telegram-bot/python-telegram-bot


Pull request description:
[FEATURE] Telegram Notifications Fixes #182

Patch:
diff --git a/README.md b/README.md
index 319fbade..5c0909d7 100644
--- a/README.md
+++ b/README.md
@@ -80,6 +80,9 @@ app:
     retry_login_interval: 600
   # Drive destination
   root: "icloud"
+  telegram:
+    # bot_token: <your Telegram bot token>
+    # chat_id: <your Telegram user or chat ID>
   smtp:
     ## If you want to receive email notifications about expired/missing 2FA credentials then uncomment
     # email: "user@test.com"
diff --git a/config.yaml b/config.yaml
index 1f0d5dcc..ed63b600 100644
--- a/config.yaml
+++ b/config.yaml
@@ -11,6 +11,9 @@ app:
     retry_login_interval: 600
   # Drive destination
   root: "icloud"
+  telegram:
+  # bot_token: <your Telegram bot token>
+  # chat_id: <your Telegram user or chat ID>
   smtp:
   # If you want to receive email notifications about expired/missing 2FA credentials then uncomment
   # email: "sender@test.com"
diff --git a/src/config_parser.py b/src/config_parser.py
index e24a65ae..e6f89660 100644
--- a/src/config_parser.py
+++ b/src/config_parser.py
@@ -392,3 +392,29 @@ def get_photos_folder_format(config):
         fmt = get_config_value(config=config, config_path=config_path)
         LOGGER.info(f"Using format {fmt}.")
     return fmt
+
+
+def get_telegram_bot_token(config):
+    """Return telegram bot token from config."""
+    bot_token = None
+    config_path = ["app", "telegram", "bot_token"]
+    if not traverse_config_path(config=config, config_path=config_path):
+        LOGGER.warning(
+            f"Warning: bot_token is not found in {config_path_to_string(config_path)}."
+        )
+    else:
+        bot_token = get_config_value(config=config, config_path=config_path)
+    return bot_token
+
+
+def get_telegram_chat_id(config):
+    """Return telegram chat id from config."""
+    chat_id = None
+    config_path = ["app", "telegram", "chat_id"]
+    if not traverse_config_path(config=config, config_path=config_path):
+        LOGGER.warning(
+            f"Warning: chat_id is not found in {config_path_to_string(config_path)}."
+        )
+    else:
+        chat_id = get_config_value(config=config, config_path=config_path)
+    return chat_id
diff --git a/src/notify.py b/src/notify.py
index 9d9be81e..b203ab85 100644
--- a/src/notify.py
+++ b/src/notify.py
@@ -2,13 +2,58 @@
 import datetime
 import smtplib
 
+import requests
+
 from src import LOGGER, config_parser
 from src.email_message import EmailMessage as Message
 
 
+def notify_telegram(config, last_send=None, dry_run=False):
+    """Send telegram notification."""
+    sent_on = None
+    bot_token = config_parser.get_telegram_bot_token(config=config)
+    chat_id = config_parser.get_telegram_chat_id(config=config)
+
+    if last_send and last_send > datetime.datetime.now() - datetime.timedelta(hours=24):
+        LOGGER.info("Throttling telegram to once a day")
+        sent_on = last_send
+    elif bot_token and chat_id:
+        sent_on = datetime.datetime.now()
+        if not dry_run:
+            # Post message to telegram bot using API
+            if not post_message_to_telegram(
+                bot_token,
+                chat_id,
+                """Two-step authentication for iCloud Drive, Photos (Docker) is required.
+                Please login to your server and authenticate. Please run -
+                `docker exec -it icloud /bin/sh -c "icloud --username=<icloud-username>
+                --session-directory=/app/session_data"`.""",
+            ):
+                sent_on = None
+    else:
+        LOGGER.warning(
+            "Not sending 2FA notification because Telegram is not configured."
+        )
+    return sent_on
+
+
+def post_message_to_telegram(bot_token, chat_id, message):
+    """Post message to telegram bot using API."""
+    url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
+    params = {"chat_id": chat_id, "text": message}
+    response = requests.post(url, params=params)
+    if response.status_code == 200:
+        return True
+    else:
+        # Log error message
+        LOGGER.error(f"Failed to send telegram notification. Response: {response.text}")
+        return False
+
+
 def send(config, last_send=None, dry_run=False):
-    """Send email."""
+    """Send notifications."""
     sent_on = None
+    notify_telegram(config=config, last_send=last_send, dry_run=dry_run)
     email = config_parser.get_smtp_email(config=config)
     to_email = config_parser.get_smtp_to_email(config=config)
     host = config_parser.get_smtp_host(config=config)


Test patch:
diff --git a/tests/test_config_parser.py b/tests/test_config_parser.py
index 1b1d8a7b..397a1c6a 100644
--- a/tests/test_config_parser.py
+++ b/tests/test_config_parser.py
@@ -463,3 +463,29 @@ class TestConfigParser(unittest.TestCase):
             config_parser.get_photos_filters(config=config)["libraries"],
             config["photos"]["filters"]["libraries"],
         )
+
+    def test_get_telegram_bot_token(self):
+        """Test for telegram bot token."""
+        config = read_config(config_path=tests.CONFIG_PATH)
+        config["app"]["telegram"] = {"bot_token": "bot_token"}
+        self.assertEqual(
+            config["app"]["telegram"]["bot_token"],
+            config_parser.get_telegram_bot_token(config=config),
+        )
+
+    def test_get_telegram_bot_token_none_config(self):
+        """None config."""
+        self.assertIsNone(config_parser.get_telegram_bot_token(config=None))
+
+    def test_get_telegram_chat_id(self):
+        """Test for telegram chat id."""
+        config = read_config(config_path=tests.CONFIG_PATH)
+        config["app"]["telegram"] = {"chat_id": "chat_id"}
+        self.assertEqual(
+            config["app"]["telegram"]["chat_id"],
+            config_parser.get_telegram_chat_id(config=config),
+        )
+
+    def test_get_telegram_chat_id_none_config(self):
+        """None config."""
+        self.assertIsNone(config_parser.get_telegram_chat_id(config=None))
diff --git a/tests/test_notify.py b/tests/test_notify.py
index 8ceee2da..ffd87586 100644
--- a/tests/test_notify.py
+++ b/tests/test_notify.py
@@ -5,6 +5,7 @@ from unittest.mock import patch
 
 from src import config_parser, notify
 from src.email_message import EmailMessage as Message
+from src.notify import notify_telegram, post_message_to_telegram
 
 
 class TestNotify(unittest.TestCase):
@@ -20,7 +21,8 @@ class TestNotify(unittest.TestCase):
                     "host": "smtp.test.com",
                     "port": "587",
                     "password": "password",
-                }
+                },
+                "telegram": {"bot_token": "bot_token", "chat_id": "chat_id"},
             }
         }
 
@@ -121,3 +123,110 @@ class TestNotify(unittest.TestCase):
             # Verify that a failure doesn't return a send_on timestamp
             sent_on = notify.send(self.config)
             self.assertEqual(None, sent_on)
+
+    def test_notify_telegram_success(self):
+        """Test for successful notification."""
+        config = {
+            "app": {
+                "telegram": {"bot_token": "your-bot-token", "chat_id": "your-chat-id"}
+            }
+        }
+
+        with patch("src.notify.post_message_to_telegram") as post_message_mock:
+            notify_telegram(config, None, False)
+
+            # Verify that post_message_to_telegram is called with the correct arguments
+            post_message_mock.assert_called_once_with(
+                config["app"]["telegram"]["bot_token"],
+                config["app"]["telegram"]["chat_id"],
+                """Two-step authentication for iCloud Drive, Photos (Docker) is required.
+                Please login to your server and authenticate. Please run -
+                `docker exec -it icloud /bin/sh -c "icloud --username=<icloud-username>
+                --session-directory=/app/session_data"`.""",
+            )
+
+    def test_notify_telegram_fail(self):
+        """Test for failed notification."""
+        config = {
+            "app": {
+                "telegram": {"bot_token": "your-bot-token", "chat_id": "your-chat-id"}
+            }
+        }
+
+        with patch("src.notify.post_message_to_telegram") as post_message_mock:
+            post_message_mock.return_value = False
+            notify_telegram(config, None, False)
+
+            # Verify that post_message_to_telegram is called with the correct arguments
+            post_message_mock.assert_called_once_with(
+                config["app"]["telegram"]["bot_token"],
+                config["app"]["telegram"]["chat_id"],
+                """Two-step authentication for iCloud Drive, Photos (Docker) is required.
+                Please login to your server and authenticate. Please run -
+                `docker exec -it icloud /bin/sh -c "icloud --username=<icloud-username>
+                --session-directory=/app/session_data"`.""",
+            )
+
+    def test_notify_telegram_throttling(self):
+        """Test for throttled notification."""
+        config = {
+            "telegram": {"bot_token": "your-bot-token", "chat_id": "your-chat-id"}
+        }
+        last_send = datetime.datetime.now() - datetime.timedelta(hours=24)
+        dry_run = False
+
+        with patch("src.notify.post_message_to_telegram") as post_message_mock:
+            notify_telegram(config, last_send, dry_run)
+
+            # Verify that post_message_to_telegram is not called when throttled
+            post_message_mock.assert_not_called()
+
+    def test_notify_telegram_dry_run(self):
+        """Test for dry run mode."""
+        config = {
+            "telegram": {"bot_token": "your-bot-token", "chat_id": "your-chat-id"}
+        }
+        last_send = datetime.datetime.now()
+        dry_run = True
+
+        with patch("src.notify.post_message_to_telegram") as post_message_mock:
+            notify_telegram(config, last_send, dry_run)
+
+            # Verify that post_message_to_telegram is not called in dry run mode
+            post_message_mock.assert_not_called()
+
+    def test_notify_telegram_no_config(self):
+        """Test for missing telegram configuration."""
+        config = {}
+        last_send = None
+        dry_run = False
+
+        with patch("src.notify.post_message_to_telegram") as post_message_mock:
+            notify_telegram(config, last_send, dry_run)
+
+            # Verify that post_message_to_telegram is not called when telegram configuration is missing
+            post_message_mock.assert_not_called()
+
+    def test_post_message_to_telegram(self):
+        """Test for successful post."""
+        with patch("requests.post") as post_mock:
+            post_mock.return_value.status_code = 200
+            post_message_to_telegram("bot_token", "chat_id", "message")
+
+            # Verify that post is called with the correct arguments
+            post_mock.assert_called_once_with(
+                "https://api.telegram.org/botbot_token/sendMessage",
+                params={"chat_id": "chat_id", "text": "message"},
+            )
+
+    def test_post_message_to_telegram_fail(self):
+        """Test for failed post."""
+        with patch("requests.post") as post_mock:
+            post_mock.return_value.status_code = 400
+            post_message_to_telegram("bot_token", "chat_id", "message")
+
+            # Verify that post is called with the correct arguments
+            post_mock.assert_called_once_with(
+                "https://api.telegram.org/botbot_token/sendMessage",
+                params={"chat_id": "chat_id", "text": "message"},
+            )