Тёмная сторона API

Часть 1: Анатомия данных

Всё, что ты видишь в браузере — это данные. Вопрос только в том, как их достать.

Сайт в браузере

Открытый API (мб будет документация)
Скрытый API (документации нету но стучать можно)
GraphQL endpoint
WebSocket поток
Серверный рендеринг (HTML с данными)
Клиентский рендеринг (JS собирает данные)
Скрытые поля, комментарии, метаданные

Если данные видны в браузере — их можно достать программно.

Часть 2: Уровни доступа к данным

Уровень 0 — Официальное API

Самое простое. Документация открыта, ключ бесплатный.

import requests
# Пример: публичное API

response = requests.get("https://api.example.com/v1/data")
data = response.json()

Плюсы: Стабильно, легально, документация
Минусы: Rate limits, неполные данные, ключ могут отозвать

Уровень 1 — Скрытый API (Reverse Engineering)

Сайты часто имеют внутренние API, которые фронтенд использует для получения данных. Документации нет, но endpoint работает.

Как найти:

Открой DevTools (F12)
Вкладка Network → XHR/Fetch
Обнови страницу
Смотри какие запросы летят
Найди endpoint с нужными данными

Пример — что видишь в Network: GET https://site.com/api/v2/products?page=1&limit=50

Воспроизводим

import requests

headers = {
    "Authorization": "Bearer eyJhbGciOiJIUzI1...",
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
    "Accept": "application/json",
    "Referer": "https://site.com/catalog"
}

response = requests.get(
    "https://site.com/api/v2/products",
    params={"page": 1, "limit": 50},
    headers=headers
)
print(response.json())

Инструменты для реверса:

DevTools (Network) — базовый
mitmproxy — перехват трафика мобильных приложений
Charles Proxy — то же, но с GUI
Postman — тестирование найденных endpoints
Frida — хук методов в мобильных приложениях
Burp Suite — Позволяет перехватывать, анализировать и модифицировать HTTP- и HTTPS-трафик между браузером

Уровень 2 — GraphQL Endpoints

Многие современные сайты используют GraphQL. Он отдаёт ровно те данные, которые запросишь.

Как обнаружить:

В DevTools ищи POST-запросы на /graphql
Или /api/graphql
Или любые POST с query в body

Интроспекция — всей схемы:

# Запрос всей схемы API
{
  __schema {
    types {
      name
      fields {
        name
        type {
          name
        }
      }
    }
  }
}

Пример запроса данных:

import requests

query = """
{
  users(first: 100) {
    id
    username
    email
    posts {
      title
      createdAt
    }
  }
}
"""

response = requests.post(
    "https://site.com/graphql",
    json={"query": query},
    headers={"Content-Type": "application/json"}
)
print(response.json())

Инструменты:

GraphiQL / Altair — IDE для GraphQL
graphql-path-enum — перебор скрытых полей
InQL (Burp extension) — автоматическая интроспекция

Уровень 3 — Парсинг HTML

Когда API нет или оно закрыто — парсим то, что видит браузер.

Базовый — requests + BeautifulSoup

import requests
from bs4 import BeautifulSoup

response = requests.get("https://example.com/products")
soup = BeautifulSoup(response.text, "html.parser")

products = soup.find_all("div", class_="product-card")

for product in products:
    title = product.find("h3").text.strip()
    price = product.find("span", class_="price").text
    link = product.find("a")["href"]
    
    print(f"{title} | {price} | {link}")

или когда например html динамический

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()
    
    # Перехватываем API-запросы страницы
    responses = []
    page.on("response", lambda res: responses.append(res) 
            if "api" in res.url else None)
    
    page.goto("https://example.com/catalog")
    page.wait_for_load_state("networkidle")
    
    # Данные из перехваченных API-запросов
    for res in responses:
        print(res.url, res.json())
    
    # Или контент из DOM
    content = page.content()
    browser.close()

Селекторы — что использовать

# CSS селекторы (быстрее, чище)
soup.select("div.product-card > h3.title")
soup.select("a[href*='/product/']")
soup.select("span[data-price]")

# XPath (мощнее для сложной логики)
soup.xpath("//div[@class='item'][position() < 10]")
soup.xpath("//a[contains(@href, 'product')]/@href")

Уровень 4 — Headless Browser (эмуляция)

Когда сайт проверяет JS, куки, поведение — эмулируем реального пользователя.

Playwright — стандарт

from playwright.sync_api import sync_playwright
import time
import random

with sync_playwright() as p:
    browser = p.chromium.launch(
        headless=False,
        args=[
            "--disable-blink-features=AutomationControlled",
            "--no-sandbox"
        ]
    )
    
    context = browser.new_context(
        viewport={"width": 1920, "height": 1080},
        user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        locale="ru-RU",
        timezone_id="Europe/Moscow"
    )
    
    page = context.new_page()
    
    # Убираем флаг webdriver
    page.add_init_script("""
        Object.defineProperty(navigator, 'webdriver', {
            get: () => undefined
        });
    """)
    
    page.goto("https://example.com")
    
    # Имитируем человека
    time.sleep(random.uniform(1, 3))
    page.mouse.move(random.randint(100, 800), random.randint(100, 600))
    
    # Скроллим как человек
    page.evaluate("""
        async () => {
            for (let i = 0; i < document.body.scrollHeight; i += 100) {
                window.scrollTo(0, i);
                await new Promise(r => setTimeout(r, 100));
            }
        }
    """)
    
    # Забираем контент
    html = page.content()
    
    browser.close()

Анти-детект: что проверяют сайты (зависит от фантазии создателя)

Основное

Уровень 5 — Мобильные API

Приложения часто используют API, недоступные через веб. Данные там чище и полнее.

Инструменты:

mitmproxy — перехват HTTPS трафика
Frida — инъекция кода в приложение
Charles Proxy — GUI-альтернатива
jadx — декомпиляция APK

Процесс:

Скачай APK (apkpure, apkmirror)
Декомпилируй (jadx)
Ищи строки с "api", "http", "endpoint"
Найди базовый URL и параметры
Перехвати трафик через mitmproxy
Воспроизведи запрос в Python

Часть 3: Защита и обход

Rate Limiting

Сайт ограничивает частоту запросов. Виды:

IP-based → ротация прокси
API Key-based → ротация ключей
Session-based → ротация сессий
Cookie-based → сброс куки
Device fingerprint → подмена отпечатка

Подход:

import time
import random
from itertools import cycle

class RateLimiter:
    def __init__(self, min_delay=1.0, max_delay=3.0):
        self.min_delay = min_delay
        self.max_delay = max_delay
    
    def wait(self):
        delay = random.uniform(self.min_delay, self.max_delay)
        time.sleep(delay)

# Использование прокси
proxies = cycle([
    "http://proxy1:port",
    "http://proxy2:port",
    "http://proxy3:port",
])

limiter = RateLimiter(1, 3)

for url in urls:
    proxy = next(proxies)
    limiter.wait()
    
    response = requests.get(url, proxiДл": proxy})
    # Дальше ваша логика...

Теперь главный враг автоматизации CAPTCHA

CAPTCHA можно проходить самыми разными способами от солвером до обучения локальных ИИ-моделей

Интеграция с сервисом решения:

import requests
import time

API_KEY = "your_2captcha_key"

def solve_recaptcha(site_key, page_url):
    # Отправляем задачу
    response = requests.post("https://api.2captcha.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": site_key,
        "pageurl": page_url,
        "json": 1
    })
    task_id = response.json()["request"]
    
    # Ждём решение
    for _ in range(30):
        time.sleep(5)
        result = requests.get("https://api.2captcha.com/res.php", params={
            "key": API_KEY,
            "action": "get",
            "id": task_id,
            "json": 1
        })
        if result.json()["status"] == 1:
            return result.json()["request"]
    
    return None

Часть 4: Инфраструктура

Прокси — виды и стратегии

1. HTTP прокси → обычные, для http запросов
2. SOCKS5 прокси → любой трафик
3. Резидентные → IP реальных людей (дорого, надёжно)
4. Датацентровые → IP серверов (дёшево, палевно)
5. Мобильные → IP мобильных операторов (самые надёжные)
6. Ротация → новый IP на каждый запрос
7. Стик → IP привязан к сессии

Правильная ротация:

import aiohttp
import asyncio
from itertools import cycle

class ProxyPool:
    def __init__(self, proxy_list):
        self.proxies = cycle(proxy_list)
        self.failed = set()
    
    def get_proxy(self):
        for _ in range(len(self.proxies)):
            proxy = next(self.proxies)
            if proxy not in self.failed:
                return proxy
        # Все прокси умерли — сбрасываем
        self.failed.clear()
        return next(self.proxies)
    
    def mark_failed(self, proxy):
        self.failed.add(proxy)

async def fetch(session, url, proxy):
    try:
        async with session.get(url, proxy=proxy, timeout=10) as resp:
            return await resp.json()
    except:
        pool.mark_failed(proxy)
        return None

async def main(urls):
    pool = ProxyPool([
        "http://proxy1:8080",
        "http://proxy2:8080",
    ])
    
    connector = aiohttp.TCPConnector(limit=20)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = []
        for url in urls:
            proxy = pool.get_proxy()
            tasks.append(fetch(session, url, proxy))
        
        results = await asyncio.gather(*tasks)
    return [r for r in results if r is not None]

Часть 5: Практические схемы

Схема 1: Сбор данных с каталога

from playwright.sync_api import sync_playwright
from bs4 import BeautifulSoup
import json
import time
import random

class CatalogParser:
    def __init__(self, base_url):
        self.base_url = base_url
        self.results = []
    
    def parse_page(self, page, url):
        page.goto(url)
        page.wait_for_load_state("networkidle")
        
        # Ждём появления товаров
        page.wait_for_selector(".product-card", timeout=10000)
        
        html = page.content()
        soup = BeautifulSoup(html, "html.parser")
        
        products = soup.select(".product-card")
        
        for product in products:
            self.results.append({
                "title": product.select_one(".title")?.text.strip(),
                "price": product.select_one(".price")?.text.strip(),
                "url": product.select_one("a")?.get("href"),
                "parsed_at": time.strftime("%Y-%m-%d %H:%M")
            })
        
        # Случайная задержка
        time.sleep(random.uniform(2, 5))
    
    def run(self, pages=10):
        with sync_playwright() as p:
            browser = p.chromium.launch(headless=True)
            page = browser.new_page()
            
            for i in range(1, pages + 1):
                url = f"{self.base_url}?page={i}"
                print(f"Страница {i}/{pages}")
                self.parse_page(page, url)
            
            browser.close()
        
        return self.results
    
    def save(self, filename="data.json"):
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(self.results, f, ensure_ascii=False, indent=2)

# Использование
parser = CatalogParser("https://example.com/catalog")
data = parser.run(pages=50)
parser.save()

Схема 2: Мониторинг изменений

import hashlib
import json
from datetime import datetime
from pathlib import Path

class ChangeMonitor:
    def __init__(self, state_file="state.json"):
        self.state_file = Path(state_file)
        self.state = self.load_state()
    
    def load_state(self):
        if self.state_file.exists():
            return json.loads(self.state_file.read_text())
        return {}
    
    def save_state(self):
        self.state_file.write_text(
            json.dumps(self.state, ensure_ascii=False, indent=2)
        )
    
    def get_hash(self, data):
        return hashlib.md5(json.dumps(data, sort_keys=True).encode()).hexdigest()
    
    def check(self, key, new_data):
        new_hash = self.get_hash(new_data)
        old_hash = self.state.get(key, {}).get("hash")
        
        if old_hash is None:
            self.state[key] = {"hash": new_hash, "data": new_data, "updated": datetime.now().isoformat()}
            return {"status": "new", "data": new_data}
        
        elif old_hash != new_hash:
            old = self.state[key]["data"]
            self.state[key] = {"hash": new_hash, "data": new_data, "updated": datetime.now().isoformat()}
            return {"status": "changed", "old": old, "new": new_data}
        
        return {"status": "unchanged"}

Схема 3: Распределённый парсер

import asyncio
import aiohttp
from dataclasses import dataclass, field
from typing import Optional
import json

@dataclass
class Task:
    url: str
    retries: int = 0
    max_retries: int = 3
    proxy: Optional[str] = None

@dataclass
class Worker:
    id: int
    session: aiohttp.ClientSession
    queue: asyncio.Queue
    results: list = field(default_factory=list)

class DistributedParser:
    def __init__(self, urls, num_workers=5, proxy_list=None):
        self.tasks = [Task(url=url) for url in urls]
        self.num_workers = num_workers
        self.proxy_list = proxy_list or []
        self.results = []
    
    async def worker(self, wid, queue, session):
        while True:
            try:
                task = await asyncio.wait_for(queue.get(), timeout=5)
            except asyncio.TimeoutError:
                break
            
            try:
                proxy = None
                if self.proxy_list:
                    proxy = self.proxy_list[task.retries % len(self.proxy_list)]
                
                async with session.get(
                    task.url, 
                    proxy=proxy,
                    timeout=aiohttp.ClientTimeout(total=15)
                ) as resp:
                    if resp.status == 200:
                        data = await resp.text()
                        self.results.append({
                            "url": task.url,
                            "data": data,
                            "worker": wid
                        })
                        print(f"[Worker {wid}] OK: {task.url}")
                    else:
                        raise Exception(f"Status {resp.status}")
                        
            except Exception as e:
                task.retries += 1
                if task.retries < task.max_retries:
                    await queue.put(task)
                else:
                    print(f"[Worker {wid}] FAILED: {task.url} - {e}")
            
            finally:
                queue.task_done()
    
    async def run(self):
        queue = asyncio.Queue()
        for task in self.tasks:
            queue.put(task)
        
        connector = aiohttp.TCPConnector(limit=self.num_workers * 2)
        
        async with aiohttp.ClientSession(connector=connector) as session:
            workers = [
                asyncio.create_task(self.worker(i, queue, session))
                for i in range(self.num_workers)
            ]
            await queue.join()
            for w in workers:
                w.cancel()
        
        return self.results

# Использование
urls = [f"https://example.com/item/{i}" for i in range(1000)]
parser = DistributedParser(
    urls=urls, 
    num_workers=10,
    proxy_list=["http://p1:8080", "http://p2:8080", "http://p3:8080"]
)
results = asyncio.run(parser.run())

Часть 6: Стек инструментов

Обязательный минимум

requests — базовые HTTP запросы
httpx — async HTTP, HTTP/2
beautifulsoup4 — парсинг HTML
lxml — быстрый парсер
playwright — headless браузер
scrapy — фреймворк для масштабных проектов

Стань лучше:

curl_cffi — имитация TLS fingerprint браузера
tls_client — обход Cloudflare через TLS
camoufox — анти-детект Firefox
undetected-chromedriver — анти-детект Chrome
aiohttp — async для скорости
celery — очереди задач
redis — кэш и состояние
postgresql — хранилище данных
docker — изоляция окружения (на уровне базовый минимум)

Инфраструктура (Может не подойти)

Прокси-провайдеры:

Bright Data (резидентные)
SmartProxy (резидентные)
Proxy-Seller (датацентровые)
IPRoyal (мобильные)

Решение капчи:

2Captcha
Anti-Captcha
CapSolver

Мониторинг:

Grafana + Prometheus
Telegram-бот алерты

Часть 7: Как НЕ надо

ДДОСить сайт парсером (1000 req/sec)
Игнорировать ошибки и повторять бесконечно
Хранить данные без структуры
Не логировать — не понимаешь что происходит
Хардкодить селекторы без фоллбеков
Не обрабатывать изменение вёрстк
Собирать всё подряд без цели
Не кэшировать — один и тот же запрос 50 раз
Писать всё в один файл без модульности
Игнорировать robots.txt и уважение к серверу

На этом все с вами был Vatican.