Fix duplicate posting, HTML escaping, restore sendRichMessage correctly
- Stop re-sending to a platform that already succeeded when the other platform failed (was causing duplicate posts on partial failure) - Escape VK post text before sending with parse_mode=HTML instead of only escaping matched bracket-link substrings (bare & < > broke sends) - Restore Telegram sendRichMessage (Bot API 10.1) ported faithfully from new_vk_parser's proven implementation, with fallback to standard send_photo/send_video/send_media_group - Restore "media unavailable" note when a video/photo fails to download instead of silently dropping it - Handle YouTube link attachments from VK posts (route into yt-dlp) - MAX: chunk media beyond 10 items into follow-up messages instead of dropping them - yt-dlp format selector no longer falls back to unconstrained height - Wire up VK_RATE_LIMIT_RPS, drop unused Pillow dep and dead config fields
This commit is contained in:
+131
-28
@@ -2,6 +2,9 @@ from __future__ import annotations
|
||||
|
||||
import html
|
||||
import re
|
||||
from html.parser import HTMLParser
|
||||
from typing import Optional
|
||||
from urllib.parse import urlparse
|
||||
try:
|
||||
from .config import settings
|
||||
except (ImportError, ValueError):
|
||||
@@ -43,33 +46,134 @@ def strip_trailing_hashtags(text: str) -> str:
|
||||
return "\n".join(lines).rstrip()
|
||||
|
||||
|
||||
# Matches VK bracket markup: [club123|Name], [id123|Name], [public123|Name],
|
||||
# [event123|Name] or a raw-URL bracket link [https://example.com|Title]
|
||||
_VK_LINK_RE = re.compile(
|
||||
r"\[(?:(?P<prefix>club|id|public|event)(?P<obj_id>\d+)|(?P<url>https?://[^\s\|\]]+))\|(?P<title>[^\]]+)\]"
|
||||
)
|
||||
|
||||
|
||||
def clean_vk_wiki_links(text: str, parse_mode: str = "html") -> str:
|
||||
"""
|
||||
Converts VK wiki links to clickable links:
|
||||
Converts VK bracket markup into real links, escaping everything else so the
|
||||
result is always safe to send with parse_mode="HTML":
|
||||
- [club12345|Name] -> <a href="https://vk.com/club12345">Name</a>
|
||||
- [id12345|Name] -> <a href="https://vk.com/id12345">Name</a>
|
||||
- [public12345|Name] -> <a href="https://vk.com/public12345">Name</a>
|
||||
- [event12345|Name] -> <a href="https://vk.com/event12345">Name</a>
|
||||
- [id12345|Name] -> <a href="https://vk.com/id12345">Name</a>
|
||||
- [https://example.com|Title] -> <a href="https://example.com">Title</a>
|
||||
Any text outside of recognized bracket markup is HTML-escaped, so stray
|
||||
"&", "<", ">" characters in real VK post text never break the parser.
|
||||
"""
|
||||
def _replace_vk(match: re.Match) -> str:
|
||||
prefix = match.group(1)
|
||||
obj_id = match.group(2)
|
||||
title = match.group(3)
|
||||
if parse_mode == "html":
|
||||
return f'<a href="https://vk.com/{prefix}{obj_id}">{html.escape(title)}</a>'
|
||||
return f"[{title}](https://vk.com/{prefix}{obj_id})"
|
||||
if parse_mode != "html":
|
||||
def _replace_plain(match: re.Match) -> str:
|
||||
title = match.group("title")
|
||||
if match.group("prefix"):
|
||||
target = f"https://vk.com/{match.group('prefix')}{match.group('obj_id')}"
|
||||
else:
|
||||
target = match.group("url")
|
||||
return f"[{title}]({target})"
|
||||
return _VK_LINK_RE.sub(_replace_plain, text)
|
||||
|
||||
def _replace_url(match: re.Match) -> str:
|
||||
url = match.group(1)
|
||||
title = match.group(2)
|
||||
if parse_mode == "html":
|
||||
return f'<a href="{html.escape(url, quote=True)}">{html.escape(title)}</a>'
|
||||
return f"[{title}]({url})"
|
||||
out: list[str] = []
|
||||
last_end = 0
|
||||
for match in _VK_LINK_RE.finditer(text):
|
||||
out.append(html.escape(text[last_end:match.start()]))
|
||||
title = html.escape(match.group("title"))
|
||||
if match.group("prefix"):
|
||||
href = f"https://vk.com/{match.group('prefix')}{match.group('obj_id')}"
|
||||
else:
|
||||
href = html.escape(match.group("url"), quote=True)
|
||||
out.append(f'<a href="{href}">{title}</a>')
|
||||
last_end = match.end()
|
||||
out.append(html.escape(text[last_end:]))
|
||||
return "".join(out)
|
||||
|
||||
text = re.sub(r"\[(club|id|public|event)(\d+)\|([^\]]+)\]", _replace_vk, text)
|
||||
text = re.sub(r"\[(https?://[^\s\|]+)\|([^\]]+)\]", _replace_url, text)
|
||||
return text
|
||||
|
||||
_ALLOWED_WRAPPER_TAGS = {"b", "strong", "i", "em", "u", "s", "code", "a"}
|
||||
|
||||
|
||||
class _SafeHTMLNormalizer(HTMLParser):
|
||||
"""Normalizes admin-supplied header/footer/tags text: keeps a small safe
|
||||
subset of inline HTML tags (with href-scheme validation for <a>) and
|
||||
HTML-escapes everything else, instead of the previous all-or-nothing
|
||||
'"<" not in text' heuristic."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__(convert_charrefs=True)
|
||||
self.out: list[str] = []
|
||||
self.open_tags: list[str] = []
|
||||
|
||||
def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> None:
|
||||
if tag not in _ALLOWED_WRAPPER_TAGS:
|
||||
return
|
||||
if tag == "a":
|
||||
href = next((v for k, v in attrs if k == "href" and v), "")
|
||||
parsed = urlparse(href)
|
||||
if parsed.scheme not in ("http", "https") or not parsed.netloc:
|
||||
return
|
||||
self.out.append(f'<a href="{html.escape(href, quote=True)}">')
|
||||
else:
|
||||
self.out.append(f"<{tag}>")
|
||||
self.open_tags.append(tag)
|
||||
|
||||
def handle_startendtag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> None:
|
||||
if tag == "br":
|
||||
self.out.append("\n")
|
||||
|
||||
def handle_endtag(self, tag: str) -> None:
|
||||
if tag == "br":
|
||||
self.out.append("\n")
|
||||
return
|
||||
if tag in self.open_tags:
|
||||
while self.open_tags:
|
||||
t = self.open_tags.pop()
|
||||
self.out.append(f"</{t}>")
|
||||
if t == tag:
|
||||
break
|
||||
|
||||
def handle_data(self, data: str) -> None:
|
||||
self.out.append(html.escape(data))
|
||||
|
||||
def normalized(self) -> str:
|
||||
while self.open_tags:
|
||||
self.out.append(f"</{self.open_tags.pop()}>")
|
||||
return "".join(self.out)
|
||||
|
||||
|
||||
def normalize_wrapper_text(text: str, parse_mode: str = "html") -> str:
|
||||
"""Safely prepares admin-configured header/footer/tags text for sending."""
|
||||
text = str(text or "").strip()
|
||||
if not text:
|
||||
return ""
|
||||
if parse_mode != "html":
|
||||
return text
|
||||
parser = _SafeHTMLNormalizer()
|
||||
parser.feed(text)
|
||||
parser.close()
|
||||
return parser.normalized()
|
||||
|
||||
|
||||
def build_media_unavailable_note(link_only_items: list, parse_mode: str = "html") -> str:
|
||||
"""Builds a short note listing media that couldn't be attached (too big,
|
||||
too long, private, download failed, etc.) with a link to the original,
|
||||
so readers aren't left with no idea that media was omitted."""
|
||||
if not link_only_items:
|
||||
return ""
|
||||
lines = []
|
||||
for item in link_only_items:
|
||||
url = str(getattr(item, "original_url", "") or "").strip()
|
||||
media_type = str(getattr(item, "media_type", "медиа") or "медиа")
|
||||
if not url:
|
||||
continue
|
||||
if parse_mode == "html":
|
||||
lines.append(f'- {html.escape(media_type)}: <a href="{html.escape(url, quote=True)}">ссылка</a>')
|
||||
else:
|
||||
lines.append(f"- {media_type}: {url}")
|
||||
if not lines:
|
||||
return ""
|
||||
header = "Не удалось прикрепить медиа, оригинал:"
|
||||
if parse_mode == "html":
|
||||
header = f"<i>{html.escape(header)}</i>"
|
||||
return header + "\n" + "\n".join(lines)
|
||||
|
||||
|
||||
def format_post_text(
|
||||
@@ -82,9 +186,9 @@ def format_post_text(
|
||||
bold_first_line: bool = True,
|
||||
vk_url: Optional[str] = None,
|
||||
) -> str:
|
||||
header = (settings.header_text if header is None else header).strip()
|
||||
footer = (settings.footer_text if footer is None else footer).strip()
|
||||
tags = (settings.common_tags if tags is None else tags).strip()
|
||||
header = normalize_wrapper_text(settings.header_text if header is None else header, parse_mode)
|
||||
footer = normalize_wrapper_text(settings.footer_text if footer is None else footer, parse_mode)
|
||||
tags = normalize_wrapper_text(settings.common_tags if tags is None else tags, parse_mode)
|
||||
|
||||
body = strip_trailing_hashtags(raw_text)
|
||||
lines = clean_dividers(body.splitlines())
|
||||
@@ -97,10 +201,9 @@ def format_post_text(
|
||||
|
||||
formatted_lines: list[str] = []
|
||||
for idx, line in enumerate(lines):
|
||||
cleaned_line = clean_vk_wiki_links(line, parse_mode=parse_mode) if parse_mode == "html" else line
|
||||
cleaned_line = clean_vk_wiki_links(line, parse_mode=parse_mode)
|
||||
if idx == title_idx and bold_first_line and cleaned_line.strip():
|
||||
if parse_mode == "html":
|
||||
# If clean_vk_wiki_links was run, keep existing <a> tags safe
|
||||
formatted_lines.append(f"<b>{cleaned_line.strip()}</b>")
|
||||
else:
|
||||
formatted_lines.append(f"**{cleaned_line.strip()}**")
|
||||
@@ -115,12 +218,12 @@ def format_post_text(
|
||||
|
||||
parts: list[str] = []
|
||||
if header:
|
||||
parts.append(html.escape(header) if parse_mode == "html" and "<" not in header else header)
|
||||
parts.append(header)
|
||||
if normalized_body:
|
||||
parts.append(normalized_body)
|
||||
if footer:
|
||||
parts.append(html.escape(footer) if parse_mode == "html" and "<" not in footer else footer)
|
||||
parts.append(footer)
|
||||
if tags:
|
||||
parts.append(html.escape(tags) if parse_mode == "html" and "<" not in tags else tags)
|
||||
parts.append(tags)
|
||||
|
||||
return "\n\n".join(part for part in parts if part).strip()
|
||||
|
||||
Reference in New Issue
Block a user