Delete app.py

app.py

@@ -1,964 +0,0 @@
-from __future__ import annotations
-
-import asyncio
-from threading import Lock
-import time
-import os
-import re
-import html as html_lib
-from typing import List, Tuple, Union
-
-import requests
-from fastapi import FastAPI
-from fastapi.middleware.cors import CORSMiddleware
-from pydantic import BaseModel
-
-try:
-    from telethon import TelegramClient
-    from telethon.sessions import StringSession
-except Exception:
-    TelegramClient = None
-    StringSession = None
-
-try:
-    from playwright.sync_api import TimeoutError as PlaywrightTimeoutError, sync_playwright
-except Exception:
-    PlaywrightTimeoutError = Exception
-    sync_playwright = None
-
-APP_NAME = "pr-tool-backend"
-
-VK_API_VERSION = os.getenv("VK_API_VERSION", "5.131")
-VK_ACCESS_TOKEN = os.getenv("VK_ACCESS_TOKEN", "")
-TELEGRAM_API_ID = int(os.getenv("TELEGRAM_API_ID", "0") or "0")
-TELEGRAM_API_HASH = os.getenv("TELEGRAM_API_HASH", "")
-TELEGRAM_STRING_SESSION = os.getenv("TELEGRAM_STRING_SESSION", "")
-REQUEST_TIMEOUT = float(os.getenv("REQUEST_TIMEOUT", "15"))
-TELEGRAM_OP_TIMEOUT = float(os.getenv("TELEGRAM_OP_TIMEOUT", "8"))
-TELEGRAM_CONCURRENCY = int(os.getenv("TELEGRAM_CONCURRENCY", "6"))
-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"))
-PLAYWRIGHT_GOTO_TIMEOUT_MS = int(os.getenv("PLAYWRIGHT_GOTO_TIMEOUT_MS", "20000"))
-ENABLE_TELEGRAM_BROWSER_FALLBACK = os.getenv("ENABLE_TELEGRAM_BROWSER_FALLBACK", "1") != "0"
-
-_TELEGRAM_BROWSER_LOCK = Lock()
-
-app = FastAPI(title=APP_NAME)
-
-# Для простоты разрешаем все источники. Можно сузить список доменов в проде.
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=False,
-    allow_methods=["POST", "GET", "OPTIONS"],
-    allow_headers=["*"]
-)
-
-
-class ParseRequest(BaseModel):
-    links: Union[List[str], str]
-
-
-class ParseResponse(BaseModel):
-    html: str
-    text: str
-    telegram_total: int
-    vk_total: int
-    errors: List[str]
-
-
-# ---------- Вспомогательные функции ----------
-
-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 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 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"
-    return ""
-
-
-def canonicalize_tg_link(link: str) -> Tuple[str | None, str | None, str | None]:
-    link = link.strip()
-    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[str | None, int | None, str | None]:
-    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 _telethon_ready() -> bool:
-    return bool(
-        TelegramClient
-        and StringSession
-        and TELEGRAM_API_ID
-        and TELEGRAM_API_HASH
-        and TELEGRAM_STRING_SESSION
-    )
-
-
-def process_telegram_batch(batch_items: List[Tuple[str, str, str]]) -> dict[str, Tuple[str, int]]:
-    if not batch_items:
-        return {}
-
-    if not _telethon_ready():
-        return {
-            canonical: process_telegram_link(username, message_id, canonical)
-            for canonical, username, message_id in batch_items
-        }
-
-    async def _run_batch() -> dict[str, Tuple[str, int]]:
-        results: dict[str, Tuple[str, int]] = {}
-        grouped: dict[str, List[Tuple[str, int]]] = {}
-
-        for canonical, username, message_id in batch_items:
-            grouped.setdefault(username, []).append((canonical, int(message_id)))
-
-        client = TelegramClient(
-            StringSession(TELEGRAM_STRING_SESSION),
-            TELEGRAM_API_ID,
-            TELEGRAM_API_HASH,
-            request_retries=2,
-            connection_retries=2,
-            timeout=REQUEST_TIMEOUT,
-        )
-
-        async with client:
-            semaphore = asyncio.Semaphore(max(1, TELEGRAM_CONCURRENCY))
-
-            async def _process_username(username: str, items: List[Tuple[str, int]]) -> None:
-                async with semaphore:
-                    try:
-                        entity = await asyncio.wait_for(
-                            client.get_entity(username),
-                            timeout=TELEGRAM_OP_TIMEOUT,
-                        )
-                    except Exception as exc:
-                        for canonical, _ in items:
-                            results[canonical] = (f"Ошибка (TG) при обработке {canonical}: {exc}", 0)
-                        return
-
-                    entity_title = getattr(entity, "title", None)
-                    if not entity_title:
-                        first_name = getattr(entity, "first_name", "")
-                        last_name = getattr(entity, "last_name", "")
-                        entity_title = (f"{first_name} {last_name}").strip()
-                    title = clean_telegram_title(entity_title or "", username)
-
-                    ids = [message_id for _, message_id in items]
-                    try:
-                        messages = await asyncio.wait_for(
-                            client.get_messages(entity, ids=ids),
-                            timeout=TELEGRAM_OP_TIMEOUT,
-                        )
-                    except Exception as exc:
-                        for canonical, _ in items:
-                            results[canonical] = (f"Ошибка (TG) при обработке {canonical}: {exc}", 0)
-                        return
-
-                    if not isinstance(messages, list):
-                        messages = [messages]
-
-                    message_map = {}
-                    for message in messages:
-                        if message is None:
-                            continue
-                        message_id = getattr(message, "id", None)
-                        if message_id is None:
-                            continue
-                        message_map[int(message_id)] = message
-
-                    for canonical, message_id in items:
-                        message = message_map.get(message_id)
-                        if message is None:
-                            results[canonical] = (f"Ошибка (TG): пост не найден — {canonical}", 0)
-                            continue
-                        views = int(getattr(message, "views", 0) or 0)
-                        results[canonical] = (title, views)
-
-            tasks = [
-                asyncio.create_task(_process_username(username, items))
-                for username, items in grouped.items()
-            ]
-            if tasks:
-                total_timeout = max(
-                    TELEGRAM_OP_TIMEOUT,
-                    (((len(tasks) + max(1, TELEGRAM_CONCURRENCY) - 1) // max(1, TELEGRAM_CONCURRENCY)) * TELEGRAM_OP_TIMEOUT * 2),
-                )
-                try:
-                    await asyncio.wait_for(
-                        asyncio.gather(*tasks),
-                        timeout=total_timeout,
-                    )
-                except asyncio.TimeoutError:
-                    for username, items in grouped.items():
-                        for canonical, _ in items:
-                            results.setdefault(
-                                canonical,
-                                (f"Ошибка (TG) при обработке {canonical}: timeout", 0),
-                            )
-
-        return results
-
-    try:
-        return asyncio.run(_run_batch())
-    except Exception:
-        return {
-            canonical: process_telegram_link(username, message_id, canonical)
-            for canonical, username, message_id in batch_items
-        }
-
-
-def process_telegram_link(channel_username: str, message_id: str, canonical_link: str) -> Tuple[str, int]:
-    # Набор альтернативных адресов (если t.me не резолвится)
-    urls = [
-        f"https://t.me/{channel_username}/{message_id}?embed=1",
-        f"https://t.me/{channel_username}/{message_id}?embed=1&single=1",
-        f"https://t.me/{channel_username}/{message_id}?embed=1&mode=tme",
-        f"https://t.me/s/{channel_username}/{message_id}",
-        f"https://t.me/{channel_username}/{message_id}",
-        f"https://r.jina.ai/http://t.me/{channel_username}/{message_id}?embed=1",
-        f"https://r.jina.ai/http://t.me/{channel_username}/{message_id}?embed=1&single=1",
-        f"https://r.jina.ai/http://t.me/s/{channel_username}/{message_id}",
-        f"https://r.jina.ai/http://t.me/{channel_username}/{message_id}",
-    ]
-    headers = {"User-Agent": "Mozilla/5.0"}
-    post_headers = {
-        "User-Agent": "Mozilla/5.0",
-        "Content-Type": "application/x-www-form-urlencoded",
-    }
-
-    def _parse_views_number(s: str) -> int:
-        if not s:
-            return 0
-        s = s.strip().replace("\u00a0", "").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_ok_responses(url_list):
-        texts: List[str] = []
-        last_err = None
-        for u in url_list:
-            try:
-                resp = requests.get(u, headers=headers, timeout=REQUEST_TIMEOUT)
-                if resp.status_code == 200 and resp.text:
-                    texts.append(resp.text)
-
-                    # Встроенный tg-виджет сам делает POST на тот же URL с _rl=1.
-                    # Иногда именно этот ответ содержит уже дорисованный widget DOM с просмотрами.
-                    if "tgme_widget_message_views" not in resp.text:
-                        post_resp = requests.post(
-                            u,
-                            headers=post_headers,
-                            data="_rl=1",
-                            timeout=REQUEST_TIMEOUT,
-                        )
-                        if post_resp.status_code == 200 and post_resp.text:
-                            texts.append(post_resp.text)
-            except Exception as exc:
-                last_err = exc
-        if texts:
-            return texts
-        raise last_err or Exception("Не удалось получить страницу Telegram")
-
-    def _strip_emoji(s: str) -> str:
-        # Убираем emoji/pictographs и variation selectors.
-        s = re.sub(
-            r"[\U0001F1E6-\U0001F1FF\U0001F300-\U0001FAFF\u2600-\u27BF\uFE0E\uFE0F]",
-            "",
-            s,
-        )
-        return re.sub(r"\s+", " ", s).strip()
-
-    def _clean_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 = _strip_emoji(title)
-        return title or fallback
-
-    def _is_placeholder_title(raw: str | None) -> bool:
-        if not raw:
-            return True
-        normalized = _clean_title(raw, "").strip().lower()
-        placeholders = {
-            "",
-            "telegram",
-            "telegram widget",
-            "telegram - a new era of messaging",
-            "telegram – a new era of messaging",
-        }
-        if normalized in placeholders:
-            return True
-        if normalized.startswith("telegram:"):
-            return True
-        if "a new era of messaging" in normalized:
-            return True
-        return False
-
-    def _is_username_like_title(raw: str | None) -> bool:
-        if not raw:
-            return False
-        normalized = _clean_title(raw, "").strip().lower()
-        username_variants = {
-            channel_username.lower(),
-            f"@{channel_username.lower()}",
-            f"t.me/{channel_username.lower()}",
-        }
-        return normalized in username_variants
-
-    def _extract_views(html: str) -> int:
-        patterns = [
-            # Виджет Telegram
-            r'class="tgme_widget_message_views[^"]*">([\d\s\.,kKmM]+)<',
-            # Резервный вариант из meta-description, но только если рядом явно "views"
-            r'content="[^"]*?([\d\s\.,kKmM]+)\s+views[^"]*"',
-            r'content="[^"]*?(?:Просмотры|просмотры|просмотров)[:\s]*([\d\s\.,kKmM]+)[^"]*"',
-            # Текстовый fallback (например, через jina)
-            r'(?:^|\n)\s*(?:Views|views|Просмотры|просмотры|Просмотров|просмотров)\s*[:\-]?\s*([\d\s\.,kKmM]+)\b',
-            r'([\d\s\.,kKmM]+)\s*(?:views|Views|просмотров|Просмотров)\b',
-            r'(?:👁|👀)\s*([\d\s\.,kKmM]+)\b',
-        ]
-        for pattern in patterns:
-            m = re.search(pattern, html, re.IGNORECASE)
-            if m:
-                parsed = _parse_views_number(m.group(1))
-                if parsed > 0:
-                    return parsed
-        return 0
-
-    def _extract_text_title(raw_text: str) -> str | None:
-        looks_like_html = bool(re.search(r"<(?:!DOCTYPE|html|head|body|meta|script|div)\b", raw_text, re.IGNORECASE))
-        patterns = [
-            r'(?:^|\n)\s*Title:\s*(.+)',
-            r'(?:^|\n)\s*Channel:\s*(.+)',
-            r'(?:^|\n)\s*#\s+(.+)',
-        ]
-        for pattern in patterns:
-            m = re.search(pattern, raw_text, re.IGNORECASE)
-            if m:
-                return m.group(1).strip()
-
-        if looks_like_html:
-            return None
-
-        for line in raw_text.splitlines():
-            line = line.strip()
-            if not line:
-                continue
-            if line.startswith("http://") or line.startswith("https://"):
-                continue
-            if "views" in line.lower() or "просмотр" in line.lower():
-                continue
-            if _is_username_like_title(line):
-                continue
-            if len(line) <= 120:
-                return line
-        return None
-
-    def _extract_title(raw_text: str) -> str | None:
-        candidates: List[str] = []
-
-        m_title_meta = re.search(
-            r"<meta[^>]*property=[\"']og:title[\"'][^>]*content=[\"']([^\"']+)[\"']",
-            raw_text,
-            re.IGNORECASE,
-        )
-        if m_title_meta:
-            candidates.append(m_title_meta.group(1).strip())
-
-        m_twitter_title = re.search(
-            r"<meta[^>]*property=[\"']twitter:title[\"'][^>]*content=[\"']([^\"']+)[\"']",
-            raw_text,
-            re.IGNORECASE,
-        )
-        if m_twitter_title:
-            candidates.append(m_twitter_title.group(1).strip())
-
-        m_site_name = re.search(
-            r"<meta[^>]*property=[\"']og:site_name[\"'][^>]*content=[\"']([^\"']+)[\"']",
-            raw_text,
-            re.IGNORECASE,
-        )
-        if m_site_name:
-            candidates.append(m_site_name.group(1).strip())
-
-        m_author = re.search(
-            r'class="tgme_widget_message_author_name".*?<span[^>]*>(.*?)</span>',
-            raw_text,
-            re.IGNORECASE | re.DOTALL,
-        )
-        if m_author:
-            candidates.append(m_author.group(1).strip())
-
-        m_owner = re.search(
-            r'class="tgme_widget_message_owner_name".*?<span[^>]*>(.*?)</span>',
-            raw_text,
-            re.IGNORECASE | re.DOTALL,
-        )
-        if m_owner:
-            candidates.append(m_owner.group(1).strip())
-
-        text_title = _extract_text_title(raw_text)
-        if text_title:
-            candidates.append(text_title)
-
-        for candidate in candidates:
-            if not _is_placeholder_title(candidate):
-                return candidate
-        return None
-
-    def _pick_widget_html(raw_text: str) -> str | None:
-        candidates = [raw_text]
-        iframe_matches = re.findall(r"<iframe\b[^>]*>(.*?)</iframe>", raw_text, re.IGNORECASE | re.DOTALL)
-        candidates.extend(iframe_matches)
-
-        for candidate in candidates:
-            if (
-                "tgme_widget_message_views" in candidate
-                or "tgme_widget_message_owner_name" in candidate
-                or "tgme_widget_message_author_name" in candidate
-            ):
-                return candidate
-        return None
-
-    def _fetch_browser_response() -> str | None:
-        if not ENABLE_TELEGRAM_BROWSER_FALLBACK or sync_playwright is None:
-            return None
-
-        browser_urls = [
-            f"https://t.me/{channel_username}/{message_id}?embed=1&single=1",
-            f"https://t.me/{channel_username}/{message_id}?embed=1",
-            f"https://t.me/{channel_username}/{message_id}",
-        ]
-
-        with _TELEGRAM_BROWSER_LOCK:
-            try:
-                with sync_playwright() as playwright:
-                    browser = playwright.chromium.launch(
-                        headless=True,
-                        args=["--no-sandbox", "--disable-dev-shm-usage"],
-                    )
-                    context = browser.new_context(
-                        user_agent=headers["User-Agent"],
-                        viewport={"width": 1280, "height": 900},
-                    )
-                    page = context.new_page()
-
-                    try:
-                        for url in browser_urls:
-                            try:
-                                page.goto(url, wait_until="domcontentloaded", timeout=PLAYWRIGHT_GOTO_TIMEOUT_MS)
-                            except PlaywrightTimeoutError:
-                                pass
-                            except Exception:
-                                continue
-
-                            try:
-                                page.wait_for_timeout(700)
-                                page.wait_for_selector(
-                                    ".tgme_widget_message_views, .tgme_widget_message_owner_name, .tgme_widget_message_author_name",
-                                    timeout=4000,
-                                )
-                            except PlaywrightTimeoutError:
-                                pass
-                            except Exception:
-                                pass
-
-                            frame_htmls: List[str] = []
-                            try:
-                                frame_htmls.append(page.content())
-                            except Exception:
-                                pass
-
-                            for frame in page.frames:
-                                if frame == page.main_frame:
-                                    continue
-                                try:
-                                    frame_htmls.append(frame.content())
-                                except Exception:
-                                    continue
-
-                            for frame_html in frame_htmls:
-                                widget_html = _pick_widget_html(frame_html)
-                                if widget_html:
-                                    return widget_html
-                    finally:
-                        context.close()
-                        browser.close()
-            except Exception:
-                return None
-
-        return None
-
-    try:
-        responses = _fetch_ok_responses(urls)
-
-        title_raw = None
-        views = 0
-        for response_text in responses:
-            candidate_title = _extract_title(response_text)
-            if candidate_title:
-                if not title_raw:
-                    title_raw = candidate_title
-                elif _is_username_like_title(title_raw) and not _is_username_like_title(candidate_title):
-                    title_raw = candidate_title
-
-            candidate_views = _extract_views(response_text)
-            if candidate_views > views:
-                views = candidate_views
-
-            if title_raw and not _is_username_like_title(title_raw) and views > 0:
-                break
-
-        if not title_raw or _is_username_like_title(title_raw) or views <= 0:
-            browser_response = _fetch_browser_response()
-            if browser_response:
-                browser_title = _extract_title(browser_response)
-                browser_views = _extract_views(browser_response)
-                if browser_title and (not title_raw or _is_username_like_title(title_raw)):
-                    title_raw = browser_title
-                if browser_views > views:
-                    views = browser_views
-
-        title = _clean_title(title_raw or "", channel_username)
-
-        return title, views
-
-    except Exception as exc:
-        return f"Ошибка (TG) при обработке {canonical_link}: {exc}", 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"
-
-    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:
-            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.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:
-        user_id = post_owner_id
-        for profile in response_data.get("profiles", []):
-            if profile.get("id") == user_id:
-                first_name = profile.get("first_name", "")
-                last_name = profile.get("last_name", "")
-                title = (first_name + " " + last_name).strip()
-                break
-    if not title:
-        title = "VK пост"
-
-    return title, views
-
-
-def process_vk_batch(
-    batch_items: List[Tuple[str, int, str]]
-) -> dict[str, Tuple[str, int]]:
-    api_url = "https://api.vk.com/method/wall.getById"
-    posts = ",".join(f"{owner_id}_{post_id}" for _, owner_id, post_id in batch_items)
-
-    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:
-            return {
-                canonical: (f"**Ошибка (VK)**: {exc} — {canonical}", 0)
-                for canonical, _, _ in batch_items
-            }
-
-        if resp.status_code != 200:
-            return {
-                canonical: (f"**Ошибка (VK)**: HTTP {resp.status_code} — {canonical}", 0)
-                for canonical, _, _ 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 {
-                canonical: (f"**Ошибка (VK)**: {error_msg or 'API error'} — {canonical}", 0)
-                for canonical, _, _ in batch_items
-            }
-        break
-    else:
-        return {
-            canonical: (f"**Ошибка (VK)**: Too many requests per second — {canonical}", 0)
-            for canonical, _, _ in batch_items
-        }
-
-    response_data = data.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[str, Tuple[str, int]] = {}
-    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:
-                first_name = profile.get("first_name", "")
-                last_name = profile.get("last_name", "")
-                title = (first_name + " " + last_name).strip()
-
-        result[canonical] = (title or "VK пост", views)
-
-    return result
-
-
-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 build_output(lines: List[Tuple[str, str, object, object]]) -> Tuple[str, str, int, int, List[str]]:
-    errors: List[str] = []
-
-    tg_groups = {}
-    vk_groups = {}
-
-    telegram_lines = [line for line in lines if line[0] == "telegram"]
-    vk_lines = [line for line in lines if line[0] == "vk"]
-    telegram_results = {}
-    vk_results = {}
-
-    if telegram_lines:
-        valid_telegram_batch: List[Tuple[str, str, str]] = []
-        for _, canonical, username, mid in telegram_lines:
-            if username and mid:
-                valid_telegram_batch.append((canonical, username, mid))
-        if valid_telegram_batch:
-            telegram_results = process_telegram_batch(valid_telegram_batch)
-
-    valid_vk_batch: List[Tuple[str, int, str]] = []
-    for _, canonical, owner_id, post_id in vk_lines:
-        if owner_id is not None and post_id is not None:
-            valid_vk_batch.append((canonical, owner_id, post_id))
-
-    if valid_vk_batch:
-        batch_size = max(1, VK_BATCH_SIZE)
-        for start in range(0, len(valid_vk_batch), batch_size):
-            chunk = valid_vk_batch[start:start + batch_size]
-            vk_results.update(process_vk_batch(chunk))
-
-    for plat, canonical, a, b in lines:
-        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 = telegram_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, process_vk_link(owner_id, post_id, canonical))
-                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))
-        else:
-            errors.append(f"Неизвестная платформа: {canonical}")
-
-    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,
-    )
-
-    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_sorted:
-        html_lines.append("<ol>")
-        for idx, (_, data) in enumerate(tg_sorted, start=1):
-            title = data["title"] or "Telegram"
-            items = data["items"]
-            first_link, _first_views = items[0]
-
-            title_html = _escape(title)
-            first_link_html = _escape(first_link)
-            line_html = f'<li><a href="{first_link_html}">{title_html}</a>'
-            if len(items) > 1:
-                for link2, _v in items[1:]:
-                    line_html += f' + <a href="{_escape(link2)}">ещё</a>'
-            views_str = " + ".join(human_format_views(v) for _, v in items)
-            line_html += f" — {views_str}</li>"
-            html_lines.append(line_html)
-
-            line_text = f"{idx}. {title} ({first_link})"
-            if len(items) > 1:
-                extra_links = ", ".join(link2 for link2, _v in items[1:])
-                line_text += f" + ещё: {extra_links}"
-            line_text += f" — {views_str}"
-            text_lines.append(line_text)
-        html_lines.append("</ol>")
-    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_sorted:
-        html_lines.append("<ol>")
-        for idx, (_, data) in enumerate(vk_sorted, start=1):
-            title = data["title"] or "VK пост"
-            items = data["items"]
-            first_link, _first_views = items[0]
-
-            title_html = _escape(title)
-            first_link_html = _escape(first_link)
-            line_html = f'<li><a href="{first_link_html}">{title_html}</a>'
-            if len(items) > 1:
-                for link2, _v in items[1:]:
-                    line_html += f' + <a href="{_escape(link2)}">ещё</a>'
-            views_str = " + ".join(human_format_views(v) for _, v in items)
-            line_html += f" — {views_str}</li>"
-            html_lines.append(line_html)
-
-            line_text = f"{idx}. {title} ({first_link})"
-            if len(items) > 1:
-                extra_links = ", ".join(link2 for link2, _v in items[1:])
-                line_text += f" + ещё: {extra_links}"
-            line_text += f" — {views_str}"
-            text_lines.append(line_text)
-        html_lines.append("</ol>")
-    else:
-        html_lines.append("<i>Нет ссылок на ВКонтакте</i>")
-        text_lines.append("Нет ссылок на ВКонтакте")
-    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
-
-
-@app.get("/health")
-def health_check():
-    return {"status": "ok"}
-
-
-@app.post("/parse", response_model=ParseResponse)
-def parse_links(payload: ParseRequest):
-    raw_lines = normalize_links(payload.links)
-    if not raw_lines:
-        return ParseResponse(html="", text="", telegram_total=0, vk_total=0, errors=["Нет ссылок для обработки."])
-
-    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 = 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))
-
-    html, text, tg_total, vk_total, errors = build_output(lines)
-    return ParseResponse(
-        html=html,
-        text=text,
-        telegram_total=tg_total,
-        vk_total=vk_total,
-        errors=errors,
-    )