480 lines
18 KiB
Python
Executable File
480 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# /// script
|
|
# dependencies = [
|
|
# "pyipp>=0.16.0",
|
|
# "requests>=2.31.0",
|
|
# "pypdf>=4.0.0",
|
|
# "reportlab>=4.0.0",
|
|
# ]
|
|
# ///
|
|
"""
|
|
Brother HL-L8240CDW Network Printer Controller & IPP Driver for OpenClaw.
|
|
Supports direct PDF printing, hardware N-Up, duplex, color modes,
|
|
and live Telegram status updates via Bot API.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import json
|
|
import argparse
|
|
import asyncio
|
|
from pathlib import Path
|
|
from typing import Optional, Dict, Any
|
|
|
|
import requests
|
|
from pypdf import PdfReader
|
|
from reportlab.lib.pagesizes import A4
|
|
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
|
|
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer
|
|
from reportlab.lib.units import cm
|
|
|
|
import pyipp
|
|
from pyipp import IPP
|
|
from pyipp.enums import IppOperation
|
|
from pyipp.tags import IppTag
|
|
import pyipp.serializer
|
|
|
|
# Register extra IPP attribute tags for Brother HL-L8240CDW
|
|
pyipp.serializer.ATTRIBUTE_TAG_MAP["print-color-mode"] = IppTag.KEYWORD
|
|
pyipp.serializer.ATTRIBUTE_TAG_MAP["page-ranges"] = IppTag.RANGE
|
|
|
|
DEFAULT_PRINTER_HOST = os.environ.get("BROTHER_PRINTER_HOST", "192.168.10.91")
|
|
DEFAULT_PRINTER_PORT = int(os.environ.get("BROTHER_PRINTER_PORT", "631"))
|
|
DEFAULT_PRINTER_URI = f"ipp://{DEFAULT_PRINTER_HOST}:{DEFAULT_PRINTER_PORT}/ipp/print"
|
|
TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Telegram Helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TelegramNotifier:
|
|
def __init__(self, token: Optional[str] = None, chat_id: Optional[str] = None, thread_id: Optional[int] = None, message_id: Optional[int] = None):
|
|
self.token = token or TELEGRAM_BOT_TOKEN
|
|
self.chat_id = chat_id
|
|
self.thread_id = int(thread_id) if thread_id else None
|
|
self.message_id = int(message_id) if message_id else None
|
|
|
|
def can_notify(self) -> bool:
|
|
return bool(self.token and self.chat_id)
|
|
|
|
def send_initial(self, text: str) -> Optional[int]:
|
|
if not self.can_notify():
|
|
return None
|
|
url = f"https://api.telegram.org/bot{self.token}/sendMessage"
|
|
payload = {
|
|
"chat_id": self.chat_id,
|
|
"text": text,
|
|
"parse_mode": "Markdown",
|
|
}
|
|
if self.thread_id:
|
|
payload["message_thread_id"] = self.thread_id
|
|
try:
|
|
resp = requests.post(url, json=payload, timeout=8)
|
|
if resp.ok:
|
|
data = resp.json()
|
|
self.message_id = data.get("result", {}).get("message_id")
|
|
return self.message_id
|
|
except Exception as e:
|
|
sys.stderr.write(f"Telegram send error: {e}\n")
|
|
return None
|
|
|
|
def update(self, text: str):
|
|
if not self.can_notify() or not self.message_id:
|
|
return
|
|
url = f"https://api.telegram.org/bot{self.token}/editMessageText"
|
|
payload = {
|
|
"chat_id": self.chat_id,
|
|
"message_id": self.message_id,
|
|
"text": text,
|
|
"parse_mode": "Markdown",
|
|
}
|
|
try:
|
|
requests.post(url, json=payload, timeout=8)
|
|
except Exception as e:
|
|
sys.stderr.write(f"Telegram edit error: {e}\n")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Text to PDF Converter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def convert_text_to_pdf(text: str, output_path: str, title: str = "Dokument") -> str:
|
|
"""Convert plain text or simple markdown into a clean A4 PDF."""
|
|
doc = SimpleDocTemplate(
|
|
output_path,
|
|
pagesize=A4,
|
|
rightMargin=2 * cm,
|
|
leftMargin=2 * cm,
|
|
topMargin=2 * cm,
|
|
bottomMargin=2 * cm,
|
|
)
|
|
styles = getSampleStyleSheet()
|
|
normal_style = styles["Normal"]
|
|
normal_style.fontSize = 11
|
|
normal_style.leading = 14
|
|
|
|
heading_style = styles["Heading1"]
|
|
heading_style.fontSize = 16
|
|
heading_style.leading = 20
|
|
|
|
story = [
|
|
Paragraph(title, heading_style),
|
|
Spacer(1, 0.5 * cm),
|
|
]
|
|
|
|
for paragraph in text.split("\n\n"):
|
|
clean_p = paragraph.replace("\n", "<br/>")
|
|
story.append(Paragraph(clean_p, normal_style))
|
|
story.append(Spacer(1, 0.3 * cm))
|
|
|
|
doc.build(story)
|
|
return output_path
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Printer Service
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class BrotherPrinter:
|
|
def __init__(self, host: str = DEFAULT_PRINTER_HOST, port: int = DEFAULT_PRINTER_PORT):
|
|
self.host = host
|
|
self.port = port
|
|
self.uri = f"ipp://{self.host}:{self.port}/ipp/print"
|
|
|
|
async def get_status(self) -> Dict[str, Any]:
|
|
"""Fetch status, markers, and printer details via IPP."""
|
|
async with IPP(self.host, port=self.port) as ipp:
|
|
printer = await ipp.printer()
|
|
markers = {}
|
|
for marker in printer.markers:
|
|
markers[marker.name] = {
|
|
"level": marker.level,
|
|
"type": marker.marker_type,
|
|
"color": marker.color,
|
|
}
|
|
return {
|
|
"name": printer.info.name,
|
|
"model": printer.info.model,
|
|
"state": printer.state.printer_state,
|
|
"reasons": printer.state.reasons,
|
|
"markers": markers,
|
|
"uri": self.uri,
|
|
}
|
|
|
|
async def validate_job(
|
|
self,
|
|
color_mode: str = "monochrome",
|
|
nup: int = 1,
|
|
duplex: str = "two-sided-long-edge",
|
|
copies: int = 1,
|
|
) -> bool:
|
|
"""Dry-run validation of job parameters without printing."""
|
|
async with IPP(self.host, port=self.port) as ipp:
|
|
msg = {
|
|
"version": (2, 0),
|
|
"operation": IppOperation.VALIDATE_JOB,
|
|
"operation-attributes-tag": {
|
|
"printer-uri": self.uri,
|
|
"requesting-user-name": "openclaw",
|
|
"document-format": "application/pdf",
|
|
},
|
|
"job-attributes-tag": {
|
|
"sides": duplex,
|
|
"number-up": nup,
|
|
"print-color-mode": color_mode,
|
|
"copies": copies,
|
|
},
|
|
}
|
|
res = await ipp.execute(IppOperation.VALIDATE_JOB, msg)
|
|
return res.get("status-code") == 0
|
|
|
|
async def send_print_job(
|
|
self,
|
|
pdf_bytes: bytes,
|
|
job_name: str = "OpenClaw Print",
|
|
color_mode: str = "monochrome",
|
|
nup: int = 1,
|
|
duplex: str = "two-sided-long-edge",
|
|
copies: int = 1,
|
|
) -> Dict[str, Any]:
|
|
"""Send a Print-Job request directly with raw PDF data."""
|
|
async with IPP(self.host, port=self.port) as ipp:
|
|
msg = {
|
|
"version": (2, 0),
|
|
"operation": IppOperation.PRINT_JOB,
|
|
"operation-attributes-tag": {
|
|
"printer-uri": self.uri,
|
|
"requesting-user-name": "openclaw",
|
|
"job-name": job_name,
|
|
"document-format": "application/pdf",
|
|
},
|
|
"job-attributes-tag": {
|
|
"sides": duplex,
|
|
"number-up": nup,
|
|
"print-color-mode": color_mode,
|
|
"copies": copies,
|
|
},
|
|
"data": pdf_bytes,
|
|
}
|
|
res = await ipp.execute(IppOperation.PRINT_JOB, msg)
|
|
return res
|
|
|
|
async def get_job_attributes(self, job_id: int) -> Dict[str, Any]:
|
|
"""Poll job status using GET_JOB_ATTRIBUTES."""
|
|
async with IPP(self.host, port=self.port) as ipp:
|
|
msg = {
|
|
"version": (2, 0),
|
|
"operation": IppOperation.GET_JOB_ATTRIBUTES,
|
|
"operation-attributes-tag": {
|
|
"printer-uri": self.uri,
|
|
"job-id": job_id,
|
|
"requesting-user-name": "openclaw",
|
|
},
|
|
}
|
|
try:
|
|
res = await ipp.execute(IppOperation.GET_JOB_ATTRIBUTES, msg)
|
|
jobs = res.get("jobs", [])
|
|
if jobs:
|
|
return jobs[0]
|
|
except Exception as e:
|
|
sys.stderr.write(f"Job query warning: {e}\n")
|
|
return {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI Command Handlers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def handle_status(args):
|
|
printer = BrotherPrinter(host=args.host, port=args.port)
|
|
data = asyncio.run(printer.get_status())
|
|
if args.json:
|
|
print(json.dumps(data, indent=2))
|
|
return
|
|
|
|
print("🖨️ Brother HL-L8240CDW Status")
|
|
print("=" * 35)
|
|
print(f"Modell: {data['model']}")
|
|
print(f"Status: {data['state']}")
|
|
print(f"URI: {data['uri']}")
|
|
print("\n📊 Tonerstände:")
|
|
for name, m in data["markers"].items():
|
|
color_label = {"BK": "Schwarz", "C": "Cyan", "M": "Magenta", "Y": "Gelb"}.get(name, name)
|
|
bar = "█" * (m["level"] // 10) + "░" * (10 - (m["level"] // 10))
|
|
print(f" [{name:2}] {color_label:8}: {bar} {m['level']}%")
|
|
|
|
|
|
def handle_validate(args):
|
|
printer = BrotherPrinter(host=args.host, port=args.port)
|
|
ok = asyncio.run(
|
|
printer.validate_job(
|
|
color_mode=args.color,
|
|
nup=args.nup,
|
|
duplex=args.duplex,
|
|
copies=args.copies,
|
|
)
|
|
)
|
|
if ok:
|
|
print("✅ Konfiguration ist gültig und wird vom Drucker unterstützt.")
|
|
else:
|
|
print("❌ Konfiguration wurde vom Drucker abgewiesen.")
|
|
sys.exit(1)
|
|
|
|
|
|
def handle_print(args):
|
|
notifier = TelegramNotifier(chat_id=args.chat_id, thread_id=args.thread_id, message_id=args.message_id)
|
|
|
|
# 1. Resolve source document
|
|
doc_title = args.title or "Dokument"
|
|
pdf_path = None
|
|
temp_files = []
|
|
|
|
if args.file:
|
|
p = Path(args.file)
|
|
if not p.exists():
|
|
err = f"❌ Datei nicht gefunden: {args.file}"
|
|
print(err)
|
|
notifier.update(err)
|
|
sys.exit(1)
|
|
pdf_path = str(p)
|
|
doc_title = args.title or p.name
|
|
elif args.url:
|
|
notifier.update(f"⏳ [1/4] Lade Dokument herunter: `{args.url[:40]}...`")
|
|
resp = requests.get(args.url, timeout=30)
|
|
tmp = f"/tmp/print_{int(time.time())}.pdf"
|
|
with open(tmp, "wb") as f:
|
|
f.write(resp.content)
|
|
temp_files.append(tmp)
|
|
pdf_path = tmp
|
|
doc_title = args.title or Path(args.url).name
|
|
elif args.text:
|
|
notifier.update("⏳ [1/4] Formatiere Text als Druck-PDF...")
|
|
tmp = f"/tmp/print_{int(time.time())}.pdf"
|
|
convert_text_to_pdf(args.text, tmp, title=doc_title)
|
|
temp_files.append(tmp)
|
|
pdf_path = tmp
|
|
else:
|
|
err = "❌ Kein Dokument angegeben (--file, --url oder --text erforderlich)."
|
|
print(err)
|
|
notifier.update(err)
|
|
sys.exit(1)
|
|
|
|
# 2. Inspect PDF properties
|
|
try:
|
|
reader = PdfReader(pdf_path)
|
|
total_pages = len(reader.pages)
|
|
except Exception as e:
|
|
total_pages = 1
|
|
|
|
# Safety guardrail: check max pages
|
|
if total_pages > args.max_pages and not args.force:
|
|
err = f"⚠️ *Druck abgebrochen*: Dokument hat {total_pages} Seiten (Limit: {args.max_pages}). Verwende `--force` zum Überschreiben."
|
|
print(err)
|
|
notifier.update(err)
|
|
sys.exit(1)
|
|
|
|
# Calculate estimated sheets
|
|
sheets = (total_pages + (args.nup - 1)) // args.nup
|
|
if args.duplex != "one-sided":
|
|
sheets = (sheets + 1) // 2
|
|
|
|
color_desc = "Farbe 🎨" if args.color == "color" else "Schwarz-Weiß 🖤"
|
|
duplex_desc = "Duplex (beidseitig) 🔄" if args.duplex != "one-sided" else "Simplex (einseitig) 📄"
|
|
nup_desc = f"{args.nup} Seite(n) pro Blatt"
|
|
|
|
# Initial status notification
|
|
status_header = (
|
|
f"🖨️ *Druckauftrag: {doc_title}*\n"
|
|
f"• *Umfang*: {total_pages} Seite(n) (~{sheets} Blatt Papier)\n"
|
|
f"• *Modus*: {color_desc} | {nup_desc} | {duplex_desc}\n"
|
|
)
|
|
|
|
if not notifier.message_id and notifier.can_notify():
|
|
notifier.send_initial(f"{status_header}• *Status*: ⏳ [1/3] Bereite Übertragung vor...")
|
|
else:
|
|
notifier.update(f"{status_header}• *Status*: ⏳ [1/3] Bereite Übertragung vor...")
|
|
|
|
# 3. Read PDF bytes
|
|
with open(pdf_path, "rb") as f:
|
|
pdf_data = f.read()
|
|
|
|
# 4. Connect to printer and send job
|
|
notifier.update(f"{status_header}• *Status*: 📡 [2/3] Sende an Brother HL-L8240CDW...")
|
|
printer = BrotherPrinter(host=args.host, port=args.port)
|
|
|
|
try:
|
|
res = asyncio.run(
|
|
printer.send_print_job(
|
|
pdf_bytes=pdf_data,
|
|
job_name=doc_title,
|
|
color_mode=args.color,
|
|
nup=args.nup,
|
|
duplex=args.duplex,
|
|
copies=args.copies,
|
|
)
|
|
)
|
|
except Exception as e:
|
|
err_msg = f"{status_header}• *Status*: ❌ Verbindungsfehler zum Drucker: `{e}`"
|
|
print(err_msg)
|
|
notifier.update(err_msg)
|
|
sys.exit(1)
|
|
|
|
# Check IPP status code
|
|
status_code = res.get("status-code", -1)
|
|
if status_code != 0:
|
|
err_msg = f"{status_header}• *Status*: ❌ Drucker meldet Fehler-Code `{status_code}`."
|
|
print(err_msg)
|
|
notifier.update(err_msg)
|
|
sys.exit(1)
|
|
|
|
# Extract Job ID
|
|
jobs = res.get("jobs", [])
|
|
job_id = jobs[0].get("job-id", "unbekannt") if jobs else "unbekannt"
|
|
print(f"Print job queued. Job ID: {job_id}")
|
|
|
|
notifier.update(f"{status_header}• *Status*: 🖨️ [3/3] Auftrag übertragen (Job #{job_id}). Druckvorgang läuft...")
|
|
|
|
# 5. Monitor Job Progress (up to 30 seconds)
|
|
if isinstance(job_id, int):
|
|
for _ in range(15):
|
|
time.sleep(2)
|
|
try:
|
|
attr = asyncio.run(printer.get_job_attributes(job_id))
|
|
state = attr.get("job-state")
|
|
# 9 = completed, 7 = canceled, 8 = aborted
|
|
if state == 9:
|
|
break
|
|
elif state in (7, 8):
|
|
notifier.update(f"{status_header}• *Status*: ⚠️ Auftrag #{job_id} wurde am Drucker abgebrochen.")
|
|
return
|
|
except Exception:
|
|
pass
|
|
|
|
# Final success update
|
|
final_msg = (
|
|
f"{status_header}"
|
|
f"• *Status*: ✅ **Erfolgreich gedruckt!**\n"
|
|
f"• *Job-ID*: `#{job_id}`\n"
|
|
f"• *Ausgabe*: Liegt im Ausgabefach bereit."
|
|
)
|
|
print("Druck erfolgreich abgeschlossen.")
|
|
notifier.update(final_msg)
|
|
|
|
# Cleanup temp files
|
|
for t in temp_files:
|
|
try:
|
|
os.remove(t)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main Entry Point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Brother HL-L8240CDW IPP Print Utility for OpenClaw")
|
|
parser.add_argument("--host", default=DEFAULT_PRINTER_HOST, help="Printer IP or Hostname")
|
|
parser.add_argument("--port", type=int, default=DEFAULT_PRINTER_PORT, help="IPP Port (default 631)")
|
|
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
# Subcommand: status
|
|
status_parser = subparsers.add_parser("status", help="Get printer status and toner levels")
|
|
status_parser.add_argument("--json", action="store_true", help="Output as JSON")
|
|
status_parser.set_defaults(func=handle_status)
|
|
|
|
# Subcommand: validate
|
|
val_parser = subparsers.add_parser("validate", help="Dry-run validate print parameters")
|
|
val_parser.add_argument("--color", choices=["monochrome", "color"], default="monochrome")
|
|
val_parser.add_argument("--nup", type=int, choices=[1, 2, 4], default=1)
|
|
val_parser.add_argument("--duplex", choices=["two-sided-long-edge", "two-sided-short-edge", "one-sided"], default="two-sided-long-edge")
|
|
val_parser.add_argument("--copies", type=int, default=1)
|
|
val_parser.set_defaults(func=handle_validate)
|
|
|
|
# Subcommand: print
|
|
print_parser = subparsers.add_parser("print", help="Print a document")
|
|
print_parser.add_argument("--file", help="Path to PDF file")
|
|
print_parser.add_argument("--url", help="URL of PDF document")
|
|
print_parser.add_argument("--text", help="Raw text to convert and print")
|
|
print_parser.add_argument("--title", help="Job or document title")
|
|
print_parser.add_argument("--color", choices=["monochrome", "color"], default="monochrome", help="Color mode (default: monochrome)")
|
|
print_parser.add_argument("--nup", type=int, choices=[1, 2, 4], default=1, help="Pages per sheet (1, 2 or 4)")
|
|
print_parser.add_argument("--duplex", choices=["two-sided-long-edge", "two-sided-short-edge", "one-sided"], default="two-sided-long-edge", help="Duplex mode (default: two-sided-long-edge)")
|
|
print_parser.add_argument("--copies", type=int, default=1, help="Number of copies")
|
|
print_parser.add_argument("--max-pages", type=int, default=30, help="Safety page limit (default 30)")
|
|
print_parser.add_argument("--force", action="store_true", help="Bypass max pages guardrail")
|
|
print_parser.add_argument("--chat-id", help="Telegram Chat ID for live updates")
|
|
print_parser.add_argument("--thread-id", type=int, help="Telegram Thread / Topic ID (for forum supergroups)")
|
|
print_parser.add_argument("--message-id", help="Telegram Message ID to edit (optional, created if omitted)")
|
|
print_parser.set_defaults(func=handle_print)
|
|
|
|
args = parser.parse_args()
|
|
args.func(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|