Шардовая передача датасетов

export DG_ENDPOINT="https://<your-workspace-host>"   # см. Настройки → API
export DG_TOKEN="dg_live_…"                           # не коммитьте это значение

{
  "dataset": "vault-labs/ru-web-corpus-2026",
  "revision": "8c1e0a2f7d3b4e61",
  "block_size": 8388608,
  "shards": [
    {
      "path": "data/shard-00001-of-00420.parquet",
      "size": 4001112064,
      "sha256": "0f6c2d…e91a",
      "blocks": ["5b9e04…71cd", "c013aa…0b42", "…"]      // 477 хэшей
    }
  ]
}

FIRST=$(curl -s -H "Authorization: Bearer $DG_TOKEN" \
  "$DG_ENDPOINT/api/v1/datasets/vault-labs/ru-web-corpus-2026/manifest" | jq -r '.shards[0].blocks[0]')

# последние 8 байт первого блока — длина футера Parquet (перед магическим "PAR1")
curl -s -L -r -8 -H "Authorization: Bearer $DG_TOKEN" \
  "$DG_ENDPOINT/storage/shards/$FIRST" | od -An -t u4 --endian=little

import hashlib
import os
from concurrent.futures import ThreadPoolExecutor

import requests

ENDPOINT = os.environ["DG_ENDPOINT"]
BLOCK_SIZE = 8 * 1024 * 1024

session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['DG_TOKEN']}"


def read_block(path, index):
    with open(path, "rb") as f:
        f.seek(index * BLOCK_SIZE)
        return f.read(BLOCK_SIZE)


def block_digests(path):
    digests = []
    with open(path, "rb") as f:
        while chunk := f.read(BLOCK_SIZE):
            digests.append(hashlib.sha256(chunk).hexdigest())
    return digests


def upload(path, dataset, target_path):
    digests = block_digests(path)
    r = session.post(f"{ENDPOINT}/api/v1/datasets/{dataset}/uploads", json={
        "path": target_path,
        "size": os.path.getsize(path),
        "blocks": digests,
    })
    r.raise_for_status()
    plan = r.json()
    missing = set(plan["missing"])
    # первый индекс каждого отсутствующего хэша (одинаковые блоки загружаются один раз)
    todo = {d: i for i, d in reversed(list(enumerate(digests))) if d in missing}

    def put(item):
        digest, index = item
        resp = session.put(
            f"{ENDPOINT}/storage/shards/{digest}",
            data=read_block(path, index),
            headers={"Content-Type": "application/octet-stream",
                     "X-DG-Upload-Id": plan["upload_id"]},
            timeout=300,
        )
        resp.raise_for_status()

    with ThreadPoolExecutor(max_workers=8) as pool:
        list(pool.map(put, todo.items()))

    commit = session.post(f"{ENDPOINT}/api/v1/datasets/{dataset}/uploads/{plan['upload_id']}/commit")
    commit.raise_for_status()
    return commit.json()["revision"]


def download(dataset, shard_path, dest, revision="main"):
    r = session.get(f"{ENDPOINT}/api/v1/datasets/{dataset}/manifest", params={"revision": revision})
    r.raise_for_status()
    manifest = r.json()
    entry = next(s for s in manifest["shards"] if s["path"] == shard_path)
    size = manifest["block_size"]

    with open(dest, "ab") as f:          # выделяем место, сохраняя уже скачанное
        f.truncate(entry["size"])

    def fetch(item):
        index, digest = item
        with open(dest, "r+b") as f:
            f.seek(index * size)
            if hashlib.sha256(f.read(size)).hexdigest() == digest:
                return                          # уже скачан: докачка
        resp = session.get(f"{ENDPOINT}/storage/shards/{digest}", timeout=300)
        resp.raise_for_status()
        if hashlib.sha256(resp.content).hexdigest() != digest:
            raise IOError(f"block {index}: digest mismatch")
        with open(dest, "r+b") as f:
            f.seek(index * size)
            f.write(resp.content)

    with ThreadPoolExecutor(max_workers=8) as pool:
        list(pool.map(fetch, enumerate(entry["blocks"])))

    h = hashlib.sha256()
    with open(dest, "rb") as f:
        while chunk := f.read(BLOCK_SIZE):
            h.update(chunk)
    if h.hexdigest() != entry["sha256"]:
        raise IOError(f"{shard_path}: shard digest mismatch")


if __name__ == "__main__":
    download("vault-labs/ru-web-corpus-2026",
             "data/shard-00001-of-00420.parquet",
             "shard-00001-of-00420.parquet")

# 1. Есть ли этот блок на сервере? (200 — да, 404 — нужно загрузить)
curl -I -H "Authorization: Bearer $DG_TOKEN" "$DG_ENDPOINT/storage/shards/$SHA"

# 2. Загрузка блока №0 шарда (байты 0 … 8 МиБ − 1)
dd if=data/shard-00001-of-00420.parquet bs=8M skip=0 count=1 status=none \
  | curl --fail -X PUT --data-binary @- \
      -H "Authorization: Bearer $DG_TOKEN" \
      -H "Content-Type: application/octet-stream" \
      -H "X-DG-Upload-Id: $UPLOAD_ID" \
      "$DG_ENDPOINT/storage/shards/$SHA"

# 3. Скачивание блока; -C - докачивает прерванную передачу через Range
curl --fail -L -C - -o "blocks/$SHA" \
  -H "Authorization: Bearer $DG_TOKEN" \
  "$DG_ENDPOINT/storage/shards/$SHA"

# 4. Проверка
echo "$SHA  blocks/$SHA" | sha256sum -c -