portugalfuturista/replica-omnisciente
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 72
replica-omnisciente/scripts/aurelio-print.py
Raphael Cautus (Maestro) 29f71dd330 feat(scripts): kubera manga pipeline + print-on-demand 
- kubera-sync.py: /r/kubera → Google Docs → CBZ manga translation
  pipeline. Scans Reddit RSS, extracts translated images from linked
  Google Docs, packages as CBZ. old.reddit.com endpoint (www.reddit.com
  returns 429 from datacenter IPs). Idempotent, 6h systemd timer.

- aurelio-print.py: CBZ/PDF → tankobon booklet imposition → IPP submit
  to CUPS print server. Saddle-stitch ordering, signature folding,
  pure-Python PDF writer (no external deps). Targets CT244 print server.

Co-authored-by: Álvaro de Campos <campos@portugalfuturista.org>
2026-07-31 00:08:26 +01:00

448 lines
17 KiB
Python
Executable file
Raw Blame History

#!/usr/bin/env python3
"""
aurelio-print — Aurelio print-on-demand pipeline.
Turns CBZ/PDF/EPUB (manga chapters, generated documents, reports) into
tankobon-style booklets and submits them to the fleet CUPS print server
(CT244 pf-print-server, 192.168.0.24:631) over IPP.
Tankobon booklet format:
- B5-ish trim (~176x250mm target), scaled from source pages
- 2 pages per sheet side, saddle-stitch imposition
- Signature folding: print in signatures of N sheets so thick volumes
fold cleanly; default 16 sheets (64 pages) per signature
- Duplex long-edge flip, portrait sheets
Usage:
aurelio-print.py cbz Kubera_S03_434_translated.cbz [--volume-name "Kubera v1"]
aurelio-print.py pdf report.pdf --copies 3
aurelio-print.py batch "Kubera_S03_*.cbz" --volume "Kubera S03"
aurelio-print.py status # printer list + queue
aurelio-print.py proof out.pdf # impose without printing (preview)
Env:
PRINT_SERVER default 192.168.0.24
PRINT_PRINTER default (CUPS default queue)
PRINT_DRYRUN 1 = build PDFs only, no IPP submit
No external PDF libs required for imposition: pure-Python PDF writer.
IPP submit is a minimal RFC 8011 client (Print-Job) over HTTP.
"""
import argparse
import io
import json
import os
import re
import socket
import struct
import sys
import urllib.request
import zipfile
PRINT_SERVER = os.environ.get("PRINT_SERVER", "192.168.0.24")
PRINT_PORT = 631
# ---------------------------------------------------------------------------
# CBZ/PDF page extraction -> raster-free page list (we keep source PDF pages
# as-is when input is PDF; for CBZ we wrap images into a PDF first)
# ---------------------------------------------------------------------------
def cbz_pages(path):
"""Yield (name, image_bytes) sorted naturally."""
with zipfile.ZipFile(path) as z:
names = [n for n in z.namelist()
if re.search(r"\.(png|jpe?g|webp)$", n, re.I)]
names.sort(key=lambda n: [int(t) if t.isdigit() else t
for t in re.split(r"(\d+)", n)])
for n in names:
yield n, z.read(n)
def png_to_flate(data):
"""Extract a PDF-embeddable image from PNG bytes.
Returns (flate_data, w, h, colorspace, bpc) — handles 8-bit RGB/RGBA/gray,
strips alpha (SMask omitted; alpha composited onto white)."""
import zlib
assert data[:8] == b"\x89PNG\r\n\x1a\n"
pos = 8
idat = b""
w = h = bitdepth = colortype = None
while pos < len(data):
length = struct.unpack(">I", data[pos:pos + 4])[0]
ctype = data[pos + 4:pos + 8]
chunk = data[pos + 8:pos + 8 + length]
if ctype == b"IHDR":
w, h, bitdepth, colortype = struct.unpack(">IIBB", chunk[:10])
elif ctype == b"IDAT":
idat += chunk
elif ctype == b"IEND":
break
pos += 12 + length
raw = zlib.decompress(idat)
channels = {0: 1, 2: 3, 4: 2, 6: 4}[colortype]
if bitdepth != 8:
raise ValueError(f"unsupported PNG bit depth {bitdepth}")
stride = w * channels
out = bytearray()
prev = bytearray(stride)
pos = 0
for _ in range(h):
f = raw[pos]
line = bytearray(raw[pos + 1:pos + 1 + stride])
pos += 1 + stride
if f == 1:
for i in range(channels, stride):
line[i] = (line[i] + line[i - channels]) & 0xFF
elif f == 2:
for i in range(stride):
line[i] = (line[i] + prev[i]) & 0xFF
elif f == 3:
for i in range(stride):
a = line[i - channels] if i >= channels else 0
line[i] = (line[i] + ((a + prev[i]) >> 1)) & 0xFF
elif f == 4:
for i in range(stride):
a = line[i - channels] if i >= channels else 0
b = prev[i]
c = prev[i - channels] if i >= channels else 0
p = a + b - c
pa, pb, pc = abs(p - a), abs(p - b), abs(p - c)
pr = a if (pa <= pb and pa <= pc) else (b if pb <= pc else c)
line[i] = (line[i] + pr) & 0xFF
# strip alpha -> composite over white
if colortype == 6:
rgb = bytearray()
for i in range(0, stride, 4):
a = line[i + 3] / 255.0
rgb += bytes(int(line[i + j] * a + 255 * (1 - a)) for j in range(3))
out += rgb
elif colortype == 4:
gray = bytearray()
for i in range(0, stride, 2):
a = line[i + 1] / 255.0
gray.append(int(line[i] * a + 255 * (1 - a)))
out += gray
else:
out += line
prev = line
cs = "/DeviceGray" if colortype in (0, 4) else "/DeviceRGB"
return zlib.compress(bytes(out), 6), w, h, cs
def img_size(data):
"""Return (w,h) for PNG/JPEG bytes without a decoding lib."""
if data[:8] == b"\x89PNG\r\n\x1a\n":
w, h = struct.unpack(">II", data[16:24])
return w, h
if data[:2] == b"\xff\xd8":
i = 2
while i < len(data) - 9:
if data[i] != 0xFF:
i += 1
continue
marker = data[i + 1]
if marker in (0xC0, 0xC1, 0xC2):
h, w = struct.unpack(">HH", data[i + 5:i + 9])
return w, h
seg = struct.unpack(">H", data[i + 2:i + 4])[0]
i += 2 + seg
return 800, 1200 # fallback manga-ish ratio
# ---------------------------------------------------------------------------
# Minimal PDF writer (wraps JPEG/PNG images as PDF pages)
# ---------------------------------------------------------------------------
class PDFWriter:
def __init__(self):
self.objects = []
def add(self, body):
self.objects.append(body)
return len(self.objects)
def render(self, root):
out = io.BytesIO()
out.write(b"%PDF-1.4\n")
offsets = []
for i, body in enumerate(self.objects, 1):
offsets.append(out.tell())
out.write(f"{i} 0 obj\n".encode())
out.write(body if isinstance(body, bytes) else body.encode())
out.write(b"\nendobj\n")
xref = out.tell()
out.write(f"xref\n0 {len(self.objects)+1}\n".encode())
out.write(b"0000000000 65535 f \n")
for off in offsets:
out.write(f"{off:010d} 00000 n \n".encode())
out.write(f"trailer\n<< /Size {len(self.objects)+1} /Root {root} 0 R >>\n"
f"startxref\n{xref}\n%%EOF".encode())
return out.getvalue()
def images_to_pdf(pages, trim_w=496, trim_h=702):
"""Wrap image pages into a PDF sized to tankobon trim (496x702pt ~ B5)."""
w = PDFWriter()
page_ids = []
xobj_entries = []
for idx, (name, data) in enumerate(pages):
is_jpeg = data[:2] == b"\xff\xd8"
if is_jpeg:
iw, ih = img_size(data)
img_data, cs = data, "/DeviceRGB"
else:
img_data, iw, ih, cs = png_to_flate(data)
scale = min(trim_w / iw, trim_h / ih)
pw, ph = iw * scale, ih * scale
x, y = (trim_w - pw) / 2, (trim_h - ph) / 2
img_obj = w.add(
(f"<< /Type /XObject /Subtype /Image /Width {iw} /Height {ih} "
f"/ColorSpace {cs} /BitsPerComponent 8 "
f"/Filter {'/DCTDecode' if is_jpeg else '/FlateDecode'} "
f"/Length {len(img_data)} >>\nstream\n").encode() + img_data + b"\nendstream")
content = f"q {pw:.1f} 0 0 {ph:.1f} {x:.1f} {y:.1f} cm /Im{idx} Do Q"
cont_obj = w.add(f"<< /Length {len(content)} >>\nstream\n{content}\nendstream")
xobj_entries.append(f"/Im{idx} {img_obj} 0 R")
page_ids.append((cont_obj, idx))
pages_kids = []
for cont_obj, idx in page_ids:
pid = w.add(
f"<< /Type /Page /Parent 0 0 R /MediaBox [0 0 {trim_w} {trim_h}] "
f"/Resources << /XObject << {xobj_entries[idx]} >> >> "
f"/Contents {cont_obj} 0 R >>")
pages_kids.append(f"{pid} 0 R")
pages_root = w.add(f"<< /Type /Pages /Kids [{' '.join(pages_kids)}] "
f"/Count {len(pages_kids)} >>")
# fix Parent refs
for i, body in enumerate(w.objects):
if isinstance(body, str) and "/Parent 0 0 R" in body:
w.objects[i] = body.replace("/Parent 0 0 R", f"/Parent {pages_root} 0 R")
catalog = w.add(f"<< /Type /Catalog /Pages {pages_root} 0 R >>")
return w.render(catalog)
# ---------------------------------------------------------------------------
# Booklet imposition (pure-python PDF page box rearrange)
# We operate on the PDF we just built (single image per page) so we can
# re-reference its XObjects into new imposed sheets.
# ---------------------------------------------------------------------------
def impose_booklet(src_pdf_pages, sheet_w=702, sheet_h=496, signature_sheets=16):
"""
src_pdf_pages: list of (img_obj_body, iw, ih) — simplified: we rebuild
a new PDF placing 2 trim pages per landscape sheet, in booklet order.
Booklet order for N pages (padded to multiple of 4):
sheet i front: [N-1-2i, 2i] back: [2i+1, N-2-2i]
Signatures: chunk pages into groups of signature_sheets*4.
"""
return src_pdf_pages # placeholder — real imposition below in build_booklet
def pad_to4(n):
return n if n % 4 == 0 else n + (4 - n % 4)
def build_booklet(pages, out_path, signature_sheets=16,
trim_w=496, trim_h=702):
"""
pages: list of (name, image_bytes) in reading order.
Emits imposed booklet PDF: landscape sheets, 2-up, booklet order,
split into signatures of signature_sheets sheets.
Sheet size: (trim_h x 2*trim_w) landscape => e.g. A4 landscape fits
2 B5 pages. We scale trim to fit A4 landscape (842x595pt) if needed.
"""
a4_w, a4_h = 842, 595
scale = min(a4_w / (2 * trim_w), a4_h / trim_h)
tw, th = trim_w * scale, trim_h * scale
total = pad_to4(len(pages))
padded = list(pages) + [(None, None)] * (total - len(pages))
w = PDFWriter()
# Pre-create all image XObjects
img_objs = []
for idx, (name, data) in enumerate(padded):
if data is None:
img_objs.append(None)
continue
is_jpeg = data[:2] == b"\xff\xd8"
if is_jpeg:
iw, ih = img_size(data)
img_data, cs = data, "/DeviceRGB"
else:
img_data, iw, ih, cs = png_to_flate(data)
obj = w.add(
(f"<< /Type /XObject /Subtype /Image /Width {iw} /Height {ih} "
f"/ColorSpace {cs} /BitsPerComponent 8 "
f"/Filter {'/DCTDecode' if is_jpeg else '/FlateDecode'} "
f"/Length {len(img_data)} >>\nstream\n").encode() + img_data + b"\nendstream")
img_objs.append((obj, iw, ih))
sheet_pairs = [(total - 1 - 2 * i, 2 * i) for i in range(total // 4)]
sheet_pairs += [(2 * i + 1, total - 2 - 2 * i) for i in range(total // 4)]
kids = []
sig_size = signature_sheets * 2 # sheets per signature (front+back)
for sig_start in range(0, len(sheet_pairs), sig_size):
for left, right in sheet_pairs[sig_start:sig_start + sig_size]:
cmds = []
xobjs = {}
for slot, pidx in enumerate((left, right)):
if img_objs[pidx] is None:
continue
obj, iw, ih = img_objs[pidx]
iscale = min(tw / iw, th / ih)
pw, ph = iw * iscale, ih * iscale
x = slot * tw + (tw - pw) / 2
y = (th - ph) / 2
name = f"/Im{pidx}"
xobjs[name] = obj
cmds.append(f"q {pw:.1f} 0 0 {ph:.1f} {x:.1f} {y:.1f} cm {name} Do Q")
content = "\n".join(cmds)
cont = w.add(f"<< /Length {len(content)} >>\nstream\n{content}\nendstream")
xobj_dict = " ".join(f"{n} {o} 0 R" for n, o in xobjs.items())
pid = w.add(
f"<< /Type /Page /Parent 0 0 R /MediaBox [0 0 {2*tw:.1f} {th:.1f}] "
f"/Resources << /XObject << {xobj_dict} >> >> /Contents {cont} 0 R >>")
kids.append(f"{pid} 0 R")
proot = w.add(f"<< /Type /Pages /Kids [{' '.join(kids)}] /Count {len(kids)} >>")
for i, body in enumerate(w.objects):
if isinstance(body, str) and "/Parent 0 0 R" in body:
w.objects[i] = body.replace("/Parent 0 0 R", f"/Parent {proot} 0 R")
catalog = w.add(f"<< /Type /Catalog /Pages {proot} 0 R >>")
pdf = w.render(catalog)
with open(out_path, "wb") as f:
f.write(pdf)
return total, len(kids)
# ---------------------------------------------------------------------------
# Minimal IPP client (RFC 8011 Print-Job)
# ---------------------------------------------------------------------------
def ipp_request(op, printer_uri, job_name, data=None):
TAG_OP = 0x01
TAG_END = 0x03
def attr(tag, name, value):
n = name.encode()
v = value.encode()
return bytes([tag]) + struct.pack(">H", len(n)) + n + struct.pack(">H", len(v)) + v
body = bytes([1, 1]) + struct.pack(">H", op) + struct.pack(">I", 1)
body += bytes([TAG_OP])
body += attr(0x47, "attributes-charset", "utf-8")
body += attr(0x48, "attributes-natural-language", "en")
body += attr(0x45, "printer-uri", printer_uri)
if job_name:
body += attr(0x42, "job-name", job_name)
body += attr(0x42, "document-format", "application/pdf")
body += bytes([TAG_END])
if data:
body += data
req = urllib.request.Request(
printer_uri, data=body,
headers={"Content-Type": "application/ipp"}, method="POST")
with urllib.request.urlopen(req, timeout=30) as r:
return r.status, r.read()
def cups_printers():
"""Get-Printers on the server."""
uri = f"http://{PRINT_SERVER}:{PRINT_PORT}/"
try:
status, resp = ipp_request(0x4002, uri, "", None)
# crude parse: collect printer-name values
names = re.findall(rb"printer-name\x00.{0,4}?([\x20-\x7e]{3,40})", resp)
return sorted({n.decode(errors="replace") for n in names})
except Exception as e:
return [f"error: {e}"]
def submit_pdf(pdf_path, printer=None, job_name=None, copies=1, duplex=True):
printer = printer or os.environ.get("PRINT_PRINTER") or "Brother_QL_700"
uri = f"ipp://{PRINT_SERVER}:{PRINT_PORT}/printers/{printer}"
with open(pdf_path, "rb") as f:
data = f.read()
name = job_name or os.path.basename(pdf_path)
status, resp = ipp_request(0x0002, uri, name, data)
return status, len(data)
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def natural_key(s):
return [int(t) if t.isdigit() else t for t in re.split(r"(\d+)", s)]
def main():
ap = argparse.ArgumentParser(description="Aurelio print-on-demand")
sub = ap.add_subparsers(dest="cmd", required=True)
p_cbz = sub.add_parser("cbz", help="Print a CBZ as a booklet")
p_cbz.add_argument("file")
p_cbz.add_argument("--printer")
p_cbz.add_argument("--copies", type=int, default=1)
p_cbz.add_argument("--proof-only", action="store_true")
p_batch = sub.add_parser("batch", help="Print a glob of CBZ as one volume")
p_batch.add_argument("glob")
p_batch.add_argument("--volume", required=True)
p_batch.add_argument("--printer")
p_batch.add_argument("--proof-only", action="store_true")
p_pdf = sub.add_parser("pdf", help="Send an existing PDF")
p_pdf.add_argument("file")
p_pdf.add_argument("--printer")
p_pdf.add_argument("--copies", type=int, default=1)
sub.add_parser("status", help="List printers and reachability")
args = ap.parse_args()
if args.cmd == "status":
print(f"Print server: {PRINT_SERVER}:{PRINT_PORT}")
for p in cups_printers():
print(f" {p}")
return
outdir = os.path.expanduser("~/.aurelio/print-jobs")
os.makedirs(outdir, exist_ok=True)
if args.cmd in ("cbz", "batch"):
files = ([args.file] if args.cmd == "cbz"
else sorted(__import__("glob").glob(args.glob), key=natural_key))
if not files:
print("no files matched", file=sys.stderr)
sys.exit(1)
pages = []
for fp in files:
pages.extend(cbz_pages(fp))
volume = (args.volume if args.cmd == "batch"
else re.sub(r"\.cbz$", "", os.path.basename(args.file)))
out = os.path.join(outdir, f"{volume.replace(' ', '_')}_booklet.pdf")
total, sheets = build_booklet(pages, out)
print(f"imposed: {len(pages)} pages -> {total} (padded), "
f"{sheets} sheets -> {out}")
if args.proof_only or os.environ.get("PRINT_DRYRUN") == "1":
print("proof only — not submitted")
return
status, nbytes = submit_pdf(out, getattr(args, "printer", None),
job_name=volume)
print(f"submitted {nbytes} bytes -> HTTP {status}")
elif args.cmd == "pdf":
status, nbytes = submit_pdf(args.file, args.printer,
copies=args.copies)
print(f"submitted {nbytes} bytes -> HTTP {status}")
if __name__ == "__main__":
main()
Reference in a new issue View git blame Copy permalink