plugin.telegram.TelegramSender Class Reference

Public Member Functions
	__init__ (self, bot_token, chat_ids, max_retries=None, initial_delay=None, max_delay=None, parse_mode=None)
	send_message (self, text)
	send_location (self, latitude, longitude)

Data Fields
	bot_token
	chat_ids
	max_retries
	initial_delay
	max_delay
	parse_mode
	msg_queue

Protected Member Functions
	_worker_loop (self)
	_send_to_telegram (self, msg_type, chat_id, content)

Protected Attributes
	_worker

Constructor & Destructor Documentation

__init__()

plugin.telegram.TelegramSender.__init__	(		self,
			bot_token,
			chat_ids,
			max_retries = None,
			initial_delay = None,
			max_delay = None,
			parse_mode = None
	)

    def __init__(self, bot_token, chat_ids, max_retries=None, initial_delay=None, max_delay=None, parse_mode=None):
        self.bot_token = bot_token
        self.chat_ids = chat_ids
        self.max_retries = max_retries if max_retries is not None else 5
        self.initial_delay = initial_delay if initial_delay is not None else 2
        self.max_delay = max_delay if max_delay is not None else 300
        self.parse_mode = parse_mode
        self.msg_queue = queue.Queue()
        self._worker = threading.Thread(target=self._worker_loop, daemon=True)
        self._worker.start()
 

Member Function Documentation

send_message()

plugin.telegram.TelegramSender.send_message	(		self,
			text
	)

    def send_message(self, text):
        for chat_id in self.chat_ids:
            self.msg_queue.put(("text", chat_id, text, 0))  # retry_count = 0
 

send_location()

plugin.telegram.TelegramSender.send_location	(		self,
			latitude,
			longitude
	)

    def send_location(self, latitude, longitude):
        for chat_id in self.chat_ids:
            self.msg_queue.put(("location", chat_id, {"latitude": latitude, "longitude": longitude}, 0))
 

_worker_loop()

plugin.telegram.TelegramSender._worker_loop (   self )

protected

    def _worker_loop(self):
        delay = self.initial_delay
 
        while True:
            try:
                msg_type, chat_id, content, retry_count = self.msg_queue.get()
 
                success, permanent_failure, custom_delay = self._send_to_telegram(msg_type, chat_id, content)
 
                if success:
                    delay = self.initial_delay
 
                elif permanent_failure:
                    logger.error("Permanenter Fehler – Nachricht wird verworfen.")
 
                elif retry_count >= self.max_retries:
                    logger.error("Maximale Wiederholungsanzahl erreicht – Nachricht wird verworfen.")
 
                else:
                    logger.warning(f"Erneutes Einreihen der Nachricht (Versuch {retry_count + 1}).")
                    self.msg_queue.put((msg_type, chat_id, content, retry_count + 1))
 
                    # use the Telegram-provided value (retry_after) if available
                    wait_time = custom_delay if custom_delay is not None else delay
                    time.sleep(wait_time)
 
                    # increase delay for the next attempt (exponential backoff)
                    delay = min(delay * 2, self.max_delay)
 
            except Exception as e:
                logger.exception(f"Fehler im Telegram-Worker: {e}")
                time.sleep(5)
 

_send_to_telegram()

plugin.telegram.TelegramSender._send_to_telegram (   self,   msg_type,   chat_id,   content  )

protected

    def _send_to_telegram(self, msg_type, chat_id, content):
        if msg_type == "text":
            url = f"https://api.telegram.org/bot{self.bot_token}/sendMessage"
            payload = {
                'chat_id': chat_id,
                'text': content,
            }
            if self.parse_mode:
                payload['parse_mode'] = self.parse_mode
        elif msg_type == "location":
            url = f"https://api.telegram.org/bot{self.bot_token}/sendLocation"
            payload = {
                'chat_id': chat_id,
                **content
            }
        else:
            logger.error("Unbekannter Nachrichtentyp.")
            return False, True, None  # unknown message type = permanent failure
 
        try:
            custom_delay = None  # standardvalue for return, except in case of 429
 
            response = requests.post(url, data=payload, timeout=10)
 
            if response.status_code == 429:
                custom_delay = response.json().get("parameters", {}).get("retry_after", 5)
                logger.warning(f"Rate Limit erreicht – warte {custom_delay} Sekunden.")
                return False, False, custom_delay  # Telegram gives exact wait time
 
            if response.status_code == 400:
                logger.error("Ungültige Parameter – Nachricht wird nicht erneut gesendet.")
                return False, True, custom_delay  # permanent failure
 
            if response.status_code == 401:
                logger.critical("Ungültiger Bot-Token – bitte prüfen!")
                return False, True, custom_delay  # permanent failure
 
            response.raise_for_status()
            logger.info(f"Erfolgreich gesendet an Chat-ID {chat_id}")
            return True, False, custom_delay
 
        except requests.RequestException as e:
            logger.warning(f"Fehler beim Senden an Telegram (Chat-ID {chat_id}): {e}")
            return False, False, custom_delay
 
 
# ===========================
# BoswatchPlugin-Class
# ===========================
 
 

Field Documentation

bot_token

plugin.telegram.TelegramSender.bot_token

chat_ids

plugin.telegram.TelegramSender.chat_ids

max_retries

plugin.telegram.TelegramSender.max_retries

initial_delay

plugin.telegram.TelegramSender.initial_delay

max_delay

plugin.telegram.TelegramSender.max_delay

parse_mode

plugin.telegram.TelegramSender.parse_mode

msg_queue

plugin.telegram.TelegramSender.msg_queue

_worker

plugin.telegram.TelegramSender._worker

protected