InvenTree: Масове створення місць зберігання через Python API-скрипт

#!/usr/bin/env python3
"""Створити підмісця Schublade під місцем Apothekerschrank у InvenTree.

Цей скрипт підключається до інстансу InvenTree через REST API і створює
набір підмісць-шухляд ("Schublade") під батьківським місцем під назвою
"Apothekerschrank". Для кожного з трьох рядів A, B і C створюються
шухляди з номерами від 1 до 19, що дає 57 підмісць загалом.
Кожному підмісцю призначається тип місця зберігання "Schublade".

Скрипт ідемпотентний: якщо підмісце вже існує під батьківським елементом,
воно пропускається, а не дублюється.

Конфігурація зчитується з ``config.yaml`` у тому ж каталозі:

.. code-block:: yaml

    inventree:
      server: https://inventree.example.com
      token: your-api-token-here

Вимоги:
    - Python 3.8+
    - requests
    - PyYAML

Використання::

    python3 create_locations.py
"""

import sys
import requests
import yaml
from pathlib import Path

# Шлях до YAML-файлу конфігурації, очікується поруч із цим скриптом.
CONFIG_PATH = Path(__file__).parent / "config.yaml"


def load_config():
    """Завантажити налаштування підключення до InvenTree з ``config.yaml``.

    Повертає dict із ключами ``server`` і ``token``.
    """
    with open(CONFIG_PATH, "r") as f:
        return yaml.safe_load(f)["inventree"]


class InvenTreeAPI:
    """Мінімальний REST API-клієнт для InvenTree.

    Обгортає :mod:`requests` з токен-автентифікацією та зручними методами
    для GET- і POST-викликів до API InvenTree.
    """

    def __init__(self, server, token):
        """Ініціалізувати API-клієнт.

        :param server: Базовий URL інстансу InvenTree, наприклад
            ``https://inventree.example.com``.
        :param token: API-токен автентифікації (згенеруйте його у
            вебінтерфейсі InvenTree у *Settings > API Keys*).
        """
        self.server = server.rstrip("/")
        self.token = token
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Token {token}",
            "Content-Type": "application/json",
        })

    def get(self, path, params=None):
        """Виконати GET-запит і повернути розібрану JSON-відповідь."""
        r = self.session.get(f"{self.server}{path}", params=params)
        r.raise_for_status()
        return r.json()

    def post(self, path, data):
        """Виконати POST-запит із JSON-тілом і повернути розібрану відповідь."""
        r = self.session.post(f"{self.server}{path}", json=data)
        if r.status_code == 400:
            print(f"  ПОМИЛКА 400: {r.json()}")
        r.raise_for_status()
        return r.json()


def find_location_type(api, name):
    """Знайти тип місця зберігання за назвою.

    Перебирає всі типи місць через пагіновані виклики API.
    Повертає первинний ключ (``pk``), якщо знайдено, інакше ``None``.
    """
    offset = 0
    while True:
        data = api.get("/api/stock/location-type/", params={"limit": 100, "offset": offset})
        results = data if isinstance(data, list) else data.get("results", [])
        for t in results:
            if t["name"] == name:
                return t["pk"]
        if isinstance(data, list) or not data.get("next"):
            break
        offset += 100
    return None


def find_location(api, name):
    """Знайти місце зберігання за назвою серед усіх місць.

    Перебирає всі місця зберігання через пагіновані виклики API.
    Повертає первинний ключ (``pk``), якщо знайдено, інакше ``None``.
    """
    offset = 0
    while True:
        data = api.get("/api/stock/location/", params={"limit": 100, "offset": offset})
        for loc in data["results"]:
            if loc["name"] == name:
                return loc["pk"]
        if not data["next"]:
            break
        offset += 100
    return None


def find_child_location(api, parent_pk, name):
    """Знайти дочірнє місце зберігання за назвою під заданим батьківським.

    Запитує API для всіх місць, чий ``parent`` відповідає
    ``parent_pk``, і повертає ``pk`` першого збігу, або ``None``.
    """
    results = api.get("/api/stock/location/", params={
        "parent": parent_pk, "limit": 1000
    })
    loc_list = results if isinstance(results, list) else results.get("results", [])
    for loc in loc_list:
        if loc["name"] == name:
            return loc["pk"]
    return None


def main():
    """Головна точка входу: знайти батьківське місце, перевірити тип місця, створити підмісця."""
    config = load_config()
    api = InvenTreeAPI(config["server"], config["token"])

    # --- Знайти або створити тип місця зберігання "Schublade" ---
    # InvenTree підтримує визначені користувачем типи місць, які можна призначати
    # окремим місцям зберігання. Ми перевикористовуємо наявний тип, якщо він існує.
    schublade_type_pk = find_location_type(api, "Schublade")
    if schublade_type_pk:
        print(f"Тип місця 'Schublade' знайдено (pk={schublade_type_pk})")
    else:
        result = api.post("/api/stock/location-type/", {
            "name": "Schublade",
            "description": "Schublade im Apothekerschrank",
            "icon": "ti:box:outline",
        })
        schublade_type_pk = result["pk"]
        print(f"Тип місця 'Schublade' створено (pk={schublade_type_pk})")

    # --- Знайти батьківське місце "Apothekerschrank" ---
    # Це місце має вже існувати в InvenTree. Скрипт завершиться
    # з помилкою, якщо його не вдасться знайти.
    apotheker_pk = find_location(api, "Apothekerschrank")
    if not apotheker_pk:
        print("ПОМИЛКА: Місце 'Apothekerschrank' не знайдено")
        sys.exit(1)
    print(f"Місце 'Apothekerschrank' знайдено (pk={apotheker_pk})")

    # --- Створити підмісця: Schublade A1..A19, B1..B19, C1..C19 ---
    # Кожне підмісце створюється як дочірнє до місця Apothekerschrank
    # і йому призначається тип місця "Schublade". Місця, що вже існують,
    # пропускаються, щоб скрипт залишався ідемпотентним.
    created = 0
    skipped = 0
    for letter in ["A", "B", "C"]:
        for number in range(1, 20):
            name = f"Schublade {letter}{number}"
            existing = find_child_location(api, apotheker_pk, name)
            if existing:
                print(f"  Пропущено (існує): {name} (pk={existing})")
                skipped += 1
                continue
            result = api.post("/api/stock/location/", {
                "name": name,
                "parent": apotheker_pk,
                "location_type": schublade_type_pk,
            })
            print(f"  Створено: {name} (pk={result['pk']})")
            created += 1

    print(f"\nГотово: {created} створено, {skipped} пропущено")


if __name__ == "__main__":
    main()