Hallo,
da wir gerade mit Dolibarr (v23) angefangen haben, wollten wir unsere alten Lieferungen von zwei Artikeln mit Seriennummer importieren.
Herausgekommen ist ein Python-Script, das eine CSV-Datei mit folgenden Spalten entgegennimmt: Kunden-ID, Lieferscheinname, Seriennummer Artikel 1, Seriennummer Artikel 2.
#!/usr/bin/env python3
„“"
Dolibarr Historischer Lieferschein-Import
Liest die CSV und legt pro Zeile folgendes an:
- Seriennummern (llx_product_lot) für Produkt 1 (hw_id) und Produkt 2 (stick_id)
- Lieferschein-Kopf (llx_expedition) mit Kundennummer und Lieferscheinnummer
- Kundenauftrag (llx_commande) mit Auftragspositionen (llx_commandedet)
- Lieferschein-Positionen (llx_expeditiondet) je Produkt
- Lagerbewegungen (llx_product_stock_movement) als Ausgang
Konfiguration anpassen, dann ausführen:
python3 dolibarr_import.py --dry-run # Nur prüfen, nichts schreiben
python3 dolibarr_import.py # Echter Import
„“"
import csv
import argparse
import sys
from datetime import datetime
try:
import mysql.connector
except ImportError:
print(„Bitte mysql-connector-python installieren: pip install mysql-connector-python“)
sys.exit(1)
============================================================
KONFIGURATION – hier anpassen
============================================================
DB_CONFIG = {
„host“: „localhost“,
„port“: 3306,
„user“: „doliuser“,
„password“: „dolipass“,
„database“: „dolibarr“,
}
CSV_FILE = „doliimport-1.csv“
ENTITY = 1 # Dolibarr-Entity (fast immer 1)
FK_USER = 1 # Dolibarr-User-ID für den Import (rowid aus llx_user)
WAREHOUSE_ID = 1 # Lager-ID (rowid aus llx_entrepot)
PRODUCT_ID_HW = 24 # Produkt-ID für hw_id (rowid aus llx_product)
PRODUCT_ID_STICK= 28 # Produkt-ID für stick_id (rowid aus llx_product)
DELIVERY_DATE = „2024-01-01“ # Historisches Lieferdatum (YYYY-MM-DD)
============================================================
def get_or_create_lot(cursor, product_id, serial, entity, dry_run):
„“„Seriennummer in llx_product_lot anlegen, falls nicht vorhanden.“„“
cursor.execute(
„SELECT rowid FROM llx_product_lot WHERE fk_product = %s AND batch = %s AND entity = %s“,
(product_id, serial, entity)
)
row = cursor.fetchone()
if row:
return row[0]
if dry_run:
print(f" [DRY] Neue Seriennummer: Produkt {product_id} / {serial}")
return -1
cursor.execute(
"""INSERT INTO llx_product_lot
(fk_product, batch, entity, tms, datec)
VALUES (%s, %s, %s, NOW(), NOW())""",
(product_id, serial, entity)
)
return cursor.lastrowid
def create_expedition(cursor, fk_soc, ref, delivery_date, entity, fk_user, dry_run):
„“„Lieferschein-Kopf in llx_expedition anlegen.“„“
if dry_run:
print(f" [DRY] Neuer Lieferschein: {ref} für Kunde {fk_soc}")
return -1
cursor.execute(
"""INSERT INTO llx_expedition
(ref, fk_soc, date_delivery, fk_statut, entity, fk_user_author, date_creation, tms)
VALUES (%s, %s, %s, 2, %s, %s, NOW(), NOW())""",
(ref, fk_soc, delivery_date, entity, fk_user)
)
return cursor.lastrowid
def create_commande(cursor, fk_soc, ref, entity, fk_user, dry_run):
„“„Kundenauftrag in llx_commande anlegen.“„“
if dry_run:
print(f" [DRY] Neuer Auftrag: {ref} für Kunde {fk_soc}")
return -1
cursor.execute(
"""INSERT INTO llx_commande
(ref, ref_client, fk_soc, entity, fk_user_author, date_creation, tms)
VALUES (%s, %s, %s, %s, %s, NOW(), NOW())""",
(ref, ref, fk_soc, entity, fk_user)
)
return cursor.lastrowid
def create_element_element(cursor, cmd_id, exp_id, dry_run):
„“„Verknüpfung Auftrag ↔ Lieferschein in llx_element_element anlegen.“„“
if dry_run:
print(f" [DRY] Verknüpfung: Auftrag {cmd_id} → Lieferschein {exp_id}")
return
cursor.execute(
"""INSERT INTO llx_element_element
(fk_source, sourcetype, fk_target, targettype)
VALUES (%s, 'commande', %s, 'shipping')""",
(cmd_id, exp_id)
)
def create_commandedet(cursor, fk_commande, fk_product, rang, dry_run):
„“„Auftragsposition in llx_commandedet anlegen.“„“
if dry_run:
print(f" [DRY] Auftragsposition: Auftrag {fk_commande} / Produkt {fk_product} / Rang {rang}")
return -1
cursor.execute(
"SELECT description FROM llx_product WHERE rowid = %s",
(fk_product,)
)
row = cursor.fetchone()
description = row[0] if row and row[0] else ""
cursor.execute(
"""INSERT INTO llx_commandedet
(fk_commande, fk_product, rang, description)
VALUES (%s, %s, %s, %s)""",
(fk_commande, fk_product, rang, description)
)
return cursor.lastrowid
def create_expedition_detail(cursor, fk_expedition, fk_product, lot_id, serial, stock_id, commandedet_id, rang, dry_run):
„“„Lieferschein-Position in llx_expeditiondet anlegen.“„“
if dry_run:
print(f" [DRY] Position: Expedition {fk_expedition} / Produkt {fk_product} / {serial} / Rang {rang}")
return
cursor.execute(
"""INSERT INTO llx_expeditiondet
(fk_expedition, fk_product, fk_entrepot, qty, rang, fk_elementdet)
VALUES (%s, %s, %s, 1, %s, %s)""",
(fk_expedition, fk_product, WAREHOUSE_ID, rang, commandedet_id)
)
det_id = cursor.lastrowid
# Seriennummer zur Position verknüpfen
if lot_id and lot_id > 0:
cursor.execute(
"""INSERT INTO llx_expeditiondet_batch
(fk_expeditiondet, batch, qty, fk_origin_stock, fk_warehouse)
VALUES (%s, %s, 1, %s, %s)""",
(det_id, serial, stock_id, WAREHOUSE_ID)
)
def create_stock_mouvement(cursor, fk_product, serial, lot_id, warehouse_id, fk_user, exp_id, dry_run):
„“„Zuführung und Auslieferung in llx_stock_mouvement anlegen.“„“
if dry_run:
print(f" [DRY] Lagerbewegung: Produkt {fk_product} / {serial}")
return -1
# Zuführung
cursor.execute(
"""INSERT INTO llx_stock_mouvement
(fk_product, fk_entrepot, value, type_mouvement, fk_origin, batch, datem, fk_user_author)
VALUES (%s, %s, 1, 0, 0, %s, NOW(), %s)""",
(fk_product, warehouse_id, serial, fk_user)
)
# Auslieferung
cursor.execute(
"""INSERT INTO llx_stock_mouvement
(fk_product, fk_entrepot, value, type_mouvement, fk_origin, origintype, batch, datem, fk_user_author)
VALUES (%s, %s, -1, 2, %s, 'shipping', %s, NOW(), %s)""",
(fk_product, warehouse_id, exp_id, serial, fk_user)
)
return cursor.lastrowid
def main():
parser = argparse.ArgumentParser()
parser.add_argument(„–dry-run“, action=„store_true“, help=„Nur simulieren, nichts in DB schreiben“)
args = parser.parse_args()
dry_run = args.dry_run
if dry_run:
print("=== DRY-RUN MODUS – keine Datenbankänderungen ===\n")
conn = mysql.connector.connect(**DB_CONFIG)
cursor = conn.cursor()
delivery_date = datetime.strptime(DELIVERY_DATE, "%Y-%m-%d").date()
errors = []
count = 0
with open(CSV_FILE, newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
for lineno, row in enumerate(reader, start=2):
fk_soc = row["location_name"].strip()
ref = row["internal_id"].strip()
hw_id = row["hw_id"].strip()
stick_id = row["stick_id"].strip()
print(f"Zeile {lineno}: {ref} (Kunde {fk_soc})")
try:
# Seriennummern anlegen
lot_hw = get_or_create_lot(cursor, PRODUCT_ID_HW, hw_id, ENTITY, dry_run)
lot_stick = get_or_create_lot(cursor, PRODUCT_ID_STICK, stick_id, ENTITY, dry_run)
# Auftrag anlegen
cmd_id = create_commande(cursor, fk_soc, ref, ENTITY, FK_USER, dry_run)
cmd_det_hw = create_commandedet(cursor, cmd_id, PRODUCT_ID_HW, 1, dry_run)
cmd_det_stick = create_commandedet(cursor, cmd_id, PRODUCT_ID_STICK, 2, dry_run)
# Lieferschein anlegen
exp_id = create_expedition(cursor, fk_soc, ref, delivery_date, ENTITY, FK_USER, dry_run)
create_element_element(cursor, cmd_id, exp_id, dry_run)
# Lagerbewegungen (zuerst, damit wir die rowid für fk_origin_stock haben)
stock_id_hw = create_stock_mouvement(cursor, PRODUCT_ID_HW, hw_id, lot_hw, WAREHOUSE_ID, FK_USER, exp_id, dry_run)
stock_id_stick = create_stock_mouvement(cursor, PRODUCT_ID_STICK, stick_id, lot_stick, WAREHOUSE_ID, FK_USER, exp_id, dry_run)
# Positionen anlegen (rang 1 = hw_id, rang 2 = stick_id)
create_expedition_detail(cursor, exp_id, PRODUCT_ID_HW, lot_hw, hw_id, stock_id_hw, cmd_det_hw, 1, dry_run)
create_expedition_detail(cursor, exp_id, PRODUCT_ID_STICK, lot_stick, stick_id, stock_id_stick, cmd_det_stick, 2, dry_run)
if not dry_run:
conn.commit()
count += 1
except Exception as e:
conn.rollback()
msg = f"Zeile {lineno} ({ref}): FEHLER – {e}"
print(f" ❌ {msg}")
errors.append(msg)
cursor.close()
conn.close()
print(f"\n{'='*50}")
print(f"Verarbeitet: {count} Datensätze")
if errors:
print(f"Fehler ({len(errors)}):")
for e in errors:
print(f" - {e}")
else:
print("Keine Fehler.")
if name == „main“:
main()
