Create app.py

app.py

@@ -0,0 +1,420 @@
+import streamlit as st
+import asyncio
+import nest_asyncio
+import re
+import os
+import requests
+import urllib.parse
+import hashlib
+import base64
+import secrets
+from telethon import TelegramClient
+from telethon.sessions import StringSession
+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 = "1ApWapzMBu8a9o9Wu1GexPdkO7Jsr6po1xym67EcfAxkEbVx1oxw6wW7LINb9Xf0VYcyEAP-b-D0RQ_4x0RDAW8Di_0P2r4bj7xNeV9xO3wBMcaeq7FJux3UMq5buq89gqG8JarZG04tIyWoVnH__dq1_6G2Zkr0GF4WJCS2UN7mmwBacIIUEjUryEA97tXYC8jD14J-1MDgFsNNZTJ6uurD5RzH2DIhphChebxOkDBdXmpLGlanEzYzVOlx4p6Mo2zYorRG968o64CxKEh-WiJaPSNrS3IEfIVWFVd_tHKWzQSPD1zHHhcw0v2_ZjCKbZnTMcT3uMYuyLwA55_dckzcw0wZgAZQ="
+
+# ----------- Настройки ВКонтакте -----------
+VK_CLIENT_ID = 53202706
+VK_CLIENT_SECRET = "piaMXHEDj0afa4QObJCD"
+VK_API_VERSION = "5.131"
+REDIRECT_URI = "https://huggingface.co/spaces/PRAnalytics/prtooltest"
+
+# Инициализация токенов в 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()
+
+    st.title("VK ID Authorization with PKCE")
+
+    # Генерация code_challenge
+    code_challenge = generate_code_challenge(st.session_state.code_verifier)
+
+    # Шаг 1: Авторизация
+    st.markdown("### 1) Авторизация")
+    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"[Перейти к авторизации]({auth_url})")
+
+    st.markdown("""
+    После авторизации VK вернёт вас на redirect_uri с параметрами:  
+    `?code=...&device_id=...&state=...`
+    """)
+
+    # Шаг 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:
+        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)
+
+            # Сохраняем токены и device_id в session_state
+            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("Токены успешно получены и сохранены!")
+        except Exception as e:
+            st.error(f"Ошибка при декодировании ответа: {e}")
+            st.text(resp.text)
+    else:
+        st.warning("Параметры code, device_id и state не найдены в URL. Пройдите авторизацию.")
+
+# Функция для обновления токена через 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)
+    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()
+        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
+
+# Инициализация Telethon-клиента (Telegram)
+def init_telethon():
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    client = TelegramClient(StringSession(TELEGRAM_STRING_SESSION), TELEGRAM_API_ID, TELEGRAM_API_HASH)
+    loop.run_until_complete(client.connect())
+    if not loop.run_until_complete(client.is_user_authorized()):
+        st.error("Telegram-сессия не авторизована. Проверьте StringSession или выполните login_telethon.py.")
+        st.stop()
+    return client, loop
+
+if "telethon_client" not in st.session_state:
+    st.session_state["telethon_client"], st.session_state["telethon_loop"] = init_telethon()
+
+tg_client = st.session_state["telethon_client"]
+tg_loop = st.session_state["telethon_loop"]
+
+# Функция для форматирования числа просмотров
+def human_format_views(num: int) -> str:
+    if num < 1000:
+        return f"{num / 1000:.1f}K"
+    elif num < 10_000:
+        return f"{num / 1000:.1f}K"
+    elif num < 1_000_000:
+        return f"{int(num / 1000)}K"
+    else:
+        return f"{num / 1_000_000:.1f}M"
+
+# Функция для обработки ссылок Telegram
+def process_telegram_link(link: str) -> tuple[str, int]:
+    link = link.strip()
+    if not link:
+        return "", 0
+
+    match = re.match(r'https?://(?:t(?:elegram)?\.me)/([^/]+)/(\d+)', link, re.IGNORECASE)
+    if not match:
+        return f"**Ошибка (TG)**: не распознал ссылку — {link}", 0
+
+    channel_username = match.group(1)
+    message_id = int(match.group(2))
+
+    try:
+        msg = tg_loop.run_until_complete(tg_client.get_messages(channel_username, ids=message_id))
+        if msg is None:
+            return f"**Ошибка (TG)**: сообщение не найдено — {link}", 0
+
+        channel_entity = tg_loop.run_until_complete(tg_client.get_entity(channel_username))
+        channel_title = channel_entity.title
+        views = msg.views or 0
+        result = f"- [{channel_title}]({link}) — {human_format_views(views)}"
+        return result, views
+    except Exception as e:
+        return f"**Ошибка (TG)** при обработке [{link}]({link}): {e}", 0
+
+# Функция для обработки ссылок VK
+def process_vk_link(link: str) -> tuple[str, int]:
+    link = link.strip()
+    if not link:
+        return "", 0
+
+    match = re.search(r'vk\.com/wall(-?\d+)_(\d+)', link)
+    if not match:
+        return f"**Ошибка (VK)**: не распознал ссылку — {link}", 0
+
+    owner_id = int(match.group(1))
+    post_id = match.group(2)
+    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": token,
+            "v": VK_API_VERSION,
+            "extended": 1
+        }
+        return requests.get(api_url, params=params)
+
+    # Первый запрос с текущим access_token
+    token = st.session_state["VK_ACCESS_TOKEN"]
+    resp = do_request(token)
+
+    if resp.status_code != 200:
+        return f"**Ошибка (VK)**: HTTP {resp.status_code} — {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():
+                # Повторяем запрос с новым токеном
+                new_token = st.session_state["VK_ACCESS_TOKEN"]
+                resp = do_request(new_token)
+                data = resp.json()
+                if "error" in data:
+                    # Если опять ошибка - выходим
+                    err_msg2 = data["error"].get("error_msg", "")
+                    return f"**Ошибка (VK)**: {err_msg2} — {link}", 0
+            else:
+                return f"**Ошибка (VK)**: не удалось обновить токен — {link}", 0
+        else:
+            return f"**Ошибка (VK)**: {error_msg} — {link}", 0
+
+    response_data = data.get("response", {})
+    items = response_data.get("items", [])
+    if not items:
+        return f"**Ошибка (VK)**: пост не найден — {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 пост"
+
+    result = f"- [{title}]({link}) — {human_format_views(views)}"
+    return result, 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 ""
+
+# Интерфейс Streamlit
+def main():
+    st.title("PR Tool made by @Dina")
+
+    # Если токены не получены, показываем авторизацию
+    if not st.session_state.get("VK_ACCESS_TOKEN"):
+        vk_auth()
+        return
+
+    # Если токены есть, показываем парсер
+    st.write("Вставьте ссылки (каждая на новой строке) — как Telegram, так и ВКонтакте. Пример:")
+    st.code("""
+    https://t.me/meduzalive/122479
+    https://vk.com/wall-12256840_3220312
+    https://telegram.me/za_traktor/27423
+    https://vk.com/wall-237426_1643234
+    """)
+
+    links_input = st.text_area("Ссылки:", height=200)
+
+    if st.button("Обработать"):
+        lines = links_input.strip().split('\n')
+        tg_results = []
+        tg_total_views = 0
+        vk_results = []
+        vk_total_views = 0
+
+        for link in lines:
+            platform = detect_platform(link)
+            if platform == "telegram":
+                res_line, views = process_telegram_link(link)
+                tg_results.append((res_line, views))
+                tg_total_views += views
+            elif platform == "vk":
+                res_line, views = process_vk_link(link)
+                vk_results.append((res_line, views))
+                vk_total_views += views
+            else:
+                st.markdown(f"**Ошибка**: не распознал платформу — {link}", unsafe_allow_html=True)
+
+        tg_results.sort(key=lambda x: x[1], reverse=True)
+        vk_results.sort(key=lambda x: x[1], reverse=True)
+
+        md_lines = []
+
+        # --- Telegram ---
+        md_lines.append("## Telegram")
+        md_lines.append(f"**Суммарный охват постов — {human_format_views(tg_total_views)} просмотров**")
+        if tg_results:
+            for r_line, _ in tg_results:
+                md_lines.append(r_line)
+        else:
+            md_lines.append("_Нет ссылок на Telegram_")
+        md_lines.append("")
+
+        # --- ВКонтакте ---
+    md_lines.append("## ВКонтакте")
+    md_lines.append(f"**Суммарный охват постов — {human_format_views(vk_total_views)} просмотров**")
+    if vk_results:
+        for r_line, _ in vk_results:
+            md_lines.append(r_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)