Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| 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 | |
| APP_NAME = "pr-tool-backend" | |
| VK_API_VERSION = os.getenv("VK_API_VERSION", "5.131") | |
| VK_ACCESS_TOKEN = os.getenv("VK_ACCESS_TOKEN", "") | |
| REQUEST_TIMEOUT = float(os.getenv("REQUEST_TIMEOUT", "15")) | |
| 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 human_format_views(num: int) -> str: | |
| if num >= 1_000_000: | |
| value = num / 1_000_000 | |
| suffix = "M" | |
| else: | |
| value = num / 1000 | |
| suffix = "K" | |
| if value.is_integer(): | |
| formatted = f"{int(value)}{suffix}" | |
| else: | |
| formatted = f"{value:.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 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/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(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_first_ok(url_list): | |
| 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: | |
| return resp.text | |
| except Exception as exc: | |
| last_err = exc | |
| 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 = html_lib.unescape((raw or "").strip()) | |
| title = re.sub(r"\s*[-|]\s*Telegram\s*$", "", title, flags=re.IGNORECASE) | |
| title = _strip_emoji(title) | |
| return title or fallback | |
| 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]+)[^"]*"', | |
| ] | |
| 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 | |
| try: | |
| html = _fetch_first_ok(urls) | |
| title_raw = None | |
| m_site_name = re.search( | |
| r"<meta[^>]*property=[\"']og:site_name[\"'][^>]*content=[\"']([^\"']+)[\"']", | |
| html, | |
| re.IGNORECASE, | |
| ) | |
| if m_site_name: | |
| title_raw = m_site_name.group(1).strip() | |
| m_title_meta = re.search( | |
| r"<meta[^>]*property=[\"']og:title[\"'][^>]*content=[\"']([^\"']+)[\"']", | |
| html, | |
| re.IGNORECASE, | |
| ) | |
| if m_title_meta and not title_raw: | |
| title_raw = m_title_meta.group(1).strip() | |
| if not title_raw: | |
| m_title = re.search( | |
| r'class="tgme_widget_message_author_name".*?<span[^>]*>(.*?)</span>', | |
| html, | |
| re.IGNORECASE | re.DOTALL, | |
| ) | |
| if m_title: | |
| title_raw = m_title.group(1).strip() | |
| if not title_raw: | |
| m_title = re.search( | |
| r'class="tgme_widget_message_owner_name".*?<span[^>]*>(.*?)</span>', | |
| html, | |
| re.IGNORECASE | re.DOTALL, | |
| ) | |
| if m_title: | |
| title_raw = m_title.group(1).strip() | |
| title = _clean_title(title_raw or "", channel_username) | |
| views = _extract_views(html) | |
| 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" | |
| 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_msg = data["error"].get("error_msg", "") | |
| return f"**Ошибка (VK)**: {error_msg or 'API error'} — {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 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 = {} | |
| 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 = process_telegram_link(username, mid, canonical) | |
| 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 = 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 | |
| def health_check(): | |
| return {"status": "ok"} | |
| 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, | |
| ) | |