exostring
d01a17146e
VK can briefly serve a not-yet-transcoded master stream on a capped-height URL right after a video is published, tripping the size check on the very first attempt. Treat that as transient (retry with a longer backoff) instead of failing permanently, and merge the fallback note into a single "Смотреть видео в VK" link instead of two separate lines. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
248 lines
8.9 KiB
Python
248 lines
8.9 KiB
Python
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):
|
|
from config import settings
|
|
|
|
|
|
def split_message_chunks(text: str, limit: int = 4000) -> list[str]:
|
|
text = str(text or "").strip()
|
|
if not text:
|
|
return []
|
|
chunks: list[str] = []
|
|
while len(text) > limit:
|
|
split_at = text.rfind("\n", 0, limit)
|
|
if split_at < limit // 2:
|
|
split_at = text.rfind(" ", 0, limit)
|
|
if split_at < limit // 2:
|
|
split_at = limit
|
|
chunks.append(text[:split_at].strip())
|
|
text = text[split_at:].strip()
|
|
if text:
|
|
chunks.append(text)
|
|
return chunks
|
|
|
|
|
|
def clean_dividers(lines: list[str]) -> list[str]:
|
|
sep_pattern = re.compile(r"^\s*[━—─\-=\*\#_]{2,}\s*$")
|
|
return ["" if sep_pattern.match(line) else line for line in lines]
|
|
|
|
|
|
# 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 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>
|
|
- [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.
|
|
"""
|
|
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)
|
|
|
|
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)
|
|
|
|
|
|
_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", vk_url: Optional[str] = None
|
|
) -> 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 media_type == "video":
|
|
target = vk_url or url
|
|
label = "Смотреть видео в VK"
|
|
else:
|
|
target = url
|
|
label = media_type
|
|
if parse_mode == "html":
|
|
lines.append(f'- <a href="{html.escape(target, quote=True)}">{html.escape(label)}</a>')
|
|
else:
|
|
lines.append(f"- {label}: {target}")
|
|
if not lines:
|
|
return ""
|
|
header = "Не поместилось в пост:"
|
|
if parse_mode == "html":
|
|
header = f"<i>{html.escape(header)}</i>"
|
|
return header + "\n" + "\n".join(lines)
|
|
|
|
|
|
def build_oversized_video_note(items: list, parse_mode: str = "html") -> str:
|
|
"""Like build_media_unavailable_note, but for videos that were deliberately
|
|
skipped for exceeding a platform's size limit (e.g. MAX's 250MB video cap) -
|
|
friendlier tone since this isn't a failure, just a known platform limit."""
|
|
if not items:
|
|
return ""
|
|
lines = []
|
|
for item in items:
|
|
url = str(getattr(item, "original_url", "") or "").strip()
|
|
if not url:
|
|
continue
|
|
if parse_mode == "html":
|
|
lines.append(f'- <a href="{html.escape(url, quote=True)}">видео по ссылке</a>')
|
|
else:
|
|
lines.append(f"- видео по ссылке: {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(
|
|
raw_text: str,
|
|
*,
|
|
parse_mode: str = "html", # "html" or "plain"
|
|
header: Optional[str] = None,
|
|
footer: Optional[str] = None,
|
|
tags: Optional[str] = None,
|
|
bold_first_line: bool = True,
|
|
vk_url: Optional[str] = None,
|
|
) -> str:
|
|
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)
|
|
|
|
lines = clean_dividers(raw_text.splitlines())
|
|
|
|
title_idx: Optional[int] = None
|
|
for idx, line in enumerate(lines):
|
|
if line.strip():
|
|
title_idx = idx
|
|
break
|
|
|
|
formatted_lines: list[str] = []
|
|
for idx, line in enumerate(lines):
|
|
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":
|
|
formatted_lines.append(f"<b>{cleaned_line.strip()}</b>")
|
|
else:
|
|
formatted_lines.append(f"**{cleaned_line.strip()}**")
|
|
# Blank line after header if next line is not empty
|
|
if idx + 1 < len(lines) and lines[idx + 1].strip() != "":
|
|
formatted_lines.append("")
|
|
else:
|
|
formatted_lines.append(cleaned_line)
|
|
|
|
raw_body = "\n".join(formatted_lines)
|
|
normalized_body = re.sub(r"\n{3,}", "\n\n", raw_body).strip()
|
|
|
|
parts: list[str] = []
|
|
if header:
|
|
parts.append(header)
|
|
if normalized_body:
|
|
parts.append(normalized_body)
|
|
if footer:
|
|
parts.append(footer)
|
|
if tags:
|
|
parts.append(tags)
|
|
|
|
return "\n\n".join(part for part in parts if part).strip()
|