654 lines
27 KiB
Python
654 lines
27 KiB
Python
"""Service d'impression — file d'attente SQLite + appel script_print.sh."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import re
|
|
import subprocess
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Literal
|
|
|
|
import aiosqlite
|
|
|
|
from backend.services.config_service import Config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PrintStatus = Literal["pending", "printing", "done", "cancelled", "error"]
|
|
|
|
CREATE_SQL = """
|
|
CREATE TABLE IF NOT EXISTS print_queue (
|
|
id TEXT PRIMARY KEY,
|
|
photo_id TEXT,
|
|
filename TEXT NOT NULL,
|
|
thumb_url TEXT,
|
|
copies INTEGER DEFAULT 1,
|
|
status TEXT DEFAULT 'pending',
|
|
printer TEXT,
|
|
requested_at REAL,
|
|
processed_at REAL,
|
|
error_msg TEXT
|
|
);
|
|
"""
|
|
|
|
|
|
class PrinterService:
|
|
def __init__(self, config: Config):
|
|
self._cfg = config.print
|
|
self._db_path: Path | None = None
|
|
self._db: aiosqlite.Connection | None = None
|
|
self._lock = asyncio.Lock()
|
|
self._led = None # injecté après démarrage via set_led_service()
|
|
|
|
def set_led_service(self, led_svc) -> None:
|
|
"""Injecte le service LED pour que execute_print puisse mettre à jour les LEDs."""
|
|
self._led = led_svc
|
|
|
|
async def init_db(self, db_path: Path):
|
|
db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._db_path = db_path
|
|
self._db = await aiosqlite.connect(str(db_path))
|
|
self._db.row_factory = aiosqlite.Row
|
|
await self._db.execute(CREATE_SQL)
|
|
# Migration : ajoute photo_id si la colonne n'existe pas encore
|
|
try:
|
|
await self._db.execute("ALTER TABLE print_queue ADD COLUMN photo_id TEXT")
|
|
logger.info("Migration : colonne photo_id ajoutée à print_queue")
|
|
except Exception:
|
|
pass # Colonne déjà présente
|
|
await self._db.commit()
|
|
logger.info("Base print_queue initialisée: %s", db_path)
|
|
|
|
async def close(self):
|
|
if self._db:
|
|
await self._db.close()
|
|
|
|
# ── File d'attente ────────────────────────────────────────────────────────
|
|
|
|
async def add_request(
|
|
self, filename: str, thumb_url: str = "", copies: int = 1, photo_id: str = ""
|
|
) -> dict:
|
|
"""Ajoute une demande d'impression dans la file. Retourne l'entrée créée."""
|
|
entry_id = str(uuid.uuid4())
|
|
now = time.time()
|
|
|
|
await self._db.execute(
|
|
"INSERT INTO print_queue (id, photo_id, filename, thumb_url, copies, status, requested_at) "
|
|
"VALUES (?, ?, ?, ?, ?, 'pending', ?)",
|
|
(entry_id, photo_id, filename, thumb_url, copies, now),
|
|
)
|
|
await self._db.commit()
|
|
|
|
entry = {
|
|
"id": entry_id,
|
|
"photo_id": photo_id,
|
|
"filename": filename,
|
|
"thumb_url": thumb_url,
|
|
"copies": copies,
|
|
"status": "pending",
|
|
"requested_at": now,
|
|
}
|
|
|
|
# Mode direct : impression immédiate sans validation
|
|
if self._cfg.mode == "direct":
|
|
asyncio.create_task(self.execute_print(entry_id, copies))
|
|
|
|
logger.info("Demande d'impression ajoutée: %s (%s)", entry_id, filename)
|
|
return entry
|
|
|
|
async def get_queue(self, status: str | None = None) -> list[dict]:
|
|
"""Liste les entrées de la file d'attente."""
|
|
if status:
|
|
cursor = await self._db.execute(
|
|
"SELECT * FROM print_queue WHERE status = ? ORDER BY requested_at DESC",
|
|
(status,),
|
|
)
|
|
else:
|
|
cursor = await self._db.execute(
|
|
"SELECT * FROM print_queue ORDER BY requested_at DESC LIMIT 100"
|
|
)
|
|
rows = await cursor.fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
async def get_pending(self) -> list[dict]:
|
|
return await self.get_queue("pending")
|
|
|
|
async def get_pending_by_photo_id(self) -> dict[str, list]:
|
|
"""Retourne {photo_id: [entries actives]}. Conservé pour compatibilité."""
|
|
stats = await self.get_print_stats_by_photo_id()
|
|
return {pid: info["requests"] for pid, info in stats.items() if info["requests"]}
|
|
|
|
async def get_print_stats_by_photo_id(self) -> dict[str, dict]:
|
|
"""Stats complètes d'impression par photo_id.
|
|
|
|
Retourne {photo_id: {requests, pending, printing, done, copies_done}}.
|
|
'requests' contient uniquement les entrées pending/printing (pour la lightbox).
|
|
'done' et 'copies_done' comptent les impressions réussies.
|
|
"""
|
|
async with self._db.execute(
|
|
"SELECT id, photo_id, filename, status, copies, requested_at, thumb_url "
|
|
"FROM print_queue WHERE status IN ('pending','printing','done') "
|
|
"ORDER BY requested_at"
|
|
) as cur:
|
|
rows = await cur.fetchall()
|
|
|
|
result: dict[str, dict] = {}
|
|
for r in rows:
|
|
key = r["photo_id"] or Path(r["filename"]).stem
|
|
if key not in result:
|
|
result[key] = {
|
|
"requests": [],
|
|
"pending": 0,
|
|
"printing": 0,
|
|
"done": 0,
|
|
"copies_done": 0,
|
|
}
|
|
if r["status"] in ("pending", "printing"):
|
|
result[key]["requests"].append(dict(r))
|
|
if r["status"] == "pending":
|
|
result[key]["pending"] += 1
|
|
elif r["status"] == "printing":
|
|
result[key]["printing"] += 1
|
|
elif r["status"] == "done":
|
|
result[key]["done"] += 1
|
|
result[key]["copies_done"] += r["copies"] or 1
|
|
return result
|
|
|
|
async def cancel_by_photo_id(self, photo_id: str) -> int:
|
|
"""Annule toutes les demandes pending pour un photo_id (UUID) donné."""
|
|
pending = await self.get_queue("pending")
|
|
cancelled = 0
|
|
for r in pending:
|
|
key = r.get("photo_id") or Path(r["filename"]).stem
|
|
if key == photo_id:
|
|
ok = await self.cancel(r["id"])
|
|
if ok:
|
|
cancelled += 1
|
|
return cancelled
|
|
|
|
async def cancel(self, entry_id: str) -> bool:
|
|
async with self._lock:
|
|
cursor = await self._db.execute(
|
|
"SELECT status FROM print_queue WHERE id = ?", (entry_id,)
|
|
)
|
|
row = await cursor.fetchone()
|
|
if not row or row["status"] not in ("pending",):
|
|
return False
|
|
await self._db.execute(
|
|
"UPDATE print_queue SET status='cancelled', processed_at=? WHERE id=?",
|
|
(time.time(), entry_id),
|
|
)
|
|
await self._db.commit()
|
|
return True
|
|
|
|
# ── Impression ────────────────────────────────────────────────────────────
|
|
|
|
async def execute_print(self, entry_id: str, copies: int = 1, printer: str = "") -> dict:
|
|
"""Lance l'impression via script_print.sh."""
|
|
async with self._lock:
|
|
cursor = await self._db.execute(
|
|
"SELECT * FROM print_queue WHERE id = ?", (entry_id,)
|
|
)
|
|
entry = await cursor.fetchone()
|
|
if not entry:
|
|
return {"success": False, "error": "Entrée introuvable"}
|
|
if entry["status"] not in ("pending",):
|
|
return {"success": False, "error": f"Statut incompatible: {entry['status']}"}
|
|
|
|
await self._db.execute(
|
|
"UPDATE print_queue SET status='printing', copies=? WHERE id=?",
|
|
(copies, entry_id),
|
|
)
|
|
await self._db.commit()
|
|
|
|
filename = entry["filename"]
|
|
script = self._cfg.script_path
|
|
|
|
# Appel async du script d'impression
|
|
try:
|
|
result = await asyncio.to_thread(
|
|
self._run_print_script, script, filename, copies
|
|
)
|
|
except Exception as e:
|
|
result = {"success": False, "error": str(e), "printer": ""}
|
|
|
|
now = time.time()
|
|
if result["success"]:
|
|
await self._db.execute(
|
|
"UPDATE print_queue SET status='done', printer=?, processed_at=? WHERE id=?",
|
|
(result.get("printer", ""), now, entry_id),
|
|
)
|
|
else:
|
|
await self._db.execute(
|
|
"UPDATE print_queue SET status='error', error_msg=?, processed_at=? WHERE id=?",
|
|
(result.get("error", ""), now, entry_id),
|
|
)
|
|
await self._db.commit()
|
|
|
|
logger.info("Impression %s: %s", entry_id, "OK" if result["success"] else result.get("error"))
|
|
|
|
# Mise à jour LED après impression (utile surtout en mode direct / background task)
|
|
if self._led:
|
|
effect = "finished" if result["success"] else "error"
|
|
self._led.play(effect)
|
|
|
|
return result
|
|
|
|
def _run_print_script(self, script: str, filename: str, copies: int) -> dict:
|
|
"""Appelle script_print.sh de façon synchrone."""
|
|
if not Path(script).exists():
|
|
return {"success": False, "error": f"Script introuvable: {script}"}
|
|
if not Path(filename).exists():
|
|
return {"success": False, "error": f"Fichier introuvable: {filename}"}
|
|
|
|
# Imprimantes exclues du load-balancing (flag excluded dans config)
|
|
excluded_list = ",".join(
|
|
self._printer_name(p)
|
|
for p in self._cfg.printers
|
|
if isinstance(p, dict) and p.get("excluded", False)
|
|
)
|
|
cmd = ["/bin/bash", script, filename, "image", "default", str(copies), excluded_list]
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd, capture_output=True, text=True, timeout=120
|
|
)
|
|
stdout = proc.stdout.strip()
|
|
# Cherche PRINTED: ou PRINT_ERROR: dans la dernière ligne significative
|
|
for line in reversed(stdout.splitlines()):
|
|
line = line.strip()
|
|
if line.startswith("PRINTED:"):
|
|
parts = line.split(":")
|
|
printer = parts[1] if len(parts) > 1 else ""
|
|
return {"success": True, "printer": printer, "output": stdout}
|
|
if line.startswith("PRINT_ERROR:"):
|
|
parts = line.split(":", 2)
|
|
printer = parts[1] if len(parts) > 1 else ""
|
|
reason = parts[2] if len(parts) > 2 else ""
|
|
err_msg = f"Erreur imprimante {printer}: {reason}" if reason else f"Erreur imprimante {printer}"
|
|
return {"success": False, "error": err_msg, "printer": printer}
|
|
# Aucun marqueur reconnu
|
|
return {"success": False, "error": proc.stderr.strip() or stdout or "Script sans sortie reconnue", "printer": ""}
|
|
except subprocess.TimeoutExpired:
|
|
return {"success": False, "error": "Timeout impression (120s)"}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
# ── IPP direct ───────────────────────────────────────────────────────────
|
|
|
|
def _get_printer_ip(self, printer_name: str) -> str:
|
|
"""Extrait l'IP de l'imprimante depuis l'URI CUPS (lpstat -v)."""
|
|
try:
|
|
r = subprocess.run(
|
|
["lpstat", "-v", printer_name],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
m = re.search(r"(\d+\.\d+\.\d+\.\d+)", r.stdout)
|
|
return m.group(1) if m else ""
|
|
except Exception:
|
|
return ""
|
|
|
|
def _query_ipp_direct(self, printer_ip: str) -> dict:
|
|
"""Interroge directement le serveur IPP natif de l'imprimante (sans auth).
|
|
|
|
Retourne un dict avec keys: state, state_reasons, accepting, markers.
|
|
"""
|
|
ipp_test = "/usr/share/cups/ipptool/get-printer-attributes.test"
|
|
empty: dict = {"state": "", "state_reasons": [], "accepting": None, "markers": []}
|
|
if not printer_ip:
|
|
return empty
|
|
try:
|
|
r = subprocess.run(
|
|
["ipptool", "-tv", f"ipp://{printer_ip}:631/ipp/print", ipp_test],
|
|
capture_output=True, text=True, timeout=8
|
|
)
|
|
out = r.stdout
|
|
except Exception as e:
|
|
logger.debug("ipptool direct échec %s: %s", printer_ip, e)
|
|
return empty
|
|
|
|
result: dict = {"state": "", "state_reasons": [], "accepting": None, "markers": []}
|
|
|
|
# printer-state
|
|
m = re.search(r"printer-state \(enum\)\s*=\s*(\S+)", out)
|
|
if m:
|
|
result["state"] = m.group(1).lower() # idle / processing / stopped
|
|
|
|
# printer-state-reasons
|
|
m = re.search(r"printer-state-reasons \([^)]+\)\s*=\s*(.+)", out)
|
|
if m:
|
|
reasons = [r.strip() for r in m.group(1).split(",")]
|
|
result["state_reasons"] = [r for r in reasons if r and r != "none"]
|
|
|
|
# printer-is-accepting-jobs
|
|
m = re.search(r"printer-is-accepting-jobs \(boolean\)\s*=\s*(\S+)", out)
|
|
if m:
|
|
result["accepting"] = m.group(1).lower() == "true"
|
|
|
|
# marker-levels (encre/ruban)
|
|
names_m = re.search(r"marker-names \([^)]+\)\s*=\s*(.+)", out)
|
|
levels_m = re.search(r"marker-levels \([^)]+\)\s*=\s*(.+)", out)
|
|
colors_m = re.search(r"marker-colors \([^)]+\)\s*=\s*(.+)", out)
|
|
types_m = re.search(r"marker-types \([^)]+\)\s*=\s*(.+)", out)
|
|
if names_m and levels_m:
|
|
names = [n.strip().strip('"') for n in names_m.group(1).split(",")]
|
|
levels = [l.strip() for l in levels_m.group(1).split(",")]
|
|
colors = [c.strip().strip('"') for c in colors_m.group(1).split(",")] if colors_m else []
|
|
types = [t.strip().strip('"') for t in types_m.group(1).split(",")] if types_m else []
|
|
for i, name in enumerate(names):
|
|
try:
|
|
lvl = int(levels[i]) if i < len(levels) else -1
|
|
except ValueError:
|
|
lvl = -1
|
|
result["markers"].append({
|
|
"name": name,
|
|
"level": lvl,
|
|
"color": colors[i] if i < len(colors) else "",
|
|
"type": types[i] if i < len(types) else "",
|
|
})
|
|
|
|
return result
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────────
|
|
|
|
def _printer_name(self, p) -> str:
|
|
"""Accepte une entrée printers qui soit un str ou un dict."""
|
|
return p["name"] if isinstance(p, dict) else str(p)
|
|
|
|
def _printer_label(self, p) -> str:
|
|
return p.get("label", self._printer_name(p)) if isinstance(p, dict) else str(p)
|
|
|
|
# ── Statut imprimantes CUPS ───────────────────────────────────────────────
|
|
|
|
async def get_printers_status(self) -> list[dict]:
|
|
"""Retourne le statut détaillé des imprimantes CUPS configurées."""
|
|
statuses = []
|
|
for p in self._cfg.printers:
|
|
name = self._printer_name(p)
|
|
excluded = isinstance(p, dict) and bool(p.get("excluded", False))
|
|
status = await asyncio.to_thread(self._get_printer_status, name)
|
|
stats = await self._get_printer_db_stats(name)
|
|
statuses.append({
|
|
"name": name,
|
|
"label": self._printer_label(p),
|
|
"excluded": excluded,
|
|
**status,
|
|
**stats,
|
|
})
|
|
return statuses
|
|
|
|
async def _get_printer_db_stats(self, printer_name: str) -> dict:
|
|
"""Retourne les stats de la file SQLite pour une imprimante donnée."""
|
|
try:
|
|
cursor = await self._db.execute(
|
|
"SELECT status, COUNT(*) as cnt FROM print_queue WHERE printer = ? GROUP BY status",
|
|
(printer_name,)
|
|
)
|
|
rows = await cursor.fetchall()
|
|
counts = {r["status"]: r["cnt"] for r in rows}
|
|
# Aussi compter les jobs sans printer assigné (mode direct)
|
|
cursor2 = await self._db.execute(
|
|
"SELECT COUNT(*) as total FROM print_queue WHERE status='done'"
|
|
)
|
|
r2 = await cursor2.fetchone()
|
|
return {
|
|
"stats_done": counts.get("done", 0),
|
|
"stats_error": counts.get("error", 0),
|
|
"stats_cancelled": counts.get("cancelled", 0),
|
|
}
|
|
except Exception:
|
|
return {"stats_done": 0, "stats_error": 0, "stats_cancelled": 0}
|
|
|
|
def _get_printer_status(self, printer_name: str) -> dict:
|
|
"""Statut complet : CUPS pour les jobs, IPP direct pour état/erreurs/encre."""
|
|
try:
|
|
# ── Etat de base via lpstat ──
|
|
r_state = subprocess.run(
|
|
["lpstat", "-p", printer_name],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
if r_state.returncode != 0 or "not found" in (r_state.stderr or "").lower():
|
|
return {"state": "offline", "accepting": False, "jobs": [], "jobs_count": 0}
|
|
|
|
out = r_state.stdout.lower()
|
|
if "idle" in out:
|
|
state = "idle"
|
|
elif "printing" in out or "processing" in out:
|
|
state = "printing"
|
|
elif "stopped" in out or "disabled" in out:
|
|
state = "disabled"
|
|
else:
|
|
state = "unknown"
|
|
|
|
# Est-ce que l'imprimante accepte les nouveaux jobs (CUPS) ?
|
|
r_accept = subprocess.run(
|
|
["lpstat", "-a", printer_name],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
accepting = "accepting" in r_accept.stdout.lower()
|
|
|
|
# Jobs CUPS en cours
|
|
jobs = self._get_cups_jobs(printer_name)
|
|
|
|
# URI du périphérique
|
|
uri = ""
|
|
printer_ip = ""
|
|
try:
|
|
r_uri = subprocess.run(
|
|
["lpstat", "-v", printer_name],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
for line in r_uri.stdout.splitlines():
|
|
if "device for" in line.lower():
|
|
parts = line.split(":", 2)
|
|
uri = (parts[1].strip() + ":" + parts[2].strip()) if len(parts) > 2 else ""
|
|
break
|
|
m = re.search(r"(\d+\.\d+\.\d+\.\d+)", r_uri.stdout)
|
|
printer_ip = m.group(1) if m else ""
|
|
except Exception:
|
|
pass
|
|
|
|
# Modèle / location via lpstat -l -p
|
|
model = ""
|
|
location = ""
|
|
reasons: list[str] = []
|
|
try:
|
|
r_info = subprocess.run(
|
|
["lpstat", "-l", "-p", printer_name],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
for line in r_info.stdout.splitlines():
|
|
l = line.strip()
|
|
if l.startswith("Description:"):
|
|
model = l.split(":", 1)[1].strip()
|
|
elif l.startswith("Location:"):
|
|
location = l.split(":", 1)[1].strip()
|
|
elif l.lower().startswith("reason:"):
|
|
raw = l.split(":", 1)[1].strip()
|
|
if raw and raw.lower() not in ("none", ""):
|
|
reasons.append(self._REASON_LABELS.get(raw, raw))
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Sondage IPP direct sur l'imprimante (source de vérité) ──
|
|
markers: list[dict] = []
|
|
if printer_ip:
|
|
ipp = self._query_ipp_direct(printer_ip)
|
|
if ipp["state"]:
|
|
# Mappe les états IPP → notre convention
|
|
ipp_state_map = {
|
|
"idle": "idle",
|
|
"processing": "printing",
|
|
"stopped": "disabled",
|
|
}
|
|
state = ipp_state_map.get(ipp["state"], ipp["state"])
|
|
if ipp["accepting"] is not None:
|
|
accepting = ipp["accepting"]
|
|
if ipp["state_reasons"]:
|
|
reasons = [
|
|
self._REASON_LABELS.get(r, r)
|
|
for r in ipp["state_reasons"]
|
|
]
|
|
if ipp["markers"]:
|
|
markers = ipp["markers"]
|
|
else:
|
|
markers = self._get_marker_levels(printer_name)
|
|
|
|
return {
|
|
"state": state,
|
|
"accepting": accepting,
|
|
"jobs": jobs,
|
|
"jobs_count": len(jobs),
|
|
"uri": uri,
|
|
"model": model,
|
|
"location": location,
|
|
"reasons": reasons,
|
|
"markers": markers,
|
|
}
|
|
except Exception as e:
|
|
return {"state": "error", "accepting": False, "jobs": [], "jobs_count": 0, "error": str(e)}
|
|
|
|
# Traduction des Reason CUPS en français
|
|
_REASON_LABELS: dict[str, str] = {
|
|
"input-tray-missing": "⚠️ Bac papier absent",
|
|
"media-empty": "❌ Plus de papier",
|
|
"media-low": "⚠️ Papier presque épuisé",
|
|
"media-needed": "⚠️ Papier requis",
|
|
"marker-supply-empty": "❌ Cartouche/ruban vide",
|
|
"marker-supply-low": "⚠️ Cartouche/ruban faible",
|
|
"marker-supply-low-warning": "⚠️ Ruban faible",
|
|
"cover-open": "❌ Capot ouvert",
|
|
"door-open": "❌ Porte ouverte",
|
|
"offline-report": "❌ Imprimante hors ligne",
|
|
"connecting-to-device": "🔄 Connexion en cours…",
|
|
"toner-empty": "❌ Toner vide",
|
|
"toner-low": "⚠️ Toner faible",
|
|
"output-tray-missing": "⚠️ Bac de sortie absent",
|
|
"output-area-full": "⚠️ Bac de sortie plein",
|
|
"paused": "⏸ Imprimante en pause",
|
|
}
|
|
|
|
def _get_marker_levels(self, printer_name: str) -> list[dict]:
|
|
"""Récupère les niveaux d'encre/ruban via lpstat ou ipptool."""
|
|
markers = []
|
|
try:
|
|
# Essai via ipptool si disponible
|
|
r = subprocess.run(
|
|
["ipptool", "-tv",
|
|
f"ipp://localhost:631/printers/{printer_name}",
|
|
"/usr/share/cups/ipptool/get-printer-attributes.test"],
|
|
capture_output=True, text=True, timeout=8
|
|
)
|
|
lines = r.stdout.splitlines()
|
|
names, levels, colors, types_ = [], [], [], []
|
|
for line in lines:
|
|
l = line.strip()
|
|
if "marker-names" in l:
|
|
names = [x.strip().strip('"') for x in l.split("=", 1)[-1].split(",") if x.strip()]
|
|
elif "marker-levels" in l:
|
|
levels = [x.strip() for x in l.split("=", 1)[-1].split(",") if x.strip()]
|
|
elif "marker-colors" in l:
|
|
colors = [x.strip().strip('"') for x in l.split("=", 1)[-1].split(",") if x.strip()]
|
|
elif "marker-types" in l:
|
|
types_ = [x.strip().strip('"') for x in l.split("=", 1)[-1].split(",") if x.strip()]
|
|
for i, name in enumerate(names):
|
|
level = int(levels[i]) if i < len(levels) else -1
|
|
markers.append({
|
|
"name": name,
|
|
"level": level,
|
|
"color": colors[i] if i < len(colors) else "",
|
|
"type": types_[i] if i < len(types_) else "",
|
|
})
|
|
except Exception:
|
|
pass
|
|
return markers
|
|
|
|
def _get_cups_jobs(self, printer_name: str) -> list[dict]:
|
|
"""Retourne la liste des jobs CUPS en cours pour une imprimante."""
|
|
try:
|
|
r = subprocess.run(
|
|
["lpstat", "-o", printer_name, "-l"],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
jobs = []
|
|
current = {}
|
|
for line in r.stdout.splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
if line.startswith(printer_name):
|
|
if current:
|
|
jobs.append(current)
|
|
parts = line.split()
|
|
current = {
|
|
"id": parts[0] if parts else "",
|
|
"user": parts[1] if len(parts) > 1 else "",
|
|
"size": parts[3] if len(parts) > 3 else "",
|
|
"date": " ".join(parts[4:7]) if len(parts) > 6 else "",
|
|
"details": [],
|
|
}
|
|
elif current and ":" in line:
|
|
current.setdefault("details", []).append(line)
|
|
if current:
|
|
jobs.append(current)
|
|
return jobs
|
|
except Exception:
|
|
return []
|
|
|
|
async def get_cups_jobs(self, printer_name: str) -> list[dict]:
|
|
return await asyncio.to_thread(self._get_cups_jobs, printer_name)
|
|
|
|
async def cancel_cups_jobs(self, printer_name: str) -> bool:
|
|
"""Annule tous les jobs CUPS pour une imprimante."""
|
|
try:
|
|
result = subprocess.run(
|
|
["cancel", "-a", printer_name],
|
|
capture_output=True, text=True, timeout=10
|
|
)
|
|
return result.returncode == 0
|
|
except Exception as e:
|
|
logger.error("Erreur cancel CUPS: %s", e)
|
|
return False
|
|
|
|
async def cancel_cups_job(self, job_id: str) -> bool:
|
|
"""Annule un job CUPS précis."""
|
|
try:
|
|
result = subprocess.run(
|
|
["cancel", job_id],
|
|
capture_output=True, text=True, timeout=10
|
|
)
|
|
return result.returncode == 0
|
|
except Exception as e:
|
|
logger.error("Erreur cancel CUPS job %s: %s", job_id, e)
|
|
return False
|
|
|
|
async def enable_printer(self, printer_name: str) -> bool:
|
|
"""Active l'imprimante CUPS (cupsenable) et accepte les nouveaux jobs."""
|
|
try:
|
|
r1 = subprocess.run(["sudo", "cupsenable", printer_name], capture_output=True, timeout=10)
|
|
r2 = subprocess.run(["sudo", "cupsaccept", printer_name], capture_output=True, timeout=10)
|
|
return r1.returncode == 0 and r2.returncode == 0
|
|
except Exception as e:
|
|
logger.error("Erreur enable printer %s: %s", printer_name, e)
|
|
return False
|
|
|
|
async def disable_printer(self, printer_name: str) -> bool:
|
|
"""Désactive l'imprimante CUPS (cupsdisable)."""
|
|
try:
|
|
result = subprocess.run(["sudo", "cupsdisable", printer_name], capture_output=True, timeout=10)
|
|
return result.returncode == 0
|
|
except Exception as e:
|
|
logger.error("Erreur disable printer %s: %s", printer_name, e)
|
|
return False
|
|
|
|
async def reject_jobs(self, printer_name: str) -> bool:
|
|
"""Refuse les nouveaux jobs (cupsreject) sans stopper l'impression en cours."""
|
|
try:
|
|
result = subprocess.run(["sudo", "cupsreject", printer_name], capture_output=True, timeout=10)
|
|
return result.returncode == 0
|
|
except Exception as e:
|
|
logger.error("Erreur reject printer %s: %s", printer_name, e)
|
|
return False
|