feat: add update notify

media_downloader.py

@@ -646,7 +646,7 @@ def main():
         for task in tasks:
             task.cancel()
         logger.info(_t("Stopped!"))
-        check_for_updates()
+        check_for_updates(app.proxy)
         logger.info(f"{_t('update config')}......")
         app.update_config()
         logger.success(

module/bot.py

@@ -38,6 +38,7 @@
 )
 from utils.format import replace_date_time, validate_title
 from utils.meta_data import MetaData
+from utils.updates import get_latest_release
 
 # pylint: disable = C0301, R0902
 
@@ -343,9 +344,33 @@ async def send_help_str(client: pyrogram.Client, chat_id):
         The help string includes information about the Telegram Media Downloader bot,
         its version, and the available commands.
     """
+
+    update_keyboard = InlineKeyboardMarkup(
+        [
+            [
+                InlineKeyboardButton(
+                    "Github",
+                    url="https://github.com/tangyoha/telegram_media_downloader/releases",
+                ),
+                InlineKeyboardButton(
+                    "Join us", url="https://t.me/TeegramMediaDownload"
+                ),
+            ]
+        ]
+    )
+
+    latest_release = get_latest_release(_bot.app.proxy)
+
+    latest_release_str = (
+        f"{_t('New Version')}: [{latest_release['name']}]({latest_release['html_url']})\n"
+        if latest_release
+        else ""
+    )
+
     msg = (
         f"`\n🤖 {_t('Telegram Media Downloader')}\n"
-        f"🌐 {_t('Version')}: {utils.__version__}`\n\n"
+        f"🌐 {_t('Version')}: {utils.__version__}`\n"
+        f"{latest_release_str}\n"
         f"{_t('Available commands:')}\n"
         f"/help - {_t('Show available commands')}\n"
         f"/get_info - {_t('Get group and user info from message link')}\n"
@@ -360,7 +385,7 @@ async def send_help_str(client: pyrogram.Client, chat_id):
         f"`[` `]` {_t('means optional, not required')}\n"
     )
 
-    await client.send_message(chat_id, msg)
+    await client.send_message(chat_id, msg, reply_markup=update_keyboard)
 
 
 async def help_command(client: pyrogram.Client, message: pyrogram.types.Message):

module/language.py

@@ -251,6 +251,12 @@ def set_language(language: Language):
     ],
     "Upload Progresses": ["上传进度", "Прогресс загрузки", "Прогрес завантаження"],
     "Download Progresses": ["下载进度", "Прогресс скачивания", "Прогрес завантаження"],
+    "New Version": ["新版本", "новая версия", "нова версія"],
+    "Stop bot download or forward": [
+        "停止机器人下载或转发",
+        "Остановить загрузку или пересылку ботом",
+        "Зупинити завантаження або пересилання ботом",
+    ],
 }
 
 

tests/test_media_downloader.py

@@ -345,7 +345,7 @@ async def edit_message_text(self, *args, **kwargs):
         return True
 
 
-def check_for_updates():
+def check_for_updates(_: dict = None):
     pass
 
 

tests/utils/test_updates.py

@@ -1,11 +1,12 @@
 """Unittest module for update checker."""
+import json
 import sys
 import unittest
 
 import mock
 
 sys.path.append("..")  # Adds higher directory to python modules path.
-from utils.updates import check_for_updates
+from utils.updates import check_for_updates, get_latest_release
 
 
 class FakeHTTPSConnection:
@@ -30,37 +31,117 @@ def read(self):
             return b"{error}"
 
 
-class UpdatesTestCase(unittest.TestCase):
-    @mock.patch(
-        "utils.updates.http.client.HTTPSConnection",
-        new=mock.MagicMock(return_value=FakeHTTPSConnection(200)),
-    )
-    @mock.patch("utils.updates.__version__", new="0.0.1")
-    @mock.patch("utils.updates.Console")
-    @mock.patch("utils.updates.Markdown")
-    def test_update(self, mock_markdown, mock_console):
-        check_for_updates()
-        name: str = "v0.0.0 2022-03-02"
-        html_url: str = (
-            "https://github.com/tangyoha/telegram_media_downloader/releases/tag/v0.0.0"
+class MocResponse:
+    def __init__(self, text: str):
+        self.text = text
+
+
+def new_request_get(*args, **kwargs):
+    return MocResponse('{"tag_name":"v0.0.0"}')
+
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+from utils import __version__
+from utils.updates import check_for_updates, get_latest_release
+
+
+class TestUpdates(unittest.TestCase):
+    @patch("requests.get")
+    def test_get_latest_release(self, mock_get):
+        # Mock the response from requests.get
+        mock_response = MagicMock()
+        mock_response.text = json.dumps(
+            {
+                "name": "v0.0.0 2022-03-02",
+                "tag_name": "v0.0.0",
+                "html_url": "https://github.com/tangyoha/telegram_media_downloader/releases/tag/v0.0.0",
+            }
         )
-        expected_message: str = (
-            f"## New version of Telegram-Media-Downloader is available - {name}\n"
-            "You are using an outdated version v0.0.1 please pull in the changes using `git pull` or download the latest release.\n\n"
-            f"Find more details about the latest release here - {html_url}"
+        mock_get.return_value = mock_response
+
+        # Call the function with a test proxy_config
+        proxy_config = {
+            "scheme": "http",
+            "hostname": "localhost",
+            "port": "8080",
+            "username": "user",
+            "password": "pass",
+        }
+        result = get_latest_release(proxy_config)
+
+        # Check the result
+        self.assertEqual(result["name"], "v0.0.0 2022-03-02")
+        self.assertEqual(result["tag_name"], "v0.0.0")
+        self.assertEqual(
+            result["html_url"],
+            "https://github.com/tangyoha/telegram_media_downloader/releases/tag/v0.0.0",
         )
-        mock_markdown.assert_called_with(expected_message)
-        mock_console.return_value.print.assert_called_once()
 
-    @mock.patch(
-        "utils.updates.http.client.HTTPSConnection",
-        new=mock.MagicMock(return_value=FakeHTTPSConnection(500)),
-    )
-    @mock.patch("utils.updates.Console")
-    def test_exception(self, mock_console):
-        check_for_updates()
-        exception_message: str = (
-            "Following error occurred when checking for updates\n"
-            "<class 'json.decoder.JSONDecodeError'>, Expecting property name enclosed in double quotes: line 1 column 2 (char 1)"
+    @patch("requests.get")
+    def test_get_latest_release_same_version(self, mock_get):
+        # Mock the response from requests.get
+        mock_response = MagicMock()
+        mock_response.text = json.dumps(
+            {
+                "name": f"v{__version__} 2022-03-02",
+                "tag_name": f"v{__version__}",
+                "html_url": "https://github.com/tangyoha/telegram_media_downloader/releases/tag/v0.0.0",
+            }
         )
-        mock_console.return_value.log.assert_called_with(exception_message)
+        mock_get.return_value = mock_response
+
+        # Call the function with a test proxy_config
+        proxy_config = {
+            "scheme": "http",
+            "hostname": "localhost",
+            "port": "8080",
+            "username": "user",
+            "password": "pass",
+        }
+        result = get_latest_release(proxy_config)
+
+        # Check the result
+        self.assertEqual(result, {})
+
+    @patch("requests.get")
+    def test_get_latest_release_exception(self, mock_get):
+        # Mock the response from requests.get to raise an exception
+        mock_get.side_effect = Exception("Test exception")
+
+        # Call the function with a test proxy_config
+        proxy_config = {
+            "scheme": "http",
+            "hostname": "localhost",
+            "port": "8080",
+            "username": "user",
+            "password": "pass",
+        }
+        result = get_latest_release(proxy_config)
+
+        # Check the result
+        self.assertEqual(result, {})
+
+    @patch("utils.updates.get_latest_release")
+    @patch("utils.updates.Console")
+    def test_check_for_updates(self, mock_console, mock_get_latest_release):
+        # Mock the response from get_latest_release
+        mock_get_latest_release.return_value = {
+            "name": "v0.0.0 2022-03-02",
+            "tag_name": "v0.0.0",
+            "html_url": "https://github.com/tangyoha/telegram_media_downloader/releases/tag/v0.0.0",
+        }
+
+        # Call the function with a test proxy_config
+        proxy_config = {
+            "scheme": "http",
+            "hostname": "localhost",
+            "port": "8080",
+            "username": "user",
+            "password": "pass",
+        }
+        check_for_updates(proxy_config)
+
+        # Check the console output
+        mock_console.return_value.print.assert_called_once()

utils/updates.py

@@ -1,41 +1,77 @@
 """Utility module to check for new release of telegram-media-downloader"""
-import http.client
 import json
 
+import requests  # type: ignore
+from loguru import logger
 from rich.console import Console
 from rich.markdown import Markdown
 
 from . import __version__
 
 
 # pylint: disable = C0301
-def check_for_updates() -> None:
-    """Checks for new releases.
+def get_latest_release(proxy_config: dict = None) -> dict:
+    """
+    Get the latest release information.
 
-    Using Github API checks for new release and prints information of new release if available.
+    :param proxy_config: A dictionary containing proxy configuration settings (default: {}).
+    :type proxy_config: dict
+    :return: A dictionary containing the latest release information.
+    :rtype: dict
     """
-    console = Console()
-    try:
-        headers: dict = {
-            "Content-Type": "application/json",
-            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.102 Safari/537.36",
+    headers: dict = {
+        "Content-Type": "application/json",
+        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.102 Safari/537.36",
+    }
+
+    if proxy_config:
+        scheme = proxy_config.get("scheme", "")
+        hostname = proxy_config.get("hostname", "")
+        port = proxy_config.get("port", "")
+        username = proxy_config.get("username")
+        password = proxy_config.get("password")
+
+    proxies = {}
+    if proxy_config:
+        proxies = {
+            "http": f"{scheme}://{hostname}:{port}",
+            "https": f"{scheme}://{hostname}:{port}",
         }
-        conn = http.client.HTTPSConnection("api.github.com")
-        conn.request(
-            method="GET",
-            url="/repos/tangyoha/telegram_media_downloader/releases/latest",
+
+        if username and password:
+            proxies["http"] = f"{scheme}://{username}:{password}@{hostname}:{port}"
+            proxies["https"] = f"{scheme}://{username}:{password}@{hostname}:{port}"
+    try:
+        response = requests.get(
+            url="https://api.github.com/repos/tangyoha/telegram_media_downloader/releases/latest",
             headers=headers,
+            proxies=proxies,
+            timeout=60,
         )
-        res = conn.getresponse()
-        latest_release: dict = json.loads(res.read().decode("utf-8"))
-        if f"v{__version__}" != latest_release["tag_name"]:
-            update_message: str = (
-                f"## New version of Telegram-Media-Downloader is available - {latest_release['name']}\n"
-                f"You are using an outdated version v{__version__} please pull in the changes using `git pull` or download the latest release.\n\n"
-                f"Find more details about the latest release here - {latest_release['html_url']}"
-            )
-            console.print(Markdown(update_message))
+
     except Exception as e:
-        console.log(
-            f"Following error occurred when checking for updates\n{e.__class__}, {e}"
+        logger.warning(f"{e}")
+        return {}
+
+    latest_release: dict = json.loads(response.text)
+
+    if f"v{__version__}" != latest_release["tag_name"]:
+        return latest_release
+
+    return {}
+
+
+def check_for_updates(proxy_config: dict = None):
+    """Checks for new releases.
+
+    Using Github API checks for new release and prints information of new release if available.
+    """
+    console = Console()
+    latest_release = get_latest_release(proxy_config)
+    if latest_release:
+        update_message: str = (
+            f"## New version of Telegram-Media-Downloader is available - {latest_release['name']}\n"
+            f"You are using an outdated version v{__version__} please pull in the changes using `git pull` or download the latest release.\n\n"
+            f"Find more details about the latest release here - {latest_release['html_url']}"
         )
+        console.print(Markdown(update_message))