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)
	shutdown (self)

Data Fields
	bot_token
	chat_ids
	max_retries
	initial_delay
	max_delay
	msg_queue
	parse_mode

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

Static Protected Member Functions
	_escape_text (text, parse_mode)
	_shorten_text (text, parse_mode)
	Shorten an escaped text to the telegram limit of 4096 characters without cutting an escape sequence or an HTML entity.

Protected Attributes
	_stop_event
	_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._stop_event = threading.Event()
        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.msg_queue = queue.Queue(maxsize=100)  # max 100 messages
        self.parse_mode = parse_mode
 
        # Start Worker Thread
        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):
        clean_text = self._escape_text(text, self.parse_mode)
        for chat_id in self.chat_ids:
            try:
                self.msg_queue.put(("text", chat_id, clean_text, 0), block=False)
            except queue.Full:
                logger.error(f"Message queue full for chat_id {chat_id}, message discarded: {clean_text[:50]}...")
 

send_location()

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

    def send_location(self, latitude, longitude):
        try:
            # Use validated float values, not original strings
            lat = float(latitude)
            lon = float(longitude)
            # check Telegram API Limits
            if not (-90 <= lat <= 90) or not (-180 <= lon <= 180):
                logger.error(f"Invalid coordinates: lat={lat}, lon={lon} (out of range)")
                return
        except (TypeError, ValueError) as e:
            logger.error(f"Invalid coordinate format: {e}, location skipped")
            return
 
        for chat_id in self.chat_ids:
            try:
                self.msg_queue.put(("location", chat_id, {"latitude": lat, "longitude": lon}, 0), block=False)
            except queue.Full:
                logger.error(f"Location queue full for chat_id {chat_id}, location discarded: {lat}, {lon}")
 

_escape_text()

plugin.telegram.TelegramSender._escape_text (   text,   parse_mode  )

staticprotected

    def _escape_text(text, parse_mode):
        if not text:
            return ""
        if parse_mode == "HTML":
            protected_tags = {}
            tag_pattern = r'<(/?)(\w+)>'
            allowed = ["b", "strong", "i", "em", "code", "pre", "u", "s"]
 
            def protect_tag(match):
                tag_full = match.group(0)
                tag_name = match.group(2)
                if tag_name in allowed:
                    placeholder = f"__TAG_{len(protected_tags)}__"
                    protected_tags[placeholder] = tag_full
                    return placeholder
                return tag_full
 
            text = re.sub(tag_pattern, protect_tag, text)
            text = text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
            for placeholder, original_tag in protected_tags.items():
                text = text.replace(placeholder, original_tag)
        elif parse_mode == "MarkdownV2":
            # Escape all MarkdownV2 special characters (Telegram API requirement)
            # See: https://core.telegram.org/bots/api#markdownv2-style
            escape_chars = r'_*[]()~`>#+-=|{}.!'
            text = "".join(['\\' + char if char in escape_chars else char for char in text])
        return TelegramSender._shorten_text(text, parse_mode)
 

_shorten_text()

plugin.telegram.TelegramSender._shorten_text (   text,   parse_mode  )

staticprotected

Shorten an escaped text to the telegram limit of 4096 characters without cutting an escape sequence or an HTML entity.

    def _shorten_text(text, parse_mode):
        r"""!Shorten an escaped text to the telegram limit of 4096 characters
        without cutting an escape sequence or an HTML entity"""
        if len(text) <= 4096:
            return text
        suffix = "\\[\\.\\.\\.\\]" if parse_mode == "MarkdownV2" else "[...]"
        text = text[:4096 - len(suffix)]
        if parse_mode == "MarkdownV2":
            # an odd number of backslashes at the end would escape the suffix
            if (len(text) - len(text.rstrip("\\"))) % 2:
                text = text[:-1]
        elif parse_mode == "HTML":
            # remove a cut HTML entity at the end
            lastAmp = text.rfind("&")
            if lastAmp != -1 and ";" not in text[lastAmp:]:
                text = text[:lastAmp]
        return text + suffix
 

_worker_loop()

plugin.telegram.TelegramSender._worker_loop (   self )

protected

    def _worker_loop(self):
        delay = self.initial_delay
        while not self._stop_event.is_set():
            try:
                msg_type, chat_id, content, retry_count = self.msg_queue.get(timeout=1)
                success, permanent_failure, custom_delay, error_details = self._send_to_telegram(msg_type, chat_id, content)
                if success:
                    delay = self.initial_delay
                elif permanent_failure or retry_count >= self.max_retries:
                    logger.warning(f"Discarding message for {chat_id}: {error_details}")
                else:
                    wait_time = custom_delay if custom_delay is not None else delay
                    time.sleep(wait_time)
                    self.msg_queue.put((msg_type, chat_id, content, retry_count + 1))
                    delay = min(delay * 2, self.max_delay)
            except queue.Empty:
                continue
            except Exception as e:
                logger.error(f"Error in 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):
        url = f"https://api.telegram.org/bot{self.bot_token}/"
        url += "sendMessage" if msg_type == "text" else "sendLocation"
        payload = {'chat_id': chat_id}
 
        if msg_type == "text":
            payload['text'] = content
            if self.parse_mode:
                payload['parse_mode'] = self.parse_mode
        else:
            payload.update(content)
 
        try:
            response = requests.post(url, data=payload, timeout=10)
            if response.status_code == 200:
                logger.info(f"Successfully sent to Chat-ID {chat_id}")
                return True, False, None, None
 
            # Rate limiting
            if response.status_code == 429:
                retry_after = response.json().get("parameters", {}).get("retry_after", 5)
                logger.warning(f"Rate limited for {chat_id}, retry after {retry_after}s")
                return False, False, retry_after, f"Rate Limit (retry after {retry_after}s)"
 
            return False, response.status_code < 500, None, f"HTTP {response.status_code}"
        except Exception as e:
            return False, False, None, str(e)
 

shutdown()

plugin.telegram.TelegramSender.shutdown

(

self

)

Graceful shutdown with queue draining

    def shutdown(self):
        r"""Graceful shutdown with queue draining"""
        logger.info("Shutting down Telegram sender...")
        # let the worker send the remaining messages before it is stopped
        timeout = time.time() + 5
        while not self.msg_queue.empty() and time.time() < timeout:
            time.sleep(0.1)
        self._stop_event.set()
        remaining = self.msg_queue.qsize()
        if remaining > 0:
            logger.warning(f"{remaining} messages in queue discarded during shutdown")
        self._worker.join(timeout=5)
 
 
# ===========================
# BoswatchPlugin-Class
# ===========================
 

Field Documentation

_stop_event

plugin.telegram.TelegramSender._stop_event

protected

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

msg_queue

plugin.telegram.TelegramSender.msg_queue

parse_mode

plugin.telegram.TelegramSender.parse_mode

_worker

plugin.telegram.TelegramSender._worker

protected