BaseLinker API – Kompletny Przewodnik z Przykładami Kodu

import requests
import json
import time


class BaseLinkerAPI:
    """Klient BaseLinker API z automatyczną obsługą sesji."""

    API_URL = "https://api.baselinker.com/connector.php"

    def __init__(self, token: str):
        self.token = token
        self.session = requests.Session()
        self.session.headers.update({
            "X-BLToken": token,
            "Content-Type": "application/x-www-form-urlencoded"
        })

    def call(self, method: str, parameters: dict = None) -> dict:
        """Wywołaj metodę API BaseLinker."""
        data = {
            "method": method,
            "parameters": json.dumps(parameters or {})
        }
        response = self.session.post(self.API_URL, data=data, timeout=30)
        response.raise_for_status()
        return response.json()


# Inicjalizacja – token z panelu BaseLinker -> Ustawienia -> Klucze API
bl = BaseLinkerAPI(token="TWOJ_TOKEN_API")
print("Połączono z BaseLinker API")

from datetime import datetime, timedelta

# Token z panelu BaseLinker
bl = BaseLinkerAPI("TWOJ_TOKEN_API")

# Pobierz zamówienia z ostatnich 7 dni
date_from = int((datetime.now() - timedelta(days=7)).timestamp())

result = bl.call("getOrders", {
    "date_from": date_from,
    "status_id": 0,        # 0 = wszystkie statusy
    "get_unconfirmed_orders": True  # Uwzględnij niepotwierdzone
})

if result["status"] == "SUCCESS":
    orders = result["orders"]
    print(f"Znaleziono {len(orders)} zamowien")

    for order in orders:
        print(
            "Zamowienie #" + str(order["order_id"]) + ": " +
            str(order["delivery_fullname"]) + " - " +
            str(order["price_brutto"]) + " " + str(order["currency"])
        )
else:
    print("Blad: " + result["error_message"])

from datetime import datetime, timedelta

def pobierz_wszystkie_zamowienia(bl: BaseLinkerAPI, days_back: int = 30) -> list:
    """
    Pobiera WSZYSTKIE zamówienia z paginacją.
    BaseLinker API zwraca max 100 zamówień na jedno zapytanie.
    """
    wszystkie = []
    date_from = int((datetime.now() - timedelta(days=days_back)).timestamp())

    while True:
        result = bl.call("getOrders", {"date_from": date_from})

        if result["status"] != "SUCCESS":
            raise Exception("Blad API: " + result["error_message"])

        partia = result["orders"]
        if not partia:
            break

        wszystkie.extend(partia)
        print(f"Pobrano {len(wszystkie)} zamowien...")

        # Mniej niż 100 = ostatnia strona
        if len(partia) < 100:
            break

        # Następna strona: timestamp ostatniego zamówienia + 1s
        date_from = max(o["date_add"] for o in partia) + 1
        time.sleep(0.65)  # Rate limit: max 100 req/min

    return wszystkie


bl = BaseLinkerAPI("TWOJ_TOKEN_API")
zamowienia = pobierz_wszystkie_zamowienia(bl, days_back=90)
print("Lacznie: " + str(len(zamowienia)) + " zamowien")

bl = BaseLinkerAPI("TWOJ_TOKEN_API")

# Krok 1: Pobierz ID katalogu i magazynu
katalogi = bl.call("getInventories")
inventory_id = katalogi["inventories"][0]["inventory_id"]

magazyny = bl.call("getInventoryWarehouses", {"inventory_id": inventory_id})
warehouse_id = magazyny["warehouses"][0]["warehouse_id"]

print(f"Katalog: {inventory_id}, Magazyn: {warehouse_id}")

# Krok 2: Sprawdź aktualne stany
stany = bl.call("getInventoryProductsStock", {
    "inventory_id": inventory_id,
    "page": 1
})

for product_id, stock_data in stany["products"].items():
    ilosc = stock_data.get(str(warehouse_id), 0)
    print("Produkt " + product_id + ": " + str(ilosc) + " szt.")

# Krok 3: Zaktualizuj stany (max 1000 produktów na raz)
wynik = bl.call("updateInventoryProductsStock", {
    "inventory_id": inventory_id,
    "products": {
        "1001": {str(warehouse_id): 150},
        "1002": {str(warehouse_id): 75},
        "1003": {str(warehouse_id): 0},  # Wyprzedany
    }
})

print("Zaktualizowano: " + str(wynik["counter"]) + " produktow")

import base64

bl = BaseLinkerAPI("TWOJ_TOKEN_API")

def stworz_i_pobierz_etykiete(order_id: int, courier_code: str = "dpd") -> bytes:
    """Tworzy przesyłkę kurierską i pobiera etykietę PDF (base64)."""

    # Krok 1: Stwórz przesyłkę w systemie kurierskim
    paczka = bl.call("createPackage", {
        "order_id":     order_id,
        "courier_code": courier_code,
        "fields": [
            {"id": "weight",  "value": "1.5"},
            {"id": "size_x",  "value": "30"},
            {"id": "size_y",  "value": "20"},
            {"id": "size_z",  "value": "15"},
        ]
    })

    if paczka["status"] != "SUCCESS":
        raise Exception("Blad tworzenia paczki: " + paczka["error_message"])

    package_id = paczka["package_id"]
    nr_paczki  = paczka.get("package_number", "brak")
    print("Paczka #" + str(package_id) + ", numer sledzenia: " + nr_paczki)

    # Krok 2: Pobierz etykietę (format: base64-encoded PDF)
    etykieta = bl.call("getLabel", {
        "courier_code": courier_code,
        "package_id":   package_id,
        "page_format":  "A4"
    })

    if etykieta["status"] != "SUCCESS":
        raise Exception("Blad etykiety: " + etykieta["error_message"])

    # Zdekoduj base64 i zapisz jako PDF
    pdf_bytes = base64.b64decode(etykieta["label"])
    filename  = "etykieta_" + str(order_id) + ".pdf"

    with open(filename, "wb") as f:
        f.write(pdf_bytes)

    print("Etykieta zapisana: " + filename)
    return pdf_bytes

# Użycie
stworz_i_pobierz_etykiete(order_id=98765, courier_code="dpd")

bl = BaseLinkerAPI("TWOJ_TOKEN_API")
inventory_id = 12345  # Twoje ID katalogu

# Pobierz listę produktów (paginacja po 1000)
lista = bl.call("getInventoryProductsList", {
    "inventory_id":     inventory_id,
    "page":             1,
    "filter_category_id": 0  # 0 = wszystkie kategorie
})

product_ids = list(lista["products"].keys())
print("Znaleziono " + str(len(product_ids)) + " produktow")

# Pobierz szczegółowe dane dla max 1000 produktów
szczegoly = bl.call("getInventoryProductsData", {
    "inventory_id": inventory_id,
    "products":     product_ids[:1000]
})

for pid, produkt in szczegoly["products"].items():
    print(
        "SKU: " + str(produkt.get("sku", "brak")) +
        " | Nazwa: " + str(produkt["text_fields"].get("name", "brak")) +
        " | EAN: " + str(produkt.get("ean", "brak"))
    )

# Dodaj / zaktualizuj produkt
nowy = bl.call("addInventoryProduct", {
    "inventory_id": inventory_id,
    "product_id":   "",           # Pusty string = nowy produkt
    "ean":          "5901234123457",
    "sku":          "POLO-L-001",
    "text_fields": {
        "name":        "Koszulka polo meska rozmiar L",
        "description": "<p>Wysokiej jakosci koszulka polo 100% bawelna.</p>",
    },
    "images": {
        "1": "https://example.com/img/polo-l-001.jpg"
    }
})
print("Produkt dodany z ID: " + str(nowy["product_id"]))

bl = BaseLinkerAPI("TWOJ_TOKEN_API")

# Pobierz listę wszystkich statusów zamówień w koncie
statusy = bl.call("getOrderStatusList")

print("Dostepne statusy:")
for status in statusy["statuses"]:
    print("  ID: " + str(status["id"]) + " | Nazwa: " + status["name"])

# Zmień status jednego zamówienia
bl.call("setOrderStatus", {
    "order_id":  98765,
    "status_id": 3  # np. ID statusu "Wyslane" z Twojego konta
})
print("Status zamowienia #98765 zmieniony")

# Zmień status wielu zamówień jednocześnie (batch)
order_ids = [98765, 98766, 98767, 98768]

wynik = bl.call("setOrderStatuses", {
    "order_ids": order_ids,
    "status_id": 5  # np. "Zrealizowane"
})
print("Zaktualizowano " + str(len(order_ids)) + " zamowien")

import requests
import json
import time
import logging

logger = logging.getLogger(__name__)


class BaseLinkerAPI:
    API_URL     = "https://api.baselinker.com/connector.php"
    MAX_RETRIES = 3
    MIN_DELAY   = 0.65  # 100 req/min -> ~1 req/s, z marginesem bezpieczenstwa

    def __init__(self, token: str):
        self.token = token
        self.session = requests.Session()
        self.session.headers.update({
            "X-BLToken": token,
            "Content-Type": "application/x-www-form-urlencoded"
        })
        self._last_call = 0.0

    def call(self, method: str, parameters: dict = None) -> dict:
        # Przestrzegaj rate limitu
        elapsed = time.time() - self._last_call
        if elapsed < self.MIN_DELAY:
            time.sleep(self.MIN_DELAY - elapsed)

        for attempt in range(self.MAX_RETRIES):
            try:
                resp = self.session.post(
                    self.API_URL,
                    data={
                        "method":     method,
                        "parameters": json.dumps(parameters or {})
                    },
                    timeout=30
                )
                resp.raise_for_status()
                self._last_call = time.time()

                data = resp.json()

                if data.get("status") == "ERROR":
                    code = data.get("error_code", 0)
                    msg  = data.get("error_message", "Nieznany blad")

                    # Kod 1001 = przekroczono rate limit -> czekaj i ponow
                    if code == 1001 and attempt < self.MAX_RETRIES - 1:
                        wait = 2 ** (attempt + 1)
                        logger.warning("Rate limit! Czekam " + str(wait) + "s...")
                        time.sleep(wait)
                        continue

                    raise Exception("BL " + str(code) + ": " + msg)

                return data

            except requests.RequestException as e:
                if attempt == self.MAX_RETRIES - 1:
                    raise
                wait = 2 ** attempt
                logger.warning("Proba " + str(attempt + 1) + "/" + str(self.MAX_RETRIES) + " nieudana: " + str(e) + ". Czekam " + str(wait) + "s")
                time.sleep(wait)

        raise Exception("Nie udalo sie wywolac " + method)

"""
getJournalList – pobiera log zdarzeń zamówień z ostatnich 3 dni.
Idealne do synchronizacji zmian statusów z zewnętrznym systemem.
"""
bl = BaseLinkerAPI("TWOJ_TOKEN_API")

# Pobierz zdarzenia od konkretnego ID (0 = wszystkie dostępne)
log = bl.call("getJournalList", {
    "last_log_id":  0,
    "logs_types":   [2, 18],  # 2=zmiana statusu, 18=nowy komentarz
    "order_id":     0         # 0 = wszystkie zamówienia
})

if log["status"] == "SUCCESS":
    for zdarzenie in log["logs"]:
        print(
            "[" + str(zdarzenie["log_type"]) + "] " +
            "Zamowienie #" + str(zdarzenie["order_id"]) +
            " | " + str(zdarzenie["object_data"])
        )

    # Zapisz ostatnie ID do następnego wywołania
    if log["logs"]:
        ostatnie_id = max(z["log_id"] for z in log["logs"])
        print("Nastepne wywolanie: last_log_id=" + str(ostatnie_id))