Spaces:
Sleeping
Sleeping
File size: 5,379 Bytes
72f0375 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | """Read public MAX posts and their counters without joining channels."""
from __future__ import annotations
import base64
import logging
import os
import re
from html.parser import HTMLParser
from pathlib import Path
from typing import Optional, Tuple
from urllib.parse import urlsplit
import json5
import requests
LOGGER = logging.getLogger(__name__)
API_BASE_URL = "https://platform-api2.max.ru"
CA_FILE = str(Path(__file__).parent / "certs" / "russian_trusted_root_ca.pem")
class PublicPostParser(HTMLParser):
def __init__(self):
super().__init__(convert_charrefs=True)
self.title = None
self.scripts = []
self._script = None
def handle_starttag(self, tag, attrs):
attrs = dict(attrs)
if tag == "meta" and attrs.get("property") == "og:title":
self.title = attrs.get("content")
if tag == "script":
self._script = []
def handle_data(self, data):
if self._script is not None:
self._script.append(data)
def handle_endtag(self, tag):
if tag == "script" and self._script is not None:
self.scripts.append("".join(self._script))
self._script = None
def parse_public_post(page_html: str, fallback: str) -> Tuple[str, Optional[int]]:
parser = PublicPostParser()
parser.feed(page_html)
title = parser.title or fallback
# Svelte serializes page data as JS literals, not strict JSON. JSON5 reads
# those literals without executing any code from the remote page.
for script in parser.scripts:
for match in re.finditer(r"\bdata\s*:\s*(\[)", script):
try:
nodes = json5.loads(
script, start=match.start(1), consume_trailing=False,
)
except ValueError:
continue
if not isinstance(nodes, list):
continue
for node in nodes:
if not isinstance(node, dict):
continue
data = node.get("data") or {}
if not isinstance(data, dict):
continue
link_info = data.get("linkInfo") or {}
if not isinstance(link_info, dict):
continue
channel = link_info.get("channel") or {}
if not isinstance(channel, dict):
continue
channel_id = channel.get("channelId")
if type(channel_id) is int and channel_id != 0:
return channel.get("title") or title, channel_id
return title, None
def message_id_from_link(channel_id: int, post_id: str) -> str:
if not re.fullmatch(r"[A-Za-z0-9_-]+", post_id):
raise ValueError("invalid_post_id")
message_bytes = base64.b64decode(
post_id + "=" * (-len(post_id) % 4), altchars=b"-_", validate=True,
)
if len(message_bytes) != 8:
raise ValueError("unsupported_post_id")
chat_bytes = (-abs(channel_id)).to_bytes(8, "big", signed=True)
return "mid." + chat_bytes.hex() + message_bytes.hex()
def get_public_post(session, slug, post_id, canonical_link, timeout):
title = slug
try:
page = session.get(canonical_link, timeout=timeout)
page.raise_for_status()
title, channel_id = parse_public_post(page.text, slug)
token = os.getenv("MAX_ACCESS_TOKEN", "").strip()
if not token:
LOGGER.warning("MAX views unavailable: token is not configured")
return title, None
if channel_id is None:
LOGGER.warning("MAX channel ID is absent for %s", canonical_link)
return title, None
mid = message_id_from_link(channel_id, post_id)
response = session.get(
f"{API_BASE_URL}/messages/{mid}",
headers={"Authorization": token},
timeout=timeout,
verify=CA_FILE,
allow_redirects=False,
)
if response.status_code != 200:
LOGGER.warning(
"MAX API returned HTTP %s for %s", response.status_code, canonical_link,
)
return title, None
message = response.json()
if not isinstance(message, dict):
return title, None
body = message.get("body") or {}
stat = message.get("stat") or {}
if not isinstance(body, dict) or not isinstance(stat, dict):
return title, None
if body.get("mid") != mid:
LOGGER.warning("MAX API returned a different message for %s", canonical_link)
return title, None
if message.get("url"):
returned_path = urlsplit(message["url"]).path.rstrip("/")
if returned_path != urlsplit(canonical_link).path.rstrip("/"):
LOGGER.warning("MAX API returned a different post URL for %s", canonical_link)
return title, None
views = stat.get("views")
if type(views) is int and views >= 0:
return title, views
LOGGER.warning("MAX API response has no views for %s", canonical_link)
except (requests.RequestException, ValueError, OverflowError) as exc:
# Log only the exception class: credential values must not enter logs.
LOGGER.warning("MAX lookup failed for %s (%s)", canonical_link, type(exc).__name__)
return title, None
|