Added custom tg message limit

action.yml

@@ -12,6 +12,9 @@ inputs:
   tg-message-thread-id:
     description: "Telegram Message Thread ID"
     required: false
+  tg-message-limit:
+    description: "Telegram Message limit"
+    required: false
   discord-webhook-url:
     description: "Discord Webhook URL"
     required: false
@@ -63,6 +66,7 @@ runs:
         TELEGRAM_BOT_TOKEN: ${{ inputs.tg-bot-token }}
         TELEGRAM_CHAT_ID: ${{ inputs.tg-chat-id }}
         TELEGRAM_MESSAGE_THREAD_ID: ${{ inputs.tg-message-thread-id }}
+        TELEGRAM_MESSAGE_LIMIT: ${{ inputs.tg-message-limit }}
         DISCORD_WEBHOOK_URL: ${{ inputs.discord-webhook-url }}
         DISCORD_THREAD_ID: ${{ inputs.discord-thread-id }}
         GITHUB_TOKEN: ${{ inputs.github-token }}
@@ -78,4 +82,4 @@ runs:
 
 branding:
   icon: "message-circle"
-  color: "blue"
+  color: "blue"

notifier/__main__.py

@@ -8,7 +8,7 @@
 from notifier.application.services import RenderService
 from notifier.infrastructure.discord_gateway import DiscordGateway
 from notifier.infrastructure.github_gateway import GithubGateway
-from notifier.infrastructure.telegram_gateway import TelegramGateway
+from notifier.infrastructure.telegram_gateway import TelegramGateway, TG_MESSAGE_LIMIT_DEFAULT
 
 
 def get_interactor(url: str) -> type[SendIssue] | type[SendPR]:
@@ -55,6 +55,7 @@ def get_interactor(url: str) -> type[SendIssue] | type[SendPR]:
             attempt_count=int(os.environ.get("ATTEMPT_COUNT", "2")),
             message_thread_id=os.environ.get("TELEGRAM_MESSAGE_THREAD_ID"),
             custom_template=html_template,
+            tg_message_limit=os.environ.get('TELEGRAM_MESSAGE_LIMIT') or TG_MESSAGE_LIMIT_DEFAULT
         )
         notifiers.append(telegram_gateway)
 
@@ -85,4 +86,4 @@ def get_interactor(url: str) -> type[SendIssue] | type[SendPR]:
     except Exception as e:
         traceback.print_exc(file=sys.stderr)
         print(f"Error processing event: {e}", file=sys.stderr)
-        sys.exit(1)
+        sys.exit(1)

notifier/infrastructure/telegram_gateway.py

@@ -5,8 +5,9 @@
 from notifier.application import interfaces
 from notifier.domain.entities import Issue, PullRequest
 from notifier.infrastructure.send_weebhook import send_webhook
+from notifier.infrastructure.truncate_html import TruncateHTML
 
-TG_MESSAGE_LIMIT: typing.Final = 4096
+TG_MESSAGE_LIMIT_DEFAULT: typing.Final = 4096
 
 ISSUE_TEMPLATE: typing.Final = (
     "🚀 <b>New issue to <a href=/{repository}>{repository}</a> by <a href=/{user}>@{user}</a> </b><br/>"
@@ -33,6 +34,7 @@ def __init__(
         chat_id: str,
         bot_token: str,
         attempt_count: int,
+        tg_message_limit: int,
         message_thread_id: str | int | None = None,
         custom_template: str = "",
     ) -> None:
@@ -41,21 +43,26 @@ def __init__(
         self._attempt_count = attempt_count
         self._message_thread_id = message_thread_id
         self._custom_template = custom_template
+        self._tg_message_limit = tg_message_limit
+
 
     def send_issue(
         self,
         issue: Issue,
         formatted_body: str,
         formatted_labels: str,
     ) -> None:
-        message = self._create_issue_message(issue, formatted_body, formatted_labels)
-        render_result = sulguk.transform_html(message, base_url="https://github.com")
 
-        if len(render_result.text) > TG_MESSAGE_LIMIT:
-            message = self._create_issue_message(issue, "<p></p>", formatted_labels)
-            render_result = sulguk.transform_html(
-                message, base_url="https://github.com"
-            )
+        message = self._create_issue_message(
+            issue=issue,
+            body=formatted_body,
+            labels=formatted_labels
+        )
+        render_result = sulguk.transform_html(
+            message, base_url="https://github.com"
+        )
+
+
         send_webhook(
             payload=self._create_payload(render_result),
             url=f"https://api.telegram.org/bot{self._bot_token}/sendMessage",
@@ -73,12 +80,6 @@ def send_pull_request(
         )
         render_result = sulguk.transform_html(message, base_url="https://github.com")
 
-        if len(render_result.text) > TG_MESSAGE_LIMIT:
-            message = self._create_pr_message(pull_request, "<p></p>", formatted_labels)
-            render_result = sulguk.transform_html(
-                message, base_url="https://github.com"
-            )
-
         send_webhook(
             payload=self._create_payload(render_result),
             url=f"https://api.telegram.org/bot{self._bot_token}/sendMessage",
@@ -94,30 +95,60 @@ def _create_payload(self, render_result: sulguk.RenderResult) -> dict:
             "entities": render_result.entities,
             "disable_web_page_preview": True,
         }
+
         payload["chat_id"] = self._chat_id
 
         if self._message_thread_id is not None:
             payload["message_thread_id"] = self._message_thread_id
 
         return payload
 
+
+    def _create_message_with_limit(
+            self,
+            template: str,
+            payload: dict,
+    ):
+
+        message = template.format(**payload)
+        message_length = len(message)
+        body = payload["body"]
+        truncate_html = TruncateHTML()
+
+        if message_length > self._tg_message_limit:
+            max_length_body = len(body) - (message_length - self._tg_message_limit)
+            payload['body'] = truncate_html.render(
+                raw_html=body,
+                max_length=max_length_body,
+            )
+            return template.format(**payload)
+
+        return message
+
+
     def _create_issue_message(self, issue: Issue, body: str, labels: str) -> str:
         template = self._custom_template or ISSUE_TEMPLATE
-        return template.format(
+        payload = dict(
             id=issue.id,
             user=issue.user,
             title=issue.title,
             labels=labels,
             url=issue.url,
             body=body,
             repository=issue.repository,
-            promo="<a href='/reagento/relator'>sent via relator</a>",
+            promo="<a href='/reagento/relator'>sent via relator</a>"
+        )
+
+        return self._create_message_with_limit(
+            template=template,
+            payload=payload,
         )
 
     def _create_pr_message(self, pr: PullRequest, body: str, labels: str) -> str:
         """Create HTML message for pull request"""
         template = self._custom_template or PR_TEMPLATE
-        return template.format(
+
+        payload = dict(
             id=pr.id,
             user=pr.user,
             title=pr.title,
@@ -131,3 +162,8 @@ def _create_pr_message(self, pr: PullRequest, body: str, labels: str) -> str:
             base_ref=pr.base_ref,
             promo="<a href='/reagento/relator'>sent via relator</a>",
         )
+
+        return self._create_message_with_limit(
+            template=template,
+            payload=payload
+        )

notifier/infrastructure/truncate_html.py

@@ -0,0 +1,69 @@
+from dataclasses import field, dataclass
+
+from bs4 import BeautifulSoup, NavigableString, Tag
+
+
+@dataclass
+class TruncateHtmlState:
+    count: int = field(default=0)
+    done: bool = field(default=False)
+
+
+class TruncateHTML:
+    def __init__(self):
+        self._ellipsis = "..."
+
+    def _walk(
+            self,
+            node: Tag | NavigableString,
+            state: TruncateHtmlState,
+            max_length: int
+    ) -> bool:
+        if isinstance(node, NavigableString):
+            text = node.text
+            if state.count + len(text) <= max_length:
+                state.count += len(text)
+                return
+
+            remaining = max_length - state.count
+            if remaining <= 0:
+                node.extract()
+                state.done = True
+                return
+
+            cut_text = text[:remaining].rstrip() + self._ellipsis
+            node.replace_with(cut_text)
+            state.done = True
+            return
+
+        if isinstance(node, Tag):
+            children = list(node.contents)
+            for i, child in enumerate(children):
+                self._walk(child, state, max_length)
+                if state.done:
+                    tags_deleted = children[i + 1:]
+                    for tag_deleted in tags_deleted:
+                        tag_deleted.extract()
+                    break
+
+    def render(
+            self,
+            raw_html: str,
+            max_length: int
+    ) -> str:
+
+        soup = BeautifulSoup(raw_html, "html.parser")
+        state = TruncateHtmlState(
+            count=0
+        )
+
+        for node in soup.contents:
+            self._walk(
+                node=node,
+                state=state,
+                max_length=max_length
+            )
+            if state.done:
+                break
+
+        return str(soup)