diff --git a/backend/services/printer_service.py b/backend/services/printer_service.py index f5f280f..58d95f3 100644 --- a/backend/services/printer_service.py +++ b/backend/services/printer_service.py @@ -232,15 +232,39 @@ class PrinterService: for p in self._cfg.printers: name = self._printer_name(p) 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), **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 CUPS complet pour une imprimante (état, jobs, accepting).""" + """Statut CUPS complet pour une imprimante (état, jobs, accepting, uri, modèle).""" try: # État de l'imprimante r_state = subprocess.run( @@ -271,15 +295,117 @@ class PrinterService: # Liste des jobs CUPS en cours jobs = self._get_cups_jobs(printer_name) + # URI du périphérique (adresse IP / protocole) + uri = "" + 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 + except Exception: + pass + + # Modèle, localisation et raisons d'erreur (lpstat -l -p) + model = "" + location = "" + reasons: list[str] = [] + state_message = "" + 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", ""): + label = self._REASON_LABELS.get(raw, raw) + reasons.append(label) + except Exception: + pass + + # Niveaux d'encre/ruban via attributs IPP (localhost CUPS) + 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: diff --git a/frontend/templates/admin/print.html b/frontend/templates/admin/print.html index 639b3f5..7cfc59d 100644 --- a/frontend/templates/admin/print.html +++ b/frontend/templates/admin/print.html @@ -143,6 +143,53 @@ {% endif %} + +
+ {% if p.uri %}🔌 {{ p.uri }}{% endif %} + {% if p.model %}🖨 {{ p.model }}{% endif %} + {% if p.location %}📍 {{ p.location }}{% endif %} +
+ + + {% if p.reasons %} +
+ {% for reason in p.reasons %} +
+ {{ reason }} +
+ {% endfor %} +
+ {% endif %} + + + {% if p.markers %} +
+ {% for m in p.markers %} + {% set lvl = m.level | int %} +
+
+ 🎨 {{ m.name }} + + {% if lvl < 0 %}N/A{% else %}{{ lvl }}%{% endif %} + +
+ {% if lvl >= 0 %} +
+
+
+ {% endif %} +
+ {% endfor %} +
+ {% endif %} + + +
+ ✅ {{ p.stats_done or 0 }} imprimé{{ 's' if (p.stats_done or 0) > 1 else '' }} + {% if p.stats_error %}❌ {{ p.stats_error }} erreur{{ 's' if p.stats_error > 1 else '' }}{% endif %} + {% if p.stats_cancelled %}✕ {{ p.stats_cancelled }} annulé{{ 's' if p.stats_cancelled > 1 else '' }}{% endif %} +
+
{% if p.state == 'disabled' %}