Read public MAX post views through Bot API

Dockerfile

@@ -6,6 +6,8 @@ COPY requirements.txt ./
 RUN pip install --no-cache-dir -r requirements.txt
 
 COPY app.py ./
+COPY max_api.py ./
+COPY certs/ ./certs/
 
 EXPOSE 7860
 

app.py

@@ -13,6 +13,8 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 
+from max_api import get_public_post
+
 APP_NAME = "pr-tool-backend"
 
 VK_API_VERSION = os.getenv("VK_API_VERSION", "5.131")
@@ -105,7 +107,7 @@ class ParseResponse(BaseModel):
     telegram_total: int
     vk_total: int
     errors: List[str]
-    max_count: int = 0  # число ссылок MAX (просмотры из MAX недоступны)
+    max_count: int = 0
 
 
 # ---------- Вспомогательные функции ----------
@@ -306,27 +308,11 @@ def process_telegram_link(channel_username: str, message_id: str, canonical_link
     return title, views
 
 
-def process_max_link(slug: str, post_id: str, canonical_link: str) -> Tuple[str, None]:
-    """MAX: подхватываем название канала из страницы поста.
-    Просмотры из MAX недоступны, поэтому всегда возвращаем None."""
-    title = slug
-    try:
-        session = _get_session()
-        resp = session.get(canonical_link, timeout=MAX_HTTP_TIMEOUT)
-        if resp.status_code == 200 and resp.text:
-            page_html = resp.text
-            m = re.search(
-                r"<meta[^>]*property=[\"']og:title[\"'][^>]*content=[\"']([^\"']+)[\"']",
-                page_html,
-                re.IGNORECASE,
-            )
-            if not m:
-                m = re.search(r"<title[^>]*>(.*?)</title>", page_html, re.IGNORECASE | re.DOTALL)
-            if m:
-                title = clean_max_title(m.group(1), slug)
-    except Exception:  # noqa: BLE001
-        title = slug  # тихо откатываемся к слагу из URL
-    return title or slug, None
+def process_max_link(slug: str, post_id: str, canonical_link: str) -> Tuple[str, Optional[int]]:
+    title, views = get_public_post(
+        _get_session(), slug, post_id, canonical_link, MAX_HTTP_TIMEOUT,
+    )
+    return clean_max_title(title, slug), views
 
 
 def process_vk_link(owner_id: int, post_id: str, canonical_link: str) -> Tuple[str, int]:
@@ -508,8 +494,7 @@ def resolve_max(max_tasks, deadline):
     fetched = _resolve_pool(
         tasks, process_max_link, MAX_CONCURRENCY, deadline, None, defaults=defaults
     )
-    # просмотры из MAX недоступны всегда
-    return {c: (t, None) for c, (t, _v) in fetched.items()}
+    return fetched
 
 
 def resolve_vk(vk_tasks, deadline):
@@ -654,13 +639,15 @@ def build_output(lines, deadline):
                 max_groups.setdefault(key, {"title": "MAX", "items": []})
                 max_groups[key]["items"].append((canonical, None))
             else:
-                title, _views = max_results.get(canonical, (slug, None))
+                title, views = max_results.get(canonical, (slug, None))
                 key = slug
                 if key not in max_groups:
                     max_groups[key] = {"title": title, "items": []}
                 if not max_groups[key].get("title"):
                     max_groups[key]["title"] = title
-                max_groups[key]["items"].append((canonical, None))
+                max_groups[key]["items"].append((canonical, views))
+                if views is None:
+                    errors.append(f"Не удалось получить просмотры MAX: {canonical}")
 
         else:
             errors.append(f"Неизвестная платформа: {canonical}")
@@ -668,6 +655,8 @@ def build_output(lines, deadline):
     tg_total_views = sum(v for g in tg_groups.values() for _, v in g["items"] if v)
     vk_total_views = sum(v for g in vk_groups.values() for _, v in g["items"] if v)
     max_count = sum(len(g["items"]) for g in max_groups.values())
+    max_total = sum(v for g in max_groups.values() for _, v in g["items"] if v is not None)
+    max_missing = sum(v is None for g in max_groups.values() for _, v in g["items"])
 
     html_lines: List[str] = []
     text_lines: List[str] = []
@@ -705,10 +694,15 @@ def build_output(lines, deadline):
     # --- MAX (секцию показываем только если есть ссылки MAX) ---
     if max_groups:
         html_lines.append("<h2>MAX</h2>")
-        html_lines.append("<i>Просмотры из MAX недоступны.</i>")
         text_lines.append("MAX")
-        text_lines.append("Просмотры из MAX недоступны.")
-        h, t = _render_groups(max_groups, has_views=False, default_title="MAX")
+        if max_missing < max_count:
+            html_lines.append(f'Суммарно посты собрали <b>{human_format_views(max_total)}</b> просмотров.')
+            text_lines.append(f"Суммарно посты собрали {human_format_views(max_total)} просмотров.")
+        if max_missing:
+            note = f"Постов без данных о просмотрах: {max_missing}. Они не включены в сумму."
+            html_lines.append(f"<i>{note}</i>")
+            text_lines.append(note)
+        h, t = _render_groups(max_groups, has_views=True, default_title="MAX")
         html_lines.append(h)
         text_lines.extend(t)
         html_lines.append("<br/>")
@@ -739,66 +733,6 @@ def health_check():
     return {"status": "ok"}
 
 
-@app.get("/_max_public_probe")
-def max_public_probe():
-    import base64
-    import hashlib
-    import json
-    import ssl
-    import urllib.error
-    import urllib.request
-
-    token = os.getenv("MAX_ACCESS_TOKEN", "").strip()
-    if not token:
-        return {"configured": False}
-    try:
-        cert_response = requests.get(
-            "https://gu-st.ru/content/Other/doc/russian_trusted_root_ca.cer",
-            timeout=(5, 10),
-        )
-        cert_response.raise_for_status()
-        pem = cert_response.text
-        fingerprint = hashlib.sha256(ssl.PEM_cert_to_DER_cert(pem)).hexdigest()
-        if fingerprint != "d26d2d0231b7c39f92cc738512ba54103519e4405d68b5bd703e9788ca8ecf31":
-            return {"configured": True, "certificate": "unexpected_fingerprint"}
-        context = ssl.create_default_context(cadata=pem)
-    except Exception as exc:
-        return {"configured": True, "certificate_error": type(exc).__name__}
-
-    def check(path):
-        request = urllib.request.Request(
-            "https://platform-api2.max.ru" + path,
-            headers={"Authorization": token},
-        )
-        try:
-            with urllib.request.urlopen(request, context=context, timeout=8) as response:
-                data = json.load(response)
-                stat = data.get("stat") or {}
-                return {"http": response.status, "code": data.get("code"),
-                        "views": stat.get("views"),
-                        "message_count": len(data.get("messages", [])),
-                        "is_bot": data.get("is_bot")}
-        except urllib.error.HTTPError as exc:
-            try:
-                data = json.load(exc)
-            except Exception:
-                data = {}
-            return {"http": exc.code, "code": data.get("code")}
-        except Exception as exc:
-            return {"error": type(exc).__name__}
-
-    post_id = "AaE-viFyB3E"
-    post_hex = base64.urlsafe_b64decode(post_id + "=").hex()
-    channel_id = 68067500157571
-    negative_hex = ((-channel_id) & ((1 << 64) - 1)).to_bytes(8, "big").hex()
-    positive_hex = channel_id.to_bytes(8, "big").hex()
-    return {"configured": True, "auth": check("/me"),
-            "channel": check(f"/messages?chat_id={-channel_id}&count=1"),
-            "negative_mid": check(f"/messages/mid.{negative_hex}{post_hex}"),
-            "positive_mid": check(f"/messages/mid.{positive_hex}{post_hex}"),
-            "short_id": check(f"/messages/{post_id}")}
-
-
 @app.post("/parse", response_model=ParseResponse)
 def parse_links(payload: ParseRequest):
     deadline = time.monotonic() + TOTAL_TIME_BUDGET

certs/README.md

@@ -0,0 +1,10 @@
+# MAX API Certificate
+
+Public Russian Trusted Root CA certificate downloaded from:
+https://gu-st.ru/content/Other/doc/russian_trusted_root_ca.cer
+
+SHA-256 certificate fingerprint:
+`D2:6D:2D:02:31:B7:C3:9F:92:CC:73:85:12:BA:54:10:35:19:E4:40:5D:68:B5:BD:70:3E:97:88:CA:8E:CF:31`
+
+Used only for HTTPS requests to `platform-api2.max.ru`. Certificate validation
+remains enabled; other HTTP clients retain their default trust settings.

certs/russian_trusted_root_ca.pem

@@ -0,0 +1,33 @@
+-----BEGIN CERTIFICATE-----
+MIIFwjCCA6qgAwIBAgICEAAwDQYJKoZIhvcNAQELBQAwcDELMAkGA1UEBhMCUlUx
+PzA9BgNVBAoMNlRoZSBNaW5pc3RyeSBvZiBEaWdpdGFsIERldmVsb3BtZW50IGFu
+ZCBDb21tdW5pY2F0aW9uczEgMB4GA1UEAwwXUnVzc2lhbiBUcnVzdGVkIFJvb3Qg
+Q0EwHhcNMjIwMzAxMjEwNDE1WhcNMzIwMjI3MjEwNDE1WjBwMQswCQYDVQQGEwJS
+VTE/MD0GA1UECgw2VGhlIE1pbmlzdHJ5IG9mIERpZ2l0YWwgRGV2ZWxvcG1lbnQg
+YW5kIENvbW11bmljYXRpb25zMSAwHgYDVQQDDBdSdXNzaWFuIFRydXN0ZWQgUm9v
+dCBDQTCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBAMfFOZ8pUAL3+r2n
+qqE0Zp52selXsKGFYoG0GM5bwz1bSFtCt+AZQMhkWQheI3poZAToYJu69pHLKS6Q
+XBiwBC1cvzYmUYKMYZC7jE5YhEU2bSL0mX7NaMxMDmH2/NwuOVRj8OImVa5s1F4U
+zn4Kv3PFlDBjjSjXKVY9kmjUBsXQrIHeaqmUIsPIlNWUnimXS0I0abExqkbdrXbX
+YwCOXhOO2pDUx3ckmJlCMUGacUTnylyQW2VsJIyIGA8V0xzdaeUXg0VZ6ZmNUr5Y
+Ber/EAOLPb8NYpsAhJe2mXjMB/J9HNsoFMBFJ0lLOT/+dQvjbdRZoOT8eqJpWnVD
+U+QL/qEZnz57N88OWM3rabJkRNdU/Z7x5SFIM9FrqtN8xewsiBWBI0K6XFuOBOTD
+4V08o4TzJ8+Ccq5XlCUW2L48pZNCYuBDfBh7FxkB7qDgGDiaftEkZZfApRg2E+M9
+G8wkNKTPLDc4wH0FDTijhgxR3Y4PiS1HL2Zhw7bD3CbslmEGgfnnZojNkJtcLeBH
+BLa52/dSwNU4WWLubaYSiAmA9IUMX1/RpfpxOxd4Ykmhz97oFbUaDJFipIggx5sX
+ePAlkTdWnv+RWBxlJwMQ25oEHmRguNYf4Zr/Rxr9cS93Y+mdXIZaBEE0KS2iLRqa
+OiWBki9IMQU4phqPOBAaG7A+eP8PAgMBAAGjZjBkMB0GA1UdDgQWBBTh0YHlzlpf
+BKrS6badZrHF+qwshzAfBgNVHSMEGDAWgBTh0YHlzlpfBKrS6badZrHF+qwshzAS
+BgNVHRMBAf8ECDAGAQH/AgEEMA4GA1UdDwEB/wQEAwIBhjANBgkqhkiG9w0BAQsF
+AAOCAgEAALIY1wkilt/urfEVM5vKzr6utOeDWCUczmWX/RX4ljpRdgF+5fAIS4vH
+tmXkqpSCOVeWUrJV9QvZn6L227ZwuE15cWi8DCDal3Ue90WgAJJZMfTshN4OI8cq
+W9E4EG9wglbEtMnObHlms8F3CHmrw3k6KmUkWGoa+/ENmcVl68u/cMRl1JbW2bM+
+/3A+SAg2c6iPDlehczKx2oa95QW0SkPPWGuNA/CE8CpyANIhu9XFrj3RQ3EqeRcS
+AQQod1RNuHpfETLU/A2gMmvn/w/sx7TB3W5BPs6rprOA37tutPq9u6FTZOcG1Oqj
+C/B7yTqgI7rbyvox7DEXoX7rIiEqyNNUguTk/u3SZ4VXE2kmxdmSh3TQvybfbnXV
+4JbCZVaqiZraqc7oZMnRoWrXRG3ztbnbes/9qhRGI7PqXqeKJBztxRTEVj8ONs1d
+WN5szTwaPIvhkhO3CO5ErU2rVdUr89wKpNXbBODFKRtgxUT70YpmJ46VVaqdAhOZ
+D9EUUn4YaeLaS8AjSF/h7UkjOibNc4qVDiPP+rkehFWM66PVnP1Msh93tc+taIfC
+EYVMxjh8zNbFuoc7fzvvrFILLe7ifvEIUqSVIC/AzplM/Jxw7buXFeGP1qVCBEHq
+391d/9RAfaZ12zkwFsl+IKwE/OZxW8AHa9i1p4GO0YSNuczzEm4=
+-----END CERTIFICATE-----

max_api.py

@@ -0,0 +1,143 @@
+"""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

requirements.txt

@@ -2,3 +2,4 @@ fastapi==0.110.0
 uvicorn[standard]==0.27.1
 requests==2.31.0
 pydantic==2.6.1
+json5==0.17.3

tests/test_max_api.py

@@ -0,0 +1,98 @@
+import os
+import time
+import unittest
+from unittest.mock import Mock, patch
+
+import app
+import max_api
+
+
+URL = "https://max.ru/vkontakte/AaE-viFyB3E"
+MID = "mid.ffffc217ccd5157d01a13ebe21720771"
+PAGE = '''<meta content="Public &amp; news" property="og:title">
+<script>app.start(element, {
+data: [null,{type:"data",data:{linkInfo:{channel:{
+title:"Public & news",description:"a } and [ inside text",
+participantsCount:5287,channelId:68067500157571},message:{text:"post"}}}}],
+form:null});</script>'''
+
+
+def page_response():
+    return Mock(status_code=200, text=PAGE)
+
+
+class MaxPublicPostTests(unittest.TestCase):
+    def test_hydration_and_message_id(self):
+        title, channel_id = max_api.parse_public_post(PAGE, "vkontakte")
+        self.assertEqual(title, "Public & news")
+        self.assertEqual(channel_id, 68067500157571)
+        self.assertEqual(max_api.message_id_from_link(channel_id, "AaE-viFyB3E"), MID)
+
+    def test_missing_hydration_preserves_meta_title(self):
+        title, channel_id = max_api.parse_public_post(
+            '<meta property="og:title" content="Public &amp; news">', "vkontakte",
+        )
+        self.assertEqual(title, "Public & news")
+        self.assertIsNone(channel_id)
+
+    def test_bad_post_id_is_rejected(self):
+        with self.assertRaises(ValueError):
+            max_api.message_id_from_link(68067500157571, "bad")
+
+    @patch.dict(os.environ, {"MAX_ACCESS_TOKEN": "test-token"})
+    def test_public_post_views_and_credentials(self):
+        session = Mock()
+        session.get.side_effect = [page_response(), Mock(status_code=200, json=lambda: {
+            "body": {"mid": MID}, "stat": {"views": 1197}, "url": URL,
+        })]
+        self.assertEqual(max_api.get_public_post(
+            session, "vkontakte", "AaE-viFyB3E", URL, (5, 8),
+        ), ("Public & news", 1197))
+        public_request, api_request = session.get.call_args_list
+        self.assertNotIn("headers", public_request.kwargs)
+        self.assertEqual(api_request.args[0], max_api.API_BASE_URL + "/messages/" + MID)
+        self.assertEqual(api_request.kwargs["headers"], {"Authorization": "test-token"})
+        self.assertEqual(api_request.kwargs["verify"], max_api.CA_FILE)
+        self.assertFalse(api_request.kwargs["allow_redirects"])
+
+    @patch.dict(os.environ, {"MAX_ACCESS_TOKEN": "test-token"})
+    def test_denied_and_missing_counter_are_unknown_not_zero(self):
+        for response in [Mock(status_code=403), Mock(status_code=200, json=lambda: {
+            "body": {"mid": MID}, "stat": {},
+        }), Mock(status_code=200, json=lambda: {
+            "body": {"mid": "different"}, "stat": {"views": 9999},
+        })]:
+            with self.subTest(status=response.status_code):
+                session = Mock()
+                session.get.side_effect = [page_response(), response]
+                self.assertEqual(max_api.get_public_post(
+                    session, "vkontakte", "AaE-viFyB3E", URL, (5, 8),
+                ), ("Public & news", None))
+
+    @patch.dict(os.environ, {"MAX_ACCESS_TOKEN": "test-token"})
+    def test_real_zero_is_preserved(self):
+        session = Mock()
+        session.get.side_effect = [page_response(), Mock(status_code=200, json=lambda: {
+            "body": {"mid": MID}, "stat": {"views": 0},
+        })]
+        self.assertEqual(max_api.get_public_post(
+            session, "vkontakte", "AaE-viFyB3E", URL, (5, 8),
+        )[1], 0)
+
+    def test_output_keeps_max_counters_and_partial_sum(self):
+        lines = [("max", URL, "vkontakte", "AaE-viFyB3E"),
+                 ("max", "https://max.ru/other/AaE-viFyB3E", "other", "AaE-viFyB3E")]
+        with patch.object(app, "resolve_max", return_value={
+            URL: ("Public & news", 1197), lines[1][1]: ("Other", None),
+        }):
+            html, text, _, _, errors, max_count = app.build_output(lines, time.monotonic() + 10)
+        self.assertIn("<b>1,2K</b>", html)
+        self.assertIn("Public &amp; news", html)
+        self.assertIn("1,2K", text)
+        self.assertIn("Постов без данных о просмотрах: 1.", text)
+        self.assertEqual(len(errors), 1)
+        self.assertEqual(max_count, 2)
+
+
+if __name__ == "__main__":
+    unittest.main()