Zablokowani w SaaS. Jak wyciągnąć 100% danych do migracji

Krótko: natywny eksport w SaaS daje ci połowę katalogu i udaje, że to całość. Reszta — metafieldy, redirecty 301, struktura kolekcji, alt-teksty, handle → URL — siedzi za API i przez przycisk „Export” nigdy nie wyjdzie. Jak zaczniesz migrację na tym okrojonym CSV-ie, po przełączeniu domeny zobaczysz spadek, którego nikt ci nie wytłumaczy.

Werdykt na start: nie klikaj eksportu. Uderz w Admin API i wyciągnij dane na twardo. Poniżej masz gotowy skrypt w Pythonie, który robi to na Shopify — pobiera cały katalog z metafieldami w jednym jobie (Bulk Operations), plus redirecty i strony CMS osobno. Wszystko przez oficjalne API, twoim własnym tokenem, do twojego własnego sklepu. Żadnego scrapingu HTML-a, żadnych sztuczek. Po prostu wchodzisz tam, gdzie leżą prawdziwe dane.

Co gubi natywny eksport (i dlaczego to boli w migracji)

Weź Shopify. Eksport produktów do CSV z panelu to wygodny plik, który wygląda kompletnie — dopóki nie porównasz go z tym, co realnie masz w sklepie. Braki, które wywalają migrację SEO:

• Metafieldy. Tu siedzi połowa twojego contentu: custom pola SEO, specyfikacje, dane do szablonów, atrybuty produktów. Natywny CSV ich nie rusza. Migrujesz i nagle połowa opisów kategorii znika.
• Redirecty 301. Cała mapa przekierowań, którą budowałeś latami, jest w URL Redirects — nie w eksporcie produktów. Zgubisz ją = zgubisz link equity i wpuszczasz użytkownika w 404.
• Struktura kolekcji. Które produkty należą do której kolekcji, smart vs manual, warunki reguł. CSV produktowy tego nie oddaje w pełni.
• Alt-teksty i mapowanie mediów. Alt-teksty grafik to sygnał SEO i dostępność. W eksporcie bywają puste albo nie ma ich wcale.
• Strony CMS i artykuły bloga. Osobny byt, osobny eksport, często pomijany.

Zestawienie na sucho:

Dane	Natywny eksport CSV	Admin API
Produkty + warianty	tak	tak
Metafieldy (SEO, custom)	nie	tak
Redirecty 301	nie	tak
Struktura kolekcji (reguły smart)	częściowo	tak
Alt-teksty mediów	niepewnie	tak
Strony CMS / artykuły bloga	osobno / niepełne	tak
SEO title + description per produkt	częściowo	tak

To nie jest „SaaS jest zły”. To jest realizm platformowy: panel jest zrobiony pod wygodę, nie pod kompletność. Kompletność jest w API — i dobrze, bo to znaczy, że da się ją wyciągnąć.

Dlaczego Bulk Operations, a nie klikanie po REST

Na Shopify masz dwie drogi. REST Admin API działa, ale jest już legacy — Shopify od dawna spycha nowe rzeczy na GraphQL i to na nim buduj. Do jednorazowego, pełnego dumpa katalogu najlepszym narzędziem są Bulk Operations (GraphQL): odpalasz jedno zapytanie, Shopify robi asynchroniczny job po swojej stronie i oddaje ci gotowy plik JSONL z całym katalogiem — razem z metafieldami — bez płacenia za to punktami rate limitu.

To jest różnica między „pętla po 250 produktów, potem osobny call po metafieldy dla każdego z osobna” (wolno, tysiące zapytań, throttling) a „jeden job, jeden plik, koniec”. Filozofia jest prosta: jak coś robisz raz, a solidnie, zrób to narzędziem, które jest do tego zaprojektowane. Bulk jest do tego zaprojektowany.

Redirecty i strony CMS to osobne connectiony — te ciągniesz zwykłą paginacją kursorową. Szybkie, więc nie ma sensu przepalać na to bulka.

Setup: custom app i token (uwaga na zmianę z 2026)

Zanim odpalisz skrypt, potrzebujesz Admin API access tokena. Kroki:

1. Utwórz custom app dla swojego sklepu. Ważne: od 1 stycznia 2026 nie utworzysz już nowej legacy custom appki bezpośrednio ze starego panelu Admin — robisz to w Shopify Dev Dashboard (dev.shopify.com) albo przez Shopify CLI. Istniejące custom appki działają dalej bez zmian.
2. Nadaj scope’y do odczytu tego, co ciągniesz: read_products, read_content (strony, artykuły, redirecty), read_online_store_pages. Do metafieldów w nowszych wersjach wystarczy scope zasobu nadrzędnego.
3. Zainstaluj appkę w sklepie i skopiuj Admin API access token (zaczyna się od shpat_). Trzymaj go jak hasło — to pełny dostęp do odczytu danych sklepu.

Token i domenę sklepu wrzucasz do zmiennych środowiskowych, nie do kodu:

export SHOPIFY_SHOP="twojsklep.myshopify.com"
export SHOPIFY_TOKEN="shpat_xxxxxxxxxxxxxxxxxxxx"
pip install requests

Skrypt

Cały skrypt niżej. Robi trzy rzeczy: odpala bulk job na produktach, ściąga JSONL i skleja go z powrotem w drzewo (bo bulk zwraca płaską listę z __parentId), a potem osobno paginuje redirecty i strony. Obsługuje throttling po stronie modelu kosztowego GraphQL.

#!/usr/bin/env python3
"""
shopify_export.py - wyciaga PELEN katalog ze sklepu Shopify przez Admin GraphQL
API (Bulk Operations), plus redirecty 301 i strony CMS.
Natywny eksport CSV tego nie zrobi. To robi.

Wymagania: python3 + requests  ->  pip install requests
Zmienne srodowiskowe:
  SHOPIFY_SHOP    np. twojsklep.myshopify.com
  SHOPIFY_TOKEN   Admin API access token (shpat_...)
"""

import os
import sys
import json
import time
import requests

SHOP = os.environ.get("SHOPIFY_SHOP")
TOKEN = os.environ.get("SHOPIFY_TOKEN")
API_VERSION = "2026-07"  # aktualna stabilna; rotuje co kwartal, pinuj swiadomie

if not SHOP or not TOKEN:
    sys.exit("Ustaw SHOPIFY_SHOP i SHOPIFY_TOKEN w zmiennych srodowiskowych.")

ENDPOINT = f"https://{SHOP}/admin/api/{API_VERSION}/graphql.json"
HEADERS = {"X-Shopify-Access-Token": TOKEN, "Content-Type": "application/json"}


def gql(query, variables=None):
    """Pojedyncze zapytanie GraphQL z obsluga throttlingu (model kosztowy)."""
    while True:
        r = requests.post(
            ENDPOINT,
            headers=HEADERS,
            json={"query": query, "variables": variables or {}},
            timeout=60,
        )
        if r.status_code == 429:
            time.sleep(float(r.headers.get("Retry-After", 2)))
            continue
        r.raise_for_status()
        data = r.json()
        if "errors" in data:
            throttled = any(
                e.get("extensions", {}).get("code") == "THROTTLED"
                for e in data["errors"]
            )
            if throttled:
                time.sleep(2)
                continue
            raise RuntimeError(json.dumps(data["errors"], ensure_ascii=False, indent=2))
        return data["data"]


def paginate(query, root_key):
    """Generyczny paginator connectionow (cursor-based)."""
    cursor, rows = None, []
    while True:
        conn = gql(query, {"cursor": cursor})[root_key]
        rows += [e["node"] for e in conn["edges"]]
        if not conn["pageInfo"]["hasNextPage"]:
            break
        cursor = conn["pageInfo"]["endCursor"]
    return rows


# --- 1. BULK: caly katalog z metafieldami w jednym jobie ---------------------

BULK_QUERY = """
{
  products {
    edges {
      node {
        id
        handle
        title
        descriptionHtml
        productType
        vendor
        status
        tags
        onlineStoreUrl
        seo { title description }
        metafields { edges { node { namespace key value type } } }
        variants { edges { node { id sku price compareAtPrice title barcode } } }
        images { edges { node { id url altText } } }
        collections { edges { node { id handle title } } }
      }
    }
  }
}
"""

START_BULK = """
mutation ($q: String!) {
  bulkOperationRunQuery(query: $q) {
    bulkOperation { id status }
    userErrors { field message }
  }
}
"""

POLL = """
{ currentBulkOperation { id status errorCode objectCount url } }
"""


def run_bulk_products(out_path="products.jsonl"):
    res = gql(START_BULK, {"q": BULK_QUERY})
    errs = res["bulkOperationRunQuery"]["userErrors"]
    if errs:
        raise RuntimeError(f"Bulk userErrors: {errs}")
    print("Bulk operation wystartowal. Czekam na zakonczenie...")

    while True:
        time.sleep(5)
        op = gql(POLL)["currentBulkOperation"]
        print(f"  status={op['status']} objectCount={op.get('objectCount')}")
        if op["status"] == "COMPLETED":
            url = op["url"]
            if not url:
                print("Brak danych (pusty katalog?).")
                return
            break
        if op["status"] in ("FAILED", "CANCELED"):
            raise RuntimeError(f"Bulk {op['status']}: {op.get('errorCode')}")

    with requests.get(url, stream=True, timeout=300) as resp:
        resp.raise_for_status()
        with open(out_path, "wb") as f:
            for chunk in resp.iter_content(chunk_size=8192):
                f.write(chunk)
    print(f"Zapisano surowy JSONL -> {out_path}")


def reassemble(jsonl_path="products.jsonl", out_path="products.json"):
    """Skleja plaski JSONL z powrotem w drzewo po __parentId."""
    products, children = {}, []
    with open(jsonl_path, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            obj = json.loads(line)
            if "__parentId" in obj:
                children.append(obj)
            elif "/Product/" in obj.get("id", ""):
                obj["_variants"], obj["_images"] = [], []
                obj["_metafields"], obj["_collections"] = [], []
                products[obj["id"]] = obj

    bucket = {
        "/ProductVariant/": "_variants",
        "/MediaImage/": "_images",
        "/ProductImage/": "_images",
        "/Metafield/": "_metafields",
        "/Collection/": "_collections",
    }
    for child in children:
        parent = products.get(child["__parentId"])
        if not parent:
            continue
        cid = child.get("id", "")
        for marker, key in bucket.items():
            if marker in cid:
                parent[key].append(child)
                break

    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(list(products.values()), f, ensure_ascii=False, indent=2)
    print(f"Sklejono {len(products)} produktow -> {out_path}")


# --- 2. Redirecty 301 (natywny eksport tego NIE daje) ------------------------

REDIRECTS = """
query ($cursor: String) {
  urlRedirects(first: 250, after: $cursor) {
    edges { node { id path target } }
    pageInfo { hasNextPage endCursor }
  }
}
"""

# --- 3. Strony CMS -----------------------------------------------------------

PAGES = """
query ($cursor: String) {
  pages(first: 100, after: $cursor) {
    edges { node { id handle title body bodySummary isPublished } }
    pageInfo { hasNextPage endCursor }
  }
}
"""


def dump(query, root_key, out_path):
    rows = paginate(query, root_key)
    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(rows, f, ensure_ascii=False, indent=2)
    print(f"Zapisano {len(rows)} rekordow ({root_key}) -> {out_path}")


def main():
    run_bulk_products()
    reassemble()
    dump(REDIRECTS, "urlRedirects", "redirects.json")
    dump(PAGES, "pages", "pages.json")
    print("Gotowe. Masz 100% katalogu + redirecty + strony. Teraz mapujesz na nowy stack.")


if __name__ == "__main__":
    main()

Odpalasz jednym poleceniem:

python3 shopify_export.py

Na wyjściu masz products.json (drzewo produkt → warianty, grafiki, metafieldy, kolekcje), redirects.json i pages.json. To jest twój materiał wejściowy do migracji — pełny, nie okrojony.

Jak to działa, w skrócie

• gql() to jedna funkcja do wszystkich zapytań GraphQL. Kluczowe: model rate limitu w GraphQL Admin jest kosztowy (punktowy, nie 2 req/s jak REST), a throttling wraca w polu errors z kodem THROTTLED. Dlatego łapiemy i 429, i THROTTLED, i czekamy — bez tego przy większym katalogu skrypt się wywali w połowie.
• Bulk działa asynchronicznie: bulkOperationRunQuery startuje job, potem pollujesz currentBulkOperation, aż status wejdzie w COMPLETED, i dopiero wtedy dostajesz signed URL do pliku JSONL (ważny ograniczony czas, pobierz od razu).
• reassemble() jest tu najważniejszy do zrozumienia. Bulk nie zwraca zagnieżdżonego JSON-a — zwraca płaską listę linii, gdzie każdy rekord podrzędny (wariant, grafika, metafield) ma __parentId wskazujący na produkt. Sklejasz to sam, po typie GID-a. To nie bug, to sposób, w jaki Shopify radzi sobie z dowolnie dużym katalogiem bez trzymania wszystkiego w pamięci.

Pułapki, na których ludzie się wykładają

• Wersja API. W skrypcie jest 2026-07, bo to teraz najnowsza stabilna. Shopify wypuszcza nową co kwartał (styczeń, kwiecień, lipiec, październik) i wspiera każdą min. 12 miesięcy. Pinuj wersję świadomie i podbijaj co kwartał — jak trafisz na wygaszoną, Shopify po cichu poda ci najstarszą wspieraną i możesz dostać inne dane, niż myślisz.
• Limit 128KB na JSON metafieldy. Od wersji 2026-04 Shopify ucina wartości metafieldów typu JSON na 128KB. Jeśli trzymasz w nich duże struktury, zaudytuj je przed migracją, żeby nie zgubić końcówek.
• Jeden bulk naraz. Na sklep leci jedna operacja bulk w danym momencie. Jak masz wiszący stary job, nowy nie wystartuje — najpierw go dokończ albo anuluj.
• Redirecty to osobny zasób. Powtarzam, bo to najczęściej gubiona rzecz w całej migracji. Nie ma ich w produktach. Wyciągnij je osobno (urlRedirects) i zmapuj 1:1 na nowej platformie, zanim przełączysz DNS.
• REST kontra GraphQL. Jak znajdziesz w sieci stary tutorial oparty o /admin/api/.../products.json z paginacją po nagłówku Link — działa, ale to legacy. Do nowej roboty bierz GraphQL.

Checklista przed przełączeniem domeny

1. Katalog wyciągnięty przez API z metafieldami — nie z CSV.
2. Pełna mapa redirectów 301 wyeksportowana i zmapowana na nowe URL-e.
3. SEO title / description i alt-teksty przeniesione, nie porzucone.
4. Kolekcje i ich reguły odtworzone na nowym stacku.
5. Strony CMS i artykuły bloga przeniesione z zachowanymi handle’ami tam, gdzie się da.
6. Stary sitemap zderzony z nowym — sprawdzasz, czy nic nie wypadło.

Migracja to nie „przerzuć produkty i przełącz DNS”. Migracja to „przenieś 100% danych, których crawler dotykał, i zmapuj każdy stary URL na nowy”. Reszta to sprzątanie po sobie w Search Console przez pół roku.