Spaces:
Sleeping
Sleeping
| """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 | |