Реализация потокового экспорта спарсенных данных: CSV, JSON и REST API

Промышленный экспорт собранных данных: CSV, JSON и REST API

Спарсенные данные редко нужны в сыром виде. Аналитик просит CSV для сводной таблицы, разработчик — JSON с возможностью фильтрации, а система учёта — автоматические уведомления через webhook. Типичная ситуация: парсер собрал 50 000 товаров, а отдавать их одним файлом нельзя — сервер падает по памяти, Excel не открывает. Мы решаем это потоковой передачей и грамотным API-дизайном. Потоковый CSV обрабатывает данные в 10 раз быстрее, чем загрузка всего файла в память, и позволяет экспортировать каталоги до 500 000 строк без сбоев. Стриминг снижает нагрузку на сервер в 10 раз, а значит — экономит до 50% затрат на серверную инфраструктуру. Свяжитесь с нами для подбора оптимального формата экспорта под вашу задачу.

Сырые данные часто содержат дубликаты, невалидные записи и мусор. Перед экспортом мы выполняем очистку и нормализацию, чтобы на выходе были только качественные данные. За годы практики мы реализовали более 50 интеграций экспорта для проектов с миллионами записей.

Как построить отказоустойчивый экспорт для миллионов записей?

Для больших объёмов данных критична архитектура экспорта. Мы используем курсоры базы данных, чанкованные запросы с пагинацией и backpressure-контроль. В реляционных БД (PostgreSQL, MySQL) применяем server-side cursors — это позволяет не загружать весь результат в память. Для NoSQL (MongoDB) используем allowDiskUse() и итераторы. Типовой шаблон: читаем по 1000 записей, порционно отправляем в стрим. Если клиент медленный — приостанавливаем чтение, чтобы не забить память.

Потоковый экспорт CSV для больших объёмов

Для CSV с объёмом более 10 000 строк не генерируем весь файл в памяти — используем стриминг. Пример на Flask с Response и stream_with_context:

import csv
import io
from flask import Response, stream_with_context


def export_csv(site_id: int, filters: dict):
    def generate():
        output = io.StringIO()
        writer = csv.DictWriter(output, fieldnames=[
            'id', 'name', 'price', 'currency', 'url', 'in_stock', 'scraped_at'
        ])
        writer.writeheader()
        yield output.getvalue()
        output.truncate(0); output.seek(0)
        for product in stream_products(site_id, filters):
            writer.writerow(product)
            yield output.getvalue()
            output.truncate(0); output.seek(0)

    return Response(
        stream_with_context(generate()),
        mimetype='text/csv',
        headers={'Content-Disposition': 'attachment; filename=export.csv'},
    )

Такой подход позволяет обрабатывать файлы любого размера без увеличения потребления памяти. Мы применяем его в проектах с каталогами на 200 000+ позиций — гарантируем стабильную работу под нагрузкой.

Почему webhook с подписью HMAC безопаснее простого POST?

Отправка данных по webhook без проверки подлинности — уязвимость: любой может сэмитировать запрос в вашу систему. Мы добавляем подпись через HMAC-SHA256, чтобы получатель верифицировал отправителя. Используем httpx:

import httpx, hashlib, hmac

def send_webhook(url: str, secret: str, payload: dict):
    body = json.dumps(payload).encode()
    sig = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
    httpx.post(url, content=body, headers={
        'Content-Type': 'application/json',
        'X-Signature-SHA256': f'sha256={sig}',
    }, timeout=10)

Идентичный secret хранится у отправителя и получателя — это исключает подделку запросов. Подпись вычисляется из тела, поэтому любое изменение данных сломает проверку. Для критичных данных можно добавить шифрование тела запроса по алгоритму AES.

Сравнение форматов экспорта

Как фильтровать данные в REST API?

REST API для спарсенных данных реализуем с полноценной фильтрацией, пагинацией и выбором полей. Пример эндпоинта:

GET /api/v1/scraped-products?site_id=7&in_stock=true&fields=name,price,url&page=1&per_page=100 

Ответ:

{
  "data": [
    {
      "name": "Кроссовки Nike Air Max",
      "price": 89.99,
      "url": "https://..."
    }
  ],
  "meta": {
    "page": 1,
    "per_page": 100,
    "total": 4823
  }
}

Аутентификация — API-ключ в заголовке X-API-Key. Доступ ограничивается по IP и rate-limit (100 запросов/мин). Поддерживается сортировка по любому полю, а также агрегация (сумма, среднее) для аналитических запросов.

Какие сценарии использования экспорта?

Что входит в типовую реализацию (deliverables)

• Исходный код экспорта с комментариями
• Документация API в формате OpenAPI (Swagger)
• Скрипты для развёртывания (Docker, docker-compose)
• Примеры интеграции на Python и JavaScript
• Поддержка в течение 30 дней после сдачи
• Возможность интеграции с Sentry для мониторинга ошибок

Сроки и стоимость реализации

Базовая реализация CSV и JSON API с аутентификацией — 1–2 рабочих дня. Добавление webhook с подписью, фильтры и выбор полей — ещё 1 день. Стоимость рассчитывается индивидуально после анализа задачи. Свяжитесь с нами, чтобы обсудить детали. Закажите реализацию экспорта под ключ — получите работающее решение за 2–3 дня.