Upload app.py

app.py

@@ -0,0 +1,623 @@
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+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()
+    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[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 {}
+    results: dict[str, Tuple[str, int]] = {}
+    worker_count = max(1, min(8, len(batch_items)))
+
+    def _resolve(item: Tuple[str, str, str]) -> Tuple[str, Tuple[str, int]]:
+        canonical, username, message_id = item
+        return canonical, process_telegram_link(username, message_id, canonical)
+
+    with ThreadPoolExecutor(max_workers=worker_count) as executor:
+        for canonical, result in executor.map(_resolve, batch_items):
+            results[canonical] = result
+
+    return results
+
+
+def process_telegram_link(channel_username: str, message_id: str, canonical_link: str) -> Tuple[str, int]:
+    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",
+        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(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 _fetch_first_ok(url_list: List[str]) -> str:
+        last_err = None
+        for url in url_list:
+            try:
+                resp = requests.get(url, headers=headers, timeout=REQUEST_TIMEOUT)
+                if resp.status_code == 200 and resp.text:
+                    return resp.text
+            except Exception as exc:
+                last_err = exc
+        raise last_err or Exception("Не удалось получить страницу Telegram")
+
+    try:
+        page_html = _fetch_first_ok(urls)
+
+        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
+    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
+                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 = 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,
+    )