grotti/grotti/store.py

"""Projekte, Dateien, Spulen und Läufe in PostgreSQL."""

import json
import uuid
from datetime import date, datetime, timedelta, timezone
from pathlib import Path

from psycopg.types.json import Json

from grotti.catalog import MACHINE_LABELS, MATERIALS
from grotti.printers import FAMILY_LABELS, family_from_name
def iso(value):
    return value.isoformat() if value else None


def project_out(row, files=None, file_count=None):
    data = {
        "id": row["id"],
        "name": row["name"],
        "notes": row["notes"],
        "status": row["status"],
        "quote": row["quote"] or {},
        "folder": (row.get("folder") or "").strip(),
        "created_at": iso(row["created_at"]),
    }
    if files is not None:
        data["files"] = files
    if file_count is not None:
        data["file_count"] = file_count
    return data


def file_folder(stored_path, source, watch_dir=None):
    """Relativpfad des Elternordners unter watch_dir; Uploads → „Upload“."""
    if (source or "") == "upload":
        return "Upload"
    if not stored_path or not watch_dir:
        return ""
    try:
        rel = Path(stored_path).resolve().relative_to(Path(watch_dir).resolve())
    except (ValueError, OSError):
        return ""
    parent = rel.parent.as_posix()
    return "" if parent in (".", "") else parent


def file_out(row, watch_dir=None):
    # stored_path bleibt intern (get_file_row / list_file_rows), nicht in der API.
    from grotti.parse import path_has_ace_trailer

    stored = row.get("stored_path") or ""
    ace_trailer = False
    if stored:
        ace_trailer = path_has_ace_trailer(stored, kind=row.get("kind") or "", plate=row.get("plate") or "")
    return {
        "id": row["id"],
        "project_id": row["project_id"],
        "spool_id": row["spool_id"],
        "name": row["name"],
        "source": row["source"],
        "kind": row["kind"],
        "printer_family": row["printer_family"],
        "filaments": row["filaments"] or [],
        "weight_g": row["weight_g"],
        "estimated_s": row["estimated_s"],
        "plate": row["plate"],
        "size_bytes": row["size_bytes"],
        "mtime": row["mtime"],
        "slice_meta": row.get("slice_meta") or {},
        "ace_trailer": ace_trailer,
        "folder": file_folder(stored, row.get("source"), watch_dir),
        "preview": "/api/v1/dateien/%s/vorschau" % row["id"] if row["thumb_entry"] else "",
    }


def spool_out(row):
    colors = row.get("colors") or []
    if isinstance(colors, str):
        colors = json.loads(colors)
    archived = row.get("archived_at")
    return {
        "id": row["id"],
        "name": row["name"],
        "brand": row["brand"],
        "material_type": row["material_type"],
        "color": row["color"],
        "colors": colors,
        "price_per_kg": row["price_per_kg"],
        "density": row["density"],
        "volumetric": row["volumetric"],
        "remaining_g": row["remaining_g"],
        "printer_host": row["printer_host"],
        "ace_slot": row.get("ace_slot"),
        "material_id": row.get("material_id"),
        "chargen_nr": row.get("chargen_nr") or "",
        "hersteller_charge": row.get("hersteller_charge") or "",
        "code": row.get("code") or "",
        "archived_at": archived.isoformat() if archived and hasattr(archived, "isoformat") else (archived or None),
        "archived": bool(archived),
    }


def next_chargen(conn):
    prefix = "S-%s-" % date.today().strftime("%Y%m%d")
    # Advisory-Lock gegen Doppelvergabe bei parallelen Worker/Requests.
    conn.execute("SELECT pg_advisory_xact_lock(%s)", (870011,))
    row = conn.execute(
        """
        SELECT chargen_nr FROM spools
        WHERE chargen_nr LIKE %s
        ORDER BY chargen_nr DESC
        LIMIT 1
        """,
        (prefix + "%",),
    ).fetchone()
    number = 1
    if row and row["chargen_nr"]:
        try:
            number = int(str(row["chargen_nr"]).rsplit("-", 1)[-1]) + 1
        except ValueError:
            number = 1
    return "%s%04d" % (prefix, number)


def ensure_spool_codes(conn):
    rows = conn.execute(
        """
        SELECT id, chargen_nr, code FROM spools
        WHERE chargen_nr IS NULL OR chargen_nr = '' OR code IS NULL OR code = ''
        """
    ).fetchall()
    for row in rows:
        code = (row["chargen_nr"] or row["code"] or "").strip() or next_chargen(conn)
        conn.execute(
            """
            UPDATE spools
            SET chargen_nr = COALESCE(NULLIF(chargen_nr, ''), %s),
                code = COALESCE(NULLIF(code, ''), %s)
            WHERE id = %s
            """,
            (code, code, row["id"]),
        )


def run_out(row):
    planned = float(row.get("weight_planned_g") or 0) or float(row.get("weight_g") or 0)
    used = float(row.get("weight_used_g") or 0)
    if used <= 0 and row.get("deducted"):
        used = float(row.get("weight_g") or 0)
    return {
        "id": row["id"],
        "printer_host": row["printer_host"],
        "printer_name": row["printer_name"],
        "filename": row["filename"],
        "file_id": row["file_id"],
        "project_id": row["project_id"],
        "spool_id": row["spool_id"],
        "state": row["state"],
        "started_at": iso(row["started_at"]),
        "ended_at": iso(row["ended_at"]),
        "duration_s": row["duration_s"],
        "weight_g": float(row.get("weight_g") or 0),
        "weight_planned_g": planned,
        "weight_used_g": used,
        "weight_source": row.get("weight_source") or "",
        "filament_used_mm": float(row.get("filament_used_mm") or 0),
        "filament_total_mm": float(row.get("filament_total_mm") or 0),
        "progress_last": float(row.get("progress_last") or 0),
        "estimated_s": float(row.get("estimated_s") or 0),
        "filament_cost": row["filament_cost"],
        "energy_cost": row["energy_cost"],
        "deducted": row["deducted"],
        "needs_spool": row["needs_spool"],
        "verdict": row["verdict"],
        "verdict_reason": row.get("verdict_reason") or "",
        "pieces_ok": int(row.get("pieces_ok") or 0),
        "pieces_bad": int(row.get("pieces_bad") or 0),
        "order_item_id": row.get("order_item_id"),
        "material_type": row["material_type"],
        "chargen_nr": row.get("chargen_nr") or "",
        "message": row.get("message") or "",
        "objects": _json_list(row.get("objects")),
        "cost_per_part": (
            (float(row.get("filament_cost") or 0) + float(row.get("energy_cost") or 0))
            / int(row["pieces_ok"])
            if int(row.get("pieces_ok") or 0) > 0
            else None
        ),
    }


def list_projects(conn):
    rows = conn.execute(
        """
        SELECT p.*, COUNT(f.id) AS file_count
        FROM projects p
        LEFT JOIN files f ON f.project_id = p.id
        GROUP BY p.id
        ORDER BY p.created_at DESC
        """
    ).fetchall()
    return [project_out(row, file_count=row["file_count"]) for row in rows]


def get_project(conn, project_id):
    row = conn.execute("SELECT * FROM projects WHERE id = %s", (project_id,)).fetchone()
    if not row:
        return None
    files = conn.execute(
        "SELECT * FROM files WHERE project_id = %s ORDER BY mtime DESC, name",
        (project_id,),
    ).fetchall()
    return project_out(row, files=[file_out(item) for item in files])


def create_project(conn, name, notes, status, folder=""):
    row = conn.execute(
        """
        INSERT INTO projects (name, notes, status, folder)
        VALUES (%s, %s, %s, %s)
        RETURNING *
        """,
        (name, notes, status, (folder or "").strip()),
    ).fetchone()
    return project_out(row, files=[], file_count=0)


def update_project(conn, project_id, fields):
    sets = []
    values = []
    for key in ("name", "notes", "status", "folder"):
        if key in fields:
            sets.append("%s = %%s" % key)
            values.append(fields[key] if key != "folder" else (fields[key] or "").strip())
    if "quote" in fields:
        sets.append("quote = %s")
        values.append(Json(fields["quote"]))
    if not sets:
        return get_project(conn, project_id)
    values.append(project_id)
    row = conn.execute(
        "UPDATE projects SET %s WHERE id = %%s RETURNING id" % ", ".join(sets),
        values,
    ).fetchone()
    if not row:
        return None
    return get_project(conn, project_id)


def list_project_folders(conn):
    """id + folder für Sync (nur Projekte mit Ordner)."""
    return conn.execute(
        """
        SELECT id, folder FROM projects
        WHERE coalesce(folder, '') <> ''
        ORDER BY length(folder) DESC, folder
        """
    ).fetchall()


def set_file_project(conn, file_id, project_id):
    conn.execute(
        "UPDATE files SET project_id = %s WHERE id = %s",
        (project_id, file_id),
    )


def delete_project(conn, project_id):
    row = conn.execute("DELETE FROM projects WHERE id = %s RETURNING id", (project_id,)).fetchone()
    return bool(row)


def list_file_rows(conn):
    """Rohzeilen inkl. stored_path — für interne Pfad-Operationen."""
    return conn.execute("SELECT * FROM files ORDER BY mtime DESC, name").fetchall()


def list_files(conn, watch_dir=None):
    return [file_out(row, watch_dir=watch_dir) for row in list_file_rows(conn)]


def get_file_row(conn, file_id):
    return conn.execute("SELECT * FROM files WHERE id = %s", (file_id,)).fetchone()


def upsert_scanned(conn, meta, force=False):
    statement = """
        INSERT INTO files (
            name, stored_path, source, kind, printer_family, filaments, weight_g,
            estimated_s, thumb_entry, plate, size_bytes, mtime, slice_meta
        ) VALUES (
            %(name)s, %(stored_path)s, 'watch', %(kind)s, %(printer_family)s, %(filaments)s,
            %(weight_g)s, %(estimated_s)s, %(thumb_entry)s, %(plate)s, %(size_bytes)s, %(mtime)s,
            %(slice_meta)s
        )
        ON CONFLICT (source, stored_path) DO UPDATE SET
            name = EXCLUDED.name,
            kind = EXCLUDED.kind,
            printer_family = EXCLUDED.printer_family,
            filaments = EXCLUDED.filaments,
            weight_g = EXCLUDED.weight_g,
            estimated_s = EXCLUDED.estimated_s,
            thumb_entry = EXCLUDED.thumb_entry,
            plate = EXCLUDED.plate,
            size_bytes = EXCLUDED.size_bytes,
            mtime = EXCLUDED.mtime,
            slice_meta = EXCLUDED.slice_meta
    """
    if not force:
        statement += """
            WHERE files.size_bytes IS DISTINCT FROM EXCLUDED.size_bytes
               OR files.mtime IS DISTINCT FROM EXCLUDED.mtime
        """
    payload = dict(meta)
    payload["filaments"] = Json(meta.get("filaments") or [])
    payload["slice_meta"] = Json(meta.get("slice_meta") or {})
    conn.execute(statement, payload)


def insert_upload(conn, meta):
    row = conn.execute(
        """
        INSERT INTO files (
            name, stored_path, source, kind, printer_family, filaments, weight_g,
            estimated_s, thumb_entry, plate, size_bytes, mtime, project_id, slice_meta
        ) VALUES (
            %(name)s, %(stored_path)s, 'upload', %(kind)s, %(printer_family)s, %(filaments)s,
            %(weight_g)s, %(estimated_s)s, %(thumb_entry)s, %(plate)s, %(size_bytes)s, %(mtime)s,
            %(project_id)s, %(slice_meta)s
        )
        RETURNING *
        """,
        {
            **meta,
            "filaments": Json(meta.get("filaments") or []),
            "slice_meta": Json(meta.get("slice_meta") or {}),
        },
    ).fetchone()
    return file_out(row)


def update_file(conn, file_id, fields, watch_dir=None):
    sets = []
    values = []
    for key in ("project_id", "spool_id", "printer_family", "weight_g", "estimated_s", "thumb_entry", "kind", "plate"):
        if key in fields:
            sets.append("%s = %%s" % key)
            values.append(fields[key])
    if "filaments" in fields:
        sets.append("filaments = %s")
        values.append(Json(fields.get("filaments") or []))
    if "slice_meta" in fields:
        sets.append("slice_meta = %s")
        values.append(Json(fields.get("slice_meta") or {}))
    if not sets:
        row = get_file_row(conn, file_id)
        return file_out(row, watch_dir=watch_dir) if row else None
    values.append(file_id)
    row = conn.execute(
        "UPDATE files SET %s WHERE id = %%s RETURNING *" % ", ".join(sets),
        values,
    ).fetchone()
    return file_out(row, watch_dir=watch_dir) if row else None


def update_file_location(conn, file_id, stored_path, name=None, size_bytes=None, mtime=None, source=None):
    """Pfad nach Verschieben auf der Platte — ID und Zuordnungen bleiben."""
    sets = ["stored_path = %s"]
    values = [stored_path]
    if name is not None:
        sets.append("name = %s")
        values.append(name)
    if size_bytes is not None:
        sets.append("size_bytes = %s")
        values.append(size_bytes)
    if mtime is not None:
        sets.append("mtime = %s")
        values.append(mtime)
    if source is not None:
        sets.append("source = %s")
        values.append(source)
    values.append(file_id)
    return conn.execute(
        "UPDATE files SET %s WHERE id = %%s RETURNING *" % ", ".join(sets),
        values,
    ).fetchone()


def delete_file(conn, file_id):
    return conn.execute("DELETE FROM files WHERE id = %s RETURNING *", (file_id,)).fetchone()


def forget_missing_watch(conn, seen):
    if seen:
        conn.execute(
            "DELETE FROM files WHERE source = 'watch' AND NOT (stored_path = ANY(%s))",
            (seen,),
        )
    else:
        conn.execute("DELETE FROM files WHERE source = 'watch'")


def list_spools(conn):
    ensure_spool_codes(conn)
    rows = conn.execute("SELECT * FROM spools ORDER BY name, id").fetchall()
    return [spool_out(row) for row in rows]


def get_spool_row(conn, spool_id):
    return conn.execute("SELECT * FROM spools WHERE id = %s", (spool_id,)).fetchone()


def get_spool_by_code(conn, code):
    code = (code or "").strip()
    if not code:
        return None
    ensure_spool_codes(conn)
    row = conn.execute(
        """
        SELECT * FROM spools
        WHERE code = %s OR chargen_nr = %s
        ORDER BY id
        LIMIT 1
        """,
        (code, code),
    ).fetchone()
    return spool_out(row) if row else None


def create_spool(conn, fields):
    payload = dict(fields)
    payload["colors"] = Json(payload.get("colors") or [])
    payload.setdefault("ace_slot", None)
    payload.setdefault("material_id", None)
    payload.setdefault("hersteller_charge", "")
    code = (payload.get("chargen_nr") or payload.get("code") or "").strip() or next_chargen(conn)
    payload["chargen_nr"] = code
    payload["code"] = code
    row = conn.execute(
        """
        INSERT INTO spools (
            name, brand, material_type, color, colors, price_per_kg, density, volumetric,
            remaining_g, printer_host, ace_slot, material_id, chargen_nr, hersteller_charge, code
        ) VALUES (
            %(name)s, %(brand)s, %(material_type)s, %(color)s, %(colors)s, %(price_per_kg)s,
            %(density)s, %(volumetric)s, %(remaining_g)s, %(printer_host)s, %(ace_slot)s, %(material_id)s,
            %(chargen_nr)s, %(hersteller_charge)s, %(code)s
        )
        RETURNING *
        """,
        payload,
    ).fetchone()
    return spool_out(row)


def update_spool(conn, spool_id, fields):
    allowed = (
        "name", "brand", "material_type", "color", "colors", "price_per_kg", "density",
        "volumetric", "remaining_g", "printer_host", "ace_slot", "material_id",
        "hersteller_charge",
    )
    sets = []
    values = []
    for key in allowed:
        if key in fields:
            sets.append("%s = %%s" % key)
            values.append(Json(fields[key]) if key == "colors" else fields[key])
    if "archived_at" in fields:
        if fields["archived_at"] is None or fields["archived_at"] is False:
            sets.append("archived_at = NULL")
        elif fields["archived_at"] is True or fields["archived_at"] == "now":
            sets.append("archived_at = now()")
        else:
            sets.append("archived_at = %s")
            values.append(fields["archived_at"])
    if not sets:
        row = get_spool_row(conn, spool_id)
        return spool_out(row) if row else None
    values.append(spool_id)
    row = conn.execute(
        "UPDATE spools SET %s WHERE id = %%s RETURNING *" % ", ".join(sets),
        values,
    ).fetchone()
    return spool_out(row) if row else None


def mark_spool_empty(conn, spool_id):
    """Rest auf 0, vom Drucker ins Regal, archivieren."""
    row = conn.execute(
        """
        UPDATE spools
        SET remaining_g = 0,
            printer_host = '',
            ace_slot = NULL,
            archived_at = COALESCE(archived_at, now())
        WHERE id = %s
        RETURNING *
        """,
        (spool_id,),
    ).fetchone()
    return spool_out(row) if row else None


def material_out(row):
    s1 = row["volumetric_s1"]
    return {
        "id": row["id"],
        "name": row["name"],
        "brand": row["brand"],
        "material_type": row["material_type"],
        "density": row["density"],
        "volumetric": row["volumetric"],
        "volumetric_s1": s1,
        "machines": row["machines"],
        "machines_label": MACHINE_LABELS.get(row["machines"], row["machines"]),
        "price_per_kg": row["price_per_kg"],
        "note": row["note"],
    }


def list_materials(conn):
    rows = conn.execute("SELECT * FROM materials ORDER BY name").fetchall()
    return [material_out(row) for row in rows]


def get_material(conn, material_id):
    row = conn.execute("SELECT * FROM materials WHERE id = %s", (material_id,)).fetchone()
    return material_out(row) if row else None


def create_material(conn, data):
    row = conn.execute(
        """
        INSERT INTO materials (
            name, brand, material_type, density, volumetric, volumetric_s1,
            machines, price_per_kg, note
        ) VALUES (
            %(name)s, %(brand)s, %(material_type)s, %(density)s, %(volumetric)s,
            %(volumetric_s1)s, %(machines)s, %(price_per_kg)s, %(note)s
        )
        RETURNING *
        """,
        data,
    ).fetchone()
    return material_out(row)


def update_material(conn, material_id, fields):
    allowed = (
        "name", "brand", "material_type", "density", "volumetric", "volumetric_s1",
        "machines", "price_per_kg", "note",
    )
    sets = []
    values = {"id": material_id}
    for key in allowed:
        if key in fields:
            sets.append("%s = %%(%s)s" % (key, key))
            values[key] = fields[key]
    if not sets:
        return get_material(conn, material_id)
    row = conn.execute(
        "UPDATE materials SET %s WHERE id = %%(id)s RETURNING *" % ", ".join(sets),
        values,
    ).fetchone()
    return material_out(row) if row else None


def delete_material(conn, material_id):
    used = conn.execute(
        "SELECT id FROM spools WHERE material_id = %s LIMIT 1",
        (material_id,),
    ).fetchone()
    if used:
        return "in_use"
    row = conn.execute(
        "DELETE FROM materials WHERE id = %s RETURNING id",
        (material_id,),
    ).fetchone()
    return "ok" if row else None


def seed_materials(conn):
    for item in MATERIALS:
        conn.execute(
            """
            INSERT INTO materials (
                name, brand, material_type, density, volumetric, volumetric_s1,
                machines, price_per_kg, note
            ) VALUES (
                %(name)s, %(brand)s, %(material_type)s, %(density)s, %(volumetric)s,
                %(volumetric_s1)s, %(machines)s, %(price_per_kg)s, %(note)s
            )
            ON CONFLICT (name) DO UPDATE SET
                brand = EXCLUDED.brand,
                material_type = EXCLUDED.material_type,
                density = EXCLUDED.density,
                volumetric = EXCLUDED.volumetric,
                volumetric_s1 = EXCLUDED.volumetric_s1,
                machines = EXCLUDED.machines,
                note = EXCLUDED.note
            """,
            item,
        )


def delete_spool(conn, spool_id):
    used = conn.execute(
        "SELECT 1 FROM run_spools WHERE spool_id = %s LIMIT 1",
        (spool_id,),
    ).fetchone()
    if used:
        from grotti.errors import GrottiError
        raise GrottiError(
            "Die Spule hat Läufe — erst archivieren oder behalten.",
            "spule",
            409,
        )
    seg = conn.execute(
        "SELECT 1 FROM run_segments WHERE spool_id = %s LIMIT 1",
        (spool_id,),
    ).fetchone()
    if seg:
        from grotti.errors import GrottiError
        raise GrottiError(
            "Die Spule steckt in Abschnitten — nicht löschen.",
            "spule",
            409,
        )
    row = conn.execute("DELETE FROM spools WHERE id = %s RETURNING id", (spool_id,)).fetchone()
    return bool(row)


def spools_for_printer(conn, host):
    return conn.execute(
        "SELECT * FROM spools WHERE printer_host = %s ORDER BY id",
        (host,),
    ).fetchall()


def spool_at_slot(conn, host, slot):
    """Eine Spule an Host + ACE-Platz (1-basiert) oder None."""
    if not host or slot is None:
        return None
    return conn.execute(
        """
        SELECT * FROM spools
        WHERE printer_host = %s AND ace_slot = %s
        ORDER BY id LIMIT 1
        """,
        (host, int(slot)),
    ).fetchone()


def file_by_name(conn, filename):
    base = (filename or "").replace("\\", "/").rsplit("/", 1)[-1]
    if not base:
        return None
    return conn.execute(
        "SELECT * FROM files WHERE name = %s ORDER BY created_at DESC LIMIT 1",
        (base,),
    ).fetchone()


def add_send(conn, printer_host, filename, file_id, project_id, spool_id, ace_map=None, bed_leveling=False):
    conn.execute(
        """
        INSERT INTO sends (printer_host, filename, file_id, project_id, spool_id, ace_map, bed_leveling)
        VALUES (%s, %s, %s, %s, %s, %s, %s)
        """,
        (
            printer_host,
            filename,
            file_id,
            project_id,
            spool_id,
            Json(ace_map if isinstance(ace_map, list) else []),
            bool(bed_leveling),
        ),
    )


def latest_send(conn, host, filename):
    base = (filename or "").replace("\\", "/").rsplit("/", 1)[-1]
    return conn.execute(
        """
        SELECT * FROM sends
        WHERE printer_host = %s AND (filename = %s OR filename = %s)
        ORDER BY created_at DESC
        LIMIT 1
        """,
        (host, filename or "", base),
    ).fetchone()


def latest_send_for_file(conn, host, file_id):
    """Letzter Sende-Eintrag für Datei auf diesem Host (für „Nochmal“)."""
    if not host or not file_id:
        return None
    row = conn.execute(
        """
        SELECT * FROM sends
        WHERE printer_host = %s AND file_id = %s
        ORDER BY created_at DESC
        LIMIT 1
        """,
        (host, file_id),
    ).fetchone()
    if not row:
        return None
    return {
        "id": row["id"],
        "printer_host": row["printer_host"],
        "filename": row["filename"],
        "file_id": row["file_id"],
        "project_id": row["project_id"],
        "spool_id": row["spool_id"],
        "ace_map": _json_list(row.get("ace_map")),
        "bed_leveling": bool(row.get("bed_leveling")),
        "created_at": iso(row["created_at"]),
    }


def upsert_run(conn, item):
    payload = {
        "weight_planned_g": float(item.get("weight_planned_g") or item.get("weight_g") or 0),
        "weight_used_g": float(item.get("weight_used_g") or 0),
        "weight_source": item.get("weight_source") or "",
        "filament_used_mm": float(item.get("filament_used_mm") or 0),
        "filament_total_mm": float(item.get("filament_total_mm") or 0),
        "progress_last": float(item.get("progress_last") or 0),
        "estimated_s": float(item.get("estimated_s") or 0),
        "message": item.get("message") or "",
        **item,
    }
    payload["objects"] = Json(item.get("objects") if item.get("objects") is not None else [])
    if "order_item_id" not in payload:
        payload["order_item_id"] = item.get("order_item_id")
    if "weight_g" not in payload or payload.get("weight_g") is None:
        payload["weight_g"] = payload["weight_planned_g"]
    return conn.execute(
        """
        INSERT INTO runs (
            printer_host, printer_name, filename, file_id, project_id, spool_id,
            state, started_at, ended_at, duration_s, weight_g, energy_cost,
            moonraker_job_id, needs_spool, material_type, order_item_id,
            weight_planned_g, weight_used_g, weight_source,
            filament_used_mm, filament_total_mm, progress_last, estimated_s,
            message, objects
        ) VALUES (
            %(printer_host)s, %(printer_name)s, %(filename)s, %(file_id)s, %(project_id)s, %(spool_id)s,
            %(state)s, %(started_at)s, %(ended_at)s, %(duration_s)s, %(weight_g)s, %(energy_cost)s,
            %(moonraker_job_id)s, %(needs_spool)s, %(material_type)s, %(order_item_id)s,
            %(weight_planned_g)s, %(weight_used_g)s, %(weight_source)s,
            %(filament_used_mm)s, %(filament_total_mm)s, %(progress_last)s, %(estimated_s)s,
            %(message)s, %(objects)s
        )
        ON CONFLICT (printer_host, moonraker_job_id) DO UPDATE SET
            printer_name = EXCLUDED.printer_name,
            filename = EXCLUDED.filename,
            file_id = COALESCE(runs.file_id, EXCLUDED.file_id),
            project_id = COALESCE(runs.project_id, EXCLUDED.project_id),
            spool_id = COALESCE(runs.spool_id, EXCLUDED.spool_id),
            order_item_id = COALESCE(runs.order_item_id, EXCLUDED.order_item_id),
            state = EXCLUDED.state,
            started_at = COALESCE(runs.started_at, EXCLUDED.started_at),
            ended_at = EXCLUDED.ended_at,
            duration_s = EXCLUDED.duration_s,
            weight_g = CASE WHEN EXCLUDED.weight_g > 0 THEN EXCLUDED.weight_g ELSE runs.weight_g END,
            weight_planned_g = CASE
                WHEN EXCLUDED.weight_planned_g > 0 THEN EXCLUDED.weight_planned_g
                ELSE runs.weight_planned_g
            END,
            weight_used_g = CASE
                WHEN EXCLUDED.weight_used_g > 0 THEN EXCLUDED.weight_used_g
                ELSE runs.weight_used_g
            END,
            weight_source = CASE
                WHEN EXCLUDED.weight_source <> '' THEN EXCLUDED.weight_source
                ELSE runs.weight_source
            END,
            filament_used_mm = GREATEST(runs.filament_used_mm, EXCLUDED.filament_used_mm),
            filament_total_mm = CASE
                WHEN EXCLUDED.filament_total_mm > 0 THEN EXCLUDED.filament_total_mm
                ELSE runs.filament_total_mm
            END,
            progress_last = GREATEST(runs.progress_last, EXCLUDED.progress_last),
            estimated_s = CASE
                WHEN EXCLUDED.estimated_s > 0 THEN EXCLUDED.estimated_s
                ELSE runs.estimated_s
            END,
            message = CASE
                WHEN EXCLUDED.message <> '' THEN EXCLUDED.message
                ELSE runs.message
            END,
            objects = CASE
                WHEN EXCLUDED.objects <> '[]'::jsonb THEN EXCLUDED.objects
                ELSE runs.objects
            END,
            energy_cost = EXCLUDED.energy_cost,
            needs_spool = CASE WHEN runs.deducted THEN false ELSE EXCLUDED.needs_spool END,
            material_type = CASE
                WHEN EXCLUDED.material_type <> '' THEN EXCLUDED.material_type
                ELSE runs.material_type
            END
        RETURNING *
        """,
        payload,
    ).fetchone()


def mark_deducted(conn, run_id, spool_id, filament_cost, energy_cost, weight_used_g=None, weight_source=""):
    return conn.execute(
        """
        UPDATE runs
        SET deducted = true, needs_spool = false, spool_id = %s,
            filament_cost = %s, energy_cost = %s,
            weight_used_g = COALESCE(%s, weight_used_g),
            weight_g = COALESCE(%s, weight_g),
            weight_source = CASE WHEN %s <> '' THEN %s ELSE weight_source END
        WHERE id = %s AND deducted = false
        RETURNING id
        """,
        (
            spool_id,
            filament_cost,
            energy_cost,
            weight_used_g,
            weight_used_g,
            weight_source or "",
            weight_source or "",
            run_id,
        ),
    ).fetchone()


def deduct_spool(conn, spool_id, grams):
    return conn.execute(
        "UPDATE spools SET remaining_g = remaining_g - %s WHERE id = %s RETURNING *",
        (grams, spool_id),
    ).fetchone()


def list_runs(conn, limit=200, since=None, until=None, host="", material="", brand=""):
    where = []
    values = []
    until_bound = exclusive_until(until)
    if since:
        where.append("r.started_at >= %s")
        values.append(since)
    if until_bound:
        where.append("r.started_at < %s")
        values.append(until_bound)
    if host:
        where.append("r.printer_host = %s")
        values.append(host)
    spool_join = ""
    if material or brand:
        spool_join = (
            "JOIN run_spools rs0 ON rs0.run_id = r.id "
            "JOIN spools s0 ON s0.id = rs0.spool_id"
        )
        if material:
            where.append("s0.material_type ILIKE %s")
            values.append(material)
        if brand:
            where.append("s0.brand ILIKE %s")
            values.append(brand)
    # Live-Zeile ausblenden, wenn derselbe Druck schon als History-Job offen ist
    where.append(
        """
        NOT (
          r.moonraker_job_id LIKE 'live:%%'
          AND r.state IN ('in_progress', 'printing', 'paused')
          AND EXISTS (
            SELECT 1 FROM runs h
            WHERE h.printer_host = r.printer_host
              AND h.filename = r.filename
              AND h.moonraker_job_id NOT LIKE 'live:%%'
              AND h.state IN ('in_progress', 'printing', 'paused')
          )
        )
        """
    )
    clause = (" WHERE " + " AND ".join(where)) if where else ""
    values.append(int(limit))
    rows = conn.execute(
        """
        SELECT DISTINCT r.*
        FROM runs r
        %s
        %s
        ORDER BY r.started_at DESC NULLS LAST, r.id DESC
        LIMIT %%s
        """ % (spool_join, clause),
        values,
    ).fetchall()
    return [run_out(row) for row in rows]


def get_run_row(conn, run_id):
    return conn.execute("SELECT * FROM runs WHERE id = %s", (run_id,)).fetchone()


def latest_run_for_host(conn, host):
    """Aktueller oder letzter Lauf mit Proben fürs Cockpit-Verlauf."""
    host = (host or "").strip()
    if not host:
        return None
    runs = list_runs(conn, limit=12, host=host)
    if not runs:
        return None
    active = [r for r in runs if (r.get("state") or "") in ("in_progress", "printing", "paused")]
    candidates = active + [r for r in runs if r not in active]
    best = None
    best_n = -1
    for run in candidates:
        row = conn.execute(
            "SELECT count(*) AS n FROM run_samples WHERE run_id = %s",
            (run["id"],),
        ).fetchone()
        n = int(row["n"] or 0) if row else 0
        if (run.get("state") or "") in ("in_progress", "printing", "paused"):
            run["_sample_count"] = n
            return run
        if n >= 2 and n > best_n:
            best = run
            best_n = n
            best["_sample_count"] = n
    return best


def get_run_by_chargen(conn, code):
    text = (code or "").strip()
    if not text:
        return None
    row = conn.execute(
        "SELECT * FROM runs WHERE chargen_nr = %s LIMIT 1",
        (text,),
    ).fetchone()
    if row:
        return row
    label = conn.execute(
        "SELECT run_id FROM run_labels WHERE chargen_nr = %s LIMIT 1",
        (text,),
    ).fetchone()
    if not label:
        return None
    return get_run_row(conn, label["run_id"])


def set_verdict(conn, run_id, verdict, verdict_reason="", pieces_ok=0, pieces_bad=0):
    from grotti import orders

    before = conn.execute(
        "SELECT order_item_id, pieces_ok, pieces_bad FROM runs WHERE id = %s",
        (run_id,),
    ).fetchone()
    row = conn.execute(
        """
        UPDATE runs
        SET verdict = %s, verdict_reason = %s, pieces_ok = %s, pieces_bad = %s
        WHERE id = %s
        RETURNING *
        """,
        (verdict, verdict_reason or "", int(pieces_ok or 0), int(pieces_bad or 0), run_id),
    ).fetchone()
    if row and before and before.get("order_item_id"):
        orders.apply_verdict_delta(
            conn,
            before["order_item_id"],
            before.get("pieces_ok"),
            before.get("pieces_bad"),
            pieces_ok,
            pieces_bad,
        )
    return run_out(row) if row else None


def run_for_abraeumen_verdict(conn, host):
    """Letzter abgeschlossener oder noch laufender Lauf des Hosts."""
    host = (host or "").strip()
    if not host:
        return None
    return conn.execute(
        """
        SELECT * FROM runs
        WHERE printer_host = %s
          AND state IN ('completed', 'in_progress', 'printing', 'paused')
        ORDER BY
            CASE WHEN state IN ('in_progress', 'printing', 'paused') THEN 0 ELSE 1 END,
            ended_at DESC NULLS LAST,
            started_at DESC NULLS LAST,
            id DESC
        LIMIT 1
        """,
        (host,),
    ).fetchone()


def assign_run_spool(conn, run_id, spool_id):
    return conn.execute(
        """
        UPDATE runs
        SET spool_id = %s
        WHERE id = %s
        RETURNING *
        """,
        (spool_id, run_id),
    ).fetchone()


def next_druck_chargen(conn):
    """Nächste Druck-Charge des Tages. Objekt-Etiketten (`D-…-01`) blockieren die Nummer."""
    day = date.today().strftime("%Y%m%d")
    prefix = "D-%s-" % day
    batch_re = "^D-%s-[0-9]{4}$" % day
    object_re = "^D-%s-[0-9]{4}-[0-9]{2}$" % day
    conn.execute("SELECT pg_advisory_xact_lock(%s)", (870012,))
    row = conn.execute(
        """
        SELECT n FROM (
            SELECT chargen_nr AS n FROM runs
            WHERE chargen_nr ~ %s
            UNION
            SELECT CASE
                WHEN chargen_nr ~ %s THEN regexp_replace(chargen_nr, '-[0-9]{2}$', '')
                ELSE chargen_nr
            END AS n
            FROM run_labels
            WHERE chargen_nr ~ %s OR chargen_nr ~ %s
        ) AS nums
        ORDER BY n DESC
        LIMIT 1
        """,
        (batch_re, object_re, batch_re, object_re),
    ).fetchone()
    number = 1
    if row and row["n"]:
        try:
            number = int(str(row["n"]).rsplit("-", 1)[-1]) + 1
        except ValueError:
            number = 1
    return "%s%04d" % (prefix, number)


def ensure_run_chargen(conn, run_id, weight_used_g=None):
    """Vergibt D-… am laufenden oder beendeten Druck. Bestehende Nummer bleibt.

    weight_used_g wird ignoriert (früher Gate); Parameter bleibt für Aufrufer.
    """
    row = get_run_row(conn, run_id)
    if not row:
        return None
    if row.get("chargen_nr"):
        return row["chargen_nr"]
    state = row.get("state") or ""
    if state not in (
        "in_progress", "printing", "paused",
        "completed", "cancelled", "error", "interrupted",
    ):
        return None
    code = next_druck_chargen(conn)
    updated = conn.execute(
        """
        UPDATE runs SET chargen_nr = %s
        WHERE id = %s AND (chargen_nr IS NULL OR chargen_nr = '')
        RETURNING chargen_nr
        """,
        (code, run_id),
    ).fetchone()
    return (updated or {}).get("chargen_nr") or code


def replace_run_spools(conn, run_id, entries):
    conn.execute("DELETE FROM run_spools WHERE run_id = %s", (run_id,))
    for item in entries:
        conn.execute(
            """
            INSERT INTO run_spools (
                run_id, spool_id, ace_slot, weight_g, weight_planned_g, cost, is_estimate
            ) VALUES (%s, %s, %s, %s, %s, %s, %s)
            """,
            (
                run_id,
                item["spool_id"],
                item.get("ace_slot"),
                float(item.get("weight_g") or 0),
                float(item.get("weight_planned_g") or 0),
                float(item.get("cost") or 0),
                bool(item.get("is_estimate")),
            ),
        )


def list_run_spools(conn, run_id):
    rows = conn.execute(
        """
        SELECT rs.*, s.name AS spool_name, s.chargen_nr AS spool_chargen,
               s.material_type, s.brand, s.color, s.colors, s.hersteller_charge,
               s.price_per_kg, s.remaining_g
        FROM run_spools rs
        JOIN spools s ON s.id = rs.spool_id
        WHERE rs.run_id = %s
        ORDER BY rs.ace_slot NULLS LAST, rs.id
        """,
        (run_id,),
    ).fetchall()
    out = []
    for row in rows:
        out.append({
            "spool_id": row["spool_id"],
            "ace_slot": row["ace_slot"],
            "weight_g": float(row["weight_g"] or 0),
            "weight_planned_g": float(row["weight_planned_g"] or 0),
            "cost": float(row["cost"] or 0),
            "is_estimate": bool(row["is_estimate"]),
            "name": row["spool_name"],
            "chargen_nr": row["spool_chargen"] or "",
            "material_type": row["material_type"] or "",
            "brand": row["brand"] or "",
            "color": row["color"] or "",
            "colors": _json_list(row.get("colors")),
            "hersteller_charge": row["hersteller_charge"] or "",
            "remaining_g": float(row["remaining_g"] or 0),
        })
    return out


def _etikett_spool_line(row, ace_slot=None):
    spool = spool_out(row)
    return {
        "spool_id": spool["id"],
        "name": spool["name"],
        "chargen_nr": spool["chargen_nr"],
        "material_type": spool["material_type"],
        "brand": spool["brand"],
        "hersteller_charge": spool["hersteller_charge"],
        "ace_slot": ace_slot if ace_slot is not None else spool.get("ace_slot"),
        "color": spool.get("color") or "",
    }


def etikett_spools_for_run(conn, run):
    """Material fürs Etikett: Abzug, sonst Send/ACE/Lauf-Spule, sonst Metadaten."""
    existing = list_run_spools(conn, run["id"])
    if existing:
        return existing
    out = []
    seen = set()

    def add_row(row, ace_slot=None):
        if not row:
            return
        sid = row.get("id") if "id" in row else row.get("spool_id")
        if sid is not None and sid in seen:
            return
        if sid is not None:
            seen.add(sid)
        out.append(_etikett_spool_line(row, ace_slot=ace_slot))

    host = (run.get("printer_host") or "").strip()
    sent = latest_send(conn, host, run.get("filename") or "") if host else None
    ace_map = _json_list((sent or {}).get("ace_map")) if sent else []
    if ace_map and host:
        for item in ace_map:
            try:
                slot = int(item.get("ams_index")) + 1
            except (TypeError, ValueError, AttributeError):
                continue
            row = conn.execute(
                """
                SELECT * FROM spools
                WHERE printer_host = %s AND ace_slot = %s
                ORDER BY id LIMIT 1
                """,
                (host, slot),
            ).fetchone()
            add_row(row, ace_slot=slot)
    spool_id = run.get("spool_id") or (sent or {}).get("spool_id")
    if spool_id:
        add_row(get_spool_row(conn, spool_id))
    if not out and host and not ace_map:
        rows = conn.execute(
            """
            SELECT * FROM spools
            WHERE printer_host = %s AND ace_slot IS NULL
            ORDER BY id
            """,
            (host,),
        ).fetchall()
        for row in rows:
            add_row(row)
    if not out and (run.get("material_type") or "").strip():
        out.append({
            "spool_id": None,
            "name": "",
            "chargen_nr": "",
            "material_type": (run.get("material_type") or "").strip(),
            "brand": "",
            "hersteller_charge": "",
            "ace_slot": None,
            "color": "",
        })
    return out


def list_spool_runs(conn, spool_id, limit=50):
    rows = conn.execute(
        """
        SELECT r.*, rs.weight_g AS spool_weight_g, rs.cost AS spool_cost,
               rs.is_estimate AS spool_is_estimate
        FROM run_spools rs
        JOIN runs r ON r.id = rs.run_id
        WHERE rs.spool_id = %s
        ORDER BY r.started_at DESC NULLS LAST, r.id DESC
        LIMIT %s
        """,
        (spool_id, int(limit)),
    ).fetchall()
    out = []
    for row in rows:
        item = run_out(row)
        item["spool_weight_g"] = float(row["spool_weight_g"] or 0)
        item["spool_cost"] = float(row["spool_cost"] or 0)
        item["spool_is_estimate"] = bool(row["spool_is_estimate"])
        out.append(item)
    return out


def spool_usage_stats(conn, spool_id, days=28):
    row = conn.execute(
        """
        SELECT
            COALESCE(SUM(rs.weight_g), 0) AS total_g,
            COALESCE(SUM(CASE WHEN r.verdict = 'ausschuss' THEN rs.weight_g ELSE 0 END), 0) AS scrap_g,
            COALESCE(SUM(rs.cost), 0) AS total_cost,
            COALESCE(SUM(CASE
                WHEN r.started_at >= now() - make_interval(days => %s) THEN rs.weight_g
                ELSE 0
            END), 0) AS recent_g
        FROM run_spools rs
        JOIN runs r ON r.id = rs.run_id
        WHERE rs.spool_id = %s
        """,
        (int(days), spool_id),
    ).fetchone()
    recent = float((row or {}).get("recent_g") or 0)
    per_day = recent / max(1, int(days))
    spool = get_spool_row(conn, spool_id)
    left = float((spool or {}).get("remaining_g") or 0)
    empty_in_days = None
    if per_day > 0.01 and left > 0:
        empty_in_days = left / per_day
    elif left <= 0:
        empty_in_days = 0.0
    return {
        "total_g": float((row or {}).get("total_g") or 0),
        "scrap_g": float((row or {}).get("scrap_g") or 0),
        "total_cost": float((row or {}).get("total_cost") or 0),
        "recent_g": recent,
        "per_day_g": per_day,
        "empty_in_days": empty_in_days,
        "fill_pct": (left / 10.0) if left else (0.0 if left == 0 else left / 10.0),
        "remaining_g": left,
    }


def find_open_history_run(conn, host, filename=""):
    """Offener Moonraker-History-Job (nicht live:) für Host/Datei."""
    host = (host or "").strip()
    if not host:
        return None
    if filename:
        row = conn.execute(
            """
            SELECT * FROM runs
            WHERE printer_host = %s
              AND filename = %s
              AND moonraker_job_id NOT LIKE 'live:%%'
              AND state IN ('in_progress', 'printing', 'paused')
            ORDER BY id DESC
            LIMIT 1
            """,
            (host, filename),
        ).fetchone()
        if row:
            return row
    return conn.execute(
        """
        SELECT * FROM runs
        WHERE printer_host = %s
          AND moonraker_job_id NOT LIKE 'live:%%'
          AND state IN ('in_progress', 'printing', 'paused')
        ORDER BY id DESC
        LIMIT 1
        """,
        (host,),
    ).fetchone()


def find_live_run(conn, host, filename=""):
    """Offener Live-Lauf für diesen Drucker (Watcher)."""
    if filename:
        row = conn.execute(
            """
            SELECT * FROM runs
            WHERE printer_host = %s
              AND filename = %s
              AND moonraker_job_id LIKE 'live:%%'
              AND state IN ('in_progress', 'printing', 'paused')
            ORDER BY id DESC
            LIMIT 1
            """,
            (host, filename),
        ).fetchone()
        if row:
            return row
    return conn.execute(
        """
        SELECT * FROM runs
        WHERE printer_host = %s
          AND moonraker_job_id LIKE 'live:%%'
          AND state IN ('in_progress', 'printing', 'paused')
        ORDER BY id DESC
        LIMIT 1
        """,
        (host,),
    ).fetchone()


def find_active_run(conn, host, filename=""):
    """Offener Lauf am Host: Live bevorzugt, sonst offener History-Job (Pause/Resume)."""
    return find_live_run(conn, host, filename) or find_open_history_run(conn, host, filename)


def update_send_spool(conn, host, filename, spool_id):
    """Spule am letzten Send dieser Datei korrigieren (Rollenwechsel während Druck)."""
    host = (host or "").strip()
    if not host or not spool_id:
        return None
    sent = latest_send(conn, host, filename or "")
    if not sent:
        return None
    if sent.get("spool_id") == spool_id:
        return sent
    return conn.execute(
        "UPDATE sends SET spool_id = %s WHERE id = %s RETURNING *",
        (spool_id, sent["id"]),
    ).fetchone()


def ensure_live_run(conn, host, printer_name, filename, spool_id=None):
    """Aktiven Lauf: History-Job bevorzugen, sonst Live — nie parallel beide offen."""
    hist = find_open_history_run(conn, host, filename)
    if hist:
        live = find_live_run(conn, host, filename)
        if live:
            adopt_live_run(conn, host, filename or hist.get("filename") or "", hist["id"])
            hist = get_run_row(conn, hist["id"]) or hist
        if spool_id and hist.get("spool_id") != spool_id:
            conn.execute("UPDATE runs SET spool_id = %s WHERE id = %s", (spool_id, hist["id"]))
            hist = get_run_row(conn, hist["id"]) or hist
        return hist
    row = find_live_run(conn, host, filename)
    if row:
        if filename and row.get("filename") != filename:
            conn.execute(
                "UPDATE runs SET filename = %s WHERE id = %s",
                (filename, row["id"]),
            )
            row = get_run_row(conn, row["id"])
        if spool_id and row and row.get("spool_id") != spool_id:
            conn.execute("UPDATE runs SET spool_id = %s WHERE id = %s", (spool_id, row["id"]))
            row = get_run_row(conn, row["id"])
        return row
    job_id = "live:%s:%s" % (host, datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S"))
    return upsert_run(conn, {
        "printer_host": host,
        "printer_name": printer_name or host,
        "filename": filename or "",
        "file_id": None,
        "project_id": None,
        "spool_id": spool_id,
        "state": "in_progress",
        "started_at": datetime.now(timezone.utc),
        "ended_at": None,
        "duration_s": 0,
        "weight_g": 0,
        "weight_planned_g": 0,
        "weight_used_g": 0,
        "weight_source": "",
        "filament_used_mm": 0,
        "filament_total_mm": 0,
        "progress_last": 0,
        "estimated_s": 0,
        "energy_cost": 0,
        "moonraker_job_id": job_id,
        "needs_spool": False,
        "material_type": "",
        "message": "",
        "objects": [],
    })


def abandon_live_runs(conn, host, keep_id=None):
    """Offene Live-Läufe beenden (z. B. vor einem neuen Druckstart)."""
    host = (host or "").strip()
    if not host:
        return 0
    rows = conn.execute(
        """
        UPDATE runs
        SET state = 'interrupted',
            ended_at = COALESCE(ended_at, now())
        WHERE printer_host = %s
          AND moonraker_job_id LIKE 'live:%%'
          AND state IN ('in_progress', 'printing', 'paused')
          AND (%s::bigint IS NULL OR id <> %s)
        RETURNING id
        """,
        (host, keep_id, keep_id),
    ).fetchall()
    return len(rows)


def complete_open_run(conn, run_id, state="completed"):
    """Offenen Lauf als beendet markieren (Abräumen / Druckende ohne History)."""
    state = (state or "completed").strip() or "completed"
    if state not in ("completed", "cancelled", "error", "interrupted"):
        state = "completed"
    return conn.execute(
        """
        UPDATE runs
        SET state = %s,
            ended_at = COALESCE(ended_at, now()),
            duration_s = CASE
                WHEN COALESCE(duration_s, 0) > 0 THEN duration_s
                WHEN started_at IS NOT NULL
                    THEN EXTRACT(EPOCH FROM (now() - started_at))
                ELSE COALESCE(duration_s, 0)
            END
        WHERE id = %s
          AND state IN ('in_progress', 'printing', 'paused')
        RETURNING *
        """,
        (state, run_id),
    ).fetchone()


def adopt_live_run(conn, host, filename, real_run_id):
    """Proben/Segmente vom Live-Lauf auf den History-Lauf umhängen."""
    live = find_live_run(conn, host, filename)
    if not live or live["id"] == real_run_id:
        return
    real = get_run_row(conn, real_run_id)
    if not real:
        return
    live_start = live.get("started_at")
    real_end = real.get("ended_at")
    # Gleiche Datei, neuer Druck: Live nicht an schon beendeten History-Job hängen.
    if live_start and real_end and live_start > real_end:
        return
    real_state = (real.get("state") or "")
    if real_state in ("completed", "cancelled", "error", "interrupted"):
        if live_start and real.get("started_at") and live_start > real["started_at"]:
            # Live begann nach dem History-Start und History ist schon durch — anderer Lauf.
            if not real_end or live_start > real_end:
                return
    live_id = live["id"]
    live_chargen = (live.get("chargen_nr") or "").strip()
    for table in ("run_samples", "run_segments", "run_spools"):
        conn.execute(
            "UPDATE %s SET run_id = %%s WHERE run_id = %%s" % table,
            (real_run_id, live_id),
        )
    conn.execute(
        "UPDATE events SET run_id = %s WHERE run_id = %s",
        (real_run_id, live_id),
    )
    conn.execute(
        """
        UPDATE runs AS dest SET
            progress_last = GREATEST(dest.progress_last, src.progress_last),
            filament_used_mm = GREATEST(dest.filament_used_mm, src.filament_used_mm),
            objects = CASE
                WHEN dest.objects::text IN ('[]', 'null') THEN src.objects
                ELSE dest.objects
            END
        FROM runs AS src
        WHERE dest.id = %s AND src.id = %s
        """,
        (real_run_id, live_id),
    )
    if live_chargen and real_state in ("completed", "cancelled", "error", "interrupted"):
        conn.execute("UPDATE runs SET chargen_nr = NULL WHERE id = %s", (live_id,))
        conn.execute(
            """
            UPDATE runs SET chargen_nr = %s
            WHERE id = %s AND (chargen_nr IS NULL OR chargen_nr = '')
            """,
            (live_chargen, real_run_id),
        )
    else:
        # Live-Charge war verfrüht — nicht übernehmen
        conn.execute("UPDATE runs SET chargen_nr = NULL WHERE id = %s", (live_id,))
    # Etiketten nur mit derselben Charge. Sonst bleiben sie am Live-Lauf und
    # fallen beim Löschen weg, statt eine fremde Nummer zu blockieren.
    real_after = get_run_row(conn, real_run_id)
    real_chargen = ((real_after or {}).get("chargen_nr") or "").strip()
    if live_chargen and real_chargen == live_chargen:
        conn.execute(
            """
            DELETE FROM run_labels AS src
            WHERE src.run_id = %s
              AND EXISTS (
                SELECT 1 FROM run_labels AS dst
                WHERE dst.run_id = %s AND dst.object_name = src.object_name
              )
            """,
            (live_id, real_run_id),
        )
        conn.execute(
            "UPDATE run_labels SET run_id = %s WHERE run_id = %s",
            (real_run_id, live_id),
        )
    conn.execute("DELETE FROM runs WHERE id = %s", (live_id,))


def update_run_progress(conn, run_id, fields):
    allowed = {
        "state", "progress_last", "filament_used_mm", "filament_total_mm",
        "weight_planned_g", "estimated_s", "message", "objects", "duration_s",
    }
    sets = []
    values = []
    for key, value in fields.items():
        if key not in allowed:
            continue
        if key == "objects":
            sets.append("objects = %s")
            values.append(Json(value if value is not None else []))
        else:
            sets.append("%s = %%s" % key)
            values.append(value)
    if not sets:
        return get_run_row(conn, run_id)
    values.append(run_id)
    return conn.execute(
        "UPDATE runs SET %s WHERE id = %%s RETURNING *" % ", ".join(sets),
        values,
    ).fetchone()


def add_run_sample(conn, run_id, sample):
    return conn.execute(
        """
        INSERT INTO run_samples (
            run_id, sampled_at, nozzle, nozzle_target, bed, bed_target, chamber,
            fan, speed_factor, extrude_factor, layer, progress,
            filament_used_mm, mmu_tool, nozzle_power, bed_power
        ) VALUES (
            %(run_id)s, COALESCE(%(sampled_at)s, now()), %(nozzle)s, %(nozzle_target)s,
            %(bed)s, %(bed_target)s, %(chamber)s, %(fan)s, %(speed_factor)s,
            %(extrude_factor)s, %(layer)s, %(progress)s, %(filament_used_mm)s, %(mmu_tool)s,
            %(nozzle_power)s, %(bed_power)s
        )
        RETURNING id
        """,
        {
            "run_id": run_id,
            "sampled_at": sample.get("sampled_at"),
            "nozzle": sample.get("nozzle"),
            "nozzle_target": sample.get("nozzle_target"),
            "bed": sample.get("bed"),
            "bed_target": sample.get("bed_target"),
            "chamber": sample.get("chamber"),
            "fan": sample.get("fan"),
            "speed_factor": sample.get("speed_factor"),
            "extrude_factor": sample.get("extrude_factor"),
            "layer": sample.get("layer"),
            "progress": sample.get("progress"),
            "filament_used_mm": sample.get("filament_used_mm"),
            "mmu_tool": sample.get("mmu_tool"),
            "nozzle_power": sample.get("nozzle_power"),
            "bed_power": sample.get("bed_power"),
        },
    ).fetchone()


def list_run_samples(conn, run_id, limit=2000):
    rows = conn.execute(
        """
        SELECT * FROM run_samples
        WHERE run_id = %s
        ORDER BY sampled_at ASC
        LIMIT %s
        """,
        (run_id, int(limit)),
    ).fetchall()
    out = []
    for row in rows:
        out.append({
            "sampled_at": iso(row["sampled_at"]),
            "nozzle": row["nozzle"],
            "nozzle_target": row["nozzle_target"],
            "bed": row["bed"],
            "bed_target": row["bed_target"],
            "chamber": row["chamber"],
            "fan": row["fan"],
            "speed_factor": row["speed_factor"],
            "extrude_factor": row["extrude_factor"],
            "layer": row["layer"],
            "progress": row["progress"],
            "filament_used_mm": row["filament_used_mm"],
            "mmu_tool": row["mmu_tool"],
            "nozzle_power": row.get("nozzle_power"),
            "bed_power": row.get("bed_power"),
        })
    return out


def open_segment(conn, run_id, spool_id, ace_slot, mm_start):
    return conn.execute(
        """
        INSERT INTO run_segments (run_id, spool_id, ace_slot, mm_start)
        VALUES (%s, %s, %s, %s)
        RETURNING *
        """,
        (run_id, spool_id, ace_slot, float(mm_start or 0)),
    ).fetchone()


def close_open_segments(conn, run_id, ace_slot, mm_end):
    """Offene Abschnitte am Platz (oder alle bei ace_slot=None) schließen."""
    if ace_slot is None:
        return conn.execute(
            """
            UPDATE run_segments
            SET mm_end = %s, closed_at = now()
            WHERE run_id = %s AND mm_end IS NULL
            RETURNING *
            """,
            (float(mm_end or 0), run_id),
        ).fetchall()
    return conn.execute(
        """
        UPDATE run_segments
        SET mm_end = %s, closed_at = now()
        WHERE run_id = %s AND mm_end IS NULL
          AND (ace_slot IS NOT DISTINCT FROM %s)
        RETURNING *
        """,
        (float(mm_end or 0), run_id, ace_slot),
    ).fetchall()


def list_run_segments(conn, run_id):
    rows = conn.execute(
        """
        SELECT seg.*, s.name AS spool_name, s.chargen_nr AS spool_chargen
        FROM run_segments seg
        JOIN spools s ON s.id = seg.spool_id
        WHERE seg.run_id = %s
        ORDER BY seg.id
        """,
        (run_id,),
    ).fetchall()
    return [
        {
            "id": row["id"],
            "spool_id": row["spool_id"],
            "ace_slot": row["ace_slot"],
            "mm_start": float(row["mm_start"] or 0),
            "mm_end": None if row["mm_end"] is None else float(row["mm_end"]),
            "spool_name": row["spool_name"],
            "spool_chargen": row["spool_chargen"] or "",
            "created_at": iso(row["created_at"]),
            "closed_at": iso(row["closed_at"]),
        }
        for row in rows
    ]


def _object_chargen(base, index):
    return "%s-%02d" % (base, index)


def _label_owner(conn, code):
    return conn.execute(
        """
        SELECT l.id, l.run_id, l.chargen_nr, COALESCE(r.chargen_nr, '') AS run_chargen
        FROM run_labels l
        JOIN runs r ON r.id = l.run_id
        WHERE l.chargen_nr = %s
        """,
        (code,),
    ).fetchone()


def _claim_object_chargen(conn, base, start, reserved, depth=0):
    """Freie Objekt-Charge ab start. Nummer einer anderen Lauf-Charge wird dorthin umgehängt."""
    if depth > 8:
        raise RuntimeError("Objekt-Charge lässt sich nicht vergeben")
    number = max(1, int(start or 1))
    while number < 1000:
        code = _object_chargen(base, number)
        if code in reserved:
            number += 1
            continue
        owner = _label_owner(conn, code)
        if not owner:
            return code
        owner_base = (owner["run_chargen"] or "").strip()
        if owner_base == base:
            number += 1
            continue
        if owner_base:
            moved = _claim_object_chargen(conn, owner_base, 1, reserved | {code}, depth + 1)
            conn.execute(
                "UPDATE run_labels SET chargen_nr = %s WHERE id = %s AND chargen_nr = %s",
                (moved, owner["id"], code),
            )
            if not _label_owner(conn, code):
                return code
        else:
            conn.execute("DELETE FROM run_labels WHERE id = %s AND chargen_nr = %s", (owner["id"], code))
            if not _label_owner(conn, code):
                return code
        number += 1
    raise RuntimeError("Keine freie Objekt-Charge")


def _insert_run_label(conn, run_id, name, code):
    from psycopg.errors import UniqueViolation

    try:
        with conn.transaction():
            conn.execute(
                """
                INSERT INTO run_labels (run_id, object_name, chargen_nr, excluded)
                VALUES (%s, %s, %s, false)
                """,
                (run_id, name, code),
            )
        return ""
    except UniqueViolation as exc:
        return (exc.diag.constraint_name if exc.diag else "") or "unique"


def ensure_run_labels(conn, run_id, objects, qty=1):
    """Objekt- oder Stückzahl-Etiketten anlegen. Charge vom Lauf vorausgesetzt."""
    run = get_run_row(conn, run_id)
    if not run or not run.get("chargen_nr"):
        return []
    base = run["chargen_nr"]
    existing = conn.execute(
        "SELECT * FROM run_labels WHERE run_id = %s ORDER BY id",
        (run_id,),
    ).fetchall()
    known_names = {
        (row["object_name"] or "").strip()
        for row in existing
        if (row["object_name"] or "").strip()
    }
    reserved = {(row["chargen_nr"] or "") for row in existing}
    names = []
    for item in objects or []:
        if isinstance(item, dict):
            name = (item.get("name") or "").strip()
            excluded = bool(item.get("excluded"))
        else:
            name = str(item).strip()
            excluded = False
        if name:
            names.append((name, excluded))
    if not existing and not names:
        count = max(1, int(qty or 1))
        for index in range(count):
            label_name = "" if count == 1 else "Teil %s" % (index + 1)
            placed = False
            for _try in range(5):
                if count == 1 and _try == 0 and base not in reserved and not _label_owner(conn, base):
                    code = base
                else:
                    code = _claim_object_chargen(conn, base, index + 1 + _try, reserved)
                clash = _insert_run_label(conn, run_id, label_name, code)
                if not clash:
                    reserved.add(code)
                    placed = True
                    break
                reserved.add(code)
                if "object_name" in clash:
                    break
            if not placed and count == 1:
                break
    elif names:
        index = len(known_names)
        for name, excluded in names:
            if excluded or name in known_names:
                continue
            index += 1
            placed = False
            for _try in range(5):
                code = _claim_object_chargen(conn, base, index + _try, reserved)
                clash = _insert_run_label(conn, run_id, name, code)
                if not clash:
                    reserved.add(code)
                    known_names.add(name)
                    placed = True
                    break
                reserved.add(code)
                if "object_name" in clash:
                    known_names.add(name)
                    break
            if not placed:
                continue
    rows = conn.execute(
        "SELECT * FROM run_labels WHERE run_id = %s ORDER BY id",
        (run_id,),
    ).fetchall()
    return [
        {
            "object_name": row["object_name"],
            "chargen_nr": row["chargen_nr"],
            "excluded": bool(row["excluded"]),
        }
        for row in rows
    ]


def list_run_labels(conn, run_id):
    rows = conn.execute(
        "SELECT * FROM run_labels WHERE run_id = %s ORDER BY id",
        (run_id,),
    ).fetchall()
    return [
        {
            "object_name": row["object_name"],
            "chargen_nr": row["chargen_nr"],
            "excluded": bool(row["excluded"]),
        }
        for row in rows
    ]


def list_run_events(conn, run_id, limit=50):
    rows = conn.execute(
        """
        SELECT * FROM events
        WHERE run_id = %s
        ORDER BY created_at ASC, id ASC
        LIMIT %s
        """,
        (run_id, int(limit)),
    ).fetchall()
    return [event_out(row) for row in rows]


def exclusive_until(until):
    """Reines Datum YYYY-MM-DD zählt inklusiv: Grenze = Folgetag."""
    if until is None or until == "":
        return None
    text = str(until).strip()
    if len(text) == 10 and text[4] == "-" and text[7] == "-":
        try:
            day = date.fromisoformat(text)
        except ValueError:
            return until
        return (day + timedelta(days=1)).isoformat()
    return until


def statistics(conn, since=None, until=None, host="", material="", brand="", color="", project_id=None):
    where = ["r.state IN ('completed', 'cancelled', 'error', 'interrupted')"]
    values = []
    until_bound = exclusive_until(until)
    if since:
        where.append("r.started_at >= %s")
        values.append(since)
    if until_bound:
        where.append("r.started_at < %s")
        values.append(until_bound)
    if host:
        where.append("r.printer_host = %s")
        values.append(host)
    if project_id:
        where.append("r.project_id = %s")
        values.append(project_id)

    spool_join = ""
    if material or brand or color:
        spool_join = (
            "JOIN run_spools rs0 ON rs0.run_id = r.id "
            "JOIN spools s0 ON s0.id = rs0.spool_id"
        )
        if material:
            where.append("s0.material_type ILIKE %s")
            values.append(material)
        if brand:
            where.append("s0.brand ILIKE %s")
            values.append(brand)
        if color:
            where.append("s0.color ILIKE %s")
            values.append(color)
    clause = " AND ".join(where)

    overview = conn.execute(
        """
        SELECT
            COUNT(DISTINCT r.id) AS jobs,
            COALESCE(SUM(DISTINCT r.duration_s), 0) AS duration_s,
            COALESCE(SUM(DISTINCT COALESCE(NULLIF(r.weight_used_g, 0), r.weight_g)), 0) AS weight_g,
            COALESCE(SUM(DISTINCT r.filament_cost + r.energy_cost), 0) AS cost,
            COUNT(DISTINCT r.id) FILTER (WHERE r.state = 'completed') AS completed,
            COUNT(DISTINCT r.id) FILTER (
                WHERE r.state IN ('cancelled', 'error', 'interrupted')
            ) AS failed,
            COUNT(DISTINCT r.id) FILTER (WHERE r.verdict = 'ausschuss') AS scrap,
            COUNT(DISTINCT r.id) FILTER (WHERE r.verdict = 'gut') AS good
        FROM runs r
        %s
        WHERE %s
        """ % (spool_join, clause),
        values,
    ).fetchone()

    # Ohne Spulen-Join erneut (sonst doppelte Zeilen bei Mehrfarb)
    run_where = ["r.state IN ('completed', 'cancelled', 'error', 'interrupted')"]
    run_values = []
    if since:
        run_where.append("r.started_at >= %s")
        run_values.append(since)
    if until_bound:
        run_where.append("r.started_at < %s")
        run_values.append(until_bound)
    if host:
        run_where.append("r.printer_host = %s")
        run_values.append(host)
    if project_id:
        run_where.append("r.project_id = %s")
        run_values.append(project_id)
    run_clause = " AND ".join(run_where)
    pieces_row = conn.execute(
        """
        SELECT
            COALESCE(SUM(r.pieces_ok), 0) AS pieces_ok,
            COALESCE(SUM(r.pieces_bad), 0) AS pieces_bad
        FROM runs r
        WHERE %s
        """ % run_clause,
        run_values,
    ).fetchone()

    by_printer = conn.execute(
        """
        SELECT r.printer_name, r.printer_host,
               COUNT(*) AS jobs,
               COALESCE(SUM(r.duration_s), 0) AS duration_s,
               COALESCE(SUM(COALESCE(NULLIF(r.weight_used_g, 0), r.weight_g)), 0) AS weight_g,
               COALESCE(SUM(r.filament_cost + r.energy_cost), 0) AS cost,
               COUNT(*) FILTER (WHERE r.state = 'completed') AS completed
        FROM runs r
        WHERE %s
        GROUP BY r.printer_name, r.printer_host
        ORDER BY duration_s DESC
        """ % run_clause,
        run_values,
    ).fetchall()

    mat_where = list(run_where)
    mat_values = list(run_values)
    if material:
        mat_where.append("s.material_type ILIKE %s")
        mat_values.append(material)
    if brand:
        mat_where.append("s.brand ILIKE %s")
        mat_values.append(brand)
    if color:
        mat_where.append("s.color ILIKE %s")
        mat_values.append(color)
    mat_clause = " AND ".join(mat_where)

    by_material = conn.execute(
        """
        SELECT s.material_type, s.brand, s.color,
               COALESCE(SUM(rs.weight_g), 0) AS weight_g,
               COALESCE(SUM(rs.cost), 0) AS cost,
               COUNT(DISTINCT rs.run_id) AS jobs
        FROM run_spools rs
        JOIN runs r ON r.id = rs.run_id
        JOIN spools s ON s.id = rs.spool_id
        WHERE %s
        GROUP BY s.material_type, s.brand, s.color
        ORDER BY weight_g DESC
        """ % mat_clause,
        mat_values,
    ).fetchall()

    estimate = conn.execute(
        """
        SELECT
            percentile_cont(0.5) WITHIN GROUP (
                ORDER BY r.duration_s / NULLIF(r.estimated_s, 0)
            ) AS median_ratio,
            COUNT(*) FILTER (WHERE r.estimated_s > 0 AND r.duration_s > 0) AS counted
        FROM runs r
        WHERE %s AND r.state = 'completed' AND r.estimated_s > 0
        """ % run_clause,
        run_values,
    ).fetchone()

    scrap_reasons = conn.execute(
        """
        SELECT COALESCE(NULLIF(r.verdict_reason, ''), 'sonstiges') AS reason,
               COUNT(*) AS n
        FROM runs r
        WHERE %s AND r.verdict = 'ausschuss'
        GROUP BY 1
        ORDER BY n DESC, reason
        """ % run_clause,
        run_values,
    ).fetchall()

    ov = overview or {}
    # Bei Spulen-Join können SUM(DISTINCT …) ungenau sein — Überblick ohne Join neu
    if spool_join:
        spool_filters = []
        spool_vals = list(run_values)
        if material:
            spool_filters.append("AND s.material_type ILIKE %s")
            spool_vals.append(material)
        if brand:
            spool_filters.append("AND s.brand ILIKE %s")
            spool_vals.append(brand)
        if color:
            spool_filters.append("AND s.color ILIKE %s")
            spool_vals.append(color)
        overview2 = conn.execute(
            """
            SELECT
                COUNT(*) AS jobs,
                COALESCE(SUM(r.duration_s), 0) AS duration_s,
                COALESCE(SUM(COALESCE(NULLIF(r.weight_used_g, 0), r.weight_g)), 0) AS weight_g,
                COALESCE(SUM(r.filament_cost + r.energy_cost), 0) AS cost,
                COUNT(*) FILTER (WHERE r.state = 'completed') AS completed,
                COUNT(*) FILTER (WHERE r.state IN ('cancelled', 'error', 'interrupted')) AS failed,
                COUNT(*) FILTER (WHERE r.verdict = 'ausschuss') AS scrap,
                COUNT(*) FILTER (WHERE r.verdict = 'gut') AS good
            FROM runs r
            WHERE %s
              AND (
                NOT EXISTS (SELECT 1 FROM run_spools rs WHERE rs.run_id = r.id)
                OR EXISTS (
                    SELECT 1 FROM run_spools rs
                    JOIN spools s ON s.id = rs.spool_id
                    WHERE rs.run_id = r.id
                    %s
                )
              )
            """ % (run_clause, " ".join(spool_filters)),
            spool_vals,
        ).fetchone()
        ov = overview2 or ov

    pieces_ok = int((pieces_row or {}).get("pieces_ok") or 0)
    pieces_bad = int((pieces_row or {}).get("pieces_bad") or 0)
    pieces_total = pieces_ok + pieces_bad
    return {
        "overview": {
            "jobs": int(ov.get("jobs") or 0),
            "duration_s": float(ov.get("duration_s") or 0),
            "weight_g": float(ov.get("weight_g") or 0),
            "cost": float(ov.get("cost") or 0),
            "completed": int(ov.get("completed") or 0),
            "failed": int(ov.get("failed") or 0),
            "scrap": int(ov.get("scrap") or 0),
            "good": int(ov.get("good") or 0),
            "pieces_ok": pieces_ok,
            "pieces_bad": pieces_bad,
            "yield_percent": (100.0 * pieces_ok / pieces_total) if pieces_total else None,
            "cost_per_part": (float(ov.get("cost") or 0) / pieces_ok) if pieces_ok else None,
        },
        "by_printer": [
            {
                "printer_name": row["printer_name"],
                "printer_host": row["printer_host"],
                "jobs": int(row["jobs"] or 0),
                "duration_s": float(row["duration_s"] or 0),
                "weight_g": float(row["weight_g"] or 0),
                "cost": float(row["cost"] or 0),
                "completed": int(row["completed"] or 0),
            }
            for row in by_printer
        ],
        "by_material": [
            {
                "material_type": row["material_type"] or "",
                "brand": row["brand"] or "",
                "color": row.get("color") or "",
                "weight_g": float(row["weight_g"] or 0),
                "cost": float(row["cost"] or 0),
                "jobs": int(row["jobs"] or 0),
            }
            for row in by_material
        ],
        "estimate_ratio": float((estimate or {}).get("median_ratio") or 0) or None,
        "estimate_count": int((estimate or {}).get("counted") or 0),
        "scrap_reasons": [
            {"reason": row["reason"] or "sonstiges", "count": int(row["n"] or 0)}
            for row in scrap_reasons
        ],
    }


def idea_out(row):
    return {
        "id": row["id"],
        "title": row["title"],
        "body": row["body"],
        "status": row["status"],
        "created_at": iso(row["created_at"]),
    }


def list_ideas(conn):
    rows = conn.execute(
        """
        SELECT * FROM ideas
        ORDER BY CASE status WHEN 'offen' THEN 0 WHEN 'spaeter' THEN 1 ELSE 2 END, created_at DESC
        """
    ).fetchall()
    return [idea_out(row) for row in rows]


def create_idea(conn, title, body, status):
    row = conn.execute(
        """
        INSERT INTO ideas (title, body, status)
        VALUES (%s, %s, %s)
        RETURNING *
        """,
        (title, body, status),
    ).fetchone()
    return idea_out(row)


def update_idea(conn, idea_id, fields):
    sets = []
    values = []
    for key in ("title", "body", "status"):
        if key in fields:
            sets.append("%s = %%s" % key)
            values.append(fields[key])
    if not sets:
        row = conn.execute("SELECT * FROM ideas WHERE id = %s", (idea_id,)).fetchone()
        return idea_out(row) if row else None
    values.append(idea_id)
    row = conn.execute(
        "UPDATE ideas SET %s WHERE id = %%s RETURNING *" % ", ".join(sets),
        values,
    ).fetchone()
    return idea_out(row) if row else None


def delete_idea(conn, idea_id):
    row = conn.execute("DELETE FROM ideas WHERE id = %s RETURNING id", (idea_id,)).fetchone()
    return bool(row)


def printer_out(row):
    family = row["family"] or ""
    sw = row.get("start_window") or {}
    if isinstance(sw, str):
        try:
            sw = json.loads(sw)
        except json.JSONDecodeError:
            sw = {}
    if not isinstance(sw, dict):
        sw = {}
    return {
        "id": str(row["id"]),
        "name": row["name"],
        "host": row["host"],
        "family": family,
        "family_label": FAMILY_LABELS.get(family, ""),
        "locked": bool(row.get("locked")),
        "start_window": sw,
    }


def list_printers(conn):
    rows = conn.execute("SELECT * FROM printers ORDER BY name").fetchall()
    return [printer_out(row) for row in rows]


def printer_by_host(conn, host):
    row = conn.execute("SELECT * FROM printers WHERE host = %s", (host,)).fetchone()
    return printer_out(row) if row else None


def printer_by_id(conn, printer_id):
    row = conn.execute("SELECT * FROM printers WHERE id = %s", (printer_id,)).fetchone()
    return row


def create_printer(conn, printer_id, name, host, family):
    row = conn.execute(
        """
        INSERT INTO printers (id, name, host, family)
        VALUES (%s, %s, %s, %s)
        RETURNING *
        """,
        (printer_id, name, host, family),
    ).fetchone()
    return printer_out(row)


def update_printer(conn, printer_id, name, host, family, locked=None, start_window=None):
    current = printer_by_id(conn, printer_id)
    if not current:
        return None
    if locked is None:
        locked = bool(current.get("locked"))
    if start_window is None:
        sw = current.get("start_window") or {}
    elif isinstance(start_window, dict):
        sw = start_window
    else:
        sw = {}
    if not isinstance(sw, dict):
        sw = {}
    row = conn.execute(
        """
        UPDATE printers
        SET name = %s, host = %s, family = %s, locked = %s, start_window = %s
        WHERE id = %s
        RETURNING *
        """,
        (name, host, family, bool(locked), Json(sw), printer_id),
    ).fetchone()
    if current["host"] != host:
        old = current["host"]
        conn.execute("UPDATE spools SET printer_host = %s WHERE printer_host = %s", (host, old))
        conn.execute("UPDATE sends SET printer_host = %s WHERE printer_host = %s", (host, old))
        conn.execute(
            "UPDATE runs SET printer_host = %s, printer_name = %s WHERE printer_host = %s",
            (host, name, old),
        )
        conn.execute("UPDATE print_queue SET host = %s WHERE host = %s", (host, old))
        conn.execute("UPDATE events SET host = %s, printer = %s WHERE host = %s", (host, name, old))
        conn.execute(
            """
            INSERT INTO bed_status (host, needs_clear, cleared_at, message)
            SELECT %s, needs_clear, cleared_at, message FROM bed_status WHERE host = %s
            ON CONFLICT (host) DO UPDATE SET
                needs_clear = EXCLUDED.needs_clear,
                cleared_at = EXCLUDED.cleared_at,
                message = EXCLUDED.message
            """,
            (host, old),
        )
        conn.execute("DELETE FROM bed_status WHERE host = %s", (old,))
    elif current["name"] != name:
        conn.execute("UPDATE runs SET printer_name = %s WHERE printer_host = %s", (name, host))
        conn.execute("UPDATE events SET printer = %s WHERE host = %s", (name, host))
    return printer_out(row)


def delete_printer(conn, printer_id):
    current = printer_by_id(conn, printer_id)
    if not current:
        return False
    host = current["host"]
    # Wartende Jobs dieses Hosts streichen; laufende bleiben nicht hängen.
    conn.execute(
        "UPDATE print_queue SET status = 'cancelled', message = 'Drucker entfernt.' WHERE host = %s AND status IN ('waiting', 'error')",
        (host,),
    )
    conn.execute(
        "UPDATE print_queue SET status = 'cancelled', message = 'Drucker entfernt.', phase = '' WHERE host = %s AND status = 'running'",
        (host,),
    )
    conn.execute("UPDATE spools SET printer_host = '', ace_slot = NULL WHERE printer_host = %s", (host,))
    conn.execute("DELETE FROM bed_status WHERE host = %s", (host,))
    row = conn.execute("DELETE FROM printers WHERE id = %s RETURNING id", (printer_id,)).fetchone()
    return bool(row)


def seed_printers(conn, path):
    count = conn.execute("SELECT COUNT(*) AS n FROM printers").fetchone()["n"]
    if count:
        return 0
    if not path.exists():
        return 0
    inserted = 0
    for row in json.loads(path.read_text(encoding="utf-8")):
        name = (row.get("name") or "").strip()
        host = (row.get("host") or "").strip()
        if not name or not host:
            continue
        family = row.get("family") or family_from_name(name)
        create_printer(conn, uuid.uuid4(), name, host, family)
        inserted += 1
    return inserted


def event_out(row):
    return {
        "id": row["id"],
        "host": row["host"] or "",
        "printer": row["printer"] or "",
        "kind": row["kind"],
        "message": row["message"] or "",
        "filename": row["filename"] or "",
        "run_id": row.get("run_id"),
        "created_at": iso(row["created_at"]),
        "read_at": iso(row["read_at"]),
        "unread": row["read_at"] is None,
    }


# Protokoll-Ereignisse, die kein Alert sind: gleich als gelesen speichern.
QUIET_EVENT_KINDS = {"ablauf", "start", "fertig", "spule", "auth"}
# Was in der Sidebar als „Meldungen“ zählt.
ALERT_EVENT_KINDS = frozenset({
    "fehler", "abbruch", "offline", "queue", "spule", "bedarf", "wartung", "auftrag",
})


def add_event(conn, host, printer, kind, message, filename="", run_id=None):
    row = conn.execute(
        """
        INSERT INTO events (host, printer, kind, message, filename, run_id, read_at)
        VALUES (%s, %s, %s, %s, %s, %s, CASE WHEN %s THEN now() ELSE NULL END)
        RETURNING *
        """,
        (
            host or "",
            printer or "",
            kind,
            message or "",
            filename or "",
            run_id,
            kind in QUIET_EVENT_KINDS,
        ),
    ).fetchone()
    return event_out(row)


def list_events(conn, limit=40, unread_only=False, host=None, kinds=None, since=None):
    where = []
    values = []
    if unread_only:
        where.append("read_at IS NULL")
    if host:
        where.append("host = %s")
        values.append(host)
    if kinds:
        where.append("kind = ANY(%s)")
        values.append(list(kinds))
    if since:
        where.append("created_at >= %s")
        values.append(since)
    values.append(int(limit))
    rows = conn.execute(
        "SELECT * FROM events %s ORDER BY created_at DESC, id DESC LIMIT %%s"
        % ("WHERE " + " AND ".join(where) if where else ""),
        values,
    ).fetchall()
    return [event_out(row) for row in rows]


def unread_event_count(conn, kinds=None):
    if kinds:
        return conn.execute(
            "SELECT COUNT(*) AS n FROM events WHERE read_at IS NULL AND kind = ANY(%s)",
            (list(kinds),),
        ).fetchone()["n"]
    return conn.execute("SELECT COUNT(*) AS n FROM events WHERE read_at IS NULL").fetchone()["n"]


def mark_events_read(conn, event_ids=None, kinds=None):
    if event_ids:
        conn.execute(
            "UPDATE events SET read_at = now() WHERE read_at IS NULL AND id = ANY(%s)",
            (list(event_ids),),
        )
    elif kinds:
        conn.execute(
            "UPDATE events SET read_at = now() WHERE read_at IS NULL AND kind = ANY(%s)",
            (list(kinds),),
        )
    else:
        conn.execute("UPDATE events SET read_at = now() WHERE read_at IS NULL")
    return unread_event_count(conn, kinds=kinds or list(ALERT_EVENT_KINDS))


def recent_event_kind(conn, host, kind, within_s=60):
    row = conn.execute(
        """
        SELECT id FROM events
        WHERE host = %s AND kind = %s AND created_at > now() - (%s || ' seconds')::interval
        LIMIT 1
        """,
        (host or "", kind, str(int(within_s))),
    ).fetchone()
    return bool(row)


def _json_list(value):
    if isinstance(value, list):
        return value
    if isinstance(value, str):
        try:
            parsed = json.loads(value)
        except json.JSONDecodeError:
            return []
        return parsed if isinstance(parsed, list) else []
    return []


STEP_LABELS = {"print": "Drucken", "cool": "Abkühlen", "sweep": "Schieben", "check": "Tasten"}


def _step_label(step):
    kind = (step or {}).get("type") if isinstance(step, dict) else ""
    return STEP_LABELS.get(kind, kind or "")


def _is_legacy_mesh_check(step):
    """Altes 3×3-Prüfen (probe_count / mesh_*) — aus UI und Ablauf raus."""
    if not isinstance(step, dict) or (step.get("type") or "").lower() != "check":
        return False
    return "probe_count" in step or "mesh_min_x" in step


def print_queue_out(row, file_name=None):
    ace = _json_list(row.get("ace_map"))
    steps = _json_list(row.get("steps"))
    visible_idx = [i for i, step in enumerate(steps) if not _is_legacy_mesh_check(step)]
    labels = [_step_label(steps[i]) for i in visible_idx]
    raw_index = int(row.get("step_index") or 0)
    display_index = 0
    for vis_i, orig_i in enumerate(visible_idx):
        if orig_i <= raw_index:
            display_index = vis_i
    family = row["family"] or ""
    return {
        "id": row["id"],
        "file_id": row["file_id"],
        "filename": file_name or row.get("filename") or "",
        "host": row["host"] or "",
        "printer_name": row.get("printer_name") or "",
        "family": family,
        "family_label": FAMILY_LABELS.get(family, ""),
        "step_labels": labels,
        "step_label_current": labels[display_index] if 0 <= display_index < len(labels) else "",
        "priority": int(row["priority"] or 100),
        "status": row["status"],
        "spool_id": row["spool_id"],
        "ace_map": ace,
        "bed_leveling": bool(row["bed_leveling"]),
        "override_machine": bool(row["override_machine"]),
        "message": row["message"] or "",
        "steps": steps,
        "step_index": int(row.get("step_index") or 0),
        "display_step_index": display_index,
        "repeat_total": int(row.get("repeat_total") or 1),
        "repeat_done": int(row.get("repeat_done") or 0),
        "phase": row.get("phase") or "",
        "phase_started_at": iso(row.get("phase_started_at")),
        "is_ablauf": bool(steps),
        "order_item_id": row.get("order_item_id"),
        "retry_count": int(row.get("retry_count") or 0),
        "wait_reason": row.get("wait_reason") or "",
        "created_at": iso(row["created_at"]),
        "updated_at": iso(row["updated_at"]),
    }


def enqueue_print(
    conn,
    file_id,
    host,
    family,
    priority,
    spool_id,
    ace_map,
    bed_leveling,
    override,
    steps=None,
    repeat_total=1,
    order_item_id=None,
):
    steps = steps if isinstance(steps, list) else []
    try:
        repeats = max(1, int(repeat_total if repeat_total is not None else 1))
    except (TypeError, ValueError):
        repeats = 1
    row = conn.execute(
        """
        INSERT INTO print_queue (
            file_id, host, family, priority, spool_id, ace_map,
            bed_leveling, override_machine, status, message,
            steps, step_index, repeat_total, repeat_done, phase, order_item_id
        )
        VALUES (
            %s, %s, %s, %s, %s, %s, %s, %s, 'waiting', %s,
            %s, 0, %s, 0, '', %s
        )
        RETURNING *
        """,
        (
            file_id,
            host or "",
            family or "",
            int(priority if priority is not None else 100),
            spool_id,
            Json(ace_map if isinstance(ace_map, list) else []),
            bool(bed_leveling),
            bool(override),
            "Ablauf wartet…" if steps else "Wartet auf freien Drucker…",
            Json(steps),
            repeats,
            order_item_id,
        ),
    ).fetchone()
    file_row = get_file_row(conn, file_id)
    file_name = (file_row or {}).get("name") or ""
    printer = printer_by_host(conn, host) if host else None
    target = (printer or {}).get("name") or (
        "nächster freier " + FAMILY_LABELS[family] if family in FAMILY_LABELS else "nächster freier Drucker"
    )
    what = "Ablauf (%d×)" % repeats if steps else "Druck"
    add_event(
        conn,
        host or "",
        (printer or {}).get("name") or "",
        "ablauf",
        "%s eingereiht: %s → %s" % (what, file_name or "Datei", target),
        file_name,
    )
    return print_queue_out(row, file_name)


def list_print_queue(conn, limit=40):
    rows = conn.execute(
        """
        SELECT q.*, f.name AS filename, p.name AS printer_name
        FROM print_queue q
        LEFT JOIN files f ON f.id = q.file_id
        LEFT JOIN printers p ON p.host = q.host AND q.host <> ''
        ORDER BY
            CASE q.status
                WHEN 'running' THEN 0
                WHEN 'waiting' THEN 1
                WHEN 'error' THEN 2
                ELSE 3
            END,
            CASE WHEN q.status IN ('waiting', 'running', 'error') THEN q.priority ELSE 0 END ASC,
            CASE WHEN q.status IN ('waiting', 'running', 'error') THEN q.id ELSE 0 END ASC,
            q.updated_at DESC NULLS LAST,
            q.id DESC
        LIMIT %s
        """,
        (int(limit),),
    ).fetchall()
    return [print_queue_out(row, row.get("filename")) for row in rows]


def queue_by_host(conn, printers):
    """Laufender Queue-Eintrag und Zahl wartender Jobs je Drucker (für die Wand)."""
    running = {}
    for row in conn.execute(
        """
        SELECT q.*, f.name AS filename, p.name AS printer_name
        FROM print_queue q
        LEFT JOIN files f ON f.id = q.file_id
        LEFT JOIN printers p ON p.host = q.host
        WHERE q.status = 'running' AND q.host <> ''
        ORDER BY q.updated_at DESC
        """
    ).fetchall():
        running.setdefault(row["host"], print_queue_out(row, row.get("filename")))
    failed = {}
    for row in conn.execute(
        """
        SELECT q.*, f.name AS filename, p.name AS printer_name
        FROM print_queue q
        LEFT JOIN files f ON f.id = q.file_id
        LEFT JOIN printers p ON p.host = q.host
        WHERE q.status = 'error' AND q.host <> '' AND jsonb_array_length(q.steps) > 0
          AND q.updated_at > now() - interval '24 hours'
        ORDER BY q.updated_at DESC
        """
    ).fetchall():
        failed.setdefault(row["host"], print_queue_out(row, row.get("filename")))
    waiting_rows = conn.execute(
        """
        SELECT q.host, q.family, f.name AS filename
        FROM print_queue q
        LEFT JOIN files f ON f.id = q.file_id
        WHERE q.status = 'waiting'
        ORDER BY q.priority ASC, q.id ASC
        """
    ).fetchall()
    out = {}
    for printer in printers:
        host = printer.get("host") or ""
        family = printer.get("family") or ""
        mine = [
            row for row in waiting_rows
            if row["host"] == host or (not row["host"] and (not row["family"] or row["family"] == family))
        ]
        out[host] = {
            "running": running.get(host),
            "failed": None if running.get(host) else failed.get(host),
            "waiting": len(mine),
            "waiting_next": (mine[0]["filename"] or "") if mine else "",
        }
    return out


def get_print_queue_row(conn, item_id):
    row = conn.execute(
        """
        SELECT q.*, f.name AS filename, p.name AS printer_name
        FROM print_queue q
        LEFT JOIN files f ON f.id = q.file_id
        LEFT JOIN printers p ON p.host = q.host AND q.host <> ''
        WHERE q.id = %s
        """,
        (item_id,),
    ).fetchone()
    return row


def update_print_queue(conn, item_id, fields):
    allowed = {
        "priority": "priority",
        "status": "status",
        "message": "message",
        "host": "host",
        "step_index": "step_index",
        "repeat_done": "repeat_done",
        "phase": "phase",
        "file_id": "file_id",
        "spool_id": "spool_id",
        "bed_leveling": "bed_leveling",
        "override_machine": "override_machine",
        "retry_count": "retry_count",
        "wait_reason": "wait_reason",
    }
    sets = []
    values = []
    for key, column in allowed.items():
        if key in fields:
            sets.append("%s = %%s" % column)
            values.append(fields[key])
    if "ace_map" in fields:
        sets.append("ace_map = %s")
        values.append(Json(fields["ace_map"] if isinstance(fields["ace_map"], list) else []))
    if "steps" in fields:
        sets.append("steps = %s")
        raw_steps = fields["steps"]
        if isinstance(raw_steps, Json):
            values.append(raw_steps)
        else:
            values.append(Json(raw_steps if isinstance(raw_steps, list) else []))
    if "phase_started_at" in fields:
        if fields["phase_started_at"] is None:
            sets.append("phase_started_at = NULL")
        elif fields["phase_started_at"] == "now":
            sets.append("phase_started_at = now()")
        else:
            sets.append("phase_started_at = %s")
            values.append(fields["phase_started_at"])
    if not sets:
        row = get_print_queue_row(conn, item_id)
        return print_queue_out(row, (row or {}).get("filename")) if row else None
    sets.append("updated_at = now()")
    values.append(item_id)
    row = conn.execute(
        "UPDATE print_queue SET %s WHERE id = %%s RETURNING id" % ", ".join(sets),
        values,
    ).fetchone()
    if not row:
        return None
    fresh = get_print_queue_row(conn, row["id"])
    return print_queue_out(fresh, fresh.get("filename")) if fresh else None


_PROFILE_VARIANTS = frozenset({"meta", "matte", "matt", "silk", "plus", "+", "hf", "cf"})
_PROFILE_TYPES = frozenset({"pla", "petg", "abs", "asa", "tpu", "pa", "pc", "hips", "pva"})


def _profile_tokens(profile):
    return (profile or "").strip().lower().replace("_", " ").replace("-", " ").split()


def _profile_has_variant(profile):
    return bool(_PROFILE_VARIANTS.intersection(_profile_tokens(profile)))


def _profile_base_type(profile):
    """Letzter Material-Typ im Profilnamen (Sunlu PLA → pla)."""
    for token in reversed(_profile_tokens(profile)):
        if token in _PROFILE_TYPES:
            return token
    return ""


def _profile_matches(need_profile, spool_name):
    """Sorte: exakter Katalogname (Meta ≠ Matte ≠ PLA). Zusatz nur hinter dem Namen."""
    a = (need_profile or "").strip().lower()
    b = (spool_name or "").strip().lower()
    if not a:
        return True
    if not b:
        return False
    if a == b:
        return True
    # Spulen-Anzeigename: "Sunlu PLA Meta schwarz" — nicht "Sunlu PLA" ⊂ "Sunlu PLA Matte"
    if b.startswith(a) and len(b) > len(a) and not b[len(a)].isalnum():
        rest = b[len(a):].lstrip(" ·-–_|/")
        # Nächstes Wort darf keine bekannte PLA-/PETG-Variante sein
        first = (rest.split() or [""])[0]
        if first in _PROFILE_VARIANTS:
            return False
        return True
    return False


def _need_label(need):
    profile = (need.get("profile") or "").strip()
    if profile:
        return profile
    kind = (need.get("material_type") or "").strip()
    color = (need.get("color") or "").strip()
    return (kind + ((" " + color) if color else "")).strip() or "Material"


def _spool_material_set(conn, host):
    rows = conn.execute(
        """
        SELECT lower(coalesce(s.material_type, '')) AS mt,
               lower(coalesce(s.color, '')) AS color,
               coalesce(s.remaining_g, 0) AS remaining_g,
               lower(coalesce(m.name, s.name, '')) AS material_name,
               s.ace_slot AS ace_slot,
               s.id AS spool_id
        FROM spools s
        LEFT JOIN materials m ON m.id = s.material_id
        WHERE s.printer_host = %s
          AND s.archived_at IS NULL
          AND coalesce(s.remaining_g, 0) > 0.5
        """,
        (host or "",),
    ).fetchall()
    return [
        {
            "material_type": (r["mt"] or "").strip(),
            "color": (r["color"] or "").strip(),
            "remaining_g": float(r["remaining_g"] or 0),
            "material_name": (r["material_name"] or "").strip(),
            "ace_slot": r["ace_slot"],
            "spool_id": r["spool_id"],
        }
        for r in rows
    ]


def _file_needs(conn, file_id):
    if not file_id:
        return [], 0.0
    row = conn.execute(
        "SELECT filaments, weight_g FROM files WHERE id = %s",
        (file_id,),
    ).fetchone()
    if not row:
        return [], 0.0
    filaments = row.get("filaments") or []
    if isinstance(filaments, str):
        try:
            import json
            filaments = json.loads(filaments)
        except Exception:
            filaments = []
    types = []
    for index, item in enumerate(filaments if isinstance(filaments, list) else []):
        if isinstance(item, dict):
            t = (item.get("type") or item.get("material_type") or "").strip().lower()
            c = (item.get("color") or "").strip().lower()
            profile = (item.get("profile") or "").strip()
            try:
                paint = int(item.get("paint_index"))
            except (TypeError, ValueError):
                paint = index
            if t or profile:
                types.append({
                    "material_type": t,
                    "color": c,
                    "profile": profile,
                    "paint_index": paint,
                })
        elif isinstance(item, str) and item.strip():
            types.append({
                "material_type": item.strip().lower(),
                "color": "",
                "profile": "",
                "paint_index": index,
            })
    return types, float(row.get("weight_g") or 0)


def _prefer_color(spools, need_color):
    """Farbe soft: Slice-Hex ignorieren; benannte Farbe bevorzugen, nie blockieren."""
    from grotti.bedarf import _is_slicer_hex

    need_c = (need_color or "").strip().lower()
    if not need_c or _is_slicer_hex(need_c) or not spools:
        return list(spools)
    preferred = [
        h for h in spools
        if not (h.get("color") or "").strip() or (h.get("color") or "").strip().lower() == need_c
    ]
    return preferred or list(spools)


def _match_spools_for_need(need, have, used_ids=None):
    used = used_ids or set()
    candidates = [h for h in have if h.get("spool_id") not in used]
    profile = (need.get("profile") or "").strip()

    def _by_type(kind):
        kind = (kind or "").strip().lower()
        return [
            h for h in candidates
            if not kind or h["material_type"] == kind
        ]

    if profile:
        by_profile = [h for h in candidates if _profile_matches(profile, h.get("material_name"))]
        if by_profile:
            return _prefer_color(by_profile, need.get("color"))
        # Basis-Profil ohne Meta/Matte/Silk: Marke egal — Typ reicht
        # (Sunlu PLA ↔ Eryone PLA Lite; Slice-Hex #26A69A blockiert nicht „Grau“).
        if not _profile_has_variant(profile):
            kind = _profile_base_type(profile) or (need.get("material_type") or "")
            if kind:
                typed = [
                    h for h in _by_type(kind)
                    if not _profile_has_variant(h.get("material_name") or "")
                ]
                return _prefer_color(typed, need.get("color"))
        return []
    return _prefer_color(_by_type(need.get("material_type") or ""), need.get("color"))


def material_fit_score(conn, host, file_id):
    """Höher = besser. -1 = warten (falsche Sorte mit Profil, oder zu wenig Rest)."""
    need, weight = _file_needs(conn, file_id)
    have = _spool_material_set(conn, host)
    if not need:
        return 0, ""
    if not have:
        # Mit Profil: ohne Spule warten. Ohne Profil: soft (wie bisher).
        if any((n.get("profile") or "").strip() for n in need):
            return -1, "braucht %s" % _need_label(need[0])
        return 0, ""

    has_profiles = any((n.get("profile") or "").strip() for n in need)
    if has_profiles:
        score = 20
        used = set()
        for n in need:
            matched = _match_spools_for_need(n, have, used)
            if not matched:
                return -1, "braucht %s" % _need_label(n)
            best = max(matched, key=lambda h: h["remaining_g"])
            used.add(best.get("spool_id"))
            if n.get("color") and any(h["color"] == n["color"] for h in matched):
                score += 5
            if weight and best["remaining_g"] + 1e-6 >= weight:
                score += 5
            elif weight and all(h["remaining_g"] + 1e-6 < weight for h in matched):
                return -1, "braucht %s (%s g fehlen)" % (
                    _need_label(n),
                    int(max(0, weight - max(h["remaining_g"] for h in matched))),
                )
        return score, ""

    have_types = {h["material_type"] for h in have if h["material_type"]}
    need_types = {n["material_type"] for n in need if n["material_type"]}
    # Ohne Profil: falscher Typ hält den Job nicht fest (Sofort-Start warnt in der UI).
    score = 0 if (need_types and have_types and need_types.isdisjoint(have_types)) else 10
    for n in need:
        matched = _match_spools_for_need(n, have)
        if matched:
            score += 5
            if weight and any(h["remaining_g"] + 1e-6 >= weight for h in matched):
                score += 5
            elif weight and all(h["remaining_g"] + 1e-6 < weight for h in matched):
                return -1, "braucht %s (%s g fehlen)" % (
                    _need_label(n),
                    int(max(0, weight - max(h["remaining_g"] for h in matched))),
                )
    return score, ""


def auto_ace_map(conn, host, file_id):
    """ACE-Map aus eingelegten Spulen: Sorte (Profil) zuerst, Farbe soft.

    Rückgabe: (mapping, reason). reason gesetzt → nicht starten, warten.
    Ohne ACE-Spulen am Host: leere Map und kein Grund (externe Rolle).
    """
    need, _weight = _file_needs(conn, file_id)
    have = [h for h in _spool_material_set(conn, host) if h.get("ace_slot")]
    if not need:
        return [], ""
    if not have:
        return [], ""
    mapping = []
    used = set()
    for n in need:
        matched = _match_spools_for_need(n, have, used)
        # Nur ACE-Plätze
        matched = [h for h in matched if h.get("ace_slot")]
        if not matched:
            return [], "braucht %s auf ACE" % _need_label(n)
        # Farbe bevorzugen, dann Rest
        matched.sort(
            key=lambda h: (
                0 if (n.get("color") and h["color"] == n["color"]) else 1,
                -h["remaining_g"],
            )
        )
        chosen = matched[0]
        used.add(chosen.get("spool_id"))
        try:
            paint = int(n.get("paint_index"))
        except (TypeError, ValueError):
            paint = len(mapping)
        ams = int(chosen["ace_slot"]) - 1
        if ams < 0 or ams > 7:
            return [], "ACE-Platz von %s ungültig" % _need_label(n)
        mapping.append({"paint_index": paint, "ams_index": ams})
    return mapping, ""


def next_waiting_print(conn, host=None, family=None):
    """Nächster wartender Druck: Termin (due_at) vor Priorität, Material-Match bevorzugt."""
    if host:
        rows = conn.execute(
            """
            SELECT q.*, f.name AS filename, o.due_at AS order_due_at
            FROM print_queue q
            LEFT JOIN files f ON f.id = q.file_id
            LEFT JOIN order_items oi ON oi.id = q.order_item_id
            LEFT JOIN orders o ON o.id = oi.order_id
            WHERE q.status = 'waiting'
              AND (q.host = %s OR (q.host = '' AND (q.family = '' OR q.family = %s)))
            ORDER BY
              CASE WHEN o.due_at IS NULL THEN 1 ELSE 0 END,
              o.due_at ASC NULLS LAST,
              q.priority ASC,
              q.id ASC
            LIMIT 20
            """,
            (host, family or ""),
        ).fetchall()
    else:
        rows = conn.execute(
            """
            SELECT q.*, f.name AS filename, o.due_at AS order_due_at
            FROM print_queue q
            LEFT JOIN files f ON f.id = q.file_id
            LEFT JOIN order_items oi ON oi.id = q.order_item_id
            LEFT JOIN orders o ON o.id = oi.order_id
            WHERE q.status = 'waiting'
            ORDER BY
              CASE WHEN o.due_at IS NULL THEN 1 ELSE 0 END,
              o.due_at ASC NULLS LAST,
              q.priority ASC,
              q.id ASC
            LIMIT 20
            """,
        ).fetchall()
    if not rows:
        return None
    if not host:
        return rows[0]

    best = None
    best_score = None
    for row in rows:
        score, reason = material_fit_score(conn, host, row.get("file_id"))
        if score < 0:
            # Wartegrund setzen, aber anderen Job bevorzugen
            if (row.get("wait_reason") or "") != reason or (row.get("message") or "") != reason:
                conn.execute(
                    "UPDATE print_queue SET wait_reason = %s, message = %s, updated_at = now() WHERE id = %s",
                    (reason, reason, row["id"]),
                )
            continue
        if reason == "" and row.get("wait_reason"):
            conn.execute(
                """
                UPDATE print_queue
                SET message = CASE WHEN message = wait_reason THEN '' ELSE message END,
                    wait_reason = '',
                    updated_at = now()
                WHERE id = %s
                """,
                (row["id"],),
            )
        if best is None or score > best_score:
            best = row
            best_score = score
    return best


def running_print_on_host(conn, host):
    if not host:
        return None
    return conn.execute(
        """
        SELECT q.*, f.name AS filename
        FROM print_queue q
        LEFT JOIN files f ON f.id = q.file_id
        WHERE q.status = 'running' AND q.host = %s
        ORDER BY q.updated_at DESC, q.id DESC
        LIMIT 1
        """,
        (host,),
    ).fetchone()


def claim_print_queue(conn, item_id, host):
    row = conn.execute(
        """
        UPDATE print_queue
        SET status = 'running', host = %s, message = 'Wird gestartet…',
            step_index = 0, repeat_done = 0, phase = '', phase_started_at = NULL,
            updated_at = now()
        WHERE id = %s AND status = 'waiting'
        RETURNING *
        """,
        (host, item_id),
    ).fetchone()
    return row


def bed_needs_clear(conn, host):
    row = conn.execute(
        "SELECT needs_clear FROM bed_status WHERE host = %s",
        (host or "",),
    ).fetchone()
    return bool(row and row["needs_clear"])


def bed_clear_map(conn):
    rows = conn.execute("SELECT host, needs_clear FROM bed_status").fetchall()
    return {row["host"]: bool(row["needs_clear"]) for row in rows}


def set_bed_needs_clear(conn, host, needs, message=""):
    host = (host or "").strip()
    if not host:
        return
    needs = bool(needs)
    conn.execute(
        """
        INSERT INTO bed_status (host, needs_clear, cleared_at, message)
        VALUES (%s, %s, CASE WHEN %s THEN NULL ELSE now() END, %s)
        ON CONFLICT (host) DO UPDATE SET
            needs_clear = EXCLUDED.needs_clear,
            cleared_at = CASE WHEN EXCLUDED.needs_clear THEN NULL ELSE now() END,
            message = EXCLUDED.message
        """,
        (host, needs, needs, message or ""),
    )


def ensure_bed_after_print(conn, host, ended_at=None, message=""):
    """Platte belegen, wenn dieser Druck nach der letzten Quittierung endete.

    Deckt verpasste Watcher-Übergänge ab (Neustart, Poll-Lücke): History und
    erster Blick auf state=complete setzen needs_clear nur, wenn cleared_at
    vor dem Druckende liegt oder noch fehlt. Ohne Endzeit und mit bestehender
    Quittierung nicht neu belegen — sonst setzt ein Neustart bereits
    Quittiertes wieder (cleared_at geht verloren). Live-Übergänge setzen
    needs_clear direkt per set_bed_needs_clear.
    """
    host = (host or "").strip()
    if not host:
        return False
    row = conn.execute(
        "SELECT needs_clear, cleared_at FROM bed_status WHERE host = %s",
        (host,),
    ).fetchone()
    if row and row["needs_clear"]:
        return False
    cleared = row["cleared_at"] if row else None
    if ended_at is not None and cleared is not None and cleared >= ended_at:
        return False
    # Neustart / erster Blick ohne Endzeit: schon quittiert → stehen lassen
    if ended_at is None and cleared is not None:
        return False
    set_bed_needs_clear(conn, host, True, message or "Druckende")
    return True


def user_count(conn):
    from grotti.db import cached_user_count
    return cached_user_count(conn)


def list_users(conn):
    return conn.execute(
        "SELECT id, username, role, active, created_at, last_login FROM users ORDER BY username"
    ).fetchall()


def get_user(conn, user_id):
    return conn.execute(
        "SELECT * FROM users WHERE id = %s",
        (user_id,),
    ).fetchone()


def get_user_by_name(conn, username):
    return conn.execute(
        "SELECT * FROM users WHERE lower(username) = lower(%s)",
        (username,),
    ).fetchone()


def create_user(conn, username, password_hash, role="operator"):
    return conn.execute(
        """
        INSERT INTO users (username, password_hash, role)
        VALUES (%s, %s, %s)
        RETURNING id, username, role, active, created_at, last_login
        """,
        (username, password_hash, role),
    ).fetchone()


def update_user(conn, user_id, *, password_hash=None, role=None, active=None):
    row = get_user(conn, user_id)
    if not row:
        return None
    new_hash = password_hash if password_hash is not None else row["password_hash"]
    new_role = role if role is not None else row["role"]
    new_active = row["active"] if active is None else bool(active)
    return conn.execute(
        """
        UPDATE users
        SET password_hash = %s, role = %s, active = %s
        WHERE id = %s
        RETURNING id, username, role, active, created_at, last_login
        """,
        (new_hash, new_role, new_active, user_id),
    ).fetchone()


def delete_user(conn, user_id):
    conn.execute("DELETE FROM users WHERE id = %s", (user_id,))


def touch_user_login(conn, user_id):
    conn.execute("UPDATE users SET last_login = now() WHERE id = %s", (user_id,))


def create_api_token(conn, user_id, name, token_hash, token_prefix, role, expires_at=None):
    return conn.execute(
        """
        INSERT INTO api_tokens (user_id, name, token_hash, token_prefix, role, expires_at)
        VALUES (%s, %s, %s, %s, %s, %s)
        RETURNING id, user_id, name, token_prefix, role, created_at, last_used, expires_at, revoked_at
        """,
        (user_id, name or "", token_hash, token_prefix, role, expires_at),
    ).fetchone()


def list_api_tokens(conn, user_id=None, include_revoked=False):
    sql = """
        SELECT t.id, t.user_id, t.name, t.token_prefix, t.role,
               t.created_at, t.last_used, t.expires_at, t.revoked_at,
               u.username
        FROM api_tokens t
        JOIN users u ON u.id = t.user_id
        WHERE 1=1
    """
    args = []
    if user_id is not None:
        sql += " AND t.user_id = %s"
        args.append(user_id)
    if not include_revoked:
        sql += " AND t.revoked_at IS NULL"
    sql += " ORDER BY t.created_at DESC"
    return conn.execute(sql, args).fetchall()


def get_api_token_by_hash(conn, token_hash):
    return conn.execute(
        """
        SELECT t.*, u.username, u.active AS user_active, u.role AS user_role
        FROM api_tokens t
        JOIN users u ON u.id = t.user_id
        WHERE t.token_hash = %s
        """,
        (token_hash,),
    ).fetchone()


def touch_api_token(conn, token_id):
    conn.execute("UPDATE api_tokens SET last_used = now() WHERE id = %s", (token_id,))


def revoke_api_token(conn, token_id, user_id=None):
    if user_id is None:
        return conn.execute(
            """
            UPDATE api_tokens SET revoked_at = now()
            WHERE id = %s AND revoked_at IS NULL
            RETURNING id, user_id, name, token_prefix, role, created_at, last_used, expires_at, revoked_at
            """,
            (token_id,),
        ).fetchone()
    return conn.execute(
        """
        UPDATE api_tokens SET revoked_at = now()
        WHERE id = %s AND user_id = %s AND revoked_at IS NULL
        RETURNING id, user_id, name, token_prefix, role, created_at, last_used, expires_at, revoked_at
        """,
        (token_id, user_id),
    ).fetchone()


def global_search(conn, query, limit=20):
    """Drucker, Dateien, Spulen, Chargen, Aufträge für Ctrl+K."""
    q = (query or "").strip()
    if not q:
        return []
    like = "%" + q + "%"
    items = []
    for row in conn.execute(
        "SELECT host, name FROM printers WHERE name ILIKE %s OR host ILIKE %s ORDER BY name LIMIT 5",
        (like, like),
    ).fetchall():
        items.append({
            "kind": "drucker",
            "title": row["name"] or row["host"],
            "subtitle": row["host"],
            "href": "/#" + row["host"],
        })
    for row in conn.execute(
        "SELECT id, name FROM files WHERE name ILIKE %s ORDER BY mtime DESC NULLS LAST, id DESC LIMIT 6",
        (like,),
    ).fetchall():
        items.append({
            "kind": "datei",
            "title": row["name"],
            "subtitle": "Datei",
            "href": "/dateien?file=" + str(row["id"]),
        })
    for row in conn.execute(
        """
        SELECT id, name, chargen_nr, code, material_type, color
        FROM spools
        WHERE name ILIKE %s OR chargen_nr ILIKE %s OR code ILIKE %s
           OR material_type ILIKE %s OR color ILIKE %s
        ORDER BY id DESC LIMIT 6
        """,
        (like, like, like, like, like),
    ).fetchall():
        items.append({
            "kind": "spule",
            "title": row["name"],
            "subtitle": " · ".join(
                x for x in [row.get("chargen_nr") or row.get("code"), row.get("material_type"), row.get("color")] if x
            ),
            "href": "/material?spool=" + str(row["id"]),
        })
    if q.upper().startswith("D-") or q.upper().startswith("S-"):
        code = q.upper()
        run = conn.execute(
            "SELECT id, chargen_nr, filename FROM runs WHERE chargen_nr ILIKE %s ORDER BY id DESC LIMIT 3",
            (code + "%",),
        ).fetchall()
        for row in run:
            items.append({
                "kind": "charge",
                "title": row["chargen_nr"],
                "subtitle": row.get("filename") or "Druck",
                "href": "/protokoll/" + str(row["id"]),
            })
    for row in conn.execute(
        "SELECT id, title, status FROM orders WHERE title ILIKE %s ORDER BY updated_at DESC LIMIT 5",
        (like,),
    ).fetchall():
        items.append({
            "kind": "auftrag",
            "title": row["title"],
            "subtitle": row["status"],
            "href": "/auftraege",
        })
    return items[: int(limit)]