InvenTree : export PDF en lot d'étiquettes d'emplacements de stock via scripting Python

#!/usr/bin/env python3
"""Télécharger et fusionner les étiquettes PDF d'emplacements de stock depuis InvenTree.

Ce script se connecte à une instance InvenTree via l'API REST, récupère
les emplacements de stock (optionnellement filtrés par motif de nom et/ou
emplacement parent), génère des étiquettes PDF en parallèle à l'aide d'un
template d'étiquette InvenTree, et fusionne les résultats dans un fichier
PDF unique.

La configuration est lue depuis ``config.yaml`` dans le même répertoire ::

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

Prérequis :
    - Python 3.8+
    - requests
    - PyYAML
    - pypdf

Exemples d'utilisation ::

    # Générer des étiquettes pour TOUS les emplacements de stock, fusionnées en un PDF
    python3 export_location_labels.py

    # Filtrer par nom : motif glob (avec *) ou sous-chaîne (sans *)
    python3 export_location_labels.py -q "Schublade A*"
    python3 export_location_labels.py -q "Schublade"

    # Filtrer par emplacement parent (nom ou pk numérique)
    python3 export_location_labels.py -p Apothekerschrank
    python3 export_location_labels.py -p 9

    # Combiner les deux filtres
    python3 export_location_labels.py -q "Schublade A*" -p Apothekerschrank

    # Exclure des emplacements spécifiques par nom (glob ou sous-chaîne, mêmes règles que -q)
    python3 export_location_labels.py -e "Test*" -e "Illerbeuren"

    # Inclure uniquement des emplacements spécifiques (prioritaire sur les exclusions et -q)
    python3 export_location_labels.py -i "Schublade A1" -i "Schublade B2"

    # Inclusion + exclusion combinées : l'inclusion est prioritaire
    python3 export_location_labels.py -i "Schublade*" -e "Schublade C*"

    # Sélectionner un template d'étiquette spécifique (raccourci -t)
    python3 export_location_labels.py -t "Lagerort Groß 62mm"

    # Chemin de sortie personnalisé pour le PDF fusionné
    python3 export_location_labels.py -o ./my_labels.pdf

    # Écrire des PDF individuels dans un répertoire au lieu de fusionner
    python3 export_location_labels.py --individual -o ./my_labels_dir/

    # Sortie nommée automatiquement : déduit le nom du fichier à partir du terme de recherche
    # "Schublade A*" -> Schublade_A.pdf (fusionné) ou Schublade_A/ (individuel)
    python3 export_location_labels.py -q "Schublade A*"

    # Ajuster le parallélisme et le timeout
    python3 export_location_labels.py --workers 16 --timeout 120

Filtrage :
    Toute correspondance de nom normalise les espaces : toute séquence de
    caractères d'espacement (espaces, tabulations, sauts de ligne, etc.)
    dans le nom de l'emplacement comme dans le motif query/exclude/include
    est réduite à un espace unique avant comparaison.

    - **Filtre par nom** (``-q``) : si la chaîne de requête contient ``*``,
      elle est traitée comme un motif glob (ex. ``"Schublade A*"`` correspond
      à ``Schublade A1``, ``Schublade A10``, etc.). Si aucun ``*`` n'est
      présent, une correspondance par sous-chaîne insensible à la casse est
      utilisée.

    - **Filtre par parent** (``-p``) : filtre les emplacements dont le parent
      correspond à la valeur donnée. La valeur peut être soit le nom de
      l'emplacement parent (ex. ``Apothekerschrank``), soit sa clé primaire
      numérique (ex. ``9``).

    - **Exclusion** (``-e``) : exclut les emplacements correspondant au motif
      donné. Peut être spécifié plusieurs fois. Mêmes règles glob/sous-chaîne
      que ``-q``. Un emplacement est exclu s'il correspond à *n'importe quel*
      motif d'exclusion.

    - **Inclusion** (``-i``) : inclut uniquement les emplacements
      correspondant au motif donné. Peut être spécifié plusieurs fois. Mêmes
      règles glob/sous-chaîne que ``-q``. Un emplacement est inclus s'il
      correspond à *n'importe quel* motif d'inclusion. **L'inclusion est
      prioritaire sur ``-q`` et ``-e``** : lorsque ``-i`` est fourni, le
      filtre de requête est ignoré et les emplacements exclus qui
      correspondent à un motif d'inclusion sont tout de même inclus.

Sortie :
    Par défaut, tous les PDF d'étiquettes générés sont fusionnés en un
    fichier PDF unique à l'aide de ``pypdf``. Lorsque ``--individual`` est
    fourni, chaque étiquette est écrite dans un fichier PDF séparé dans un
    répertoire.

    Le chemin de sortie est déterminé comme suit :

    1. Si ``-o`` est fourni, il est utilisé directement (chemin de fichier
       en mode fusionné, chemin de répertoire en mode individuel).
    2. Si ``-o`` n'est pas fourni, le nom est déduit automatiquement :
       - À partir du terme de recherche ``-q`` en supprimant les caractères
         glob (``*``, ``?``) et en remplaçant les espaces par des tirets
         bas, ex. ``"Schublade A*"`` -> ``Schublade_A.pdf``.
       - Si pas de ``-q`` mais ``-i`` fourni, à partir du premier motif
         d'inclusion (même suppression).
       - Si ni l'un ni l'autre, ``StockLocationLabels.pdf`` (fusionné) ou
         ``StockLocationLabels/`` (individuel).

Fonctionnement :
    1. Récupère tous les emplacements de stock depuis l'API InvenTree (paginé).
    2. Applique les filtres optionnels de nom, parent, inclusion et exclusion.
    3. Récupère les templates d'étiquette ``stocklocation`` disponibles.
    4. Soumet un job d'impression d'étiquette pour chaque emplacement
       correspondant, en parallèle, via ``POST /api/label/print/``.
    5. Interroge ``GET /api/data-output/<pk>/`` jusqu'à ce que chaque job
       soit terminé.
    6. Télécharge le PDF généré depuis le chemin de sortie.
    7. Fusionne tous les PDF individuels en un seul à l'aide de ``pypdf``
       (ou les écrit individuellement si ``--individual`` est fourni).
"""

import argparse
import fnmatch
import io
import re
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

import requests
import yaml
from pypdf import PdfWriter, PdfReader

CONFIG_PATH = Path(__file__).parent / "config.yaml"


def load_config():
    with open(CONFIG_PATH, "r") as f:
        return yaml.safe_load(f)["inventree"]


class InvenTreeAPI:
    def __init__(self, server, token):
        self.server = server.rstrip("/")
        self.token = token
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Token {token}",
        })

    def get(self, path, params=None):
        r = self.session.get(f"{self.server}{path}", params=params)
        r.raise_for_status()
        return r

    def post(self, path, data=None, json=None):
        r = self.session.post(f"{self.server}{path}", data=data, json=json)
        if r.status_code == 400:
            print(f"  ERROR 400: {r.text}")
        r.raise_for_status()
        return r


def get_all_locations(api):
    """Récupère tous les emplacements de stock via des appels API paginés."""
    locations = []
    offset = 0
    while True:
        r = api.get("/api/stock/location/", params={
            "limit": 100, "offset": offset,
        })
        data = r.json()
        locations.extend(data["results"])
        if not data["next"]:
            break
        offset += 100
    return locations


def get_location_templates(api):
    """Récupère tous les templates d'étiquette activés pour les emplacements de stock."""
    templates = []
    offset = 0
    while True:
        r = api.get("/api/label/template/", params={
            "limit": 100, "offset": offset,
            "model_type": "stocklocation", "enabled": True,
        })
        data = r.json()
        templates.extend(data["results"])
        if not data["next"]:
            break
        offset += 100
    return templates


def print_and_download_label(api, template_pk, item_pks, timeout=60):
    """Soumet un job d'impression d'étiquette, interroge jusqu'à complétion, télécharge le PDF.

    Renvoie le contenu du PDF sous forme de bytes.
    """
    r = api.post("/api/label/print/", json={
        "template": template_pk,
        "items": item_pks,
    })
    result = r.json()
    output_pk = result["pk"]

    deadline = time.time() + timeout
    while time.time() < deadline:
        r = api.get(f"/api/data-output/{output_pk}/")
        data = r.json()
        if data.get("complete"):
            output_path = data.get("output")
            if not output_path:
                raise RuntimeError(
                    f"Sortie d'étiquette {output_pk} terminée mais aucun chemin de sortie"
                )
            pdf_url = f"{api.server}{output_path}"
            pr = api.session.get(pdf_url)
            pr.raise_for_status()
            return pr.content
        time.sleep(0.5)

    raise TimeoutError(
        f"La sortie d'étiquette {output_pk} ne s'est pas terminée dans les {timeout}s"
    )


def sanitize_filename(name):
    """Rend une chaîne sûre pour être utilisée comme nom de fichier."""
    for ch in r'<>:"/\\|?*':
        name = name.replace(ch, "_")
    return name.strip()


def normalize_ws(s):
    """Réduit toutes les séquences d'espacement d'une chaîne à des espaces simples."""
    return re.sub(r"\s+", " ", s).strip()


def name_matches(name, pattern):
    """Vérifie si un nom d'emplacement correspond à un motif.

    Glob si le motif contient *, sinon sous-chaîne insensible à la casse.
    Les espacements sont normalisés dans les deux avant comparaison.
    """
    name = normalize_ws(name)
    pattern = normalize_ws(pattern)
    if "*" in pattern:
        return fnmatch.fnmatch(name, pattern)
    return pattern.lower() in name.lower()


def filter_locations(locations, query, parent, includes, excludes):
    """Filtre les emplacements par motif de nom, parent, listes d'inclusion/exclusion.

    - query : motif glob (si contient *) ou sous-chaîne insensible à la casse
    - parent : nom d'emplacement parent ou pk (chaîne d'entier)
    - includes : liste de motifs ; si non vide, seuls les emplacements
      correspondants sont conservés (prioritaire sur query et excludes)
    - excludes : liste de motifs ; les emplacements correspondants sont retirés
    """
    if includes:
        filtered = [l for l in locations if any(name_matches(l.get("name", ""), p) for p in includes)]
    else:
        filtered = locations

        if query:
            filtered = [l for l in filtered if name_matches(l.get("name", ""), query)]

    if parent:
        parent_pk = None
        if parent.isdigit():
            parent_pk = int(parent)
        else:
            for l in locations:
                if normalize_ws(l.get("name", "")) == normalize_ws(parent):
                    parent_pk = l["pk"]
                    break
            if parent_pk is None:
                print(f"ERREUR : Emplacement parent '{parent}' introuvable")
                sys.exit(1)
        filtered = [l for l in filtered if l.get("parent") == parent_pk]

    if excludes and not includes:
        filtered = [l for l in filtered if not any(name_matches(l.get("name", ""), p) for p in excludes)]
    elif excludes and includes:
        filtered = [l for l in filtered if not any(name_matches(l.get("name", ""), p) for p in excludes) or any(name_matches(l.get("name", ""), p) for p in includes)]

    return filtered


def main():
    parser = argparse.ArgumentParser(
        description="Télécharger et fusionner les étiquettes PDF d'emplacements de stock depuis InvenTree"
    )
    parser.add_argument(
        "-q", "--query", default=None,
        help="Filtrer les emplacements par nom (glob si contient *, sinon sous-chaîne)",
    )
    parser.add_argument(
        "-p", "--parent", default=None,
        help="Filtrer par emplacement parent (nom ou pk numérique)",
    )
    parser.add_argument(
        "-e", "--exclude", action="append", default=[],
        help="Exclure les emplacements correspondant à ce motif (glob ou sous-chaîne). Peut être donné plusieurs fois.",
    )
    parser.add_argument(
        "-i", "--include", action="append", default=[],
        help="Inclure uniquement les emplacements correspondant à ce motif (glob ou sous-chaîne). Prioritaire sur -q et -e. Peut être donné plusieurs fois.",
    )
    parser.add_argument(
        "-t", "--template", default=None,
        help="Nom du template d'étiquette à utiliser (par défaut : le premier disponible)",
    )
    parser.add_argument(
        "-o", "--output", default=None,
        help="Chemin de sortie : fichier PDF (mode fusionné) ou répertoire (mode individuel). "
             "Si non fourni, déduit automatiquement du terme de recherche ou du motif d'inclusion.",
    )
    parser.add_argument(
        "--individual", action="store_true",
        help="Écrire des PDF individuels dans un répertoire au lieu de fusionner en un PDF",
    )
    parser.add_argument(
        "--workers", type=int, default=8,
        help="Nombre de jobs d'impression en parallèle (par défaut : 8)",
    )
    parser.add_argument(
        "--timeout", type=int, default=60,
        help="Timeout en secondes par job d'étiquette (par défaut : 60)",
    )
    args = parser.parse_args()

    config = load_config()
    api = InvenTreeAPI(config["server"], config["token"])

    # --- Récupération des templates d'étiquette ---
    templates = get_location_templates(api)
    if not templates:
        print("ERREUR : Aucun template d'étiquette activé trouvé pour model_type 'stocklocation'")
        sys.exit(1)

    print(f"{len(templates)} template(s) d'étiquette d'emplacement de stock trouvé(s) :")
    for t in templates:
        print(f"  - {t['name']} (pk={t['pk']}, {t['width']}x{t['height']}mm)")

    selected = None
    if args.template:
        for t in templates:
            if t["name"] == args.template:
                selected = t
                break
        if not selected:
            print(f"ERREUR : Template '{args.template}' introuvable")
            sys.exit(1)
    else:
        selected = templates[0]
    print(f"\nTemplate utilisé : {selected['name']} (pk={selected['pk']})")

    # --- Récupération et filtrage des emplacements ---
    locations = get_all_locations(api)
    print(f"{len(locations)} emplacements de stock au total")

    locations = filter_locations(locations, args.query, args.parent, args.include, args.exclude)
    print(f"Après filtrage : {len(locations)} emplacements")

    if not locations:
        print("Aucun emplacement ne correspond aux critères de filtrage.")
        return

    for loc in locations:
        print(f"  {loc['name']} (pk={loc['pk']})")

    # --- Impression des étiquettes en parallèle ---
    print(f"\nGénération de {len(locations)} étiquettes en parallèle "
          f"({args.workers} workers)...")

    results = {}  # pk -> (nom, pdf_bytes ou None)
    errors = {}

    def _print_one(loc):
        name = loc.get("name", f"location_{loc['pk']}")
        pk = loc["pk"]
        try:
            pdf = print_and_download_label(
                api, selected["pk"], [pk], timeout=args.timeout
            )
            return pk, name, pdf, None
        except Exception as e:
            return pk, name, None, str(e)

    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futures = {pool.submit(_print_one, loc): loc for loc in locations}
        for fut in as_completed(futures):
            pk, name, pdf, err = fut.result()
            if err:
                print(f"  ÉCHEC : {name} (pk={pk}) : {err}")
                errors[pk] = err
            else:
                print(f"  OK : {name} (pk={pk}, {len(pdf)} bytes)")
                results[pk] = (name, pdf)

    if not results:
        print("\nERREUR : Aucune étiquette n'a été générée avec succès.")
        sys.exit(1)

    # --- Détermination du chemin de sortie ---
    def derive_name():
        """Déduit automatiquement le nom de sortie à partir de la requête ou du premier motif d'inclusion."""
        source = None
        if args.query:
            source = args.query
        elif args.include:
            source = args.include[0]
        if source:
            # Supprimer les caractères glob, normaliser les espacements, remplacer les espaces par _
            cleaned = re.sub(r"[*?]", "", source)
            cleaned = normalize_ws(cleaned).replace(" ", "_")
            cleaned = sanitize_filename(cleaned)
            return cleaned if cleaned else "StockLocationLabels"
        return "StockLocationLabels"

    if args.output:
        output_path = Path(args.output)
    else:
        base_name = derive_name()
        if args.individual:
            output_path = Path(base_name)
        else:
            output_path = Path(f"{base_name}.pdf")

    # --- Écriture de la sortie ---
    if args.individual:
        output_path.mkdir(parents=True, exist_ok=True)
        print(f"\nÉcriture de {len(results)} PDF individuels dans {output_path}/...")
        for pk in sorted(results.keys()):
            name, pdf_bytes = results[pk]
            safe_name = sanitize_filename(normalize_ws(name).replace(" ", "_"))
            pdf_path = output_path / f"{safe_name}.pdf"
            pdf_path.write_bytes(pdf_bytes)
            print(f"  {pdf_path.name} ({len(pdf_bytes)} bytes)")
        print(f"\nTerminé : {len(results)} étiquettes écrites, {len(errors)} échouées")
        print(f"Répertoire de sortie : {output_path.resolve()}")
    else:
        print(f"\nFusion de {len(results)} PDF dans {output_path}...")
        writer = PdfWriter()
        for pk in sorted(results.keys()):
            name, pdf_bytes = results[pk]
            reader = PdfReader(io.BytesIO(pdf_bytes))
            for page in reader.pages:
                writer.add_page(page)

        output_path.parent.mkdir(parents=True, exist_ok=True)
        with open(output_path, "wb") as f:
            writer.write(f)

        print(f"\nTerminé : {len(results)} étiquettes fusionnées, {len(errors)} échouées")
        print(f"Sortie : {output_path.resolve()}")


if __name__ == "__main__":
    main()