Массовый экспорт карт: автоматизация через API

Экспортировать одну карту вручную — просто. Но что если нужно выгрузить 50 городов в GeoJSON, 200 кварталов в DXF или ежедневно обновлять данные по 10 регионам? Для этого существует REST API с поддержкой массового экспорта.

Обзор API

API osm2cdr.ru построен на REST-принципах. Основные эндпоинты:

Для доступа к API нужен API-ключ, который можно получить в личном кабинете. Он передаётся в заголовке X-API-Key.

Одиночный экспорт: curl

Начнём с простого запроса на экспорт одной области. Обратите внимание на форму тела: bbox — объект с полями min_lon/min_lat/max_lon/max_lat, formats — список (от 1 до 5 форматов за раз).

curl -X POST https://osm2cdr.ru/api/render \
  -H "X-API-Key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "bbox": {"min_lon": 37.59, "min_lat": 55.73, "max_lon": 37.65, "max_lat": 55.76},
    "formats": ["geojson"],
    "map_style": "standard",
    "detail_level": 4
  }'

Ответ несёт идентификатор задачи — сам файл придёт позже, отдельным запросом:

{
  "task_id": "abc123-def456",
  "status": "pending",
  "created_at": "2026-08-07T13:56:03.726383Z",
  "formats": ["geojson"]
}

Проверка статуса и скачивание:

# Проверить статус
curl -H "X-API-Key: YOUR_API_KEY" \
  https://osm2cdr.ru/api/status/abc123-def456

# Скачать результат (когда status = "completed"); формат обязателен в пути
curl -H "X-API-Key: YOUR_API_KEY" \
  -o moscow_center.geojson \
  https://osm2cdr.ru/api/download/abc123-def456/geojson

Массовый экспорт: batch endpoint

Batch endpoint принимает массив областей и возвращает один batch_id на весь пакет. Это эффективнее, чем отправлять запросы по одному — сервер оптимизирует выполнение и может переиспользовать кэш PostGIS, а результат отдаётся одним архивом.

Каждый элемент — это своя область со своим списком форматов; общие для всех настройки рендера кладутся в config элемента.

curl -X POST https://osm2cdr.ru/api/batch/export \
  -H "X-API-Key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "items": [
      {"bbox": {"min_lon": 37.59, "min_lat": 55.73, "max_lon": 37.65, "max_lat": 55.76}, "formats": ["geojson"]},
      {"bbox": {"min_lon": 30.28, "min_lat": 59.92, "max_lon": 30.36, "max_lat": 59.96}, "formats": ["geojson"]},
      {"bbox": {"min_lon": 37.59, "min_lat": 55.73, "max_lon": 37.65, "max_lat": 55.76}, "formats": ["dxf"]}
    ],
    "archive_format": "zip"
  }'

Ответ (HTTP 202):

{
  "batch_id": "b7f1c0e2-9a44-4f2b-8d31-0e5a1c9b7a20",
  "status": "pending",
  "items_total": 3,
  "items_completed": 0,
  "items_failed": 0
}

Лимиты batch:

Дополнительно: до 5 форматов на одну область, archive_format — zip или tar.gz. Без аутентификации batch отвечает 403 — анонимный доступ к пакетному экспорту закрыт.

Автоматизация на Python

Экспорт списка городов

import requests
import time
from pathlib import Path

API_URL = "https://osm2cdr.ru/api"
API_KEY = "YOUR_API_KEY"
HEADERS = {"X-API-Key": API_KEY, "Content-Type": "application/json"}

# Список городов с bbox: [min_lon, min_lat, max_lon, max_lat]
cities = {
    "moscow": [37.32, 55.57, 37.95, 55.92],
    "saint_petersburg": [30.08, 59.83, 30.55, 60.09],
    "berlin": [13.08, 52.34, 13.76, 52.68],
    "paris": [2.22, 48.81, 2.47, 48.90],
    "london": [-0.35, 51.38, 0.15, 51.60],
}

def as_bbox(values: list) -> dict:
    """[min_lon, min_lat, max_lon, max_lat] -> объект bbox, как ждёт API."""
    keys = ("min_lon", "min_lat", "max_lon", "max_lat")
    return dict(zip(keys, values))

def export_city(name: str, bbox: list, fmt: str = "geojson") -> str:
    """Создать задачу экспорта и дождаться результата."""
    # Создать задачу
    resp = requests.post(f"{API_URL}/render", headers=HEADERS, json={
        "bbox": as_bbox(bbox),
        "formats": [fmt],
        "map_style": "standard",
        "detail_level": 4
    })
    resp.raise_for_status()
    task_id = resp.json()["task_id"]
    print(f"[{name}] Task created: {task_id}")

    # Ожидание завершения
    for _ in range(120):  # Максимум 10 минут
        status = requests.get(f"{API_URL}/status/{task_id}", headers=HEADERS).json()
        if status["status"] == "completed":
            break
        if status["status"] == "failed":
            raise RuntimeError(f"Export failed: {status.get('error')}")
        time.sleep(5)

    # Скачивание — формат обязателен в пути
    output = Path(f"output/{name}.{fmt}")
    output.parent.mkdir(exist_ok=True)
    dl = requests.get(f"{API_URL}/download/{task_id}/{fmt}", headers=HEADERS)
    output.write_bytes(dl.content)
    print(f"[{name}] Saved to {output} ({len(dl.content)} bytes)")
    return str(output)

# Экспорт всех городов
for city, bbox in cities.items():
    export_city(city, bbox, "geojson")

Batch-экспорт: один пакет, один архив

Пакетный экспорт устроен иначе, чем цикл одиночных задач: вы отдаёте список областей одним запросом, получаете один batch_id и в конце забираете один архив со всеми файлами.

def batch_export(cities: dict, fmt: str = "shp", archive: str = "zip"):
    """Массовый экспорт через batch endpoint: пакет -> архив."""
    items = [
        {"bbox": as_bbox(bbox), "formats": [fmt]}
        for bbox in cities.values()
    ]

    # Отправить пакет
    resp = requests.post(f"{API_URL}/batch/export", headers=HEADERS, json={
        "items": items,
        "archive_format": archive
    })
    resp.raise_for_status()
    batch = resp.json()
    batch_id = batch["batch_id"]
    print(f"Batch created: {batch_id}, {batch['items_total']} items")

    # Ожидание всего пакета
    for _ in range(240):  # Максимум 20 минут
        st = requests.get(f"{API_URL}/batch/status/{batch_id}", headers=HEADERS).json()
        print(f"{st['items_completed']}/{st['items_total']} готово, "
              f"ошибок {st['items_failed']}")
        if st["status"] in ("completed", "failed"):
            break
        time.sleep(5)

    # Скачивание архива целиком
    suffix = "zip" if archive == "zip" else "tar.gz"
    path = Path(f"output/batch_{batch_id}.{suffix}")
    path.parent.mkdir(exist_ok=True)
    dl = requests.get(f"{API_URL}/batch/download/{batch_id}", headers=HEADERS)
    path.write_bytes(dl.content)
    print(f"Saved to {path} ({len(dl.content)} bytes)")
    return path

batch_export(cities, "shp")

В ответе статуса поле results несёт по элементу на область: item_index (позиция в вашем исходном списке), status, files и error — по item_index и сопоставляйте выдачу со своим списком городов.

Параллельный поллинг множества задач

Для долгих задач (большие области, сложные форматы) удобно отслеживать прогресс через параллельный поллинг REST-эндпоинта /api/status/{task_id}. Это путь для набора самостоятельных задач /api/render; у пакета из /api/batch/export свой сводный статус — /api/batch/status/{batch_id}. WebSocket-эндпоинт /ws/task/{task_id} существует как заглушка (немедленно закрывается с кодом 1001) и зарезервирован для будущей реализации через Celery -> Redis Pub/Sub — не полагайтесь на него в production-скриптах.

Базовый шаблон поллинга одной задачи:

import requests
import time

def poll_until_done(task_id: str, interval: int = 5, timeout: int = 600) -> dict:
    """Поллинг GET /api/status/{task_id} до completed/failed."""
    deadline = time.time() + timeout
    while time.time() < deadline:
        st = requests.get(f"{API_URL}/status/{task_id}", headers=HEADERS).json()
        progress = st.get("progress", 0)
        step = st.get("step", "")
        print(f"[{task_id[:8]}] {st['status']} {progress}% — {step}")
        if st["status"] == "completed":
            return st
        if st["status"] == "failed":
            raise RuntimeError(f"Export failed: {st.get('error')}")
        time.sleep(interval)
    raise TimeoutError(f"Task {task_id} did not finish in {timeout}s")

Параллельный мониторинг батча через ThreadPoolExecutor:

from concurrent.futures import ThreadPoolExecutor, as_completed

def poll_batch(task_ids: list[str], max_workers: int = 5) -> dict:
    """Параллельный поллинг списка task_id; возвращает {task_id: final_status}."""
    results = {}
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {pool.submit(poll_until_done, tid): tid for tid in task_ids}
        for fut in as_completed(futures):
            tid = futures[fut]
            try:
                results[tid] = fut.result()
            except Exception as e:
                results[tid] = {"status": "failed", "error": str(e)}
    return results

# Использование: собрали task_id от нескольких вызовов /api/render — и ждём всех параллельно
# results = poll_batch(task_ids, max_workers=5)

При HTTP 429 (rate limit) увеличьте interval или уменьшите max_workers — sliding window rate limiter учитывает все запросы по API-ключу, включая /api/status/{task_id}.

Rate Limiting

API использует sliding window rate limiting. При превышении лимита возвращается HTTP 429 с заголовком Retry-After:

resp = requests.post(f"{API_URL}/render", headers=HEADERS, json=payload)
if resp.status_code == 429:
    wait = int(resp.headers.get("Retry-After", 60))
    print(f"Rate limited. Waiting {wait}s...")
    time.sleep(wait)

Форматы для автоматизации

Не все форматы одинаково подходят для batch-сценариев:

Для массовых операций рекомендуем GeoJSON (универсальный) или FlatGeobuf (быстрый и компактный).

Пример: ежедневное обновление данных

import schedule

def daily_update():
    """Обновить данные по ключевым регионам."""
    regions = {
        "moscow_roads": {"bbox": [37.3, 55.5, 37.9, 55.9], "format": "geojson"},
        "spb_buildings": {"bbox": [30.1, 59.8, 30.5, 60.1], "format": "shp"},
    }
    for name, params in regions.items():
        try:
            export_city(name, params["bbox"], params["format"])
        except Exception as e:
            print(f"Error exporting {name}: {e}")

schedule.every().day.at("03:00").do(daily_update)

while True:
    schedule.run_pending()
    time.sleep(60)

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