Как создать полную копию сайта Fragment.com через авторизованную сессию

Как создать полную копию сайта Fragment.com через авторизованную сессию, учитывая что:

  1. Fragment.com использует авторизацию через Telegram без публичного API.
  2. Стандартные методы скачивания сайтов не работают с персональными разделами.
  3. Необходимо обойти ограничения доступа к защищённым страницам.
  4. Требуется исключить скачивание определенных разделов (username, number, phone, gift)
  5. Нужно сохранить функциональность авторизованной сессии включая куки подключенного wallet.

Допустим, я написал базовый код назовём его (client.py):

import json
import re
import requests
from typing import Dict, Optional
from urllib.parse import urlparse, parse_qs, quote
from base64 import b64encode
from bs4 import BeautifulSoup

class Client:
    """Универсальный клиент для работы с Fragment.com и Telegram OAuth"""
    
    def __init__(self):
        self.api = "https://oauth.telegram.org/auth{}".format
        self._api = "https://fragment.com{}".format
        self.session = requests.Session()
        self.timeout = 10
        self.headers = {
            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36",
        }
        self.params = {
            "bot_id": "5444323279",
            "origin": self._api(''),
            "request_access": "write",
            "return_to": self._api(''),
        }

    def auth_post(self, path: str, cookies: dict, data: dict, headers: dict):
        return requests.post(
            self.api(path),
            cookies=cookies,
            params=self.params,
            data=data,
            headers={**self.headers, **headers},
            timeout=self.timeout
        )

    def auth_get(self, path: str, cookies: dict, headers: dict):
        return requests.get(
            self.api(path),
            cookies=cookies,
            params=self.params,
            headers={**self.headers, **headers},
            timeout=self.timeout
        )

    def fr_post(self, path: str, data: dict, cookies: dict, headers: dict):
        return requests.post(
            self._api(path),
            cookies=cookies,
            data=data,
            headers={**self.headers, **headers},
            timeout=self.timeout
        )

    def fr_get(self, path: str, cookies: dict, headers: dict):
        return requests.get(
            self._api(path),
            cookies=cookies,
            headers={**self.headers, **headers},
            timeout=self.timeout
        )

class Stel_tsession:
    def __init__(self, res):
        self.status = 'true' in res.text
        self.stel_tsession = res.cookies.get('stel_tsession') if self.status else None

class Confirm_:
    def __init__(self, res):
        self.status = 'true' in res.text
        self.stel_token = res.cookies.get('stel_token') if self.status else None

class Link_:
    def __init__(self, hash, confirm_url):
        self.status = hash is not False
        self.hash = hash
        self.link = f"https://oauth.telegram.org{confirm_url}&allow_write=1" if
self.status else None

class Con_Log:
    def __init__(self, auth_code):
        self.status = auth_code is not False
        self.auth_code = auth_code if self.status else None

class G_hash:
    def __init__(self, main_h):
        self.status = main_h is not False
        self.hash = main_h if self.status else None

class C_code:
    def __init__(self, res):
        try:
            self.status = True
            self.stel_token = res.cookies.get('stel_token')
        except AttributeError:
            self.status = False

class Telegram(Client):
    def __init__(self, phone_number: str):
        Client.__init__(self)
        self.number = phone_number
        self.stel_ssid = self.auth_get('', {}, {}).cookies.get('stel_ssid')

    def send_num(self) -> Stel_tsession:
        res = self.auth_post(
            '/request',
            cookies={"stel_ssid": self.stel_ssid},
            headers={},
            data={"phone": self.number}
        )
        return Stel_tsession(res)
    
    def confirm(self, stel_tsession: str) -> Confirm_:
        res = self.auth_post(
            '/login',
            headers={"Cookie": f"stel_ssid={self.stel_ssid}; stel_tsession={stel_tsession}"},
            data={},
            cookies={}
        )
        return Confirm_(res)

    def confirm_link(self, stel_token: str) -> Link_:
        try:
            res = self.auth_get(
                '',
                headers={"Referer": "https://oauth.telegram.org/auth?bot_id=5444323279&origin=https%3A%2F%2Ffragment.com&request_access=write&return_to=https%3A%2F%2Ffragment.com%2F"},
                cookies={'stel_ssid': self.stel_ssid, 'stel_token': stel_token}
            )
            
            script_tag = BeautifulSoup(res.text, "html.parser").findAll("script")[1]
            confirm_url = script_tag.string[script_tag.string.find("var confirm_url = ") +
len("var confirm_url = ")+1: script_tag.string.find("',")]
            
            if confirm_url:
                hash_ = parse_qs(urlparse(confirm_url).query).get("hash")[0]
                return Link_(hash_, confirm_url)
            return Link_(False, False)
        except (IndexError, AttributeError, TypeError) as e:
            return Link_(False, False)
    
    def confirm_login(self, link: str, stel_token: str) -> Con_Log:
        res = self.session.get(
            link,
            cookies={"stel_ssid": self.stel_ssid, "stel_token": stel_token},
            headers={**self.headers}
        )
        
        try:
            soup = BeautifulSoup(res.text, 'html.parser')
            scr_ = soup.find('script').string
            j_ = scr_[scr_.find('result: ') + len('result: '):scr_.find('}, origin:') + 1]
            en_ = quote(json.dumps(json.loads(j_)))
            auth_code = b64encode(en_.encode('utf-8')).decode('utf-8')
            return Con_Log(auth_code)
        except (AttributeError, IndexError):
            return Con_Log(False)
    
    def get_hash(self) -> G_hash:
        res = self.fr_get('', cookies={"stel_ssid": self.stel_ssid}, headers={})
        try:
            main_h = re.search(r"hash=([A-Za-z0-9]{15,32})", res.text).group(1)
            return G_hash(main_h)
        except (AttributeError, IndexError):
            return G_hash(False)    

    def try_code(self, m_hash, auth_code: str) -> C_code:
        res = self.session.post(
            f'{self._api("")}/api',
            headers={"Referer": self._api("")},
            params={"hash": m_hash},
            cookies={"stel_ssid": self.stel_ssid, "stel_dt": "-180"},
            data={"auth": auth_code, "method": "logIn"}
        )
        return C_code(res)

class FragmentSiteCopier:
    """Класс для создания полной копии сайта через авторизованную сессию"""
    
    def __init__(self, telegram_client: Telegram):
        self.client = telegram_client
        self.session = telegram_client.session
        self.downloaded_pages = set()
        self.excluded_paths = {'username', 'number', 'phone', 'gift'}
    
    def download_page(self, path: str, save_path: str):
        """Скачивание отдельной страницы"""
        if any(excluded in path for excluded in self.excluded_paths):
            return False
        
        response = self.client.fr_get(path, cookies=self.session.cookies, headers={})
        
        if response.status_code == 200:
            with open(save_path, 'w', encoding='utf-8') as f:
                f.write(response.text)
            return True
        return False
    
    def extract_links(self, html_content: str):
        """Извлечение всех ссылок со страницы"""
        soup = BeautifulSoup(html_content, 'html.parser')
        links = set()
        
        for link in soup.find_all('a', href=True):
            href = link['href']
            if href.startswith('/') and not href.startswith('//'):
                links.add(href)
        
        return links
    
    def crawl_site(self, start_path: str = '', max_depth: int = 3):
        """Рекурсивный обход сайта с заданной глубиной"""
        queue = [(start_path, 0)]
        
        while queue:
            path, depth = queue.pop(0)
            
            if depth > max_depth or path in self.downloaded_pages:
                continue
            
            print(f"Downloading: {path}")
            if self.download_page(path, f"./site_copy{path or '/index.html'}"):
                self.downloaded_pages.add(path)
                
                # Получаем HTML содержимое для извлечения ссылок
                response = self.client.fr_get(path, cookies=self.session.cookies, headers={})
                if response.status_code == 200:
                    links = self.extract_links(response.text)
                    for link in links:
                        if link not in self.downloaded_pages:
                            queue.append((link, depth + 1))

# Пример использования if __name__ == "__main__":
    # Инициализация клиента
    phone_number = "+1234567890"  # Замените на реальный номер
    client = Telegram(phone_number)
    
    # Процесс авторизации
    session_token = client.send_num().stel_tsession
    input("Подтвердите вход в Telegram и нажмите Enter...")
    
    confirm_token = client.confirm(session_token).stel_token
    auth_link = client.confirm_link(confirm_token).link
    auth_code = client.confirm_login(auth_link, confirm_token).auth_code
    site_hash = client.get_hash().hash
    
    # Финальная авторизация на Fragment.com
    final_token = client.try_code(site_hash, auth_code).stel_token
    
    if final_token:
        print("Успешная авторизация!")
        
        # Создание копии сайта
        copier = FragmentSiteCopier(client)
        copier.crawl_site('/my/account', max_depth=2)
        
        print("Копирование завершено!")
    else:
        print("Ошибка авторизации")

Вполне возможно, если использовать DevTools можно будет увидеть какие запросы происходят со стороны сервера, и какие данные передает клиент. В случае с web.telegram.org, там всё работает через webSocket, с использованием apiws1 и специальных методов (эндпоинтов)


Ответы (0 шт):