feat: enable HTML rich messages in Telegram poster
- Strip separator lines (━, —, -, =, etc.) automatically - Format first line as bold HTML title (<b>Title</b>) - Keep body text escaped and formatted - Append hashtags at the end (#category #source) - Pass parse_mode='HTML' in Telegram API requests
This commit is contained in:
@@ -31,12 +31,47 @@ def publication_hashtags(category_tag: str, source_tag: str) -> str:
|
|||||||
return " ".join(f"#{tag}" for tag in tags if tag)
|
return " ".join(f"#{tag}" for tag in tags if tag)
|
||||||
|
|
||||||
|
|
||||||
def build_publication_text(text: str, category_tag: str, source_tag: str) -> str:
|
import html
|
||||||
|
|
||||||
|
|
||||||
|
def build_publication_text(
|
||||||
|
text: str,
|
||||||
|
category_tag: str,
|
||||||
|
source_tag: str,
|
||||||
|
format_title: bool = False,
|
||||||
|
parse_mode: str | None = None,
|
||||||
|
) -> str:
|
||||||
base = strip_trailing_hashtag_line(text)
|
base = strip_trailing_hashtag_line(text)
|
||||||
|
lines = base.splitlines()
|
||||||
|
sep_pattern = re.compile(r"^\s*[━—─\-=\*\#_]{2,}\s*$")
|
||||||
|
filtered_lines = [line for line in lines if not sep_pattern.match(line)]
|
||||||
|
if not filtered_lines:
|
||||||
|
return publication_hashtags(category_tag, source_tag)
|
||||||
|
title_idx = None
|
||||||
|
for idx, line in enumerate(filtered_lines):
|
||||||
|
if line.strip():
|
||||||
|
title_idx = idx
|
||||||
|
break
|
||||||
|
formatted_lines = []
|
||||||
|
for idx, line in enumerate(filtered_lines):
|
||||||
|
if idx == title_idx and format_title:
|
||||||
|
if parse_mode == "html":
|
||||||
|
escaped_title = html.escape(line.strip())
|
||||||
|
formatted_lines.append(f"<b>{escaped_title}</b>")
|
||||||
|
elif parse_mode == "markdown":
|
||||||
|
formatted_lines.append(f"**{line.strip()}**")
|
||||||
|
else:
|
||||||
|
formatted_lines.append(line.strip())
|
||||||
|
else:
|
||||||
|
if parse_mode == "html":
|
||||||
|
formatted_lines.append(html.escape(line))
|
||||||
|
else:
|
||||||
|
formatted_lines.append(line)
|
||||||
|
body = "\n".join(formatted_lines).strip()
|
||||||
hashtags = publication_hashtags(category_tag, source_tag)
|
hashtags = publication_hashtags(category_tag, source_tag)
|
||||||
if not hashtags:
|
if hashtags:
|
||||||
return base
|
return f"{body}\n\n{hashtags}" if body else hashtags
|
||||||
return f"{base}\n\n{hashtags}" if base else hashtags
|
return body
|
||||||
|
|
||||||
|
|
||||||
def parse_categories(value: object) -> list[str]:
|
def parse_categories(value: object) -> list[str]:
|
||||||
|
|||||||
@@ -159,8 +159,8 @@ class TelegramPoster:
|
|||||||
)
|
)
|
||||||
return sorted(rows, key=lambda row: row["time"])
|
return sorted(rows, key=lambda row: row["time"])
|
||||||
|
|
||||||
def chat_kwargs(self) -> dict[str, int]:
|
def chat_kwargs(self) -> dict[str, Any]:
|
||||||
kwargs = {"chat_id": int(self.chat_id)}
|
kwargs: dict[str, Any] = {"chat_id": int(self.chat_id), "parse_mode": "HTML"}
|
||||||
if self.thread_id:
|
if self.thread_id:
|
||||||
kwargs["message_thread_id"] = int(self.thread_id)
|
kwargs["message_thread_id"] = int(self.thread_id)
|
||||||
return kwargs
|
return kwargs
|
||||||
@@ -318,7 +318,7 @@ class TelegramPoster:
|
|||||||
def build_text(self, post: dict[str, Any]) -> str:
|
def build_text(self, post: dict[str, Any]) -> str:
|
||||||
category_tag = str(post.get("final_category_tag") or post.get("rewrite_category_tag") or post.get("final_category") or post.get("rewrite_category") or "")
|
category_tag = str(post.get("final_category_tag") or post.get("rewrite_category_tag") or post.get("final_category") or post.get("rewrite_category") or "")
|
||||||
source_tag = str(post.get("final_source_tag") or post.get("rewrite_source_tag") or post.get("source_tag") or "")
|
source_tag = str(post.get("final_source_tag") or post.get("rewrite_source_tag") or post.get("source_tag") or "")
|
||||||
return build_publication_text(post.get("final_text") or post.get("rewritten_text") or "", category_tag, source_tag)
|
return build_publication_text(post.get("final_text") or post.get("rewritten_text") or "", category_tag, source_tag, format_title=True, parse_mode="html")
|
||||||
|
|
||||||
async def send_text(self, text: str) -> list[int]:
|
async def send_text(self, text: str) -> list[int]:
|
||||||
message_ids: list[int] = []
|
message_ids: list[int] = []
|
||||||
|
|||||||
Reference in New Issue
Block a user