Repurpose PR_Tool as isolated ReportBridge backend; preserve MiniApp

Dockerfile

@@ -1,20 +1,8 @@
-FROM python:3.13.5-slim
-
+FROM python:3.11-slim
 WORKDIR /app
-
-RUN apt-get update && apt-get install -y \
-    build-essential \
-    curl \
-    git \
-    && rm -rf /var/lib/apt/lists/*
-
 COPY requirements.txt ./
-COPY src/ ./src/
-
-RUN pip3 install -r requirements.txt
-
-EXPOSE 8501
-
-HEALTHCHECK CMD curl --fail http://localhost:8501/_stcore/health
-
-ENTRYPOINT ["streamlit", "run", "src/streamlit_app.py", "--server.port=8501", "--server.address=0.0.0.0"]
+RUN pip install --no-cache-dir -r requirements.txt
+COPY *.py ./
+COPY certs/ ./certs/
+EXPOSE 7860
+CMD ["uvicorn", "service:app", "--host", "0.0.0.0", "--port", "7860", "--workers", "1"]

README.md

@@ -1,19 +1,60 @@
 ---
-title: VKPRTool
-emoji: 🚀
-colorFrom: red
-colorTo: red
+title: ReportBridge
+emoji: 📋
+colorFrom: green
+colorTo: blue
 sdk: docker
-app_port: 8501
-tags:
-- streamlit
+app_port: 7860
 pinned: false
-short_description: Streamlit template space
 ---
 
-# Welcome to Streamlit!
+# ReportBridge
 
-Edit `/src/streamlit_app.py` to customize this app to your heart's desire. :heart:
+Отдельный экспериментальный бэкенд расширения Медиалогии. Production Space
+`PRAnalytics/MiniApp` не вызывается, не изменяется и не перезапускается.
+`legacy_parser.py`, `max_api.py` и сертификат скопированы из локального снимка
+MiniApp, commit `88344382f5b6a32564ef2546f0332f0562a30baf`. Секреты не копировались.
 
-If you have any questions, checkout our [documentation](https://docs.streamlit.io) and [community
-forums](https://discuss.streamlit.io).
+HF Settings -> Secrets:
+
+- `BRIDGE_API_KEY`: отдельный длинный случайный пароль для расширения; НЕ токен HF.
+- `VK_ACCESS_TOKEN`: токен VK, если нужны просмотры VK.
+- `MAX_ACCESS_TOKEN`: токен бота Макс, если нужны просмотры Макс.
+- `JINA_API_KEY`: необязательно, как в старом парсере.
+
+`GET /health` сообщает только наличие настроек, не значения.
+`POST /jobs` принимает `{request_id: UUID, links: [URL, ...]}`.
+`GET /jobs/{id}` возвращает фактический прогресс и результат.
+Для обоих `/jobs` нужен заголовок `X-Bridge-Key`.
+Идентификатор запроса предотвращает дублирование запуска при сетевых повторах.
+500 ссылок на задание, максимум 5 активных заданий, 1 задание одновременно,
+3 параллельных запроса Telegram/Макс. Очередь и результаты хранятся только
+в памяти до перезапуска; готовые задания удаляются при новом запуске через час.
+
+Исходящие обращения только к публичным страницам Telegram/Макс, Jina и VK API.
+Telegram с нулём из старого HTML-парсера считается неподтверждённым счётчиком,
+а не успешным результатом. Ошибки исключаются из суммы; отчёт явно неполный.
+Внешние API могут быть недоступны или менять разметку; очередь не отменяет
+эти ограничения. Для защищённых материалов `/c/` нужен отдельный сценарий,
+они не поддерживаются.
+
+## Статус запуска
+
+27 октября 2026 репозиторий создан отдельно, но HF остановил выделение CPU
+для нового Space: `PAUSED`, `Quota exceeded for flavor cpu-basic
+(requested=1): current=2, limit=0`. Платные ресурсы не подключались.
+На этапе первоначального создания существующие Spaces не изменялись. Для работы облачного
+API нужно разрешённое HF выделение CPU в этой организации.
+Бэкенд отдельно проверен локально; реальный запрос Telegram получил просмотры.
+Тесты сервиса и обработчиков не требуют доступа к рабочему MiniApp.
+
+
+## Перенос в PR_Tool
+
+По явному разрешению владельца 27 октября 2026 бэкенд переносится в старый
+`PRAnalytics/PR_Tool`, заменяя его Streamlit-интерфейс. `MiniApp` по-прежнему
+не изменяется. Облачная сборка расширения использует
+`https://pranalytics-pr-tool.hf.space`. Старые файлы и ревизия
+`e2c307ec07d010e9b03a9c04ae2e65ed957e7c66` сохранены для отката.
+В PR_Tool до переноса не было Secrets и Variables; токены VK и Макс
+из другого Space автоматически не переносились.

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-----

legacy_parser.py

@@ -0,0 +1,794 @@
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor, wait
+import threading
+import time
+import os
+import re
+import html as html_lib
+from typing import List, Optional, Tuple, Union
+
+import requests
+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")
+VK_ACCESS_TOKEN = os.getenv("VK_ACCESS_TOKEN", "")
+
+# --- Таймауты и параллелизм (всё настраивается через переменные окружения) ---
+# Таймаут одного HTTP-запроса задаём кортежем (connect, read), чтобы зависший
+# коннект не съедал всё время.
+# С датацентр-IP (Hugging Face) Telegram блокирует/глушит прямые запросы к t.me,
+# поэтому основной путь — прокси r.jina.ai (см. process_telegram_link). Прямой
+# t.me остаётся резервом и потому держится «fail-fast».
+# max(..., floor) — защита от слишком маленьких значений в env (например,
+# случайно выставленного TELEGRAM_OP_TIMEOUT=3.05), которые рвут TG-запросы.
+_TG_CONNECT_TIMEOUT = max(float(os.getenv("TELEGRAM_CONNECT_TIMEOUT", "8")), 4.0)
+_TG_READ_TIMEOUT = max(float(os.getenv("TELEGRAM_OP_TIMEOUT", "8")), 5.0)
+TG_HTTP_TIMEOUT = (_TG_CONNECT_TIMEOUT, _TG_READ_TIMEOUT)
+# Прокси ходит к Telegram со своей стороны — даём ему больше времени.
+_TG_PROXY_TIMEOUT = max(float(os.getenv("TELEGRAM_PROXY_TIMEOUT", "25")), 10.0)
+TG_PROXY_HTTP_TIMEOUT = (10.0, _TG_PROXY_TIMEOUT)
+_MAX_CONNECT_TIMEOUT = max(float(os.getenv("MAX_CONNECT_TIMEOUT", "10")), 8.0)
+_MAX_READ_TIMEOUT = max(float(os.getenv("MAX_OP_TIMEOUT", "12")), 8.0)
+MAX_HTTP_TIMEOUT = (_MAX_CONNECT_TIMEOUT, _MAX_READ_TIMEOUT)
+REQUEST_TIMEOUT = float(os.getenv("REQUEST_TIMEOUT", "15"))  # для VK API
+
+TELEGRAM_CONCURRENCY = int(os.getenv("TELEGRAM_CONCURRENCY", "6"))
+MAX_CONCURRENCY = int(os.getenv("MAX_CONCURRENCY", "8"))
+
+VK_MAX_RETRIES = int(os.getenv("VK_MAX_RETRIES", "5"))
+VK_RETRY_DELAY = float(os.getenv("VK_RETRY_DELAY", "0.45"))
+VK_BATCH_SIZE = int(os.getenv("VK_BATCH_SIZE", "100"))
+
+# Прокси r.jina.ai — основной путь для Telegram на HF (прямой t.me заблокирован).
+# Включён по умолчанию; можно отключить через ENABLE_JINA_FALLBACK=0.
+# X-Return-Format: html заставляет jina вернуть СЫРОЙ HTML виджета t.me
+# (с tgme_widget_message_views и owner_name), который и парсит наш код.
+ENABLE_JINA_FALLBACK = os.getenv("ENABLE_JINA_FALLBACK", "1") != "0"
+_JINA_HEADERS = {"X-Return-Format": "html"}
+_JINA_API_KEY = os.getenv("JINA_API_KEY", "").strip()
+if _JINA_API_KEY:
+    # Ключ снимает строгие лимиты бесплатного тарифа r.jina.ai.
+    _JINA_HEADERS["Authorization"] = f"Bearer {_JINA_API_KEY}"
+
+# Общий бюджет времени на весь запрос. По истечении возвращаем то, что успели
+# обработать, а остальное помечаем как «превышено время» — вместо того, чтобы
+# мини-приложение отвалилось по таймауту целиком.
+TOTAL_TIME_BUDGET = float(os.getenv("TOTAL_TIME_BUDGET", "25"))
+
+# Предохранитель от слишком больших списков. Лишнее отрезаем с пометкой.
+MAX_INPUT_LINKS = int(os.getenv("MAX_INPUT_LINKS", "1200"))
+
+app = FastAPI(title=APP_NAME)
+
+# Для простоты разрешаем все источники. Можно сузить список доменов в проде.
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=False,
+    allow_methods=["POST", "GET", "OPTIONS"],
+    allow_headers=["*"],
+)
+
+
+# ---------- HTTP-сессии с переиспользованием соединений ----------
+# Отдельная сессия на поток (потокобезопасно) + пул соединений, чтобы не
+# открывать TCP/TLS заново на каждую ссылку.
+_thread_local = threading.local()
+
+
+def _get_session() -> requests.Session:
+    sess = getattr(_thread_local, "session", None)
+    if sess is None:
+        sess = requests.Session()
+        adapter = requests.adapters.HTTPAdapter(
+            pool_connections=4, pool_maxsize=4, max_retries=0
+        )
+        sess.mount("http://", adapter)
+        sess.mount("https://", adapter)
+        sess.headers.update({"User-Agent": "Mozilla/5.0"})
+        _thread_local.session = sess
+    return sess
+
+
+class ParseRequest(BaseModel):
+    links: Union[List[str], str]
+
+
+class ParseResponse(BaseModel):
+    html: str
+    text: str
+    telegram_total: int
+    vk_total: int
+    errors: List[str]
+    max_count: int = 0
+
+
+# ---------- Вспомогательные функции ----------
+
+def clean_telegram_title(raw: str, fallback: str) -> str:
+    title = re.sub(
+        r"<i\b[^>]*class=[\"'][^\"']*emoji[^\"']*[\"'][^>]*>.*?</i>",
+        "",
+        (raw or "").strip(),
+        flags=re.IGNORECASE | re.DOTALL,
+    )
+    title = re.sub(r"<[^>]+>", "", title)
+    title = html_lib.unescape(title)
+    title = re.sub(r"\s*[-|]\s*Telegram\s*$", "", title, flags=re.IGNORECASE)
+    title = re.sub(
+        r"[\U0001F1E6-\U0001F1FF\U0001F300-\U0001FAFF\u2600-\u27BF\uFE0E\uFE0F]",
+        "",
+        title,
+    )
+    title = re.sub(r"\s+", " ", title).strip()
+    return title or fallback
+
+
+def clean_max_title(raw: str, fallback: str) -> str:
+    title = re.sub(r"<[^>]+>", "", (raw or "").strip())
+    title = html_lib.unescape(title)
+    # убираем хвосты вида " — MAX", " | MAX", " - MAX"
+    title = re.sub(r"\s*[-—|·]\s*MAX\s*$", "", title, flags=re.IGNORECASE)
+    title = re.sub(r"\s+", " ", title).strip()
+    return title or fallback
+
+
+def human_format_views(num: int) -> str:
+    if num >= 1_000_000:
+        value = num / 1_000_000
+        suffix = "M"
+    else:
+        value = num / 1000
+        suffix = "K"
+
+    rounded = round(value, 1)
+    if rounded.is_integer():
+        formatted = f"{int(rounded)}{suffix}"
+    else:
+        formatted = f"{rounded:.1f}{suffix}"
+
+    return formatted.replace(".", ",")
+
+
+def _parse_views_number(raw: str) -> int:
+    if not raw:
+        return 0
+    normalized = raw.strip().replace("\u00a0", "").replace(" ", "").replace(",", ".")
+    match = re.match(r"([\d\.]+)\s*([kKmM]?)", normalized)
+    if not match:
+        return 0
+    value = float(match.group(1))
+    suffix = match.group(2).lower()
+    if suffix == "k":
+        value *= 1_000
+    elif suffix == "m":
+        value *= 1_000_000
+    return int(value)
+
+
+def detect_platform(link: str) -> str:
+    link = link.strip().lower()
+    if "t.me/" in link or "telegram.me/" in link:
+        return "telegram"
+    if "vk.com/wall" in link or "vk.ru/wall" in link:
+        return "vk"
+    if "max.ru/" in link:
+        return "max"
+    return ""
+
+
+def canonicalize_tg_link(link: str) -> Tuple[Optional[str], Optional[str], Optional[str]]:
+    link = link.strip()
+    private_match = re.match(r"https?://(?:t(?:elegram)?\.me)/c/(\d+)/(\d+)", link, re.IGNORECASE)
+    if private_match:
+        return link, None, None
+    m = re.match(r"https?://(?:t(?:elegram)?\.me)/(?:s/)?([^/]+)/(?P<id>\d+)", link, re.IGNORECASE)
+    if not m:
+        return None, None, None
+    username = m.group(1)
+    message_id = m.group("id")
+    canonical = f"https://t.me/{username}/{message_id}"
+    return canonical, username, message_id
+
+
+def canonicalize_vk_link(link: str) -> Tuple[Optional[str], Optional[int], Optional[str]]:
+    link = link.strip()
+    m = re.search(r"(?:vk\.com|vk\.ru)/wall(-?\d+)_(\d+)", link)
+    if not m:
+        return None, None, None
+    owner_id = int(m.group(1))
+    post_id = m.group(2)
+    canonical = f"https://vk.com/wall{owner_id}_{post_id}"
+    return canonical, owner_id, post_id
+
+
+def canonicalize_max_link(link: str) -> Tuple[Optional[str], Optional[str], Optional[str]]:
+    link = link.strip()
+    m = re.match(r"https?://(?:[\w-]+\.)?max\.ru/([^/?#]+)/([^/?#]+)", link, re.IGNORECASE)
+    if not m:
+        return None, None, None
+    slug = m.group(1)
+    post_id = m.group(2)
+    canonical = f"https://max.ru/{slug}/{post_id}"
+    return canonical, slug, post_id
+
+
+# ---------- Обработчики одной ссылки ----------
+
+def process_telegram_link(channel_username: str, message_id: str, canonical_link: str) -> Tuple[str, int]:
+    urls = []
+    if ENABLE_JINA_FALLBACK:
+        urls.extend([
+            (
+                f"https://r.jina.ai/https://t.me/{channel_username}/{message_id}?embed=1",
+                _JINA_HEADERS,
+                TG_PROXY_HTTP_TIMEOUT,
+            ),
+            (
+                f"https://r.jina.ai/http://t.me/{channel_username}/{message_id}?embed=1",
+                _JINA_HEADERS,
+                TG_PROXY_HTTP_TIMEOUT,
+            ),
+            (
+                f"https://r.jina.ai/http://t.me/s/{channel_username}/{message_id}",
+                _JINA_HEADERS,
+                TG_PROXY_HTTP_TIMEOUT,
+            ),
+            (
+                f"https://r.jina.ai/http://telegram.me/{channel_username}/{message_id}?embed=1",
+                _JINA_HEADERS,
+                TG_PROXY_HTTP_TIMEOUT,
+            ),
+        ])
+
+    urls.extend([
+        (f"https://t.me/{channel_username}/{message_id}?embed=1", None, TG_HTTP_TIMEOUT),
+        (f"https://t.me/s/{channel_username}/{message_id}", None, TG_HTTP_TIMEOUT),
+        (f"https://telegram.me/{channel_username}/{message_id}?embed=1", None, TG_HTTP_TIMEOUT),
+    ])
+
+    session = _get_session()
+    page_html = None
+    last_err: Optional[Exception] = None
+    for url, headers, timeout in urls:
+        try:
+            resp = session.get(url, headers=headers, timeout=timeout)
+            if resp.status_code == 200 and resp.text:
+                page_html = resp.text
+                break
+        except Exception as exc:  # noqa: BLE001
+            last_err = exc
+
+    if page_html is None:
+        return f"Ошибка (TG) при обработке {canonical_link}: {last_err or 'нет ответа'}", 0
+
+    title = None
+    meta_title = re.search(
+        r"<meta[^>]*property=[\"']og:title[\"'][^>]*content=[\"']([^\"']+)[\"']",
+        page_html,
+        re.IGNORECASE,
+    )
+    if meta_title:
+        title = meta_title.group(1).strip()
+    if not title:
+        owner_title = re.search(
+            r'class="tgme_widget_message_owner_name".*?<span[^>]*>(.*?)</span>',
+            page_html,
+            re.IGNORECASE | re.DOTALL,
+        )
+        if owner_title:
+            title = owner_title.group(1).strip()
+    title = clean_telegram_title(title or "", channel_username)
+
+    views = 0
+    widget_views = re.search(
+        r'class="tgme_widget_message_views[^"]*">([\d\s\.,kKmM]+)<',
+        page_html,
+        re.IGNORECASE,
+    )
+    if widget_views:
+        views = _parse_views_number(widget_views.group(1))
+    else:
+        candidates = re.findall(
+            r"(\d[\d\s\.,]*)([kKmM]?)\s*(?:views|просмотр|просмотра|просмотров|переглядів|visualizações|visualizzazioni|ansichten|visninger|visitas)",
+            page_html,
+            re.IGNORECASE,
+        )
+        if candidates:
+            last_num, last_suffix = candidates[-1]
+            views = _parse_views_number(last_num + last_suffix)
+
+    return title, views
+
+
+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]:
+    """Одиночный запрос к VK (запасной путь; основной — батч process_vk_batch)."""
+    posts_param = f"{owner_id}_{post_id}"
+    api_url = "https://api.vk.com/method/wall.getById"
+
+    data = None
+    for attempt in range(1, VK_MAX_RETRIES + 1):
+        params = {"posts": posts_param, "v": VK_API_VERSION, "extended": 1}
+        if VK_ACCESS_TOKEN:
+            params["access_token"] = VK_ACCESS_TOKEN
+        try:
+            resp = requests.get(api_url, params=params, timeout=REQUEST_TIMEOUT)
+        except Exception as exc:  # noqa: BLE001
+            return f"**Ошибка (VK)**: {exc} — {canonical_link}", 0
+        if resp.status_code != 200:
+            return f"**Ошибка (VK)**: HTTP {resp.status_code} — {canonical_link}", 0
+        data = resp.json()
+        if "error" in data:
+            error = data["error"]
+            error_msg = error.get("error_msg", "")
+            error_code = error.get("error_code")
+            if error_code == 6 or "too many requests per second" in error_msg.lower():
+                time.sleep(VK_RETRY_DELAY * attempt)
+                continue
+            return f"**Ошибка (VK)**: {error_msg or 'API error'} — {canonical_link}", 0
+        break
+    else:
+        return f"**Ошибка (VK)**: Too many requests per second — {canonical_link}", 0
+
+    response_data = (data or {}).get("response", {})
+    items = response_data.get("items", [])
+    if not items:
+        return f"**Ошибка (VK)**: пост не найден — {canonical_link}", 0
+
+    post = items[0]
+    views = post.get("views", {}).get("count", 0)
+    post_owner_id = post.get("owner_id", owner_id)
+
+    title = None
+    if post_owner_id < 0:
+        group_id = -post_owner_id
+        for group in response_data.get("groups", []):
+            if group.get("id") == group_id:
+                title = group.get("name")
+                break
+    else:
+        for profile in response_data.get("profiles", []):
+            if profile.get("id") == post_owner_id:
+                title = (profile.get("first_name", "") + " " + profile.get("last_name", "")).strip()
+                break
+    if not title:
+        title = "VK пост"
+    return title, views
+
+
+def process_vk_batch(batch_items: List[Tuple[str, int, str]]) -> dict:
+    api_url = "https://api.vk.com/method/wall.getById"
+    posts = ",".join(f"{owner_id}_{post_id}" for _, owner_id, post_id in batch_items)
+
+    data = None
+    for attempt in range(1, VK_MAX_RETRIES + 1):
+        params = {"posts": posts, "v": VK_API_VERSION, "extended": 1}
+        if VK_ACCESS_TOKEN:
+            params["access_token"] = VK_ACCESS_TOKEN
+        try:
+            resp = requests.get(api_url, params=params, timeout=REQUEST_TIMEOUT)
+        except Exception as exc:  # noqa: BLE001
+            return {c: (f"**Ошибка (VK)**: {exc} — {c}", 0) for c, _, _ in batch_items}
+
+        if resp.status_code != 200:
+            return {c: (f"**Ошибка (VK)**: HTTP {resp.status_code} — {c}", 0) for c, _, _ in batch_items}
+
+        data = resp.json()
+        if "error" in data:
+            error = data["error"]
+            error_msg = error.get("error_msg", "")
+            error_code = error.get("error_code")
+            if error_code == 6 or "too many requests per second" in error_msg.lower():
+                time.sleep(VK_RETRY_DELAY * attempt)
+                continue
+            return {c: (f"**Ошибка (VK)**: {error_msg or 'API error'} — {c}", 0) for c, _, _ in batch_items}
+        break
+    else:
+        return {c: (f"**Ошибка (VK)**: Too many requests per second — {c}", 0) for c, _, _ in batch_items}
+
+    response_data = (data or {}).get("response", {})
+    items = response_data.get("items", [])
+    groups = {group.get("id"): group for group in response_data.get("groups", [])}
+    profiles = {profile.get("id"): profile for profile in response_data.get("profiles", [])}
+    item_map = {
+        f"{item.get('owner_id')}_{item.get('id')}": item
+        for item in items
+        if item.get("owner_id") is not None and item.get("id") is not None
+    }
+
+    result: dict = {}
+    for canonical, owner_id, post_id in batch_items:
+        key = f"{owner_id}_{post_id}"
+        post = item_map.get(key)
+        if not post:
+            result[canonical] = (f"**Ошибка (VK)**: пост не найден — {canonical}", 0)
+            continue
+
+        views = post.get("views", {}).get("count", 0)
+        post_owner_id = post.get("owner_id", owner_id)
+        title = None
+        if post_owner_id < 0:
+            group = groups.get(-post_owner_id)
+            if group:
+                title = group.get("name")
+        else:
+            profile = profiles.get(post_owner_id)
+            if profile:
+                title = (profile.get("first_name", "") + " " + profile.get("last_name", "")).strip()
+
+        result[canonical] = (title or "VK пост", views)
+
+    return result
+
+
+# ---------- Параллельное разрешение ссылок с бюджетом времени ----------
+
+def _resolve_pool(tasks, worker, max_workers, deadline, timeout_label, defaults=None):
+    """tasks: список (canonical, args_tuple). worker(*args) -> (title, views).
+    Возвращаем dict canonical -> (title, views). Незавершённые к дедлайну
+    помечаем timeout_label (или оставляем значение из defaults, если оно есть)."""
+    results = dict(defaults) if defaults else {}
+    if not tasks:
+        return results
+
+    workers = max(1, min(max_workers, len(tasks)))
+    ex = ThreadPoolExecutor(max_workers=workers)
+    fut_map = {ex.submit(worker, *args): canonical for canonical, args in tasks}
+
+    timeout = max(0.0, deadline - time.monotonic())
+    done, not_done = wait(list(fut_map.keys()), timeout=timeout)
+
+    for fut in done:
+        canonical = fut_map[fut]
+        try:
+            results[canonical] = fut.result()
+        except Exception as exc:  # noqa: BLE001
+            if canonical not in results:
+                results[canonical] = (f"Ошибка при обработке {canonical}: {exc}", 0)
+
+    for fut in not_done:
+        canonical = fut_map[fut]
+        fut.cancel()
+        if canonical not in results:
+            results[canonical] = (timeout_label, 0)
+
+    # Не блокируемся на «зависших» запросах — отдаём управление сразу.
+    try:
+        ex.shutdown(wait=False, cancel_futures=True)
+    except TypeError:  # Python < 3.9
+        ex.shutdown(wait=False)
+
+    return results
+
+
+def resolve_telegram(tg_tasks, deadline):
+    # tg_tasks: список (canonical, username, message_id)
+    tasks = [(c, (u, mid, c)) for (c, u, mid) in tg_tasks if u and mid]
+    return _resolve_pool(
+        tasks,
+        process_telegram_link,
+        TELEGRAM_CONCURRENCY,
+        deadline,
+        "Ошибка (TG): превышено время обработки",
+    )
+
+
+def resolve_max(max_tasks, deadline):
+    # max_tasks: список (canonical, slug, post_id)
+    defaults = {c: (slug, None) for (c, slug, _post) in max_tasks}
+    tasks = [(c, (slug, post_id, c)) for (c, slug, post_id) in max_tasks]
+    fetched = _resolve_pool(
+        tasks, process_max_link, MAX_CONCURRENCY, deadline, None, defaults=defaults
+    )
+    return fetched
+
+
+def resolve_vk(vk_tasks, deadline):
+    # vk_tasks: список (canonical, owner_id, post_id)
+    results = {}
+    valid = [(c, o, p) for (c, o, p) in vk_tasks if o is not None and p is not None]
+    batch_size = max(1, VK_BATCH_SIZE)
+    for start in range(0, len(valid), batch_size):
+        if time.monotonic() > deadline:
+            for c, _, _ in valid[start:]:
+                results[c] = ("**Ошибка (VK)**: превышено время обработки", 0)
+            break
+        chunk = valid[start:start + batch_size]
+        results.update(process_vk_batch(chunk))
+    return results
+
+
+def normalize_links(links: Union[List[str], str]) -> List[str]:
+    if isinstance(links, str):
+        raw = links.splitlines()
+    else:
+        raw = []
+        for item in links:
+            raw.extend(str(item).splitlines())
+    return [line.strip() for line in raw if line.strip()]
+
+
+def _escape(text: str) -> str:
+    return html_lib.escape(text, quote=True)
+
+
+def _fmt_views(v) -> str:
+    return "—" if v is None else human_format_views(v)
+
+
+def _render_groups(groups, has_views, default_title):
+    """Сортируем группы и рендерим в HTML (<ol>) и текст. Возвращаем (html_str, text_lines)."""
+    if has_views:
+        ordered = sorted(
+            groups.items(),
+            key=lambda kv: sum((v or 0) for _, v in kv[1]["items"]),
+            reverse=True,
+        )
+    else:
+        ordered = sorted(groups.items(), key=lambda kv: (kv[1]["title"] or "").lower())
+
+    html_lines = ["<ol>"]
+    text_lines: List[str] = []
+    for idx, (_, data) in enumerate(ordered, start=1):
+        title = data["title"] or default_title
+        items = data["items"]
+        first_link, _first_views = items[0]
+
+        line_html = f'<li><a href="{_escape(first_link)}">{_escape(title)}</a>'
+        if len(items) > 1:
+            for link2, _v in items[1:]:
+                line_html += f' + <a href="{_escape(link2)}">ещё</a>'
+
+        if has_views:
+            views_str = " + ".join(_fmt_views(v) for _, v in items)
+            line_html += f" — {views_str}</li>"
+        else:
+            line_html += " — </li>"
+        html_lines.append(line_html)
+
+        line_text = f"{idx}. {title} ({first_link})"
+        if len(items) > 1:
+            line_text += " + ещё: " + ", ".join(link2 for link2, _v in items[1:])
+        if has_views:
+            line_text += f" — {views_str}"
+        else:
+            line_text += " —"
+        text_lines.append(line_text)
+
+    html_lines.append("</ol>")
+    return "\n".join(html_lines), text_lines
+
+
+def build_output(lines, deadline):
+    errors: List[str] = []
+
+    tg_lines = [ln for ln in lines if ln[0] == "telegram"]
+    vk_lines = [ln for ln in lines if ln[0] == "vk"]
+    max_lines = [ln for ln in lines if ln[0] == "max"]
+
+    tg_fetch = [(c, a, b) for _, c, a, b in tg_lines if a and b]
+    vk_fetch = [(c, a, b) for _, c, a, b in vk_lines if a is not None and b is not None]
+    max_fetch = [(c, a, b) for _, c, a, b in max_lines if a and b]
+
+    # Telegram, VK и MAX обрабатываем параллельно (раньше шли друг за другом).
+    tg_results, vk_results, max_results = {}, {}, {}
+    with ThreadPoolExecutor(max_workers=3) as top:
+        f_tg = top.submit(resolve_telegram, tg_fetch, deadline)
+        f_vk = top.submit(resolve_vk, vk_fetch, deadline)
+        f_max = top.submit(resolve_max, max_fetch, deadline)
+        tg_results = f_tg.result()
+        vk_results = f_vk.result()
+        max_results = f_max.result()
+
+    tg_groups: dict = {}
+    vk_groups: dict = {}
+    max_groups: dict = {}
+
+    for plat, canonical, a, b in lines:
+        if plat == "telegram":
+            username, mid = a, b
+            if not username or not mid:
+                key = canonical
+                invalid_title = "Ошибка (TG): ссылка Telegram недоступна для публичного парсинга"
+                if re.search(r"https?://(?:t(?:elegram)?\.me)/c/\d+/\d+", canonical, re.IGNORECASE):
+                    invalid_title = "Ошибка (TG): ссылки вида t.me/c/... не поддерживаются"
+                tg_groups.setdefault(key, {"title": invalid_title, "items": []})
+                tg_groups[key]["items"].append((canonical, 0))
+            else:
+                title, views = tg_results.get(canonical, (username, 0))
+                key = username
+                if key not in tg_groups:
+                    tg_groups[key] = {"title": title, "items": []}
+                if not tg_groups[key].get("title") or str(tg_groups[key]["title"]).startswith("Ошибка"):
+                    tg_groups[key]["title"] = title
+                tg_groups[key]["items"].append((canonical, views))
+
+        elif plat == "vk":
+            owner_id, post_id = a, b
+            if owner_id is None or post_id is None:
+                key = canonical
+                vk_groups.setdefault(key, {"title": "VK пост", "items": []})
+                vk_groups[key]["items"].append((canonical, 0))
+            else:
+                title, views = vk_results.get(canonical, (f"**Ошибка (VK)**: нет данных — {canonical}", 0))
+                key = str(owner_id)
+                if key not in vk_groups:
+                    vk_groups[key] = {"title": title, "items": []}
+                if not vk_groups[key].get("title") or str(vk_groups[key]["title"]).startswith("**Ошибка"):
+                    vk_groups[key]["title"] = title
+                vk_groups[key]["items"].append((canonical, views))
+
+        elif plat == "max":
+            slug, post_id = a, b
+            if not slug or not post_id:
+                key = canonical
+                max_groups.setdefault(key, {"title": "MAX", "items": []})
+                max_groups[key]["items"].append((canonical, None))
+            else:
+                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, views))
+                if views is None:
+                    errors.append(f"Не удалось получить просмотры MAX: {canonical}")
+
+        else:
+            errors.append(f"Неизвестная платформа: {canonical}")
+
+    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] = []
+
+    # --- Telegram ---
+    html_lines.append("<h2>Telegram</h2>")
+    html_lines.append(f'Суммарно посты собрали <b>{human_format_views(tg_total_views)}</b> просмотров.')
+    text_lines.append("Telegram")
+    text_lines.append(f"Суммарно посты собрали {human_format_views(tg_total_views)} просмотров.")
+    if tg_groups:
+        h, t = _render_groups(tg_groups, has_views=True, default_title="Telegram")
+        html_lines.append(h)
+        text_lines.extend(t)
+    else:
+        html_lines.append("<i>Нет ссылок на Telegram</i>")
+        text_lines.append("Нет ссылок на Telegram")
+    html_lines.append("<br/>")
+    text_lines.append("")
+
+    # --- VK ---
+    html_lines.append("<h2>ВКонтакте</h2>")
+    html_lines.append(f'Суммарно посты собрали <b>{human_format_views(vk_total_views)}</b> просмотров.')
+    text_lines.append("ВКонтакте")
+    text_lines.append(f"Суммарно посты собрали {human_format_views(vk_total_views)} просмотров.")
+    if vk_groups:
+        h, t = _render_groups(vk_groups, has_views=True, default_title="VK пост")
+        html_lines.append(h)
+        text_lines.extend(t)
+    else:
+        html_lines.append("<i>Нет ссылок на ВКонтакте</i>")
+        text_lines.append("Нет ссылок на ВКонтакте")
+    html_lines.append("<br/>")
+    text_lines.append("")
+
+    # --- MAX (секцию показываем только если есть ссылки MAX) ---
+    if max_groups:
+        html_lines.append("<h2>Макс</h2>")
+        text_lines.append("Макс")
+        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/>")
+        text_lines.append("")
+
+    if errors:
+        html_lines.append("<h2>Ошибки</h2>")
+        html_lines.append("<ul>")
+        for err in errors:
+            html_lines.append(f"<li>{_escape(err)}</li>")
+            text_lines.append(f"- {err}")
+        html_lines.append("</ul>")
+        html_lines.append("<br/>")
+        text_lines.append("")
+
+    return (
+        "\n".join(html_lines),
+        "\n".join(text_lines),
+        tg_total_views,
+        vk_total_views,
+        errors,
+        max_count,
+    )
+
+
+@app.get("/health")
+def health_check():
+    return {"status": "ok"}
+
+
+@app.post("/parse", response_model=ParseResponse)
+def parse_links(payload: ParseRequest):
+    deadline = time.monotonic() + TOTAL_TIME_BUDGET
+
+    raw_lines = normalize_links(payload.links)
+    if not raw_lines:
+        return ParseResponse(html="", text="", telegram_total=0, vk_total=0, errors=["Нет ссылок для обрабо��ки."])
+
+    truncated = False
+    if len(raw_lines) > MAX_INPUT_LINKS:
+        raw_lines = raw_lines[:MAX_INPUT_LINKS]
+        truncated = True
+
+    seen = set()
+    lines: List[Tuple[str, str, object, object]] = []
+    for link in raw_lines:
+        plat = detect_platform(link)
+        if plat == "telegram":
+            canonical, username, mid = canonicalize_tg_link(link)
+            if not canonical:
+                canonical, username, mid = link, None, None
+            if canonical in seen:
+                continue
+            seen.add(canonical)
+            lines.append(("telegram", canonical, username, mid))
+        elif plat == "vk":
+            canonical, owner_id, post_id = canonicalize_vk_link(link)
+            if not canonical:
+                canonical = link
+            if canonical in seen:
+                continue
+            seen.add(canonical)
+            lines.append(("vk", canonical, owner_id, post_id))
+        elif plat == "max":
+            canonical, slug, post_id = canonicalize_max_link(link)
+            if not canonical:
+                canonical, slug, post_id = link, None, None
+            if canonical in seen:
+                continue
+            seen.add(canonical)
+            lines.append(("max", canonical, slug, post_id))
+        else:
+            lines.append(("unknown", link, None, None))
+
+    html, text, tg_total, vk_total, errors, max_count = build_output(lines, deadline)
+
+    if truncated:
+        note = f"Обработаны первые {MAX_INPUT_LINKS} ссылок (список был длиннее)."
+        errors = errors + [note]
+        text = text + ("\n" if text else "") + "- " + note
+
+    return ParseResponse(
+        html=html,
+        text=text,
+        telegram_total=tg_total,
+        vk_total=vk_total,
+        errors=errors,
+        max_count=max_count,
+    )

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

reports.py

@@ -0,0 +1,91 @@
+"""Normalize public links and render structured parser results."""
+import html
+import re
+from collections import OrderedDict
+from urllib.parse import parse_qs, urlsplit
+
+
+def normalize(raw):
+    try:
+        url = urlsplit(raw.strip())
+        if url.scheme not in ("https", "http") or url.username or url.password or url.port:
+            raise ValueError("Нужна обычная ссылка http/https")
+        host = url.hostname.lower().removeprefix("www.").removeprefix("m.")
+        path = url.path.rstrip("/")
+        if host in ("t.me", "telegram.me"):
+            if path.startswith("/c/"):
+                raise ValueError("Закрытая ссылка Telegram: нужен публичный адрес поста")
+            match = re.fullmatch(r"/(?:s/)?([A-Za-z][A-Za-z0-9_]{2,})/([1-9]\d*)", path)
+            if match:
+                slug, post = match.groups()
+                slug = slug.lower()
+                return {"url": f"https://t.me/{slug}/{post}", "platform": "telegram", "group": slug, "a": slug, "b": post}
+        elif host in ("vk.com", "vk.ru"):
+            wall = path
+            if "w" in parse_qs(url.query):
+                wall = "/" + parse_qs(url.query)["w"][0]
+            match = re.fullmatch(r"/wall(-?[1-9]\d*)_([1-9]\d*)", wall)
+            if match and "reply" not in parse_qs(url.query):
+                owner, post = match.groups()
+                return {"url": f"https://vk.com/wall{owner}_{post}", "platform": "vk", "group": owner, "a": int(owner), "b": post}
+        elif host == "max.ru":
+            match = re.fullmatch(r"/([A-Za-z0-9_]+)/([A-Za-z0-9_-]+)", path)
+            if match:
+                slug, post = match.groups()
+                return {"url": f"https://max.ru/{slug}/{post}", "platform": "max", "group": slug.lower(), "a": slug, "b": post}
+        raise ValueError("Площадка или формат ссылки пока не поддерживаются")
+    except (AttributeError, TypeError):
+        raise ValueError("Некорректная ссылка") from None
+
+
+def human_views(value):
+    if value < 1000:
+        return str(value)
+    divisor, suffix = (1_000_000, "M") if value >= 1_000_000 else (1000, "K")
+    return f"{value / divisor:.1f}".rstrip("0").rstrip(".").replace(".", ",") + suffix
+
+
+def clean_title(value):
+    # Strip legacy HTML emoji representations as well as unicode emoji clusters.
+    value = re.sub(r"<[^>]+>", "", html.unescape(str(value)))
+    value = re.sub(r"[\U0001F000-\U0001FAFF\u2600-\u27BF\uFE0E\uFE0F\u200D]", "", value)
+    value = re.sub(r"[#*0-9]\u20e3", "", value)
+    return re.sub(r"\s+", " ", value).strip()
+
+
+def render(records):
+    chunks, plain, errors = [], [], []
+    for platform, label in (("telegram", "Telegram"), ("vk", "ВКонтакте"), ("max", "Макс")):
+        selected = [r for r in records if r["platform"] == platform]
+        if not selected:
+            continue
+        good = [r for r in selected if r.get("views") is not None and not r.get("error")]
+        groups = OrderedDict()
+        for record in good:
+            groups.setdefault(record["group"], []).append(record)
+        total = sum(r["views"] for r in good)
+        chunks.extend([f"<h2>{label}</h2>", f"<p>Суммарно посты собрали <b>{human_views(total)}</b> просмотров.</p>"])
+        plain.extend([label, f"Суммарно посты собрали {human_views(total)} просмотров."])
+        if len(good) != len(selected):
+            note = f"Неполный итог: получены просмотры {len(good)} из {len(selected)} постов."
+            chunks.append(f"<p><i>{note}</i></p>")
+            plain.append(note)
+        chunks.append("<ol>")
+        ordered = sorted(groups.values(), key=lambda items: sum(r["views"] for r in items), reverse=True)
+        for index, items in enumerate(ordered, 1):
+            title = clean_title(items[0]["title"]) or items[0]["group"]
+            links = [f'<a href="{html.escape(r["url"], quote=True)}">{html.escape(title if i == 0 else "ещё")}</a>' for i, r in enumerate(items)]
+            views = " + ".join(human_views(r["views"]) for r in items)
+            chunks.append(f'<li>{" + ".join(links)} — {views}</li>')
+            plain.append(f'{index}. {title}{" + ещё" * (len(items) - 1)} — {views}')
+        chunks.append("</ol>")
+        plain.append("")
+    for record in records:
+        if record.get("error"):
+            errors.append(f'{record["url"]}: {record["error"]}')
+    if errors:
+        chunks.append("<h2>Не удалось обработать</h2><ul>")
+        chunks.extend(f"<li>{html.escape(error)}</li>" for error in errors)
+        chunks.append("</ul>")
+        plain.extend(["Не удалось обработать", *errors])
+    return {"html": "\n".join(chunks), "text": "\n".join(plain), "errors": errors, "records": records}

requirements.txt

@@ -1,4 +1,5 @@
-streamlit==1.36.0
-requests==2.31.0
-nest-asyncio==1.6.0
-pyaes==1.6.0
+fastapi==0.162.0
+uvicorn[standard]==0.54.0
+requests==2.34.2
+pydantic==2.14.1
+json5==0.17.3

service.py

@@ -0,0 +1,144 @@
+"""Isolated report service. Never calls or modifies the production MiniApp Space."""
+import hmac
+import logging
+import os
+import threading
+import time
+import uuid
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from typing import List
+
+from fastapi import Depends, FastAPI, Header, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel, Field
+
+import legacy_parser as engine
+from reports import normalize, render
+
+log = logging.getLogger(__name__)
+API_KEY = os.getenv("BRIDGE_API_KEY", "")
+app = FastAPI(title="ReportBridge", version="0.1.1")
+app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=False,
+                   allow_methods=["GET", "POST", "DELETE"], allow_headers=["Content-Type", "X-Bridge-Key"])
+jobs, lock = {}, threading.RLock()
+queue = ThreadPoolExecutor(max_workers=1, thread_name_prefix="reports")
+
+
+def authorize(x_bridge_key: str = Header(default="")):
+    if not API_KEY:
+        raise HTTPException(503, "В бэкенде расширения не настроен BRIDGE_API_KEY")
+    if not hmac.compare_digest(x_bridge_key.encode(), API_KEY.encode()):
+        raise HTTPException(401, "Неверный ключ ReportBridge")
+
+
+class JobRequest(BaseModel):
+    request_id: str = Field(min_length=16, max_length=64, pattern=r"^[a-zA-Z0-9-]+$")
+    links: List[str] = Field(min_length=1, max_length=500)
+
+
+def fetch_one(record):
+    result = dict(record)
+    try:
+        if record["platform"] == "telegram":
+            title, views = engine.process_telegram_link(record["a"], record["b"], record["url"])
+        else:
+            title, views = engine.process_max_link(record["a"], record["b"], record["url"])
+        if "ошибка" in title.lower() or views is None:
+            raise ValueError("Источник не вернул просмотры. Попробуйте повторить позже.")
+        # The legacy HTML parser returns 0 when it cannot find a counter. Avoid a false zero in new reports.
+        if record["platform"] == "telegram" and views == 0:
+            raise ValueError("Не найден счётчик просмотров Telegram; это не подтверждённый ноль")
+        result.update(title=title, views=views)
+    except Exception:
+        result.update(title=record["group"], views=None, error="Не удалось получить просмотры публичного поста")
+    return result
+
+
+def run_job(job_id, records):
+    result_map = {}
+    with lock:
+        jobs[job_id]["status"] = "running"
+
+    def store_result(record):
+        with lock:
+            result_map[record["url"]] = record
+            jobs[job_id]["completed"] = len(result_map)
+
+    try:
+        with ThreadPoolExecutor(max_workers=3) as pool:
+            futures = {pool.submit(fetch_one, r): r for r in records if r["platform"] != "vk"}
+            vk = [r for r in records if r["platform"] == "vk"]
+            for start in range(0, len(vk), 100):
+                batch = vk[start:start + 100]
+                if not engine.VK_ACCESS_TOKEN:
+                    data = {}
+                else:
+                    data = engine.process_vk_batch([(r["url"], r["a"], r["b"]) for r in batch])
+                for record in batch:
+                    title, views = data.get(record["url"], ("Ошибка: VK_ACCESS_TOKEN не настроен", None))
+                    error = ("Не настроен VK_ACCESS_TOKEN в Secrets PR_Tool" if not engine.VK_ACCESS_TOKEN
+                             else "Не удалось получить просмотры VK") if "ошибка" in title.lower() else None
+                    store_result(dict(record, title=title if not error else record["group"], views=None if error else views, error=error))
+                if start + 100 < len(vk):
+                    time.sleep(0.4)
+            for future in as_completed(futures):
+                store_result(future.result())
+        output = render([result_map[r["url"]] for r in records])
+        with lock:
+            jobs[job_id].update(status="complete", result=output)
+    except Exception:
+        log.exception("ReportBridge job failed")
+        with lock:
+            jobs[job_id].update(status="failed", error="Ошибка нового бэкенда. Рабочий MiniApp не затронут.")
+
+
+@app.get("/")
+def root():
+    return {"service": "ReportBridge", "production_miniapp": "not used", "health": "/health"}
+
+
+@app.get("/health")
+def health():
+    return {"status": "ok", "service": "ReportBridge", "configured": {
+        "bridge": bool(API_KEY), "vk": bool(engine.VK_ACCESS_TOKEN), "max": bool(os.getenv("MAX_ACCESS_TOKEN"))}}
+
+
+@app.post("/jobs", dependencies=[Depends(authorize)])
+def create_job(payload: JobRequest):
+    records, seen = [], set()
+    for raw in payload.links:
+        if len(raw) > 2048:
+            raise HTTPException(422, "Слишком длинная ссылка")
+        try:
+            record = normalize(raw)
+        except ValueError as exc:
+            raise HTTPException(422, str(exc)) from None
+        if record["url"] not in seen:
+            seen.add(record["url"])
+            records.append(record)
+    canonical = [r["url"] for r in records]
+    with lock:
+        now = time.time()
+        expired = [k for k, j in jobs.items() if j["status"] not in ("queued", "running") and now - j["created"] > 3600]
+        for key in expired:
+            del jobs[key]
+        for job in jobs.values():
+            if job["request_id"] == payload.request_id:
+                if job["links"] != canonical:
+                    raise HTTPException(409, "Этот идентификатор уже используется для другого списка")
+                return {"id": job["id"], "status": job["status"]}
+        if sum(j["status"] in ("queued", "running") for j in jobs.values()) >= 5 or len(jobs) >= 100:
+            raise HTTPException(429, "Очередь нового Space заполнена. Попробуйте позже.")
+        job_id = str(uuid.uuid4())
+        jobs[job_id] = dict(id=job_id, request_id=payload.request_id, links=canonical,
+                            status="queued", completed=0, total=len(records), created=now)
+        queue.submit(run_job, job_id, records)
+    return {"id": job_id, "status": "queued"}
+
+
+@app.get("/jobs/{job_id}", dependencies=[Depends(authorize)])
+def get_job(job_id: str):
+    with lock:
+        if job_id not in jobs:
+            raise HTTPException(404, "Задание не найдено. Возможно, новый Space перезапустился; запустите обработку заново.")
+        return {k: v for k, v in jobs[job_id].items() if k not in ("links", "request_id")}

src/streamlit_app.py

@@ -1,593 +0,0 @@
-import streamlit as st
-import asyncio
-import nest_asyncio
-import re
-import requests
-import urllib.parse
-import hashlib
-import base64
-import secrets
-import streamlit.components.v1 as components
-
-# Применяем nest_asyncio для работы с asyncio в Streamlit
-nest_asyncio.apply()
-
-# ----------- Настройки Telegram -----------
-TELEGRAM_API_ID = 23646321
-TELEGRAM_API_HASH = "102edf31d404cb72d17ee4831d4e8c7b"
-TELEGRAM_STRING_SESSION = "1ApWapzMBu3AR6vZZynrQwc59LD9XDRzbMtXVii-kfgNPeguoA2KOwyFqhsi1X_zDAsBh1DGvmVv46DmtZqk0KCd0jg9XXeS8T05E51D3Al5Ps8NNOF7fMWnpli2ROM2lou-t_8KpYbPoYnNXBAO2SzYkEhvyqQPSD7xGTWVkCLAVWm_s8Lqi-SCpXeOi341K5CR9nUCzaDkBmD3eG8SfqUp6GVO2QksGDZt3YMftWpXDwxsjuFPeeExoIawVD0Kxpjp20V3nS26SlR1-tm9lvOs9ATI_6QWPcCScT3UR5FikvrudyHNE55ZLpwbt3TRGuDMHv936DBtdHwGIQaNWSUf_5lEhIb0="
-
-# ----------- Настройки ВКонтакте -----------
-VK_CLIENT_ID = 53202706
-VK_CLIENT_SECRET = "piaMXHEDj0afa4QObJCD"
-VK_API_VERSION = "5.131"
-REDIRECT_URI = "https://huggingface.co/spaces/PRAnalytics/PR_Tool"
-
-# Инициализация токенов в session_state
-if "VK_ACCESS_TOKEN" not in st.session_state:
-    st.session_state["VK_ACCESS_TOKEN"] = ""
-if "VK_REFRESH_TOKEN" not in st.session_state:
-    st.session_state["VK_REFRESH_TOKEN"] = ""
-if "VK_DEVICE_ID" not in st.session_state:
-    st.session_state["VK_DEVICE_ID"] = ""
-
-# Функции для генерации code_verifier и code_challenge
-def generate_code_verifier(length=128):
-    return secrets.token_urlsafe(length)[:length]
-
-def generate_code_challenge(code_verifier):
-    hashed = hashlib.sha256(code_verifier.encode('utf-8')).digest()
-    return base64.urlsafe_b64encode(hashed).decode('utf-8').rstrip('=')
-
-# Функция для авторизации через VK ID
-def vk_auth():
-    if 'code_verifier' not in st.session_state:
-        st.session_state.code_verifier = generate_code_verifier()
-    
-    # Проверяем наличие токена сразу после перезагрузки
-    if st.session_state.get("VK_ACCESS_TOKEN"):
-        return  # Прерываем авторизацию, если токен уже есть
-    
-    # Шаг 2: Извлечение параметров из URL
-    query_params = st.experimental_get_query_params()
-    user_code = query_params.get("code", [None])[0]
-    user_device_id = query_params.get("device_id", [None])[0]
-    user_state = query_params.get("state", [None])[0]
-    
-    if user_code and user_state:
-        with st.expander("Технические подробности", expanded=False):
-            st.write("Автоматически извлечены параметры из URL:")
-            st.write(f"code: {user_code}")
-            st.write(f"device_id: {user_device_id}")
-            st.write(f"state: {user_state}")
-
-            # Используем code_verifier из параметра state
-            code_verifier = user_state
-
-            url = "https://id.vk.com/oauth2/auth"
-            data = {
-                "grant_type": "authorization_code",
-                "client_id": VK_CLIENT_ID,
-                "client_secret": VK_CLIENT_SECRET,
-                "redirect_uri": REDIRECT_URI,
-                "code": user_code,
-                "code_verifier": code_verifier
-            }
-            if user_device_id:
-                data["device_id"] = user_device_id
-
-            st.write("Отправляем запрос на:", url)
-            st.write("Параметры запроса:", data)
-
-            # Отправка запроса на обмен кода на токены
-            resp = requests.post(url, data=data)
-            st.write("Status code:", resp.status_code)
-            st.write("Response text:", resp.text)
-
-            try:
-                tokens = resp.json()
-                st.write("Ответ (JSON):")
-                st.json(tokens)
-
-                # Сохраняем токены
-                st.session_state["VK_ACCESS_TOKEN"] = tokens["access_token"]
-                st.session_state["VK_REFRESH_TOKEN"] = tokens["refresh_token"]
-                st.session_state["VK_DEVICE_ID"] = user_device_id
-                st.success("Токены успешно получены и сохранены!")
-                
-                # Закрываем технические подробности
-                st.experimental_set_query_params()
-                
-                # Немедленный переход к парсеру без перезагрузки
-                st.rerun()
-            except Exception as e:
-                st.error(f"Ошибка: {e}")
-        return  # Важно: прерываем выполнение после обработки кода
-    
-    # Если дошли сюда, показываем кнопку авторизации
-    st.markdown("### Сначала пройдём авторизацию")
-    code_challenge = generate_code_challenge(st.session_state.code_verifier)
-    auth_params = {
-        "client_id": VK_CLIENT_ID,
-        "redirect_uri": REDIRECT_URI,
-        "response_type": "code",
-        "scope": "wall",
-        "state": st.session_state.code_verifier,
-        "code_challenge": code_challenge,
-        "code_challenge_method": "S256",
-        "response_mode": "query"
-    }
-    auth_url = "https://id.vk.com/authorize?" + urllib.parse.urlencode(auth_params, safe="-._~")
-    st.markdown(
-        f"""
-        <a href="{auth_url}" style="
-            display: inline-block;
-            background-color: #4a76a8;
-            color: white;
-            text-decoration: none;
-            padding: 10px 20px;
-            border-radius: 4px;
-            font-family: -apple-system,BlinkMacSystemFont,Roboto,Helvetica Neue,sans-serif;
-            font-weight: 500;
-            font-size: 16px;
-            margin: 10px 0;
-            text-align: center;
-        ">
-            Авторизоваться через VK ID
-        </a>
-        """, 
-        unsafe_allow_html=True
-    )
-
-# Функция для обновления токена через refresh_token
-def refresh_vk_token() -> bool:
-    st.write("Пытаемся обновить токен...")
-
-    current_refresh = st.session_state["VK_REFRESH_TOKEN"]
-    device_id = st.session_state.get("VK_DEVICE_ID", "")
-
-    if not current_refresh or not VK_CLIENT_ID or not VK_CLIENT_SECRET or not device_id:
-        st.error("Невозможно обновить токен: нет refresh_token, client_id, client_secret или device_id.")
-        return False
-
-    url = "https://id.vk.com/oauth2/auth"
-    data = {
-        "grant_type": "refresh_token",
-        "client_id": VK_CLIENT_ID,
-        "client_secret": VK_CLIENT_SECRET,
-        "refresh_token": current_refresh,
-        "device_id": device_id
-    }
-
-    headers = {"Content-Type": "application/x-www-form-urlencoded"}
-
-    resp = requests.post(url, data=data, headers=headers)
-    with st.expander("Посмотреть логи обновления токена"):
-        st.write(f"Статус ответа: {resp.status_code}")
-        st.write(f"Текст ответа: {resp.text}")
-
-    if resp.status_code != 200:
-        st.error(f"Ошибка обновления токена (HTTP {resp.status_code}): {resp.text}")
-        return False
-
-    try:
-        result = resp.json()
-        with st.expander("Посмотреть JSON-ответ обновления токена"):
-            st.write(f"Ответ от VK API: {result}")
-    except Exception as e:
-        st.error(f"Не удалось распарсить ответ при обновлении токена: {e}")
-        return False
-
-    if "access_token" not in result:
-        err_desc = result.get("error_description", result)
-        st.error(f"Не удалось получить новый access_token: {err_desc}")
-        return False
-
-    # Сохраняем новые значения
-    new_access = result["access_token"]
-    st.session_state["VK_ACCESS_TOKEN"] = new_access
-
-    if "refresh_token" in result:
-        new_refresh = result["refresh_token"]
-        st.session_state["VK_REFRESH_TOKEN"] = new_refresh
-        st.write(f"Новый refresh_token: {new_refresh}")
-    else:
-        st.warning("Новый refresh_token не был получен. Убедитесь, что это ожидаемо.")
-
-    st.success("VK-токен обновлён успешно!")
-    st.write(f"Новый access_token: {new_access}")
-
-    return True
-
-# Функция для форматирования числа просмотров
-def human_format_views(num: int) -> str:
-    if num >= 1_000_000:
-        formatted = f"{num / 1_000_000:.1f}M"
-    else:
-        formatted = f"{num / 1000:.1f}K"
-    return formatted.replace('.', ',')
-
-# Функция для обработки ссылок Telegram
-def process_telegram_link(channel_username: str, message_id: str, canonical_link: str) -> tuple[str, int]:
-    import re, requests
-
-    # Набор альтернативных адресов (если t.me не резолвится)
-    urls = [
-        f"https://t.me/{channel_username}/{message_id}?embed=1",
-        f"https://t.me/s/{channel_username}/{message_id}",
-        f"https://telegram.me/{channel_username}/{message_id}?embed=1",
-        # зеркала через r.jina.ai (обходит проблемы DNS/блокировок)
-        f"https://r.jina.ai/http://t.me/{channel_username}/{message_id}?embed=1",
-        f"https://r.jina.ai/http://t.me/s/{channel_username}/{message_id}",
-        f"https://r.jina.ai/http://telegram.me/{channel_username}/{message_id}?embed=1",
-    ]
-    headers = {"User-Agent": "Mozilla/5.0"}
-
-    def _parse_views_number(s: str) -> int:
-        if not s:
-            return 0
-        s = s.strip().replace(" ", "").replace(",", ".")
-        m2 = re.match(r'([\d\.]+)\s*([kKmM]?)', s)
-        if not m2:
-            return 0
-        val = float(m2.group(1))
-        suf = m2.group(2).lower()
-        if suf == "k":
-            val *= 1_000
-        elif suf == "m":
-            val *= 1_000_000
-        return int(val)
-
-    def _fetch_first_ok(url_list):
-        last_err = None
-        for u in url_list:
-            try:
-                resp = requests.get(u, headers=headers, timeout=15)
-                if resp.status_code == 200 and resp.text:
-                    return resp.text
-            except Exception as e:
-                last_err = e
-        raise last_err or Exception("Не удалось получить страницу Telegram")
-
-    try:
-        html = _fetch_first_ok(urls)
-
-        # Title
-        title = None
-        m_title_meta = re.search(r'<meta[^>]*property=["\']og:title["\'][^>]*content=["\']([^"\']+)["\']', html, re.IGNORECASE)
-        if m_title_meta:
-            title = m_title_meta.group(1).strip()
-        if not title:
-            m_title = re.search(
-                r'class="tgme_widget_message_owner_name".*?<span[^>]*>(.*?)</span>',
-                html, re.IGNORECASE | re.DOTALL
-            )
-            if m_title:
-                title = m_title.group(1).strip()
-        if not title:
-            title = channel_username
-
-        # Views
-        views = 0
-        m_views = re.search(
-            r'class="tgme_widget_message_views[^"]*">([\d\s\.,kKmM]+)<',
-            html, re.IGNORECASE
-        )
-        if m_views:
-            views = _parse_views_number(m_views.group(1))
-        else:
-            candidates = re.findall(
-                r'(\d[\d\s\.,]*)([kKmM]?)\s*(?:views|просмотр|просмотра|просмотров|переглядів|visualizações|visualizzazioni|ansichten|visninger|visitas)?',
-                html, re.IGNORECASE
-            )
-            if candidates:
-                last_num, last_suf = candidates[-1]
-                views = _parse_views_number(last_num + last_suf)
-
-        return title, views
-
-    except Exception as e:
-        # Возвращаем «технический» title и 0, чтобы не ломать общий вывод
-        return f"**Ошибка (TG)** при обработке [{canonical_link}]({canonical_link}): {e}", 0
-
-def process_vk_link(owner_id: int, post_id: str, canonical_link: str) -> tuple[str, int]:
-    posts_param = f"{owner_id}_{post_id}"
-    api_url = "https://api.vk.com/method/wall.getById"
-
-    def do_request(token):
-        params = {
-            "posts": posts_param,
-            "access_token": st.session_state["VK_ACCESS_TOKEN"],
-            "v": VK_API_VERSION,
-            "extended": 1
-        }
-        return requests.get(api_url, params=params)
-
-    resp = do_request(st.session_state["VK_ACCESS_TOKEN"])
-    if resp.status_code != 200:
-        return f"**Ошибка (VK)**: HTTP {resp.status_code} — {canonical_link}", 0
-
-    data = resp.json()
-    if "error" in data:
-        error_msg = data["error"].get("error_msg", "")
-        if "access_token has expired" in error_msg.lower():
-            if refresh_vk_token():
-                resp = do_request(st.session_state["VK_ACCESS_TOKEN"])
-                data = resp.json()
-                if "error" in data:
-                    err_msg2 = data["error"].get("error_msg", "")
-                    return f"**Ошибка (VK)**: {err_msg2} — {canonical_link}", 0
-            else:
-                return f"**Ошибка (VK)**: не удалось обновить токен — {canonical_link}", 0
-        else:
-            return f"**Ошибка (VK)**: {error_msg} — {canonical_link}", 0
-
-    response_data = data.get("response", {})
-    items = response_data.get("items", [])
-    if not items:
-        return f"**Ошибка (VK)**: пост не найден — {canonical_link}", 0
-
-    post = items[0]
-    views = post.get("views", {}).get("count", 0)
-    post_owner_id = post.get("owner_id", owner_id)
-
-    title = None
-    if post_owner_id < 0:
-        group_id = -post_owner_id
-        for g in response_data.get("groups", []):
-            if g["id"] == group_id:
-                title = g["name"]
-                break
-    else:
-        user_id = post_owner_id
-        for p in response_data.get("profiles", []):
-            if p["id"] == user_id:
-                first_name = p.get("first_name", "")
-                last_name = p.get("last_name", "")
-                title = (first_name + " " + last_name).strip()
-                break
-    if not title:
-        title = "VK пост"
-
-    return title, views
-
-# Функция для определения платформы
-def detect_platform(link: str) -> str:
-    link = link.strip().lower()
-    if "t.me/" in link or "telegram.me/" in link:
-        return "telegram"
-    elif "vk.com/wall" in link:
-        return "vk"
-    return ""
-
-def canonicalize_tg_link(link: str):
-    """
-    Возвращает (canonical_link, channel_username, message_id) или (None, None, None)
-    Нормализуем и t.me, и telegram.me, и t.me/s/...
-    """
-    link = link.strip()
-    m = re.match(r'https?://(?:t(?:elegram)?\.me)/(?:s/)?([^/]+)/(\d+)', link, re.IGNORECASE)
-    if not m:
-        return None, None, None
-    username = m.group(1)
-    message_id = m.group(2)
-    canonical = f"https://t.me/{username}/{message_id}"
-    return canonical, username, message_id
-
-def canonicalize_vk_link(link: str):
-    """
-    Возвращает (canonical_link, owner_id(int), post_id(str)) или (None, None, None)
-    """
-    link = link.strip()
-    m = re.search(r'vk\.com/wall(-?\d+)_(\d+)', link)
-    if not m:
-        return None, None, None
-    owner_id = int(m.group(1))
-    post_id = m.group(2)
-    canonical = f"https://vk.com/wall{owner_id}_{post_id}"
-    return canonical, owner_id, post_id
-
-# Функция для отображения интерфейса парсера
-def show_parser_interface():
-    st.write("Вставьте ссылки (каждая на новой строке) — как Telegram, так и ВКонтакте. Пример:")
-    st.code("""
-    https://t.me/meduzalive/130606
-    https://t.me/meduzalive/130604
-    https://vk.com/wall-12256840_3220312
-    https://vk.com/wall-237426_1643234
-    """)
-
-    links_input = st.text_area("Ссылки:", height=200)
-
-    if st.button("Обработать"):
-        # Сырые ссылки
-        raw_lines = [link.strip() for link in links_input.strip().split('\n') if link.strip()]
-        if not raw_lines:
-            st.warning("Нет ссылок для обработки.")
-            return
-
-        # Дедупликация по каноническим ссылкам
-        seen = set()
-        lines = []
-        for link in raw_lines:
-            plat = detect_platform(link)
-            if plat == "telegram":
-                canonical, username, mid = canonicalize_tg_link(link)
-                if not canonical:
-                    # оставляем как есть, чтобы позже показать ошибку
-                    canonical = link
-                    username = None
-                    mid = None
-                if canonical in seen:
-                    continue  # дубликат — пропускаем
-                seen.add(canonical)
-                lines.append(("telegram", canonical, username, mid))
-            elif plat == "vk":
-                canonical, owner_id, post_id = canonicalize_vk_link(link)
-                if not canonical:
-                    canonical = link
-                if canonical in seen:
-                    continue
-                seen.add(canonical)
-                lines.append(("vk", canonical, owner_id, post_id))
-            else:
-                lines.append(("unknown", link, None, None))
-
-        total_links = len(lines)
-        progress_bar = st.progress(0)
-        status_text = st.empty()
-
-        # Группы: {group_key: {"title": str, "items": [(canonical_link, views), ...]}}
-        tg_groups = {}
-        vk_groups = {}
-
-        for i, (plat, canonical, a, b) in enumerate(lines):
-            try:
-                status_text.text(f"Обработка {i+1}/{total_links}: {canonical}")
-
-                if plat == "telegram":
-                    username, mid = a, b
-                    if not username or not mid:
-                        # неправильная ссылка — формируем "ошибочную" запись как отдельную группу по канону
-                        key = canonical
-                        tg_groups.setdefault(key, {"title": "Telegram", "items": []})
-                        tg_groups[key]["items"].append((canonical, 0))
-                    else:
-                        title, views = process_telegram_link(username, mid, canonical)
-                        key = username  # группируем по username канала
-                        if key not in tg_groups:
-                            tg_groups[key] = {"title": title, "items": []}
-                        # title обновим только если он выглядит «нормально»
-                        if not tg_groups[key].get("title") or tg_groups[key]["title"].startswith("**Ошибка"):
-                            tg_groups[key]["title"] = title
-                        tg_groups[key]["items"].append((canonical, views))
-
-                elif plat == "vk":
-                    owner_id, post_id = a, b
-                    if owner_id is None or post_id is None:
-                        key = canonical
-                        vk_groups.setdefault(key, {"title": "VK пост", "items": []})
-                        vk_groups[key]["items"].append((canonical, 0))
-                    else:
-                        title, views = process_vk_link(owner_id, post_id, canonical)
-                        key = str(owner_id)  # группируем по owner_id
-                        if key not in vk_groups:
-                            vk_groups[key] = {"title": title, "items": []}
-                        if not vk_groups[key].get("title") or vk_groups[key]["title"].startswith("**Ошибка"):
-                            vk_groups[key]["title"] = title
-                        vk_groups[key]["items"].append((canonical, views))
-
-                else:
-                    st.error(f"Неизвестная платформа: {canonical}")
-
-                progress_bar.progress((i + 1) / total_links)
-
-            except Exception as e:
-                st.error(f"Ошибка при обработке {canonical}: {str(e)}")
-
-        progress_bar.empty()
-        status_text.empty()
-
-        # Подсчитываем суммарные просмотры по платформам
-        tg_total_views = sum(v for g in tg_groups.values() for _, v in g["items"])
-        vk_total_views = sum(v for g in vk_groups.values() for _, v in g["items"])
-
-        # Сортируем группы по суммарным просмотрам по убыванию
-        tg_sorted = sorted(
-            tg_groups.items(),
-            key=lambda kv: sum(v for _, v in kv[1]["items"]),
-            reverse=True
-        )
-        vk_sorted = sorted(
-            vk_groups.items(),
-            key=lambda kv: sum(v for _, v in kv[1]["items"]),
-            reverse=True
-        )
-
-        md_lines = []
-
-        # --- Telegram ---
-        md_lines.append("## Telegram")
-        md_lines.append(f"**Суммарный охват постов — {human_format_views(tg_total_views)} просмотров**")
-        if tg_sorted:
-            for idx, (_, data) in enumerate(tg_sorted, start=1):
-                title = data["title"] or "Telegram"
-                items = data["items"]
-                # первая ссылка — та, что встретилась первой
-                first_link, first_views = items[0]
-                line = f"{idx}. [{title}]({first_link})"
-                # остальные «ещё»
-                if len(items) > 1:
-                    for link2, _v in items[1:]:
-                        line += f" + [ещё]({link2})"
-                # просмотры как «A + B + C»
-                views_str = " + ".join(human_format_views(v) for _, v in items)
-                line += f" — {views_str}"
-                md_lines.append(line)
-        else:
-            md_lines.append("_Нет ссылок на Telegram_")
-        md_lines.append("")
-
-        # --- ВКонтакте ---
-        md_lines.append("## ВКонтакте")
-        md_lines.append(f"**Суммарный охват постов — {human_format_views(vk_total_views)} просмотров**")
-        if vk_sorted:
-            for idx, (_, data) in enumerate(vk_sorted, start=1):
-                title = data["title"] or "VK пост"
-                items = data["items"]
-                first_link, first_views = items[0]
-                line = f"{idx}. [{title}]({first_link})"
-                if len(items) > 1:
-                    for link2, _v in items[1:]:
-                        line += f" + [ещё]({link2})"
-                views_str = " + ".join(human_format_views(v) for _, v in items)
-                line += f" — {views_str}"
-                md_lines.append(line)
-        else:
-            md_lines.append("_Нет ссылок на ВКонтакте_")
-        md_lines.append("")
-
-        # Вывод
-        st.markdown("### Результат:")
-        for line in md_lines:
-            st.markdown(line, unsafe_allow_html=True)
-
-        markdown_result = "\n".join(md_lines)
-
-        # Кнопка копирования
-        copy_button_html = f"""
-        <html>
-          <head>
-            <script>
-              function copyToClipboard() {{
-                const btn = document.getElementById("copy-btn");
-                const text = `{markdown_result}`;
-                navigator.clipboard.writeText(text).then(function() {{
-                  btn.innerText = "Скопировано успешно";
-                }}, function(err) {{
-                  btn.innerText = "Ошибка при копировании";
-                }});
-              }}
-            </script>
-          </head>
-          <body>
-            <button id="copy-btn" onclick="copyToClipboard()" style="padding: 8px 16px; font-size: 16px;">Скопировать код</button>
-          </body>
-        </html>
-        """
-        components.html(copy_button_html, height=80)
-
-# Интерфейс Streamlit
-def main():
-    st.title("PR Tool made by @Dina")
-    
-    # Первым делом проверяем токен
-    if not st.session_state.get("VK_ACCESS_TOKEN"):
-        vk_auth()
-    else:
-        # Показываем интерфейс парсера
-        show_parser_interface()
-
-# Запуск приложения
-if __name__ == "__main__":
-    main()

tests/test_max_api.py

@@ -0,0 +1,98 @@
+import os
+import time
+import unittest
+from unittest.mock import Mock, patch
+
+import legacy_parser as 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()

tests/test_reports.py

@@ -0,0 +1,38 @@
+import unittest
+from reports import normalize, human_views, render
+
+
+class ReportTests(unittest.TestCase):
+    def test_normalize(self):
+        self.assertEqual(normalize('http://vk.ru/wall-123_456')['url'], 'https://vk.com/wall-123_456')
+        self.assertEqual(normalize('https://telegram.me/s/Channel/123')['url'], 'https://t.me/channel/123')
+        for link in ('https://t.me/c/123/456', 'https://vk.ru.evil.test/wall-123_456', 'https://vk.com/club123'):
+            with self.assertRaises(ValueError):
+                normalize(link)
+
+    def test_format_without_zero(self):
+        for value, expected in ((2000, '2K'), (10600, '10,6K'), (1000000, '1M'), (35, '35'), (0, '0')):
+            self.assertEqual(human_views(value), expected)
+
+    def test_errors_are_not_zero_views(self):
+        first = dict(normalize('https://t.me/channel/1'), title='Имя 🗿', views=2000)
+        second = dict(normalize('https://t.me/channel/2'), title='Имя', views=None, error='Нет счётчика')
+        r = render([first, second])
+        self.assertIn('<b>2K</b>', r['html'])
+        self.assertNotIn('🗿', r['html'])
+        self.assertNotIn('0K', r['html'])
+        self.assertIn('Неполный итог', r['html'])
+        self.assertEqual(len(r['errors']), 1)
+
+    def test_grouping_and_html_escaping(self):
+        records = [dict(normalize('https://max.ru/channel/AaE-viFyB3E'), title='<script>alert(1)</script> Имя', views=1000),
+                   dict(normalize('https://max.ru/channel/AaE-viFyB3F'), title='Имя', views=2000)]
+        r = render(records)
+        self.assertIn('<h2>Макс</h2>', r['html'])
+        self.assertIn('ещё</a>', r['html'])
+        self.assertNotIn('<script>', r['html'])
+        self.assertIn('<b>3K</b>', r['html'])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_service.py

@@ -0,0 +1,68 @@
+import time
+import unittest
+from unittest.mock import Mock, patch
+
+from fastapi.testclient import TestClient
+import service
+from reports import normalize
+
+
+class ServiceTests(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(service.app)
+        self.key_patch = patch.object(service, 'API_KEY', 'test-bridge-key')
+        self.key_patch.start()
+        self.addCleanup(self.key_patch.stop)
+        self.queue_patch = patch.object(service, 'queue', Mock())
+        self.queue = self.queue_patch.start()
+        self.addCleanup(self.queue_patch.stop)
+        service.jobs.clear()
+        self.headers = {'X-Bridge-Key': 'test-bridge-key'}
+        self.payload = {'request_id': '12345678-abcd-efgh', 'links': ['http://vk.ru/wall-123_456']}
+
+    def test_authentication_and_production_route_absent(self):
+        self.assertEqual(self.client.post('/jobs', json=self.payload).status_code, 401)
+        self.assertEqual(self.client.post('/jobs', json=self.payload, headers={'X-Bridge-Key': 'wrong-key'}).status_code, 401)
+        self.assertEqual(self.client.get('/health').json()['service'], 'ReportBridge')
+        self.assertEqual(self.client.post('/parse', json={'links': []}).status_code, 404)
+
+    def test_idempotency_and_canonical_dedup(self):
+        self.payload['links'].append('https://vk.com/wall-123_456')
+        a = self.client.post('/jobs', json=self.payload, headers=self.headers)
+        b = self.client.post('/jobs', json=self.payload, headers=self.headers)
+        self.assertEqual(a.status_code, 200)
+        self.assertEqual(a.json()['id'], b.json()['id'])
+        self.queue.submit.assert_called_once()
+        job = self.client.get('/jobs/' + a.json()['id'], headers=self.headers).json()
+        self.assertEqual(job['total'], 1)
+        self.assertNotIn('links', job)
+        self.payload['links'] = ['https://t.me/channel/1']
+        self.assertEqual(self.client.post('/jobs', json=self.payload, headers=self.headers).status_code, 409)
+
+    def test_private_and_arbitrary_urls_never_enqueue(self):
+        for url in ['https://t.me/c/123/456', 'http://127.0.0.1/secret', 'https://vk.com.evil.com/wall-123_456']:
+            self.payload['links'] = [url]
+            self.assertEqual(self.client.post('/jobs', json=self.payload, headers=self.headers).status_code, 422)
+        self.queue.submit.assert_not_called()
+
+    def test_queue_limits(self):
+        for i in range(5):
+            self.payload['request_id'] = f'12345678-abcd-{i:06d}'
+            self.assertEqual(self.client.post('/jobs', json=self.payload, headers=self.headers).status_code, 200)
+        self.payload['request_id'] = '12345678-abcd-999999'
+        self.assertEqual(self.client.post('/jobs', json=self.payload, headers=self.headers).status_code, 429)
+
+    def test_real_progress_and_rendering(self):
+        records = [normalize('https://t.me/channel/1'), normalize('https://t.me/channel/2')]
+        service.jobs['test'] = dict(status='queued', completed=0, total=2, created=time.time())
+        with patch.object(service, 'fetch_one', side_effect=lambda r: dict(r, title='Имя', views=2000)):
+            service.run_job('test', records)
+        job = service.jobs['test']
+        self.assertEqual(job['status'], 'complete')
+        self.assertEqual(job['completed'], 2)
+        self.assertIn('<b>4K</b>', job['result']['html'])
+        self.assertIn('ещё</a>', job['result']['html'])
+
+
+if __name__ == '__main__':
+    unittest.main()