From 023627892914d41cf54c1cccdcb18df59c34e039 Mon Sep 17 00:00:00 2001 From: Dominic Gundel Date: Sun, 18 Oct 2026 00:47:23 +0200 Subject: [PATCH] Grotti 3.0 Phase 1: Migrationen, Pool, Queue- und Abzug-Fixes. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nummerierte SQL-Migrationen, psycopg-Pool, waitress, hängende Queue-Jobs, sichere History-Abzüge, Chargen-Locks, Upload-/Host-Validierung, ProxyFix. --- CHANGELOG.md | 20 ++ grotti/anycubic_lan.py | 2 +- grotti/config.py | 17 ++ grotti/db.py | 187 ++++++++++++++++-- grotti/jobs.py | 57 ++++-- grotti/library.py | 5 +- grotti/mcp_server.py | 8 +- grotti/migrations/001_base.sql | 305 ++++++++++++++++++++++++++++++ grotti/migrations/002_indexes.sql | 6 + grotti/moonraker.py | 22 ++- grotti/print_queue.py | 61 +++++- grotti/schema.sql | 3 + grotti/send_queue.py | 30 ++- grotti/store.py | 44 ++++- grotti/watcher.py | 23 ++- server.py | 79 +++++++- tests/test_jobs_spool.py | 65 +++++++ tests/test_library_safe.py | 21 ++ web/app.py | 112 +++++++++-- 19 files changed, 986 insertions(+), 81 deletions(-) create mode 100644 grotti/migrations/001_base.sql create mode 100644 grotti/migrations/002_indexes.sql create mode 100644 tests/test_jobs_spool.py create mode 100644 tests/test_library_safe.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1bc3cc2..92a3c94 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,26 @@ Alle Änderungen an Grotti. Format nach [Keep a Changelog](https://keepachangelo ## [Unreleased] +### Neu + +- Datenbank: nummerierte Migrationen unter `grotti/migrations/`, Connection-Pool, Indizes für Sends/Läufe/Dateien/Events, Retention für `run_samples` (90 Tage). +- Server: waitress statt Flask-Dev-Server; Hintergrund-Threads starten erst in `server.main()`. + +### Geändert + +- Spulenabzug aus Moonraker-History: keine Rate mehr ohne Send-Eintrag; Spulen, die nach Druckende angelegt wurden, werden ignoriert. +- Chargen-Vergabe (`S-…` / `D-…`) mit Advisory-Lock; Druck-Charge nach lokaler Tagesgrenze. +- MCP-Steuerung: Parameter `temp`/`percent` (Alias `value` bleibt in Moonraker). +- Upload nur `.gcode`/`.3mf`/`.gcode.3mf`; Drucker-Host validiert; `?token=` nur für Kamera/Vorschau; ProxyFix; Settings-Lock. + +### Behoben + +- Hängende Queue-Jobs (`phase=starting` ohne Worker) nach Neustart werden nach 45 s auf Fehler gesetzt. +- Queue-Worker rollen die Transaktion bei Fehlern zurück; Exceptions werden geloggt. +- `snapshot()` fängt alle Verbindungsfehler ab, damit ein kaputter Drucker den Watcher nicht stoppt. +- `prepare_send` erhält die ACE-Map schon beim Einreihen in die Send-Queue. +- Spulenwarnung im Watcher berücksichtigt ACE-Plätze. + ## [2.10.0] – 2026-10-18 ### Neu diff --git a/grotti/anycubic_lan.py b/grotti/anycubic_lan.py index aa07aa4..77d5a6c 100644 --- a/grotti/anycubic_lan.py +++ b/grotti/anycubic_lan.py @@ -24,7 +24,7 @@ from cryptography.hazmat.primitives import padding from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes CRED_TTL = 300 -INFO_TTL = 10 +INFO_TTL = 45 FAIL_TTL = 45 _CRED_CACHE = {} _INFO_CACHE = {} diff --git a/grotti/config.py b/grotti/config.py index b0f847e..e124e6b 100644 --- a/grotti/config.py +++ b/grotti/config.py @@ -2,6 +2,7 @@ import json import secrets +import threading from pathlib import Path from grotti.errors import GrottiError @@ -10,6 +11,7 @@ ROOT = Path(__file__).resolve().parent.parent LOOPBACK = {"127.0.0.1", "::1", "localhost"} CONFIG_PATH = ROOT / "config.json" DEFAULT_WATCH_REL = "slice" +_SETTINGS_LOCK = threading.Lock() def _energy(raw): @@ -42,6 +44,7 @@ def _alerts(raw): "fehler": bool(alerts.get("fehler", True)), "offline": bool(alerts.get("offline", True)), "queue": bool(alerts.get("queue", True)), + "spule": bool(alerts.get("spule", True)), "ablauf": bool(alerts.get("ablauf", False)), "start": bool(alerts.get("start", False)), } @@ -196,6 +199,19 @@ def settings_public(cfg): def save_settings(cfg, energy=None, telegram=None, alerts=None, auto_queue=None, fqdn=None, watch_dir=None): """Schreibt Einstellungen zurück nach config.json und aktualisiert cfg.""" + with _SETTINGS_LOCK: + return _save_settings_locked( + cfg, + energy=energy, + telegram=telegram, + alerts=alerts, + auto_queue=auto_queue, + fqdn=fqdn, + watch_dir=watch_dir, + ) + + +def _save_settings_locked(cfg, energy=None, telegram=None, alerts=None, auto_queue=None, fqdn=None, watch_dir=None): path = cfg.get("_path") or CONFIG_PATH raw = dict(cfg.get("_raw") or {}) if path.exists(): @@ -243,6 +259,7 @@ def save_settings(cfg, energy=None, telegram=None, alerts=None, auto_queue=None, "fehler": bool(alerts.get("fehler", True)), "offline": bool(alerts.get("offline", True)), "queue": bool(alerts.get("queue", True)), + "spule": bool(alerts.get("spule", True)), "ablauf": bool(alerts.get("ablauf", False)), "start": bool(alerts.get("start", False)), } diff --git a/grotti/db.py b/grotti/db.py index da06a54..640b9c3 100644 --- a/grotti/db.py +++ b/grotti/db.py @@ -1,6 +1,9 @@ -"""PostgreSQL-Verbindung und Schema.""" +"""PostgreSQL-Verbindung, Pool und Migrationen.""" + +from __future__ import annotations import re +import threading from contextlib import contextmanager from pathlib import Path from urllib.parse import urlparse, urlunparse @@ -8,22 +11,105 @@ from urllib.parse import urlparse, urlunparse import psycopg from psycopg import sql from psycopg.rows import dict_row +from psycopg_pool import ConnectionPool -SCHEMA = (Path(__file__).parent / "schema.sql").read_text(encoding="utf-8") +MIGRATIONS_DIR = Path(__file__).parent / "migrations" +SCHEMA_LEGACY = Path(__file__).parent / "schema.sql" + +_pool: ConnectionPool | None = None +_pool_lock = threading.Lock() +_user_count_cache = {"n": None, "at": 0.0} +_USER_COUNT_TTL = 5.0 def database_name(url): name = urlparse(url).path.lstrip("/") if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name or ""): - raise SystemExit("Der Datenbankname in database_url darf nur Buchstaben, Zahlen und Unterstrich enthalten.") + raise SystemExit( + "Der Datenbankname in database_url darf nur Buchstaben, Zahlen und Unterstrich enthalten." + ) return name -def apply_schema(conn): - for statement in SCHEMA.split(";"): +def _migration_files(): + files = sorted(MIGRATIONS_DIR.glob("*.sql")) + out = [] + for path in files: + match = re.match(r"^(\d+)_", path.name) + if not match: + continue + out.append((int(match.group(1)), path)) + return out + + +def _split_statements(text): + """Einfache Trennung an Semikolon — Migrationen ohne Funktionen mit ; im Körper.""" + parts = [] + for statement in text.split(";"): cleaned = statement.strip() if cleaned: - conn.execute(cleaned) + parts.append(cleaned) + return parts + + +def _ensure_version_table(conn): + conn.execute( + """ + CREATE TABLE IF NOT EXISTS schema_version ( + version integer PRIMARY KEY, + name text NOT NULL, + applied_at timestamptz NOT NULL DEFAULT now() + ) + """ + ) + + +def _applied_versions(conn): + rows = conn.execute("SELECT version FROM schema_version ORDER BY version").fetchall() + return {int(row["version"] if isinstance(row, dict) else row[0]) for row in rows} + + +def _legacy_schema_present(conn): + row = conn.execute( + """ + SELECT 1 FROM information_schema.tables + WHERE table_schema = 'public' AND table_name = 'printers' + """ + ).fetchone() + return bool(row) + + +def apply_migrations(conn): + """Wendet fehlende Migrationen an. Bestehende Installationen ohne Version bekommen 001 gutgeschrieben.""" + _ensure_version_table(conn) + applied = _applied_versions(conn) + migrations = _migration_files() + if not migrations and SCHEMA_LEGACY.exists(): + # Fallback: altes schema.sql einmalig + for statement in _split_statements(SCHEMA_LEGACY.read_text(encoding="utf-8")): + conn.execute(statement) + return + + if not applied and _legacy_schema_present(conn): + # Schema war schon da (vor Migrationssystem) → Basis als angewandt markieren. + base = next((m for m in migrations if m[0] == 1), None) + if base: + conn.execute( + "INSERT INTO schema_version (version, name) VALUES (%s, %s) ON CONFLICT DO NOTHING", + (1, base[1].name), + ) + applied.add(1) + + for version, path in migrations: + if version in applied: + continue + text = path.read_text(encoding="utf-8") + for statement in _split_statements(text): + conn.execute(statement) + conn.execute( + "INSERT INTO schema_version (version, name) VALUES (%s, %s)", + (version, path.name), + ) def prepare(url): @@ -31,14 +117,16 @@ def prepare(url): admin = urlunparse(urlparse(url)._replace(path="/postgres")) try: with psycopg.connect(admin, autocommit=True, connect_timeout=5) as conn: - exists = conn.execute("SELECT 1 FROM pg_database WHERE datname = %s", (name,)).fetchone() + exists = conn.execute( + "SELECT 1 FROM pg_database WHERE datname = %s", (name,) + ).fetchone() if not exists: conn.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(name))) - except psycopg.Error: - pass + except psycopg.Error as exc: + print("Datenbank anlegen: %s" % exc, flush=True) try: - with psycopg.connect(url, connect_timeout=5) as conn: - apply_schema(conn) + with psycopg.connect(url, row_factory=dict_row, connect_timeout=5) as conn: + apply_migrations(conn) conn.commit() except psycopg.Error as exc: raise SystemExit( @@ -50,9 +138,50 @@ def prepare(url): ) from exc +def init_pool(url, min_size=2, max_size=16): + global _pool + with _pool_lock: + if _pool is not None: + return _pool + _pool = ConnectionPool( + conninfo=url, + min_size=min_size, + max_size=max_size, + kwargs={"row_factory": dict_row, "connect_timeout": 5}, + open=True, + name="grotti", + ) + return _pool + + +def close_pool(): + global _pool + with _pool_lock: + if _pool is not None: + _pool.close() + _pool = None + + +def get_pool(): + if _pool is None: + raise RuntimeError("Datenbank-Pool ist nicht gestartet.") + return _pool + + @contextmanager -def connect(url): - conn = psycopg.connect(url, row_factory=dict_row) +def connect(url=None): + """Verbindung aus dem Pool, sonst einmalig direkt (Tests / prepare).""" + if _pool is not None: + with _pool.connection() as conn: + try: + yield conn + except Exception: + conn.rollback() + raise + return + if not url: + raise RuntimeError("Keine database_url und kein Pool.") + conn = psycopg.connect(url, row_factory=dict_row, connect_timeout=5) try: yield conn except Exception: @@ -60,3 +189,35 @@ def connect(url): raise finally: conn.close() + + +def cached_user_count(conn): + """Zählt Benutzer mit kurzem Cache — vermeidet count(*) bei jedem Request.""" + import time + + now = time.monotonic() + cached = _user_count_cache["n"] + if cached is not None and (now - _user_count_cache["at"]) < _USER_COUNT_TTL: + return cached + n = int(conn.execute("SELECT COUNT(*) AS n FROM users").fetchone()["n"]) + _user_count_cache["n"] = n + _user_count_cache["at"] = now + return n + + +def invalidate_user_count(): + _user_count_cache["n"] = None + _user_count_cache["at"] = 0.0 + + +def prune_run_samples(conn, keep_days=90): + """Alte Temperaturproben löschen (Retention).""" + days = max(7, int(keep_days)) + row = conn.execute( + """ + DELETE FROM run_samples + WHERE sampled_at < now() - (%s || ' days')::interval + """, + (str(days),), + ) + return row.rowcount if row is not None else 0 diff --git a/grotti/jobs.py b/grotti/jobs.py index 4f6dda8..b7a0f84 100644 --- a/grotti/jobs.py +++ b/grotti/jobs.py @@ -43,14 +43,7 @@ def _targets_from_ace(conn, host, ace): targets = [] for item in ace: slot = int(item["ams_index"]) + 1 - row = conn.execute( - """ - SELECT * FROM spools - WHERE printer_host = %s AND ace_slot = %s - ORDER BY id LIMIT 1 - """, - (host, slot), - ).fetchone() + row = store.spool_at_slot(conn, host, slot) if row: targets.append(row) return targets @@ -62,12 +55,18 @@ def unix_to_dt(value): return datetime.fromtimestamp(float(value), timezone.utc) -def choose_spool(conn, host, material, hinted_id): +def choose_spool(conn, host, material, hinted_id, ended_at=None): + """Spule für History-Abzug. Ohne Hint nur Rollen, die vor Druckende existierten.""" if hinted_id: row = store.get_spool_row(conn, hinted_id) - if row: + if row and _spool_valid_for_end(row, ended_at): return row - rows = store.spools_for_printer(conn, host) + if row and not _spool_valid_for_end(row, ended_at): + return None + rows = [ + row for row in store.spools_for_printer(conn, host) + if _spool_valid_for_end(row, ended_at) + ] if len(rows) == 1: return rows[0] if material: @@ -77,6 +76,24 @@ def choose_spool(conn, host, material, hinted_id): return None +def _spool_valid_for_end(spool, ended_at): + """Verhindert Abzug alter Moonraker-Jobs auf neu eingelegte Spulen.""" + if ended_at is None or not spool: + return True + created = spool.get("created_at") + if created is None: + return True + end = ended_at + if getattr(created, "tzinfo", None) is None and getattr(end, "tzinfo", None) is not None: + created = created.replace(tzinfo=timezone.utc) + if getattr(end, "tzinfo", None) is None and getattr(created, "tzinfo", None) is not None: + end = end.replace(tzinfo=timezone.utc) + try: + return created <= end + except TypeError: + return True + + def file_material_types(file_row): types = [] for item in (file_row or {}).get("filaments") or []: @@ -93,14 +110,7 @@ def resolve_spools_at_host(conn, host, ace_map=None): if mapping: for item in mapping: slot = int(item["ams_index"]) + 1 - row = conn.execute( - """ - SELECT * FROM spools - WHERE printer_host = %s AND ace_slot = %s - ORDER BY id LIMIT 1 - """, - (host, slot), - ).fetchone() + row = store.spool_at_slot(conn, host, slot) if row: found.append(row) return found @@ -329,7 +339,12 @@ def sync_history(conn, cfg): file_id = (sent or {}).get("file_id") or (known or {}).get("id") project_id = (sent or {}).get("project_id") or (known or {}).get("project_id") hinted = (sent or {}).get("spool_id") or (known or {}).get("spool_id") - spool = choose_spool(conn, printer["host"], material, hinted) + ended_at = unix_to_dt(job.get("end_time")) + # Ohne Send-Eintrag keine Rate — sonst zieht eine neue Spule alle alten Jobs ab. + if not sent and not hinted: + spool = None + else: + spool = choose_spool(conn, printer["host"], material, hinted, ended_at=ended_at) terminal = state in TERMINAL live = store.find_live_run(conn, printer["host"], filename) progress = float((live or {}).get("progress_last") or 0) @@ -503,7 +518,7 @@ def send_file( progress("pruefen", 15, "Drucker prüfen…") try: live = moonraker.query(host) - except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError) as exc: + except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, OSError, ValueError) as exc: raise GrottiError("Der Drucker ist nicht erreichbar.", "offline") from exc if live.get("state") in ("printing", "paused"): raise GrottiError("Der Drucker druckt schon.", "belegt", 409) diff --git a/grotti/library.py b/grotti/library.py index d1c7414..f019970 100644 --- a/grotti/library.py +++ b/grotti/library.py @@ -100,7 +100,10 @@ def refresh_file_meta(conn, force=False): def safe_filename(name): base = Path(name).name.replace("\\", "_").replace("/", "_").replace('"', "") - return base or "datei.gcode" + base = base.strip().strip(".") + if not base or base in (".", ".."): + return "datei.gcode" + return base def unique_path(directory, filename): diff --git a/grotti/mcp_server.py b/grotti/mcp_server.py index 1cdeb48..2f2d2cb 100644 --- a/grotti/mcp_server.py +++ b/grotti/mcp_server.py @@ -220,11 +220,11 @@ def main(): out = [] if nozzle is not None: out.append(_api("POST", "/drucker/%s/steuern" % urllib.parse.quote(host), { - "action": "nozzle", "value": nozzle, + "action": "nozzle", "temp": nozzle, })) if bed is not None: out.append(_api("POST", "/drucker/%s/steuern" % urllib.parse.quote(host), { - "action": "bed", "value": bed, + "action": "bed", "temp": bed, })) if not out: raise RuntimeError("nozzle und/oder bed angeben.") @@ -236,7 +236,7 @@ def main(): _write_guard() return json.dumps( _api("POST", "/drucker/%s/steuern" % urllib.parse.quote(host), { - "action": "speed", "value": speed, + "action": "speed", "percent": speed, }), ensure_ascii=False, ) @@ -247,7 +247,7 @@ def main(): _write_guard() return json.dumps( _api("POST", "/drucker/%s/steuern" % urllib.parse.quote(host), { - "action": "fan", "value": fan, + "action": "fan", "percent": fan, }), ensure_ascii=False, ) diff --git a/grotti/migrations/001_base.sql b/grotti/migrations/001_base.sql new file mode 100644 index 0000000..09d143f --- /dev/null +++ b/grotti/migrations/001_base.sql @@ -0,0 +1,305 @@ +CREATE TABLE IF NOT EXISTS projects ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + name text NOT NULL, + notes text NOT NULL DEFAULT '', + status text NOT NULL DEFAULT 'idee', + quote jsonb NOT NULL DEFAULT '{}'::jsonb, + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS spools ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + name text NOT NULL, + brand text NOT NULL DEFAULT '', + material_type text NOT NULL DEFAULT '', + color text NOT NULL DEFAULT '', + price_per_kg double precision NOT NULL DEFAULT 0, + density double precision NOT NULL DEFAULT 1.24, + volumetric double precision NOT NULL DEFAULT 12, + remaining_g double precision NOT NULL DEFAULT 1000, + printer_host text NOT NULL DEFAULT '', + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS files ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + project_id bigint REFERENCES projects (id) ON DELETE SET NULL, + spool_id bigint REFERENCES spools (id) ON DELETE SET NULL, + name text NOT NULL, + stored_path text NOT NULL, + source text NOT NULL, + kind text NOT NULL, + printer_family text NOT NULL DEFAULT '', + filaments jsonb NOT NULL DEFAULT '[]'::jsonb, + weight_g double precision NOT NULL DEFAULT 0, + estimated_s double precision NOT NULL DEFAULT 0, + thumb_entry text NOT NULL DEFAULT '', + plate text NOT NULL DEFAULT '', + size_bytes bigint NOT NULL DEFAULT 0, + mtime double precision NOT NULL DEFAULT 0, + created_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (source, stored_path) +); + +ALTER TABLE files ADD COLUMN IF NOT EXISTS slice_meta jsonb NOT NULL DEFAULT '{}'::jsonb; + +CREATE TABLE IF NOT EXISTS sends ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + printer_host text NOT NULL, + filename text NOT NULL, + file_id bigint REFERENCES files (id) ON DELETE SET NULL, + project_id bigint REFERENCES projects (id) ON DELETE SET NULL, + spool_id bigint REFERENCES spools (id) ON DELETE SET NULL, + created_at timestamptz NOT NULL DEFAULT now() +); + +ALTER TABLE sends ADD COLUMN IF NOT EXISTS ace_map jsonb NOT NULL DEFAULT '[]'::jsonb; +ALTER TABLE sends ADD COLUMN IF NOT EXISTS bed_leveling boolean NOT NULL DEFAULT false; + +CREATE TABLE IF NOT EXISTS runs ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + printer_host text NOT NULL, + printer_name text NOT NULL DEFAULT '', + filename text NOT NULL DEFAULT '', + file_id bigint REFERENCES files (id) ON DELETE SET NULL, + project_id bigint REFERENCES projects (id) ON DELETE SET NULL, + spool_id bigint REFERENCES spools (id) ON DELETE SET NULL, + state text NOT NULL DEFAULT '', + started_at timestamptz, + ended_at timestamptz, + duration_s double precision NOT NULL DEFAULT 0, + weight_g double precision NOT NULL DEFAULT 0, + filament_cost double precision NOT NULL DEFAULT 0, + energy_cost double precision NOT NULL DEFAULT 0, + moonraker_job_id text NOT NULL, + deducted boolean NOT NULL DEFAULT false, + needs_spool boolean NOT NULL DEFAULT false, + verdict text NOT NULL DEFAULT '', + material_type text NOT NULL DEFAULT '', + UNIQUE (printer_host, moonraker_job_id) +); + +CREATE INDEX IF NOT EXISTS files_project ON files (project_id); + +CREATE INDEX IF NOT EXISTS runs_started ON runs (started_at DESC); + +CREATE TABLE IF NOT EXISTS printers ( + id uuid PRIMARY KEY, + name text NOT NULL, + host text NOT NULL UNIQUE, + family text NOT NULL DEFAULT '', + created_at timestamptz NOT NULL DEFAULT now() +); + +ALTER TABLE printers ADD COLUMN IF NOT EXISTS locked boolean NOT NULL DEFAULT false; + +CREATE TABLE IF NOT EXISTS ideas ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + title text NOT NULL, + body text NOT NULL DEFAULT '', + status text NOT NULL DEFAULT 'offen', + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS materials ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + name text NOT NULL UNIQUE, + brand text NOT NULL DEFAULT '', + material_type text NOT NULL DEFAULT '', + density double precision NOT NULL, + volumetric double precision NOT NULL, + volumetric_s1 double precision, + machines text NOT NULL DEFAULT 'beide', + price_per_kg double precision NOT NULL DEFAULT 0, + note text NOT NULL DEFAULT '' +); + +ALTER TABLE spools ADD COLUMN IF NOT EXISTS colors jsonb NOT NULL DEFAULT '[]'::jsonb; +ALTER TABLE spools ADD COLUMN IF NOT EXISTS ace_slot smallint; +ALTER TABLE spools ADD COLUMN IF NOT EXISTS material_id bigint REFERENCES materials (id) ON DELETE SET NULL; +ALTER TABLE spools ADD COLUMN IF NOT EXISTS chargen_nr text; +ALTER TABLE spools ADD COLUMN IF NOT EXISTS hersteller_charge text NOT NULL DEFAULT ''; +ALTER TABLE spools ADD COLUMN IF NOT EXISTS code text; + +CREATE UNIQUE INDEX IF NOT EXISTS spools_chargen_nr ON spools (chargen_nr) WHERE chargen_nr IS NOT NULL AND chargen_nr <> ''; +CREATE UNIQUE INDEX IF NOT EXISTS spools_code ON spools (code) WHERE code IS NOT NULL AND code <> ''; +-- Ein ACE-Platz gehört höchstens einer Spule. +CREATE UNIQUE INDEX IF NOT EXISTS spools_host_slot + ON spools (printer_host, ace_slot) + WHERE printer_host <> '' AND ace_slot IS NOT NULL; + +CREATE TABLE IF NOT EXISTS events ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + host text NOT NULL DEFAULT '', + printer text NOT NULL DEFAULT '', + kind text NOT NULL, + message text NOT NULL DEFAULT '', + filename text NOT NULL DEFAULT '', + created_at timestamptz NOT NULL DEFAULT now(), + read_at timestamptz +); + +CREATE INDEX IF NOT EXISTS events_unread ON events (read_at, created_at DESC); + +CREATE TABLE IF NOT EXISTS print_queue ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + file_id bigint NOT NULL REFERENCES files (id) ON DELETE CASCADE, + host text NOT NULL DEFAULT '', + family text NOT NULL DEFAULT '', + priority integer NOT NULL DEFAULT 100, + status text NOT NULL DEFAULT 'waiting', + spool_id bigint REFERENCES spools (id) ON DELETE SET NULL, + ace_map jsonb NOT NULL DEFAULT '[]'::jsonb, + bed_leveling boolean NOT NULL DEFAULT false, + override_machine boolean NOT NULL DEFAULT false, + message text NOT NULL DEFAULT '', + steps jsonb NOT NULL DEFAULT '[]'::jsonb, + step_index integer NOT NULL DEFAULT 0, + repeat_total integer NOT NULL DEFAULT 1, + repeat_done integer NOT NULL DEFAULT 0, + phase text NOT NULL DEFAULT '', + phase_started_at timestamptz, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() +); + +ALTER TABLE print_queue ADD COLUMN IF NOT EXISTS steps jsonb NOT NULL DEFAULT '[]'::jsonb; +ALTER TABLE print_queue ADD COLUMN IF NOT EXISTS step_index integer NOT NULL DEFAULT 0; +ALTER TABLE print_queue ADD COLUMN IF NOT EXISTS repeat_total integer NOT NULL DEFAULT 1; +ALTER TABLE print_queue ADD COLUMN IF NOT EXISTS repeat_done integer NOT NULL DEFAULT 0; +ALTER TABLE print_queue ADD COLUMN IF NOT EXISTS phase text NOT NULL DEFAULT ''; +ALTER TABLE print_queue ADD COLUMN IF NOT EXISTS phase_started_at timestamptz; + +CREATE INDEX IF NOT EXISTS print_queue_waiting ON print_queue (status, priority, id); +CREATE INDEX IF NOT EXISTS print_queue_running_host ON print_queue (host, status) WHERE status = 'running'; + +-- Nach dem Druck: Platte manuell quittieren, bevor „wenn frei“ den nächsten Job startet. +CREATE TABLE IF NOT EXISTS bed_status ( + host text PRIMARY KEY, + needs_clear boolean NOT NULL DEFAULT false, + cleared_at timestamptz, + message text NOT NULL DEFAULT '' +); + +CREATE TABLE IF NOT EXISTS users ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + username text NOT NULL UNIQUE, + password_hash text NOT NULL, + role text NOT NULL DEFAULT 'operator', + active boolean NOT NULL DEFAULT true, + created_at timestamptz NOT NULL DEFAULT now(), + last_login timestamptz +); + +CREATE TABLE IF NOT EXISTS api_tokens ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + user_id bigint NOT NULL REFERENCES users (id) ON DELETE CASCADE, + name text NOT NULL DEFAULT '', + token_hash text NOT NULL UNIQUE, + token_prefix text NOT NULL DEFAULT '', + role text NOT NULL DEFAULT 'operator', + created_at timestamptz NOT NULL DEFAULT now(), + last_used timestamptz, + revoked_at timestamptz, + expires_at timestamptz +); + +CREATE INDEX IF NOT EXISTS api_tokens_user ON api_tokens (user_id) WHERE revoked_at IS NULL; + +-- OIDC später: Identitäten an lokale User binden. +CREATE TABLE IF NOT EXISTS oauth_identities ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + user_id bigint NOT NULL REFERENCES users (id) ON DELETE CASCADE, + provider text NOT NULL, + subject text NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (provider, subject) +); + +-- Druckbericht: Soll/Ist, Charge, Live-Fortschritt +ALTER TABLE runs ADD COLUMN IF NOT EXISTS weight_planned_g double precision NOT NULL DEFAULT 0; +ALTER TABLE runs ADD COLUMN IF NOT EXISTS weight_used_g double precision NOT NULL DEFAULT 0; +ALTER TABLE runs ADD COLUMN IF NOT EXISTS weight_source text NOT NULL DEFAULT ''; +ALTER TABLE runs ADD COLUMN IF NOT EXISTS filament_used_mm double precision NOT NULL DEFAULT 0; +ALTER TABLE runs ADD COLUMN IF NOT EXISTS filament_total_mm double precision NOT NULL DEFAULT 0; +ALTER TABLE runs ADD COLUMN IF NOT EXISTS progress_last double precision NOT NULL DEFAULT 0; +ALTER TABLE runs ADD COLUMN IF NOT EXISTS estimated_s double precision NOT NULL DEFAULT 0; +ALTER TABLE runs ADD COLUMN IF NOT EXISTS chargen_nr text; +ALTER TABLE runs ADD COLUMN IF NOT EXISTS message text NOT NULL DEFAULT ''; +ALTER TABLE runs ADD COLUMN IF NOT EXISTS objects jsonb NOT NULL DEFAULT '[]'::jsonb; + +CREATE UNIQUE INDEX IF NOT EXISTS runs_chargen_nr + ON runs (chargen_nr) WHERE chargen_nr IS NOT NULL AND chargen_nr <> ''; + +ALTER TABLE events ADD COLUMN IF NOT EXISTS run_id bigint REFERENCES runs (id) ON DELETE SET NULL; +CREATE INDEX IF NOT EXISTS events_run ON events (run_id) WHERE run_id IS NOT NULL; + +-- Abzug je Spule (Messwert oder Schätzwert) +CREATE TABLE IF NOT EXISTS run_spools ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + run_id bigint NOT NULL REFERENCES runs (id) ON DELETE CASCADE, + spool_id bigint NOT NULL REFERENCES spools (id) ON DELETE RESTRICT, + ace_slot smallint, + weight_g double precision NOT NULL DEFAULT 0, + weight_planned_g double precision NOT NULL DEFAULT 0, + cost double precision NOT NULL DEFAULT 0, + is_estimate boolean NOT NULL DEFAULT false, + UNIQUE (run_id, spool_id, ace_slot) +); + +CREATE INDEX IF NOT EXISTS run_spools_spool ON run_spools (spool_id); +CREATE INDEX IF NOT EXISTS run_spools_run ON run_spools (run_id); + +-- Abschnitte bei Rollenwechsel mitten im Druck +CREATE TABLE IF NOT EXISTS run_segments ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + run_id bigint NOT NULL REFERENCES runs (id) ON DELETE CASCADE, + spool_id bigint NOT NULL REFERENCES spools (id) ON DELETE RESTRICT, + ace_slot smallint, + mm_start double precision NOT NULL DEFAULT 0, + mm_end double precision, + created_at timestamptz NOT NULL DEFAULT now(), + closed_at timestamptz +); + +CREATE INDEX IF NOT EXISTS run_segments_run ON run_segments (run_id); +CREATE INDEX IF NOT EXISTS run_segments_open + ON run_segments (run_id) WHERE mm_end IS NULL; + +-- Temperatur / Fortschritt während des Drucks +CREATE TABLE IF NOT EXISTS run_samples ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + run_id bigint NOT NULL REFERENCES runs (id) ON DELETE CASCADE, + sampled_at timestamptz NOT NULL DEFAULT now(), + nozzle double precision, + nozzle_target double precision, + bed double precision, + bed_target double precision, + chamber double precision, + fan double precision, + speed_factor double precision, + extrude_factor double precision, + layer integer, + progress double precision, + filament_used_mm double precision, + mmu_tool smallint, + nozzle_power double precision, + bed_power double precision +); + +CREATE INDEX IF NOT EXISTS run_samples_run_time ON run_samples (run_id, sampled_at); + +ALTER TABLE run_samples ADD COLUMN IF NOT EXISTS nozzle_power double precision; +ALTER TABLE run_samples ADD COLUMN IF NOT EXISTS bed_power double precision; + +-- Objekt-Chargen für Etiketten +CREATE TABLE IF NOT EXISTS run_labels ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + run_id bigint NOT NULL REFERENCES runs (id) ON DELETE CASCADE, + object_name text NOT NULL DEFAULT '', + chargen_nr text NOT NULL, + excluded boolean NOT NULL DEFAULT false, + UNIQUE (run_id, object_name) +); + +CREATE UNIQUE INDEX IF NOT EXISTS run_labels_chargen ON run_labels (chargen_nr); diff --git a/grotti/migrations/002_indexes.sql b/grotti/migrations/002_indexes.sql new file mode 100644 index 0000000..544cf05 --- /dev/null +++ b/grotti/migrations/002_indexes.sql @@ -0,0 +1,6 @@ +-- Heiße Abfragepfade für Sends, Läufe, Dateien und Ereignisse. +CREATE INDEX IF NOT EXISTS sends_host_filename ON sends (printer_host, filename); +CREATE INDEX IF NOT EXISTS sends_host_file ON sends (printer_host, file_id); +CREATE INDEX IF NOT EXISTS runs_host_state ON runs (printer_host, state); +CREATE INDEX IF NOT EXISTS files_name ON files (name); +CREATE INDEX IF NOT EXISTS events_host_kind_created ON events (host, kind, created_at DESC); diff --git a/grotti/moonraker.py b/grotti/moonraker.py index 938b218..eb34e86 100644 --- a/grotti/moonraker.py +++ b/grotti/moonraker.py @@ -17,6 +17,16 @@ SLOT_CACHE = {} MESH_CACHE = {} OBJECT_CACHE = {} THUMB_RE_LIMIT = 180_000 +_CACHE_MAX = 128 + + +def _trim_cache(cache, max_items=_CACHE_MAX): + """Einfache Größenbegrenzung (älteste Keys zuerst bei dict-Reihenfolge).""" + while len(cache) > max_items: + try: + cache.pop(next(iter(cache))) + except StopIteration: + break def matrix_rows(points): @@ -343,6 +353,11 @@ def steuern(host, action, params=None): """Whitelist-Steuerung ohne freie Konsole.""" params = params if isinstance(params, dict) else {} action = (action or "").strip() + # MCP und ältere Clients schicken "value" statt temp/percent. + if "temp" not in params and params.get("value") is not None and action in ("nozzle", "bed", "dryer_start"): + params = {**params, "temp": params["value"]} + if "percent" not in params and params.get("value") is not None and action in ("speed", "flow", "fan"): + params = {**params, "percent": params["value"]} def pct(key, default=100, lo=0, hi=200): try: @@ -492,6 +507,7 @@ def metadata(host, path): except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError): data = {} META_CACHE[key] = (now, data) + _trim_cache(META_CACHE) return data @@ -518,9 +534,11 @@ def embedded_preview(host, path): text = response.read().decode("utf-8", "replace") except (urllib.error.URLError, TimeoutError): EMBEDDED[key] = None + _trim_cache(EMBEDDED) return None png = embedded_png(text) EMBEDDED[key] = png + _trim_cache(EMBEDDED) return png @@ -1117,7 +1135,9 @@ def snapshot(printers): } try: card.update(query(printer["host"])) - except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError) as error: + except Exception as error: + # Auch RemoteDisconnected / IncompleteRead / ValueError — sonst fällt der ganze Tick aus. + print("snapshot %s: %s" % (printer.get("host"), error), flush=True) card.update({ "ok": False, "state": "offline", diff --git a/grotti/print_queue.py b/grotti/print_queue.py index 75908b2..4d32ce8 100644 --- a/grotti/print_queue.py +++ b/grotti/print_queue.py @@ -20,6 +20,8 @@ BUSY = {"printing", "paused", "heating", "error"} KEEPALIVE_S = 120 # Nach Start ohne print_active: Upload/MQTT/LeviQ brauchen Zeit, aber nicht ewig PRINT_START_TIMEOUT_S = 600 +# Einfacher Queue-Start ohne Worker (typisch Grotti-Neustart) +SIMPLE_START_STUCK_S = 45 def _ablauf_needs_motors(item): @@ -120,6 +122,41 @@ class PrintQueueDispatcher: thread = self._workers.get(item_id) return bool(thread and thread.is_alive()) + def _item_phase_age_s(self, item): + """Alter von phase_started_at, sonst updated_at (für hängende Starts nach Neustart).""" + started = item.get("phase_started_at") + if started: + if getattr(started, "tzinfo", None) is None: + started = started.replace(tzinfo=timezone.utc) + return (datetime.now(timezone.utc) - started).total_seconds() + updated = item.get("updated_at") + if updated: + if getattr(updated, "tzinfo", None) is None: + updated = updated.replace(tzinfo=timezone.utc) + return (datetime.now(timezone.utc) - updated).total_seconds() + return 0 + + def _fail_stuck_simple_start(self, conn, item, printer): + host = printer["host"] + store.update_print_queue( + conn, + item["id"], + { + "status": "error", + "message": "Start abgebrochen (Neustart)", + "phase": "", + "phase_started_at": None, + }, + ) + self._event( + conn, + host, + printer["name"], + "queue", + "%s: Start abgebrochen (Neustart)" % printer["name"], + item.get("filename") or "", + ) + def _spawn_worker(self, item_id, target, *args): with self._worker_lock: existing = self._workers.get(item_id) @@ -129,6 +166,8 @@ class PrintQueueDispatcher: def run(): try: target(*args) + except Exception as exc: + print("Queue-Worker %s: %s" % (item_id, exc), flush=True) finally: with self._worker_lock: self._workers.pop(item_id, None) @@ -190,7 +229,16 @@ class PrintQueueDispatcher: continue if state not in FREE: continue - if store.running_print_on_host(conn, host): + running = store.running_print_on_host(conn, host) + if running: + if ( + not _has_ablauf(running) + and (running.get("phase") or "").lower() == "starting" + and not self._worker_busy(running["id"]) + and self._item_phase_age_s(running) > SIMPLE_START_STUCK_S + ): + self._fail_stuck_simple_start(conn, running, printer) + conn.commit() continue if store.bed_needs_clear(conn, host): continue @@ -276,6 +324,7 @@ class PrintQueueDispatcher: ) wconn.commit() except GrottiError as exc: + wconn.rollback() store.update_print_queue( wconn, item_id, @@ -291,6 +340,7 @@ class PrintQueueDispatcher: ) wconn.commit() except Exception as exc: + wconn.rollback() store.update_print_queue( wconn, item_id, @@ -684,10 +734,12 @@ class PrintQueueDispatcher: self._step_event(wconn, row, printer_copy, "druckt %s" % name) wconn.commit() except GrottiError as exc: + wconn.rollback() row = store.get_print_queue_row(wconn, item_id) or item self._fail_ablauf(wconn, row, printer_copy, exc.message) wconn.commit() except Exception as exc: + wconn.rollback() row = store.get_print_queue_row(wconn, item_id) or item self._fail_ablauf(wconn, row, printer_copy, str(exc) or "Druckstart fehlgeschlagen.") wconn.commit() @@ -742,6 +794,7 @@ class PrintQueueDispatcher: return state def _start_cool_step(self, conn, item, printer, step): + # TODO: _try_cool_park (bis 60 s G-code) blockiert tick() — ggf. in Worker auslagern. host = printer["host"] cool = eject.normalize_cool(step) try: @@ -908,14 +961,17 @@ class PrintQueueDispatcher: self._next_step(wconn, row, printer_copy, {"state": "standby", "ok": True}) wconn.commit() except GrottiError as exc: + wconn.rollback() row = store.get_print_queue_row(wconn, item_id) or item self._fail_ablauf(wconn, row, printer_copy, exc.message) wconn.commit() except moonraker.PrinterError as exc: + wconn.rollback() row = store.get_print_queue_row(wconn, item_id) or item self._fail_ablauf(wconn, row, printer_copy, eject.explain_printer_error(exc.message)) wconn.commit() except Exception as exc: + wconn.rollback() row = store.get_print_queue_row(wconn, item_id) or item self._fail_ablauf(wconn, row, printer_copy, "Schieben: %s" % exc) wconn.commit() @@ -1013,14 +1069,17 @@ class PrintQueueDispatcher: self._next_step(wconn, row, printer_copy, {"state": "standby", "ok": True}) wconn.commit() except GrottiError as exc: + wconn.rollback() row = store.get_print_queue_row(wconn, item_id) or item self._fail_ablauf(wconn, row, printer_copy, exc.message) wconn.commit() except moonraker.PrinterError as exc: + wconn.rollback() row = store.get_print_queue_row(wconn, item_id) or item self._fail_ablauf(wconn, row, printer_copy, eject.explain_printer_error(exc.message)) wconn.commit() except Exception as exc: + wconn.rollback() row = store.get_print_queue_row(wconn, item_id) or item self._fail_ablauf(wconn, row, printer_copy, "Tasten: %s" % exc) wconn.commit() diff --git a/grotti/schema.sql b/grotti/schema.sql index 09d143f..5233c19 100644 --- a/grotti/schema.sql +++ b/grotti/schema.sql @@ -1,3 +1,6 @@ +-- Referenz-Schema. Laufende Migrationen liegen in grotti/migrations/*.sql +-- (001_base.sql entspricht diesem Stand, danach nummerierte Änderungen). + CREATE TABLE IF NOT EXISTS projects ( id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, name text NOT NULL, diff --git a/grotti/send_queue.py b/grotti/send_queue.py index 418ba1f..e0b9454 100644 --- a/grotti/send_queue.py +++ b/grotti/send_queue.py @@ -17,12 +17,19 @@ class SendQueue: def __init__(self, database_url, cfg, workers=4): self.database_url = database_url self.cfg = cfg + self._workers_n = max(1, int(workers)) self._lock = threading.Lock() self._pending = Queue() self._jobs = {} self._order = deque(maxlen=60) self._threads = [] - for index in range(max(1, int(workers))): + self._stop = threading.Event() + + def start(self): + if self._threads: + return + self._stop.clear() + for index in range(self._workers_n): thread = threading.Thread( target=self._worker, name="grotti-send-queue-%s" % (index + 1), @@ -31,6 +38,14 @@ class SendQueue: thread.start() self._threads.append(thread) + def stop(self): + self._stop.set() + for _ in self._threads: + try: + self._pending.put_nowait(None) + except Exception: + pass + def enqueue( self, conn, @@ -82,9 +97,14 @@ class SendQueue: want_level = bool(bed_leveling) created = [] for host in unique: - printer, row, chosen = jobs.prepare_send(conn, file_id, host, True, override, spool_id) + # Mapping zuerst, damit Materialprüfung die ACE-Spulen sieht. + mapping = ace_maps.get(host) or [] + printer, row, chosen = jobs.prepare_send( + conn, file_id, host, True, override, spool_id, ace_map=mapping + ) + if not mapping: + mapping = ace_maps.get(printer["name"]) or [] job_id = uuid.uuid4().hex[:12] - mapping = ace_maps.get(host) or ace_maps.get(printer["name"]) or [] job = { "id": job_id, "status": "pending", @@ -143,11 +163,13 @@ class SendQueue: job["updated_at"] = time.time() def _worker(self): - while True: + while not self._stop.is_set(): try: item = self._pending.get(timeout=1) except Empty: continue + if item is None: + break self._run(item) def _run(self, item): diff --git a/grotti/store.py b/grotti/store.py index 2eebce9..84234bd 100644 --- a/grotti/store.py +++ b/grotti/store.py @@ -100,6 +100,8 @@ def spool_out(row): 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 @@ -544,6 +546,28 @@ def seed_materials(conn): 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) @@ -555,6 +579,20 @@ def spools_for_printer(conn, 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: @@ -871,10 +909,11 @@ def assign_run_spool(conn, run_id, spool_id): def next_druck_chargen(conn): """Nächste Druck-Charge des Tages. Objekt-Etiketten (`D-…-01`) blockieren die Nummer.""" - day = datetime.now(timezone.utc).strftime("%Y%m%d") + 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 ( @@ -2531,7 +2570,8 @@ def ensure_bed_after_print(conn, host, ended_at=None, message=""): def user_count(conn): - return int(conn.execute("SELECT count(*) AS n FROM users").fetchone()["n"]) + from grotti.db import cached_user_count + return cached_user_count(conn) def list_users(conn): diff --git a/grotti/watcher.py b/grotti/watcher.py index 085dd44..4f1cf42 100644 --- a/grotti/watcher.py +++ b/grotti/watcher.py @@ -362,22 +362,29 @@ class StateWatcher: if need <= 0 or not filename: return False sent = store.latest_send(conn, host, filename) - spool_id = (sent or {}).get("spool_id") - if not spool_id: + ace = (sent or {}).get("ace_map") or [] + from grotti import jobs as grotti_jobs + + at_host = grotti_jobs.resolve_spools_at_host(conn, host, ace) + if not at_host: + spool_id = (sent or {}).get("spool_id") + if spool_id: + row = store.get_spool_row(conn, spool_id) + if row: + at_host = [row] + if not at_host: return False - spool = store.get_spool_row(conn, spool_id) - if not spool: - return False - left = float(spool["remaining_g"] or 0) + left = sum(float(s.get("remaining_g") or 0) for s in at_host) if left + 0.5 >= need: return False + label = " / ".join(s.get("name") or "?" for s in at_host[:3]) self._emit( conn, host, name, "spule", - "%s: Spule „%s“ hat nur %.0f g, Druck braucht %.0f g." - % (name, spool["name"], left, need), + "%s: %s hat nur %.0f g, Druck braucht %.0f g." + % (name, label, left, need), filename, ) return True diff --git a/server.py b/server.py index 90cd51a..36aba31 100644 --- a/server.py +++ b/server.py @@ -1,19 +1,26 @@ """Grotti starten. Kern und Seiten liegen daneben.""" +import atexit +import signal import threading +import time -from grotti import jobs, library, store +from waitress import serve + +from grotti import db, jobs, library, store from grotti.config import load -from grotti.db import connect, prepare from web.app import create_app def background(cfg): def work(): try: - with connect(cfg["database_url"]) as conn: + with db.connect(cfg["database_url"]) as conn: library.sync_watch(conn, cfg, force=False) jobs.sync_history(conn, cfg) + deleted = db.prune_run_samples(conn, keep_days=90) + if deleted: + print("Proben gelöscht: %s" % deleted, flush=True) conn.commit() except Exception as exc: print("Abgleich: %s" % exc, flush=True) @@ -21,19 +28,79 @@ def background(cfg): threading.Thread(target=work, name="grotti-abgleich", daemon=True).start() +def _retention_loop(cfg, stop_event): + """Nächtlich alte run_samples entfernen.""" + while not stop_event.wait(3600): + hour = time.localtime().tm_hour + if hour != 3: + continue + try: + with db.connect(cfg["database_url"]) as conn: + n = db.prune_run_samples(conn, keep_days=90) + conn.commit() + if n: + print("Retention: %s Proben gelöscht." % n, flush=True) + except Exception as exc: + print("Retention: %s" % exc, flush=True) + stop_event.wait(3600) + + def main(): cfg = load() cfg["data_dir"].mkdir(parents=True, exist_ok=True) (cfg["data_dir"] / "eingang").mkdir(exist_ok=True) - prepare(cfg["database_url"]) - with connect(cfg["database_url"]) as conn: + db.prepare(cfg["database_url"]) + db.init_pool(cfg["database_url"], min_size=2, max_size=16) + with db.connect(cfg["database_url"]) as conn: store.seed_printers(conn, cfg["printers_file"]) store.seed_materials(conn) conn.commit() + app = create_app(cfg) + watcher = app.extensions["watcher"] + dispatcher = app.extensions["print_queue"] + send_queue = app.extensions["send_queue"] + watcher.start() + dispatcher.start() + if hasattr(send_queue, "start"): + send_queue.start() + + stop_event = threading.Event() + retention = threading.Thread( + target=_retention_loop, + args=(cfg, stop_event), + name="grotti-retention", + daemon=True, + ) + retention.start() background(cfg) + + def shutdown(*_args): + stop_event.set() + try: + watcher.stop() + except Exception: + pass + try: + dispatcher.stop() + except Exception: + pass + try: + if hasattr(send_queue, "stop"): + send_queue.stop() + except Exception: + pass + db.close_pool() + + atexit.register(shutdown) + for sig in (signal.SIGINT, signal.SIGTERM): + try: + signal.signal(sig, lambda *_: shutdown()) + except (ValueError, OSError): + pass + print("Grotti: http://%s:%s/" % (cfg["host"], cfg["port"]), flush=True) - app.run(host=cfg["host"], port=cfg["port"], threaded=True, use_reloader=False) + serve(app, host=cfg["host"], port=cfg["port"], threads=16, channel_timeout=120) if __name__ == "__main__": diff --git a/tests/test_jobs_spool.py b/tests/test_jobs_spool.py new file mode 100644 index 0000000..32363bc --- /dev/null +++ b/tests/test_jobs_spool.py @@ -0,0 +1,65 @@ +"""Unit-Tests für Spulenwahl und Steuer-Parameter-Aliase.""" + +import unittest +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock, patch + +from grotti import jobs +from grotti import moonraker + + +class ChooseSpoolTests(unittest.TestCase): + def test_rejects_spool_created_after_print_end(self): + end = datetime(2026, 10, 1, 12, 0, tzinfo=timezone.utc) + old = { + "id": 1, + "material_type": "PLA", + "created_at": end - timedelta(days=2), + } + new = { + "id": 2, + "material_type": "PLA", + "created_at": end + timedelta(hours=1), + } + conn = MagicMock() + with patch("grotti.jobs.store.spools_for_printer", return_value=[new]): + with patch("grotti.jobs.store.get_spool_row", return_value=new): + self.assertIsNone(jobs.choose_spool(conn, "h", "PLA", 2, ended_at=end)) + with patch("grotti.jobs.store.spools_for_printer", return_value=[old]): + self.assertEqual(jobs.choose_spool(conn, "h", "PLA", None, ended_at=end)["id"], 1) + + def test_spool_valid_for_end_naive(self): + end = datetime(2026, 10, 1, 12, 0, tzinfo=timezone.utc) + spool = {"created_at": datetime(2026, 9, 1, 12, 0)} + self.assertTrue(jobs._spool_valid_for_end(spool, end)) + + +class SteuernAliasTests(unittest.TestCase): + @patch("grotti.moonraker.gcode_script", return_value={"ok": True}) + def test_value_alias_for_temp(self, gcode): + moonraker.steuern("10.0.0.1", "nozzle", {"value": 210}) + script = gcode.call_args[0][1] + self.assertIn("TARGET=210.0", script) + + @patch("grotti.moonraker.gcode_script", return_value={"ok": True}) + def test_percent_alias_for_speed(self, gcode): + moonraker.steuern("10.0.0.1", "speed", {"value": 120}) + script = gcode.call_args[0][1] + self.assertIn("M220 S120", script) + + +class HostValidationTests(unittest.TestCase): + def test_validate_printer_host(self): + from web.app import validate_printer_host + from grotti.errors import GrottiError + + self.assertEqual(validate_printer_host("10.10.1.60"), "10.10.1.60") + self.assertEqual(validate_printer_host("kobra.local:80"), "kobra.local:80") + with self.assertRaises(GrottiError): + validate_printer_host("http://evil") + with self.assertRaises(GrottiError): + validate_printer_host("../etc/passwd") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_library_safe.py b/tests/test_library_safe.py new file mode 100644 index 0000000..431e207 --- /dev/null +++ b/tests/test_library_safe.py @@ -0,0 +1,21 @@ +"""Pfad- und Dateiname-Grenzen.""" + +import unittest + +from grotti import library +from grotti.errors import GrottiError + + +class SafeFilenameTests(unittest.TestCase): + def test_rejects_dotdot(self): + self.assertEqual(library.safe_filename(".."), "datei.gcode") + self.assertEqual(library.safe_filename("../x.gcode"), "x.gcode") + + def test_normalize_folder_blocks_escape(self): + with self.assertRaises(GrottiError): + library.normalize_folder("../secret") + self.assertEqual(library.normalize_folder("Gridfinity/A"), "Gridfinity/A") + + +if __name__ == "__main__": + unittest.main() diff --git a/web/app.py b/web/app.py index a2e5c1a..08cfb96 100644 --- a/web/app.py +++ b/web/app.py @@ -14,9 +14,11 @@ import psycopg from flask import Flask, abort, g, jsonify, redirect, render_template, request, session, url_for from markupsafe import Markup from psycopg.rows import dict_row +from werkzeug.middleware.proxy_fix import ProxyFix from grotti import auth as grotti_auth from grotti import config as grotti_config +from grotti import db as grotti_db from grotti import eject, jobs, library, moonraker, notify, store from grotti.catalog import color_catalog from grotti.cost import calculate @@ -35,12 +37,30 @@ PROJECT_STATUS = {"idee", "bereit", "druckt", "fertig"} VERDICTS = {"", "gut", "ausschuss"} IDEA_STATUS = {"offen", "spaeter", "erledigt"} PRINTER_FAMILIES = set(FAMILY_LABELS) | {""} +UPLOAD_SUFFIXES = (".gcode", ".3mf", ".gcode.3mf") +HOST_RE = re.compile(r"^[A-Za-z0-9.\-]+(?::\d{1,5})?$") +TOKEN_QUERY_PATHS = ( + "/api/v1/kamera", + "/api/v1/vorschau", +) STEER_ACTIONS = { "nozzle", "bed", "speed", "flow", "fan", "emergency_stop", "clear_error", "exclude_object", "dryer_start", "dryer_stop", "slot_setzen", } +def validate_printer_host(host): + host = (host or "").strip() + if not host or not HOST_RE.fullmatch(host): + raise GrottiError( + "Host nur als Hostname oder IP, optional mit Port (z. B. 10.10.1.60).", + "host", + ) + if ".." in host: + raise GrottiError("Ungültiger Host.", "host") + return host + + def create_app(cfg): app = Flask( __name__, @@ -56,13 +76,14 @@ def create_app(cfg): if cfg.get("fqdn", "").startswith("https://"): app.config["SESSION_COOKIE_SECURE"] = True app.json.ensure_ascii = False + # Hinter Reverse Proxy: X-Forwarded-For für Rate-Limits. + app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1) + # Threads starten in server.main() — Tests können create_app ohne Netz/DB-Worker nutzen. app.extensions["send_queue"] = SendQueue(cfg["database_url"], cfg) watcher = StateWatcher(cfg["database_url"], cfg) dispatcher = PrintQueueDispatcher(cfg["database_url"], cfg, watcher=watcher) app.extensions["print_queue"] = dispatcher app.extensions["watcher"] = watcher - watcher.start() - dispatcher.start() PUBLIC_PATHS = { "/login", @@ -86,7 +107,13 @@ def create_app(cfg): def conn(): if "db" not in g: - g.db = psycopg.connect(cfg["database_url"], row_factory=dict_row) + try: + pool = grotti_db.get_pool() + g.db = pool.getconn() + g.db_from_pool = True + except RuntimeError: + g.db = psycopg.connect(cfg["database_url"], row_factory=dict_row) + g.db_from_pool = False return g.db def ensure_csrf(): @@ -176,7 +203,11 @@ def create_app(cfg): header = request.headers.get("Authorization", "") if header.startswith("Bearer "): presented = header[7:].strip() - elif request.method == "GET" and request.path.startswith("/api/"): + elif ( + request.method == "GET" + and any(request.path.startswith(p) for p in TOKEN_QUERY_PATHS) + ): + # Token in der URL nur für Bild-Endpunkte (img-src ohne Header). presented = (request.args.get("token") or "").strip() user = None @@ -233,24 +264,60 @@ def create_app(cfg): @app.teardown_appcontext def close_db(exc): - conn = g.pop("db", None) - if not conn: + db_conn = g.pop("db", None) + from_pool = g.pop("db_from_pool", False) + if not db_conn: return try: if exc: - conn.rollback() + db_conn.rollback() else: - conn.commit() + db_conn.commit() + except Exception: + try: + db_conn.rollback() + except Exception: + pass finally: - conn.close() + if from_pool: + try: + grotti_db.get_pool().putconn(db_conn) + except Exception: + try: + db_conn.close() + except Exception: + pass + else: + db_conn.close() @app.errorhandler(GrottiError) def grotti_error(err): - db = g.get("db") - if db: - db.rollback() + db_conn = g.get("db") + if db_conn: + try: + db_conn.rollback() + except Exception: + pass return jsonify({"error": err.message, "code": err.code}), err.status + @app.errorhandler(psycopg.Error) + def psycopg_error(err): + db_conn = g.get("db") + if db_conn: + try: + db_conn.rollback() + except Exception: + pass + message = "Datenbankfehler." + if isinstance(err, psycopg.errors.ForeignKeyViolation): + message = "Der Eintrag hängt noch an anderen Daten." + elif isinstance(err, psycopg.errors.UniqueViolation): + message = "Diesen Eintrag gibt es schon." + elif isinstance(err, psycopg.errors.InvalidTextRepresentation): + message = "Ungültiger Wert in der Anfrage." + print("DB: %s" % err, flush=True) + return jsonify({"error": message, "code": "db"}), 500 + def body(): data = request.get_json(silent=True) if not isinstance(data, dict): @@ -334,6 +401,7 @@ def create_app(cfg): username = grotti_auth.validate_username(data.get("username") or "") password = grotti_auth.validate_password(data.get("password") or "") row = store.create_user(conn(), username, grotti_auth.hash_password(password), "admin") + grotti_db.invalidate_user_count() store.touch_user_login(conn(), row["id"]) session.clear() session["user_id"] = row["id"] @@ -392,6 +460,7 @@ def create_app(cfg): if store.get_user_by_name(conn(), username): raise GrottiError("Benutzer gibt es schon.", "user", 409) row = store.create_user(conn(), username, grotti_auth.hash_password(password), role) + grotti_db.invalidate_user_count() audit("auth", "Benutzer %s (%s) angelegt" % (username, role)) return jsonify(grotti_auth.public_user(row)), 201 @@ -779,11 +848,9 @@ def create_app(cfg): def printer_create(): data = body() name = (data.get("name") or "").strip() - host = (data.get("host") or "").strip() + host = validate_printer_host(data.get("host") or "") if not name: raise GrottiError("Der Drucker braucht einen Namen.", "name") - if not host: - raise GrottiError("Der Drucker braucht eine IP oder einen Host.", "host") family = (data.get("family") or "").strip() or family_from_name(name) if family not in PRINTER_FAMILIES: raise GrottiError("Unbekannte Druckerfamilie.", "family") @@ -799,14 +866,12 @@ def create_app(cfg): raise GrottiError("Diesen Drucker kennt Grotti nicht.", "drucker", 404) name = (data.get("name") if "name" in data else current["name"]) or "" name = str(name).strip() - host = (data.get("host") if "host" in data else current["host"]) or "" - host = str(host).strip() + host_raw = (data.get("host") if "host" in data else current["host"]) or "" + host = validate_printer_host(str(host_raw).strip()) family = data.get("family") if "family" in data else current["family"] family = (family or "").strip() or family_from_name(name) if not name: raise GrottiError("Der Drucker braucht einen Namen.", "name") - if not host: - raise GrottiError("Der Drucker braucht eine IP oder einen Host.", "host") if family not in PRINTER_FAMILIES: raise GrottiError("Unbekannte Druckerfamilie.", "family") other = store.printer_by_host(conn(), host) @@ -957,6 +1022,15 @@ def create_app(cfg): upload = request.files.get("file") if not upload or not upload.filename: raise GrottiError("Keine Datei angehängt.", "datei") + low = upload.filename.lower().replace("\\", "/") + base = low.rsplit("/", 1)[-1] + if base in (".", "..") or not base.strip("."): + raise GrottiError("Ungültiger Dateiname.", "datei") + if not any(base.endswith(suf) for suf in UPLOAD_SUFFIXES): + raise GrottiError( + "Nur .gcode, .3mf oder .gcode.3mf sind erlaubt.", + "datei", + ) project_id = optional_id(request.form.get("project_id")) saved = library.save_upload(conn(), upload.filename, upload.read(), cfg, project_id) return jsonify(saved), 201