#!/usr/bin/env python3 PK۸k] Vuu __main__.py"""Agente di stampa per una stampante collegata al PC (USB o di rete). Gira sul PC del locale, chiede al server le comande da stampare e le manda alla stampante. Usa solo Python standard; per le stampanti installate in Windows serve anche: pip install pywin32 Per provare senza stampante termica c'è il metodo "A4": stampa le comande su una normale stampante da ufficio (scontrino a grandezza reale da 80 mm, con grassetti e caratteri grandi; pywin32 facoltativo). Uso: python agente_stampa.py (al primo avvio chiede il codice di abbinamento di 6 cifre) python agente_stampa.py --abbina (collega di nuovo questo PC con un nuovo codice) python agente_stampa.py --abbina --server https://... (collega questo PC a un altro server, es. quello in cloud) python agente_stampa.py --config (vecchio metodo: impostazioni a mano, stampante per stampante) python agente_stampa.py --elenco (mostra le stampanti che l'agente trova) Con l'abbinamento non serve altro sul PC: le stampanti si scelgono dall'area admin, nella sezione Stampanti, tra quelle che l'agente ha trovato (installate in Windows o termiche in rete sulla porta 9100). """ import json, os, socket, sys, threading, time, urllib.error, urllib.request from concurrent.futures import ThreadPoolExecutor # Indirizzo del server già scritto dentro l'agente (nella versione .exe per i clienti). # Vuoto = lo chiede al primo abbinamento. SERVER_PREDEFINITO = "https://comandefacili.it" VERSIONE = "1.0" # Versione per i clienti: .exe per Windows (Nuitka o PyInstaller) oppure .pyz per Linux e Raspberry Pi. # Le impostazioni vanno in %APPDATA%\ComandeFacili (Windows) o in ~/.config/comandefacili (Linux), # così restano anche quando si sostituisce il programma con uno più nuovo. WINDOWS = os.name == "nt" PYZ = sys.argv[0].endswith(".pyz") CONGELATO = bool(getattr(sys, "frozen", False)) or "__compiled__" in globals() or PYZ if CONGELATO: if WINDOWS: CARTELLA = os.path.join(os.environ.get("APPDATA") or os.path.expanduser("~"), "ComandeFacili") else: CARTELLA = os.path.join(os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config"), "comandefacili") os.makedirs(CARTELLA, exist_ok=True) else: CARTELLA = os.path.dirname(os.path.abspath(__file__)) CFG = os.path.join(CARTELLA, "agente_config.json") NOME_AVVIO = "ComandeFacili Stampa" RUN_KEY = r"Software\Microsoft\Windows\CurrentVersion\Run" class Scollegato(Exception): """Il PC è stato scollegato dall'area admin: va abbinato di nuovo.""" def log(*a): print(time.strftime("%H:%M:%S"), *a, flush=True) def percorso_exe(): return os.path.abspath(sys.argv[0]) SERVIZIO = "comandefacili-stampa.service" def comando_avvio(): """Il comando che avvia questo programma (per l'avvio automatico).""" if PYZ: return [sys.executable, percorso_exe()] return [percorso_exe()] def _linux_servizio(attiva): """Linux: servizio di sistema dell'utente (systemd), parte anche senza schermo, ad esempio su un Raspberry Pi.""" import subprocess, shlex cart = os.path.join(os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config"), "systemd", "user") f = os.path.join(cart, SERVIZIO) if not attiva: subprocess.run(["systemctl", "--user", "disable", "--now", SERVIZIO], capture_output=True) try: os.remove(f) except OSError: pass return os.makedirs(cart, exist_ok=True) with open(f, "w", encoding="utf-8") as fh: fh.write("[Unit]\nDescription=Comande Facili - programma di stampa\nAfter=network-online.target\nWants=network-online.target\n\n" f"[Service]\nExecStart={' '.join(shlex.quote(x) for x in comando_avvio())}\nRestart=always\nRestartSec=15\n" "Environment=PYTHONUNBUFFERED=1\n\n[Install]\nWantedBy=default.target\n") subprocess.run(["systemctl", "--user", "daemon-reload"], check=True, capture_output=True) subprocess.run(["systemctl", "--user", "enable", SERVIZIO], check=True, capture_output=True) # senza "linger" il servizio partirebbe solo quando l'utente entra: così parte all'accensione subprocess.run(["loginctl", "enable-linger", os.environ.get("USER") or os.getlogin()], capture_output=True) def avvio_automatico(attiva): """Avvia il programma di stampa all'accensione del computer, per l'utente attuale.""" if not WINDOWS: return _linux_servizio(attiva) import winreg with winreg.OpenKey(winreg.HKEY_CURRENT_USER, RUN_KEY, 0, winreg.KEY_SET_VALUE) as k: if attiva: winreg.SetValueEx(k, NOME_AVVIO, 0, winreg.REG_SZ, f'cmd /c start "" /min "{percorso_exe()}"') else: try: winreg.DeleteValue(k, NOME_AVVIO) except FileNotFoundError: pass def avvio_automatico_attivo(): if not WINDOWS: import subprocess try: return subprocess.run(["systemctl", "--user", "is-enabled", SERVIZIO], capture_output=True, text=True).stdout.strip() == "enabled" except Exception: return False try: import winreg with winreg.OpenKey(winreg.HKEY_CURRENT_USER, RUN_KEY) as k: v = winreg.QueryValueEx(k, NOME_AVVIO)[0] return percorso_exe().lower() in v.lower() except Exception: return False def chiedi_avvio_automatico(): """Restituisce True se su Linux il programma ora gira come servizio (questa finestra può chiudersi).""" if not CONGELATO: return False r = input("Vuoi che il programma di stampa parta da solo quando accendi il computer? [S/n]: ").strip().lower() try: avvio_automatico(not r.startswith("n")) print(" Avvio automatico " + ("NON attivo." if r.startswith("n") else "attivo: non devi più aprirlo a mano.")) except Exception as e: print(f" Non sono riuscito a impostare l'avvio automatico ({e}).") return False if WINDOWS or r.startswith("n"): return False import subprocess subprocess.run(["systemctl", "--user", "restart", SERVIZIO], capture_output=True) print("\nFatto: il programma di stampa ora lavora in sottofondo e riparte da solo a ogni accensione.") print("Per vedere cosa sta facendo: journalctl --user -u comandefacili-stampa -f") return True _MUTEX = None def una_sola_copia(): """Evita due programmi di stampa aperti insieme sullo stesso PC (stamperebbero tutto due volte).""" global _MUTEX if not WINDOWS: import fcntl _MUTEX = open(os.path.join(CARTELLA, ".in_uso"), "w") try: fcntl.flock(_MUTEX, fcntl.LOCK_EX | fcntl.LOCK_NB); return True except OSError: return False import ctypes _MUTEX = ctypes.windll.kernel32.CreateMutexW(None, False, "Local\\ComandeFaciliStampa") return ctypes.windll.kernel32.GetLastError() != 183 # 183 = ERROR_ALREADY_EXISTS def titolo(t): if os.name == "nt": try: import ctypes; ctypes.windll.kernel32.SetConsoleTitleW(t) except Exception: pass def windows_printers(): import win32print return [p[2] for p in win32print.EnumPrinters(win32print.PRINTER_ENUM_LOCAL | win32print.PRINTER_ENUM_CONNECTIONS)] def ask_printer(n): print(f"\n--- Stampante {n} ---") c = {"codice": input("Codice della stampante (area admin, sezione Stampanti): ").strip()} print("Come è collegata?\n 1) termica in rete LAN/Wi-Fi (indirizzo IP, porta 9100)\n 2) termica installata in Windows (USB)" "\n 3) schermo (prova senza stampante)\n 4) stampante A4 normale (prova)") m = input("Scelta [1/2/3/4]: ").strip() if m in ("2", "4"): c["metodo"] = "windows" if m == "2" else "a4" try: for i, nm in enumerate(windows_printers(), 1): print(f" {i}) {nm}") except ImportError: print(" (per vedere l'elenco installa prima pywin32: pip install pywin32)") c["stampante"] = input("Nome esatto della stampante (o il numero dell'elenco" + ("; vuoto = stampante predefinita" if m == "4" else "") + "): ").strip() if c["stampante"].isdigit(): try: c["stampante"] = windows_printers()[int(c["stampante"]) - 1] except Exception: pass elif m == "3": c["metodo"] = "schermo" else: c["metodo"] = "rete" c["stampante"] = input("Indirizzo IP della stampante (es. 192.168.1.50 o 192.168.1.50:9100): ").strip() return c def setup(): print("\n--- Impostazione agente di stampa ---") cfg = {"server": input("Indirizzo del server (es. http://192.168.1.10:8000 o https://comande.miosito.it): ").strip().rstrip("/"), "stampanti": [ask_printer(1)]} while input("\nAggiungere un'altra stampante su questo PC? [s/N]: ").strip().lower().startswith("s"): cfg["stampanti"].append(ask_printer(len(cfg["stampanti"]) + 1)) with open(CFG, "w", encoding="utf-8") as f: json.dump(cfg, f, indent=2) print(f"Impostazioni salvate in {CFG}\n") return cfg def abbina(server=None): print("\n--- Collegamento di questo computer al programma Comande ---") if not server: server = SERVER_PREDEFINITO if not server: server = input("Indirizzo del server [invio = http://localhost:8000]: ").strip().rstrip("/") or "http://localhost:8000" print("Nell'area admin, sezione Stampanti, premi \"Collega il computer del locale\": compare un codice di 6 cifre.") while True: cod = input("Codice di 6 cifre: ").strip().replace(" ", "") try: r = json.loads(call(server + "/pc/abbina", json.dumps({"codice": cod, "nome": socket.gethostname()}).encode()).read()) break except urllib.error.HTTPError as e: try: msg = json.loads(e.read()).get("detail") except Exception: msg = None print(" ", msg if isinstance(msg, str) else f"Codice non accettato (errore {e.code}). Riprova.") except Exception as e: print(f" Server non raggiungibile ({e}). Controlla l'indirizzo e la connessione, poi riprova.") cfg = {"server": server, "pc_token": r["token"]} with open(CFG, "w", encoding="utf-8") as f: json.dump(cfg, f, indent=2) print(f"\nFatto! Questo computer è collegato al locale \"{r['locale']}\".") print("Ora torna nell'area admin: le stampanti trovate compariranno tra pochi secondi.\n") if chiedi_avvio_automatico(): raise SystemExit(0) print("\nLascia aperta questa finestra (puoi ridurla a icona): finché è aperta le comande vengono stampate.\n") return cfg def stampanti_cups(): """Linux: le stampanti installate nel sistema di stampa (CUPS).""" import subprocess for cmd in (["lpstat", "-e"], ["lpstat", "-a"]): try: r = subprocess.run(cmd, capture_output=True, text=True, timeout=15) nomi = [x.split()[0] for x in r.stdout.splitlines() if x.strip()] if r.returncode == 0: return nomi except Exception: pass return [] def stampanti_windows(): if not WINDOWS: return stampanti_cups() try: return windows_printers() except ImportError: pass import subprocess try: r = subprocess.run(["powershell", "-NoProfile", "-NonInteractive", "-Command", "Get-Printer | Select-Object -ExpandProperty Name"], capture_output=True, text=True, timeout=30, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) return [x.strip() for x in r.stdout.splitlines() if x.strip()] except Exception: return [] VIRTUALI = ("pdf", "xps", "onenote", "fax", "anydesk", "send to", "invia a") # non sono stampanti vere def ip_locali(): ips = set() try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM); s.connect(("10.255.255.255", 1)); ips.add(s.getsockname()[0]); s.close() except OSError: pass try: for ip in socket.gethostbyname_ex(socket.gethostname())[2]: ips.add(ip) except OSError: pass return [ip for ip in ips if not ip.startswith(("127.", "169.254."))] def cerca_in_rete(porta=9100): """Cerca le stampanti termiche in rete: dispositivi della rete locale con la porta 9100 aperta.""" def prova(ip): try: with socket.create_connection((ip, porta), timeout=0.5): return ip except OSError: return None reti = {ip.rsplit(".", 1)[0] for ip in ip_locali()} candidati = [f"{r}.{i}" for r in reti for i in range(1, 255)] with ThreadPoolExecutor(64) as ex: return sorted((ip for ip in ex.map(prova, candidati) if ip), key=lambda x: tuple(map(int, x.split(".")))) def cerca_stampanti(): out = [] for ip in cerca_in_rete(): out.append({"metodo": "rete", "stampante": f"{ip}:9100", "nome": f"Termica in rete {ip}"}) for n in stampanti_windows(): if any(v in n.lower() for v in VIRTUALI): continue out.append({"metodo": "windows", "stampante": n, "nome": f"{n} (termica)"}) out.append({"metodo": "a4", "stampante": n, "nome": f"{n} (foglio A4, prova)"}) out.append({"metodo": "schermo", "stampante": "", "nome": "Nessuna stampante: mostra sullo schermo (prova)"}) return out def has_gdi(): try: import win32ui # noqa: F401 return True except ImportError: return False def fmt_for(c): if c["metodo"] == "schermo": return "text" if c["metodo"] == "a4": return "json" return "" # Misure di uno scontrino su carta da 80 mm (font standard 12x24 punti a 203 dpi): le stampe di prova # su A4 riproducono la striscia di carta a grandezza reale, in alto a sinistra del foglio. PAPER_MM, PRINT_MM = 80.0, 72.0 # larghezza carta e area stampabile CHAR_W, CHAR_H, LINE_MM = 1.5, 3.0, 3.75 # carattere 1,5 x 3 mm, interlinea 3,75 mm PAGE_X, PAGE_Y = 15.0, 15.0 # margine del foglio A4 def a4_rows(doc): """Trasforma le righe dello scontrino in (x_mm, y_mm, testo, grassetto, molt_larghezza, molt_altezza), con coordinate relative all'angolo in alto a sinistra della carta. Restituisce anche la lunghezza in mm.""" left = (PAPER_MM - PRINT_MM) / 2 cols = doc.get("cols", 48) out, y = [], 3.0 for op in doc["ops"]: if op[0] == "f": y += LINE_MM * op[1]; continue _, t, bold, big, tall, al = op wm, hm = (2 if big else 1), (2 if (big or tall) else 1) tw = len(t) * CHAR_W * wm free = cols * CHAR_W - tw x = left + max(0.0, {"c": free / 2, "r": free}.get(al, 0.0)) out.append((x, y, t, bool(bold), wm, hm)) y += LINE_MM * hm return out, y + 6.0 def print_a4_gdi(name, doc): """Stampa su una stampante normale (con pywin32) lo scontrino a grandezza reale da 80 mm.""" import win32ui, win32con dc = win32ui.CreateDC() dc.CreatePrinterDC(name or None) dpx, dpy = dc.GetDeviceCaps(88) / 25.4, dc.GetDeviceCaps(90) / 25.4 # punti per mm (X, Y) offx, offy = dc.GetDeviceCaps(112), dc.GetDeviceCaps(113) # margine non stampabile page_h = dc.GetDeviceCaps(10) X = lambda mm: int(round(mm * dpx)) - offx Y = lambda mm: int(round(mm * dpy)) - offy fonts = {} def font(bold, wm, hm): k = (bold, wm, hm) if k not in fonts: fonts[k] = win32ui.CreateFont({"name": "Consolas", "height": -int(round(CHAR_H * hm * dpy)), "width": int(round(CHAR_W * wm * dpx)), "weight": 700 if bold else 400}) return fonts[k] rows, length = a4_rows(doc) usable = (page_h + offy) / dpy - 2 * PAGE_Y # mm disponibili per pagina pen = win32ui.CreatePen(win32con.PS_DOT, 1, 0x999999) dc.StartDoc("Comanda"); dc.StartPage(); dc.SetBkMode(win32con.TRANSPARENT) def border(top, h): dc.SelectObject(pen) x0, x1 = X(PAGE_X), X(PAGE_X + PAPER_MM) y0, y1 = Y(PAGE_Y + top), Y(PAGE_Y + top + h) for a, b in (((x0, y0), (x1, y0)), ((x1, y0), (x1, y1)), ((x1, y1), (x0, y1)), ((x0, y1), (x0, y0))): dc.MoveTo(a); dc.LineTo(b) shift = 0.0 # mm già usati nelle pagine precedenti for x, y, t, bold, wm, hm in rows: if y - shift + CHAR_H * hm > usable: # non ci sta: nuova pagina border(0, usable); dc.EndPage(); dc.StartPage(); dc.SetBkMode(win32con.TRANSPARENT); shift = y - 3.0 dc.SelectObject(font(bold, wm, hm)) dc.TextOut(X(PAGE_X + x), Y(PAGE_Y + y - shift), t) border(0, min(length - shift, usable)) dc.EndPage(); dc.EndDoc(); dc.DeleteDC() # Senza pywin32: stessa impaginazione disegnata con .NET da PowerShell (niente finestre, niente pacchetti). PS_SCRIPT = r""" param([string]$Json, [string]$Printer) Add-Type -AssemblyName System.Drawing $rows = Get-Content -LiteralPath $Json -Raw -Encoding UTF8 | ConvertFrom-Json $doc = New-Object System.Drawing.Printing.PrintDocument $doc.DocumentName = 'Comanda' if ($Printer) { $doc.PrinterSettings.PrinterName = $Printer } if (-not $doc.PrinterSettings.IsValid) { throw "Stampante non trovata: $Printer" } $doc.OriginAtMargins = $false $script:i = 0 $doc.add_PrintPage({ param($s, $e) $g = $e.Graphics $g.PageUnit = [System.Drawing.GraphicsUnit]::Millimeter $g.TextRenderingHint = [System.Drawing.Text.TextRenderingHint]::AntiAliasGridFit $hx = $e.PageSettings.HardMarginX * 0.254; $hy = $e.PageSettings.HardMarginY * 0.254 $g.TranslateTransform([single]($rows.px - $hx), [single]($rows.py - $hy)) $fmt = [System.Drawing.StringFormat]::GenericTypographic $usable = $e.PageBounds.Height * 0.254 - 2 * $rows.py $shift = $null; $last = 0 while ($script:i -lt $rows.r.Count) { $r = $rows.r[$script:i] if ($shift -eq $null) { $shift = [double]$r[1] - 3.0 } $y = [double]$r[1] - $shift if ($y + $rows.ch * $r[5] -gt $usable -and $script:i -gt 0 -and $y -gt 3.1) { break } $style = if ($r[3]) { [System.Drawing.FontStyle]::Bold } else { [System.Drawing.FontStyle]::Regular } $f = New-Object System.Drawing.Font('Consolas', [single]($rows.ch * $r[5]), $style, [System.Drawing.GraphicsUnit]::Millimeter) $nat = $g.MeasureString('MMMMMMMMMM', $f, 10000, $fmt).Width / 10 $st = $g.Save() $g.TranslateTransform([single]$r[0], [single]$y) $g.ScaleTransform([single]($rows.cw * $r[4] / $nat), 1) $g.DrawString([string]$r[2], $f, [System.Drawing.Brushes]::Black, 0, 0, $fmt) $g.Restore($st); $f.Dispose() $last = $y + $rows.lh * $r[5] $script:i++ } $h = if ($script:i -ge $rows.r.Count) { [Math]::Min($rows.len - $shift, $usable) } else { $usable } $pen = New-Object System.Drawing.Pen([System.Drawing.Color]::Gray, 0.2) $pen.DashStyle = [System.Drawing.Drawing2D.DashStyle]::Dot $g.DrawRectangle($pen, 0, 0, [single]$rows.pw, [single]$h) $e.HasMorePages = $script:i -lt $rows.r.Count }) $doc.Print() """ def print_a4_ps(name, doc): import subprocess, tempfile rows, length = a4_rows(doc) data = {"r": [list(r) for r in rows], "len": length, "pw": PAPER_MM, "px": PAGE_X, "py": PAGE_Y, "cw": CHAR_W, "ch": CHAR_H, "lh": LINE_MM} tmp = tempfile.mkdtemp(prefix="comanda_") jpath, spath = os.path.join(tmp, "c.json"), os.path.join(tmp, "stampa.ps1") with open(jpath, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False) with open(spath, "w", encoding="utf-8-sig") as f: f.write(PS_SCRIPT) try: r = subprocess.run(["powershell", "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-File", spath, "-Json", jpath, "-Printer", name or ""], capture_output=True, text=True, timeout=120, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) if r.returncode != 0: msg = (r.stderr or r.stdout).strip() raise RuntimeError(msg.splitlines()[0] if msg else f"PowerShell ha restituito {r.returncode}") finally: for pth in (jpath, spath): try: os.remove(pth) except OSError: pass try: os.rmdir(tmp) except OSError: pass def testo_scontrino(doc): """Linux, stampa A4 di prova: lo scontrino come testo a larghezza fissa.""" cols, righe = doc.get("cols", 48), [] for op in doc["ops"]: if op[0] == "f": righe.extend([""] * int(op[1])); continue _, t, bold, big, tall, al = op righe.append(t.center(cols) if al == "c" else t.rjust(cols) if al == "r" else t) return "\n".join(righe) + "\n" def lp(stampante, dati, *opzioni): import subprocess cmd = ["lp", "-s", *opzioni] + (["-d", stampante] if stampante else []) try: r = subprocess.run(cmd, input=dati, capture_output=True, timeout=60) except FileNotFoundError: raise RuntimeError("manca il sistema di stampa CUPS: installalo con sudo apt install cups") if r.returncode != 0: msg = (r.stderr or r.stdout).decode("utf-8", "replace").strip() raise RuntimeError(msg.splitlines()[0] if msg else f"lp ha restituito {r.returncode}") def send(c, data): if c["metodo"] == "a4": doc = json.loads(data) if not WINDOWS: lp(c.get("stampante"), testo_scontrino(doc).encode("utf-8"), "-o", "cpi=12", "-o", "lpi=7", "-o", "page-left=40") elif has_gdi(): print_a4_gdi(c.get("stampante"), doc) else: print_a4_ps(c.get("stampante"), doc) return if c["metodo"] == "schermo": print("-" * 48 + "\n" + data.decode("utf-8", "replace") + "-" * 48); return if c["metodo"] == "rete": host, _, port = c["stampante"].partition(":") with socket.create_connection((host, int(port or 9100)), timeout=10) as s: s.sendall(data) return if not WINDOWS: return lp(c["stampante"], data, "-o", "raw") import win32print h = win32print.OpenPrinter(c["stampante"]) try: win32print.StartDocPrinter(h, 1, ("Comanda", None, "RAW")) win32print.StartPagePrinter(h); win32print.WritePrinter(h, data) win32print.EndPagePrinter(h); win32print.EndDocPrinter(h) finally: win32print.ClosePrinter(h) def call(url, data=None, timeout=20): req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json", "User-Agent": "comande-agente/1"}) return urllib.request.urlopen(req, timeout=timeout) def load(): cfg = json.load(open(CFG, encoding="utf-8")) if "pc_token" in cfg: return cfg if "stampanti" not in cfg: # vecchio formato con una sola stampante cfg = {"server": cfg["server"], "stampanti": [{k: v for k, v in cfg.items() if k != "server"}]} return cfg def poll(server, c, state): """Controlla una stampante; restituisce True se ha stampato qualcosa.""" base = f"{server}/agent/{c['codice']}" name = c.get("stampante") or ("stampante predefinita" if c["metodo"] == "a4" else c["metodo"]) try: f = fmt_for(c) r = call(base + "/next" + (f"?fmt={f}" if f else "")) if state.get("down"): log(f"[{name}] server di nuovo raggiungibile."); state["down"] = False if r.status == 204: return False jid, data = r.headers.get("X-Job-Id"), r.read() try: send(c, data); ok, err = True, "" log(f"[{name}] stampato lavoro {jid}") except Exception as e: ok, err = False, str(e) or e.__class__.__name__ log(f"[{name}] ERRORE di stampa lavoro {jid}: {err}") call(f"{base}/done/{jid}", json.dumps({"ok": ok, "error": err}).encode()) return ok except urllib.error.HTTPError as e: msg = e.read().decode("utf-8", "replace") if state.get("err") != msg: log(f"[{name}] il server risponde {e.code}: {msg}") if e.code == 404: log("Codice stampante non valido: rifai le impostazioni con python agente_stampa.py --config") state["err"] = msg except Exception as e: if not state.get("down"): log(f"[{name}] server non raggiungibile ({e}); riprovo...") state["down"] = True return False def main_abbinato(cfg): """Nuovo metodo: il PC è abbinato al locale, le stampanti da servire arrivano dal server.""" server, tok = cfg["server"], cfg["pc_token"] log(f"Programma di stampa {VERSIONE} avviato su {socket.gethostname()}, server {server}. Per fermarlo chiudi la finestra.") ricerca = {"risultato": None, "in_corso": False, "ultima": 0.0} def avvia_ricerca(): if ricerca["in_corso"]: return ricerca["in_corso"] = True def lavoro(): log("Cerco le stampanti (Windows e rete)...") try: r = cerca_stampanti() log(f"Trovate {len([x for x in r if x['metodo'] != 'a4']) - 1} stampanti.") ricerca["risultato"] = r finally: ricerca["in_corso"], ricerca["ultima"] = False, time.time() threading.Thread(target=lavoro, daemon=True).start() avvia_ricerca() stampanti, states, down, ultimo, avvisato = [], {}, False, 0.0, False while True: if time.time() - ultimo > 10 or ricerca["risultato"] is not None: body = {"nome": socket.gethostname()} if ricerca["risultato"] is not None: body["trovate"] = ricerca["risultato"] try: r = json.loads(call(f"{server}/pc/{tok}/stato", json.dumps(body).encode()).read()) if "trovate" in body: ricerca["risultato"] = None if down: log("Server di nuovo raggiungibile."); down = False nuove = r["stampanti"] if [(x["id"], x["metodo"], x["stampante"]) for x in nuove] != [(x["id"], x["metodo"], x["stampante"]) for x in stampanti]: for x in nuove: log(f"Stampo \"{x['nome']}\" su: {x['stampante'] or x['metodo']} ({x['metodo']})") if not nuove: log("Nessuna stampante assegnata a questo PC: sceglila nell'area admin, sezione Stampanti.") stampanti = nuove if r.get("attivo") is False and not avvisato: log("ATTENZIONE: il programma non è attivo per questo locale (prova finita o abbonamento scaduto). " "Le comande non arrivano finché non lo attivi dall'area admin, in Impostazioni.") avvisato = r.get("attivo") is False if r.get("cerca") or time.time() - ricerca["ultima"] > 600: avvia_ricerca() except urllib.error.HTTPError as e: if e.code == 404: log("Questo computer è stato scollegato dall'area admin: va collegato di nuovo.") raise Scollegato() else: log(f"Il server risponde {e.code}") except Exception as e: if not down: log(f"Server non raggiungibile ({e}); riprovo...") down = True ultimo = time.time() busy = False for c in stampanti: busy = poll(server, c, states.setdefault(c["id"], {})) or busy if not busy: time.sleep(2) def main(): titolo("Comande Facili - Stampa") if "--avvio-automatico" in sys.argv[:-1]: v = sys.argv[sys.argv.index("--avvio-automatico") + 1].lower() avvio_automatico(v in ("si", "sì", "1", "on")); print("Avvio automatico", "attivo" if v in ("si", "sì", "1", "on") else "disattivato"); return if "--elenco" in sys.argv: for x in cerca_stampanti(): print(f"{x['metodo']:8} {x['stampante']:30} {x['nome']}") return if not una_sola_copia(): print("Il programma di stampa è già aperto su questo computer" + (" (guarda nella barra delle applicazioni)." if WINDOWS else ".")) if WINDOWS: print("Chiudi l'altra finestra se vuoi riaprirlo da qui.") else: print("Se lavora in sottofondo e vuoi ricollegarlo: systemctl --user stop comandefacili-stampa e poi rilancialo con --abbina") if CONGELATO and sys.stdin.isatty(): input("\nPremi Invio per chiudere...") return if not sys.stdin.isatty() and not os.path.exists(CFG): log("Questo computer non è ancora collegato: apri un terminale e lancia il programma per inserire il codice di 6 cifre.") time.sleep(300); return try: vecchio = load() if os.path.exists(CFG) else {} except Exception: vecchio = {} # impostazioni rovinate: si ricollega if "--abbina" in sys.argv or not vecchio: srv = sys.argv[sys.argv.index("--server") + 1].rstrip("/") if "--server" in sys.argv[:-1] else vecchio.get("server") cfg = abbina(srv) elif "pc_token" in vecchio and "--config" not in sys.argv: cfg = vecchio if CONGELATO and sys.stdin.isatty() and not avvio_automatico_attivo(): log("Suggerimento: per farlo partire da solo all'accensione, apri il programma con --avvio-automatico si") else: cfg = None while cfg: try: return main_abbinato(cfg) except Scollegato: if not sys.stdin.isatty(): # in sottofondo non si può chiedere il codice: si aspetta un nuovo collegamento try: os.remove(CFG) except OSError: pass log("Apri un terminale e lancia il programma per collegare di nuovo questo computer.") time.sleep(300); return cfg = abbina(cfg["server"]) if "--config" in sys.argv: cfg = setup() else: cfg = load() for c in cfg["stampanti"]: log(f"Stampante: {c['metodo']} {c.get('stampante', '')}") log(f"Agente avviato, server {cfg['server']}. Ctrl+C per fermare.") states = [{} for _ in cfg["stampanti"]] while True: busy = False for c, st in zip(cfg["stampanti"], states): busy = poll(cfg["server"], c, st) or busy if not busy: time.sleep(2) if __name__ == "__main__": try: main() except KeyboardInterrupt: print("\nProgramma di stampa fermato.") except Exception as e: print(f"\nErrore imprevisto: {e!r}") if CONGELATO and WINDOWS and sys.stdin.isatty(): input("Premi Invio per chiudere...") raise PK۸k] Vuu __main__.pyPK9u