michael/P-Touch-Web
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add persistent print queue

Browse Source
This commit is contained in:
Mikei386
2026-06-22 14:56:20 +02:00
parent 7ce97443b1
commit 34f0161350
8 changed files with 170 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/app.py
+124 -9
View File
@@ -2,12 +2,14 @@ from __future__ import annotations
import base64
import io
import json
import signal
import os
import re
import shlex
import subprocess
import threading
import time
import uuid
from pathlib import Path
from typing import Any
@@ -31,10 +33,14 @@ except Exception: # pragma: no cover - optional at runtime until installed
BASE_DIR = Path(__file__).resolve().parent.parent
LABEL_DIR = Path(os.environ.get("LABEL_DIR", BASE_DIR / "labels"))
LABEL_DIR.mkdir(parents=True, exist_ok=True)
QUEUE_DIR = LABEL_DIR / "queue"
QUEUE_DIR.mkdir(parents=True, exist_ok=True)
DEFAULT_PRINT_COMMAND = "ptouch-print --image {image}"
PRINT_COMMAND = os.environ.get("PRINT_COMMAND", DEFAULT_PRINT_COMMAND)
PRINT_ENABLED = os.environ.get("PRINT_ENABLED", "1").lower() in {"1", "true", "yes", "on"}
PRINT_QUEUE_ENABLED = os.environ.get("PRINT_QUEUE_ENABLED", "1").lower() in {"1", "true", "yes", "on"}
PRINT_QUEUE_RETRY_SECONDS = int(os.environ.get("PRINT_QUEUE_RETRY_SECONDS", "30"))
DPI = int(os.environ.get("LABEL_DPI", "180"))
PTOUCH_SUPPORTS_COPIES: bool | None = None
@@ -51,6 +57,7 @@ PRINTABLE_TAPE_PX = {
app = Flask(__name__)
app.secret_key = os.environ.get("SECRET_KEY", "change-me")
PRINT_LOCK = threading.Lock()
QUEUE_WAKE = threading.Event()
def mm_to_px(mm: float) -> int:
@@ -369,6 +376,89 @@ def print_label(image_path: Path, copies: int, design: dict[str, Any]) -> list[s
return [run_print_command(image_path, copies, design)]
def run_print_job(image_path: Path, copies: int, design: dict[str, Any]) -> None:
if not PRINT_LOCK.acquire(blocking=False):
raise RuntimeError("Es laeuft bereits ein Druckjob. Bitte warten oder haengenden ptouch-print-Prozess beenden.")
try:
results = print_label(image_path, copies, design)
failed = next((result for result in results if result.returncode != 0), None)
if failed:
raise RuntimeError(format_print_error(failed))
finally:
PRINT_LOCK.release()
def queued_job_path(job_id: str) -> Path:
return QUEUE_DIR / f"{job_id}.json"
def enqueue_print_job(image_path: Path, copies: int, design: dict[str, Any]) -> dict[str, Any]:
now = time.time()
job = {
"id": uuid.uuid4().hex,
"image": image_path.name,
"copies": copies,
"design": design,
"createdAt": now,
"nextAttemptAt": now,
"attempts": 0,
"lastError": "",
}
target = queued_job_path(job["id"])
temp = target.with_suffix(".tmp")
temp.write_text(json.dumps(job, ensure_ascii=True, indent=2), encoding="utf-8")
temp.replace(target)
QUEUE_WAKE.set()
return job
def load_queue_jobs() -> list[dict[str, Any]]:
jobs: list[dict[str, Any]] = []
for path in sorted(QUEUE_DIR.glob("*.json")):
try:
job = json.loads(path.read_text(encoding="utf-8"))
if isinstance(job, dict):
jobs.append(job)
except Exception:
continue
return sorted(jobs, key=lambda item: float(item.get("createdAt", 0)))
def save_queue_job(job: dict[str, Any]) -> None:
queued_job_path(str(job["id"])).write_text(json.dumps(job, ensure_ascii=True, indent=2), encoding="utf-8")
def process_queue_once() -> None:
if not PRINT_ENABLED:
return
now = time.time()
for job in load_queue_jobs():
if float(job.get("nextAttemptAt", 0)) > now:
continue
image_path = LABEL_DIR / str(job.get("image", ""))
try:
if not image_path.exists():
raise RuntimeError(f"Queue-Bild fehlt: {image_path.name}")
run_print_job(image_path, int(job.get("copies", 1)), job.get("design") if isinstance(job.get("design"), dict) else {})
queued_job_path(str(job["id"])).unlink(missing_ok=True)
except Exception as exc:
job["attempts"] = int(job.get("attempts", 0)) + 1
job["lastError"] = str(exc)
job["nextAttemptAt"] = time.time() + PRINT_QUEUE_RETRY_SECONDS
save_queue_job(job)
break
def print_queue_worker() -> None:
while True:
try:
process_queue_once()
except Exception:
pass
QUEUE_WAKE.wait(PRINT_QUEUE_RETRY_SECONDS)
QUEUE_WAKE.clear()
def should_render_copies_as_strip(design: dict[str, Any], copies: int) -> bool:
if copies <= 1:
return False
@@ -569,15 +659,12 @@ def api_print():
print_design = dict(design)
print_design["cutMode"] = "cutmark" if design.get("cutMode") == "cutmark-chain" else "auto"
if PRINT_ENABLED:
if not PRINT_LOCK.acquire(blocking=False):
raise RuntimeError("Es laeuft bereits ein Druckjob. Bitte warten oder haengenden ptouch-print-Prozess beenden.")
try:
results = print_label(image_path, print_copies, print_design)
failed = next((result for result in results if result.returncode != 0), None)
if failed:
raise RuntimeError(format_print_error(failed))
finally:
PRINT_LOCK.release()
if PRINT_QUEUE_ENABLED:
job = enqueue_print_job(image_path, print_copies, print_design)
queue_len = len(load_queue_jobs())
message = f"Etikett wurde in die Druckwarteschlange gelegt. Position: {queue_len}."
return jsonify({"ok": True, "message": message, "image": f"/labels/{image_path.name}", "queued": True, "jobId": job["id"], "queueLength": queue_len})
run_print_job(image_path, print_copies, print_design)
message = "Etikett wurde an den Drucker gesendet."
else:
message = f"Testmodus: Etikett erstellt, nicht gedruckt: {image_path.name}"
@@ -597,6 +684,34 @@ def api_tape_info():
return jsonify({"ok": False, "error": str(exc)}), 400
@app.get("/api/queue")
def api_queue():
jobs = load_queue_jobs()
return jsonify(
{
"ok": True,
"enabled": PRINT_QUEUE_ENABLED,
"length": len(jobs),
"jobs": [
{
"id": job.get("id"),
"image": job.get("image"),
"copies": job.get("copies"),
"attempts": job.get("attempts", 0),
"lastError": job.get("lastError", ""),
"createdAt": job.get("createdAt"),
"nextAttemptAt": job.get("nextAttemptAt"),
}
for job in jobs
],
}
)
@app.get("/labels/<path:filename>")
def labels(filename: str):
return send_from_directory(LABEL_DIR, filename)
if PRINT_QUEUE_ENABLED:
threading.Thread(target=print_queue_worker, name="ptouch-print-queue", daemon=True).start()