init commit

.gitignore

@@ -0,0 +1,2 @@
+.venv
+__pycache__

LICENCE

@@ -0,0 +1,19 @@
+Copyright (c) 2018 The Python Packaging Authority
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,12 @@
+# Logging custom handler for telegram notification
+## Description
+Custom logging handler. Used to send CRITICAL level messages to my telegram chat.
+
+## Installation
+``pip install git+https://git.lesha.spb.ru/alex/logging_telegram_handler.git``
+
+## Contacts
+- **e-mail**: lesha.spb@gmail.com
+- **telegram**: https://t.me/lesha_spb
+
+

pyproject.toml

@@ -0,0 +1,31 @@
+[build-system]
+requires = ["setuptools>=75.3.0"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "LoggingTelegramHandler"
+description = "Send messages in telegram chat"
+version = "1.0.0"
+authors = [
+    { name = "Aleksei Zosimov", email = "lesha.spb@gmail.com" }
+]
+readme = "README.md"
+
+requires-python = ">=3.12"
+dependencies = [
+    "requests"
+]
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[project.urls]
+Homepage = "https://git.lesha.spb.ru/alex/logging_telegram_handler"
+Documentation = "https://git.lesha.spb.ru/alex/logging_telegram_handler"
+Repository = "https://git.lesha.spb.ru/alex/logging_telegram_handler.git"
+
+
+[tool.pytest.ini_options]
+addopts = [
+    "--import-mode=importlib",
+]

requirements.txt

@@ -0,0 +1 @@
+telegram-send @ git+https://bitbucket.org/zosimovaa/telegram_send.git

src/logging_telegram_handler.egg-info/PKG-INFO

@@ -0,0 +1,28 @@
+Metadata-Version: 2.4
+Name: logging_telegram_handler
+Version: 0.0.2
+Summary: Logging telegram handler for important notification.
+Home-page: https://bitbucket.org/zosimovaa/logging_telegram_handler/src/master/
+Author: Aleksei Zosimov
+Author-email: lesha.spb@gmail.com
+Project-URL: Bug Tracker, https://bitbucket.org/zosimovaa/logging_telegram_handler/src/master/
+Classifier: Programming Language :: Python :: 3
+Classifier: License :: OSI Approved :: MIT License
+Classifier: Operating System :: OS Independent
+Requires-Python: >=3.6
+Description-Content-Type: text/markdown
+License-File: LICENCE
+Dynamic: license-file
+
+# Logging custom handler for telegram notification
+## Description
+Custom logging handler. Used to send CRITICAL level messages to my telegram chat.
+
+## Installation
+``pip install git+https://zosimovaa@bitbucket.org/zosimovaa/logging_telegram_handler.git``
+
+## Contacts
+- **e-mail**: lesha.spb@gmail.com
+- **telegram**: https://t.me/lesha_spb
+
+

src/logging_telegram_handler.egg-info/SOURCES.txt

@@ -0,0 +1,12 @@
+LICENCE
+README.md
+pyproject.toml
+setup.cfg
+src/logging_telegram_handler/__init__.py
+src/logging_telegram_handler/handler.py
+src/logging_telegram_handler/spam_filter.py
+src/logging_telegram_handler/telegram.py
+src/logging_telegram_handler.egg-info/PKG-INFO
+src/logging_telegram_handler.egg-info/SOURCES.txt
+src/logging_telegram_handler.egg-info/dependency_links.txt
+src/logging_telegram_handler.egg-info/top_level.txt

src/logging_telegram_handler.egg-info/dependency_links.txt

@@ -0,0 +1 @@
+

src/logging_telegram_handler.egg-info/top_level.txt

@@ -0,0 +1 @@
+logging_telegram_handler

src/logging_telegram_handler/__init__.py

@@ -0,0 +1 @@
+from .handler import TelegramHandler

src/logging_telegram_handler/handler.py

@@ -0,0 +1,20 @@
+import os
+import logging
+from logging import StreamHandler
+from .telegram import TelegramSend
+
+logger = logging.getLogger(__name__)
+
+
+class TelegramHandler(StreamHandler):
+    def __init__(self, alias="Test", chat_id='211945135'):
+        StreamHandler.__init__(self)
+        api_key = os.getenv("TELEGRAM_API_KEY")
+        self.telegram = TelegramSend(alias, chat_id=chat_id, api_key=api_key)
+        self.telegram.start()
+
+    def emit(self, msg):
+        self.telegram.send(self.format(msg))
+
+    def stop(self):
+        self.telegram.stop()

src/logging_telegram_handler/spam_filter.py

@@ -0,0 +1,51 @@
+"""
+The spam filter allows you to limit the distribution of messages in case of
+a large number of messages in a short period of time.
+"""
+
+import time
+import logging
+import collections
+
+logger = logging.getLogger(__name__)
+
+
+class SpamFilter:
+    SPAM_MESSAGE = "Spam detected, take a brake :)"
+
+    def __init__(self, threshold=10, control_period=120, silence_period=1800):
+        self.threshold = threshold
+        self.control_period = control_period
+        self.silence_period = silence_period
+        self.spam_buffer = collections.deque(maxlen=self.threshold)
+        self.blocked = False
+
+    def check_block(self):
+        now = time.time()
+
+        # Not initialized yet
+        if len(self.spam_buffer) < self.threshold:
+            self.spam_buffer.append(now)
+            logger.debug("Not a spam. Current len buffer: {}".format(len(self.spam_buffer)))
+
+        # unblocked
+        elif not self.blocked:
+            self.spam_buffer.append(now)
+            timeout = self.spam_buffer[-1] - self.spam_buffer[0]
+            logger.debug("check_block() - not blocked. timeout: {}".format(timeout))
+            if timeout < self.control_period:
+                message = self.SPAM_MESSAGE
+                self.blocked = True
+                logger.warning("Spam detected")
+
+        # blocked
+        else:
+            timeout = now - self.spam_buffer[-1]
+            logger.debug("check_block() - blocked. timeout: {}".format(timeout))
+            if timeout > self.silence_period:
+                self.blocked = False
+                self.spam_buffer = collections.deque(maxlen=self.threshold)
+                self.spam_buffer.append(now)
+                logger.debug("Spam filter unblock")
+
+        return self.blocked

src/logging_telegram_handler/telegram.py

@@ -0,0 +1,72 @@
+"""
+The TelegramJustSend class provides sending messages to telegram without guarantees of delivery.
+"""
+import os
+import time
+import queue
+import logging
+import datetime
+import requests
+import threading
+
+from .spam_filter import SpamFilter
+
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.ERROR)
+
+
+class TelegramSend(threading.Thread):
+    SILENT_TIME_SINCE = 21
+    SILENT_TIME_BEFORE = 9
+    MAX_BUFFER_SIZE = 20
+    TIMEOUT = 1
+    URL = "https://api.telegram.org/bot{0}/sendMessage?chat_id={1}&text={2}&disable_notification={3}"
+
+    def __init__(self, alias, chat_id, api_key):
+        threading.Thread.__init__(self, daemon=True)
+        self.alias = alias
+        self.chat_id = chat_id
+        self.api_key = api_key
+        self.spam_filter = SpamFilter()
+        self.halt = threading.Event()
+        self.queue = queue.Queue(maxsize=self.MAX_BUFFER_SIZE)
+
+    def __send(self, message):
+        blocked = self.spam_filter.check_block()
+        if not blocked:
+            silent = self.check_silent()
+            message = "[{0}]: {1}".format(self.alias, message)
+            url = self.URL.format(self.api_key, self.chat_id, message, silent)
+            resp = requests.get(url).json()
+            logger.debug(resp)
+        else:
+            self.queue = queue.Queue(maxsize=self.MAX_BUFFER_SIZE)
+
+    def run(self):
+        pid = os.getpid()
+        logger.error(f"Telegram send process started with pid {pid}")
+        while True:
+            try:
+                message = self.queue.get()
+                logger.debug("got message for send: {0}".format(message))
+                self.__send(message)
+            except Exception as e:
+                logger.error(e)
+            finally:
+                time.sleep(self.TIMEOUT)
+
+    def stop(self):
+        self.halt.set()
+        self.join()
+        logger.warning('Stopped')
+
+    def send(self, message):
+        logger.debug("send: {}".format(message))
+        self.queue.put_nowait(message)
+
+    def check_silent(self):
+        now = datetime.datetime.now()
+        silent = True if (now.hour > self.SILENT_TIME_SINCE) or (now.hour < self.SILENT_TIME_BEFORE) else False
+        logger.debug("check_silent(): silent: {0}, hour: {1}".format(silent, now.hour))
+        return silent

tests/logger_test.py

@@ -0,0 +1,35 @@
+import unittest
+import time
+
+import logging
+
+from logging_telegram_handler import TelegramHandler
+
+
+class MyTestCase(unittest.TestCase):
+    def test_something(self):
+
+        chat_id = "211945135"
+        alias = "[TEST logging_telegram_handler]"
+
+        logging.basicConfig(level=logging.WARNING)
+
+        logger = logging.getLogger(__name__)
+
+        telegram_handler = TelegramHandler(chat_id, alias)
+        telegram_handler.setLevel(logging.WARNING)
+        logger.addHandler(telegram_handler)
+
+        logger.critical("critical")
+        logger.error("error")
+        logger.warning("warning")
+        logger.info("info")
+        logger.debug("debug")
+
+        time.sleep(2)
+
+        self.assertEqual(True, True)  # add assertion here
+
+
+if __name__ == '__main__':
+    unittest.main()