338 lines
11 KiB
Python
338 lines
11 KiB
Python
"""
|
|
Scribus Bridge Runner — runs inside the Scribus Python scripting environment.
|
|
|
|
Listens on a Unix socket for JSON commands from the Flask bridge server
|
|
and dispatches them to the Scribus Python API.
|
|
|
|
Launched via: scribus --python-script scribus_runner.py
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import socket
|
|
import sys
|
|
import threading
|
|
import traceback
|
|
from pathlib import Path
|
|
|
|
try:
|
|
import scribus
|
|
except ImportError:
|
|
# Running outside Scribus for testing
|
|
scribus = None
|
|
print("[runner] WARNING: scribus module not available (running outside Scribus?)")
|
|
|
|
SOCKET_PATH = "/tmp/scribus_bridge.sock"
|
|
DESIGNS_DIR = Path("/data/designs")
|
|
SCREENSHOT_DIR = Path("/tmp/scribus_screenshots")
|
|
|
|
|
|
def _ensure_dirs():
|
|
DESIGNS_DIR.mkdir(parents=True, exist_ok=True)
|
|
SCREENSHOT_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
# ── Command handlers ──
|
|
|
|
def cmd_new_document(args: dict) -> dict:
|
|
"""Create a new Scribus document."""
|
|
width = args.get("width", 210) # mm, A4 default
|
|
height = args.get("height", 297)
|
|
margins = args.get("margins", 10)
|
|
pages = args.get("pages", 1)
|
|
unit = args.get("unit", 0) # 0=points, 1=mm, 2=inches, 3=picas
|
|
|
|
if scribus:
|
|
# newDocument(width, height, topMargin, leftMargin, rightMargin, bottomMargin, ..., unit, pages, ...)
|
|
scribus.newDocument(
|
|
(width, height),
|
|
(margins, margins, margins, margins),
|
|
scribus.PORTRAIT, pages, unit, scribus.FACINGPAGES, scribus.FIRSTPAGELEFT, 1
|
|
)
|
|
return {"ok": True, "message": f"Created {width}x{height}mm document with {pages} page(s)"}
|
|
|
|
|
|
def cmd_add_text_frame(args: dict) -> dict:
|
|
"""Create a text frame and optionally set its content."""
|
|
x = args.get("x", 10)
|
|
y = args.get("y", 10)
|
|
w = args.get("width", 100)
|
|
h = args.get("height", 30)
|
|
text = args.get("text", "")
|
|
font_size = args.get("fontSize", 12)
|
|
font_name = args.get("fontName", "Liberation Sans")
|
|
name = args.get("name")
|
|
|
|
if scribus:
|
|
frame = scribus.createText(x, y, w, h, name or "")
|
|
if text:
|
|
scribus.setText(text, frame)
|
|
scribus.setFontSize(font_size, frame)
|
|
try:
|
|
scribus.setFont(font_name, frame)
|
|
except Exception:
|
|
scribus.setFont("Liberation Sans", frame)
|
|
return {"ok": True, "frame": frame}
|
|
|
|
frame_name = name or f"text_{x}_{y}"
|
|
return {"ok": True, "frame": frame_name, "simulated": True}
|
|
|
|
|
|
def cmd_add_image_frame(args: dict) -> dict:
|
|
"""Create an image frame, optionally loading an image from a path or URL."""
|
|
x = args.get("x", 10)
|
|
y = args.get("y", 10)
|
|
w = args.get("width", 100)
|
|
h = args.get("height", 100)
|
|
image_path = args.get("imagePath", "")
|
|
name = args.get("name")
|
|
|
|
if scribus:
|
|
frame = scribus.createImage(x, y, w, h, name or "")
|
|
if image_path and os.path.exists(image_path):
|
|
scribus.loadImage(image_path, frame)
|
|
scribus.setScaleImageToFrame(True, True, frame)
|
|
return {"ok": True, "frame": frame}
|
|
|
|
frame_name = name or f"image_{x}_{y}"
|
|
return {"ok": True, "frame": frame_name, "simulated": True}
|
|
|
|
|
|
def cmd_add_shape(args: dict) -> dict:
|
|
"""Create a geometric shape (rectangle or ellipse)."""
|
|
shape_type = args.get("shapeType", "rect")
|
|
x = args.get("x", 10)
|
|
y = args.get("y", 10)
|
|
w = args.get("width", 50)
|
|
h = args.get("height", 50)
|
|
fill = args.get("fill")
|
|
name = args.get("name")
|
|
|
|
if scribus:
|
|
if shape_type == "ellipse":
|
|
frame = scribus.createEllipse(x, y, w, h, name or "")
|
|
else:
|
|
frame = scribus.createRect(x, y, w, h, name or "")
|
|
if fill:
|
|
# Define and set fill color
|
|
color_name = f"fill_{frame}"
|
|
r, g, b = _parse_color(fill)
|
|
scribus.defineColorRGB(color_name, r, g, b)
|
|
scribus.setFillColor(color_name, frame)
|
|
return {"ok": True, "frame": frame}
|
|
|
|
return {"ok": True, "frame": name or f"{shape_type}_{x}_{y}", "simulated": True}
|
|
|
|
|
|
def cmd_get_doc_state(args: dict) -> dict:
|
|
"""Return a full snapshot of the current document state."""
|
|
if not scribus:
|
|
return {"error": "No scribus module", "simulated": True}
|
|
|
|
try:
|
|
page_count = scribus.pageCount()
|
|
except Exception:
|
|
return {"pages": [], "frames": [], "message": "No document open"}
|
|
|
|
pages = []
|
|
for p in range(1, page_count + 1):
|
|
scribus.gotoPage(p)
|
|
w, h = scribus.getPageSize()
|
|
pages.append({"number": p, "width": w, "height": h})
|
|
|
|
frames = []
|
|
all_objects = scribus.getAllObjects()
|
|
for obj_name in all_objects:
|
|
obj_type = scribus.getObjectType(obj_name)
|
|
x, y = scribus.getPosition(obj_name)
|
|
w, h = scribus.getSize(obj_name)
|
|
frame_info = {
|
|
"name": obj_name,
|
|
"type": obj_type,
|
|
"x": x, "y": y,
|
|
"width": w, "height": h,
|
|
}
|
|
if obj_type == "TextFrame":
|
|
try:
|
|
frame_info["text"] = scribus.getText(obj_name)
|
|
frame_info["fontSize"] = scribus.getFontSize(obj_name)
|
|
frame_info["fontName"] = scribus.getFont(obj_name)
|
|
except Exception:
|
|
pass
|
|
frames.append(frame_info)
|
|
|
|
return {"pages": pages, "frames": frames}
|
|
|
|
|
|
def cmd_screenshot(args: dict) -> dict:
|
|
"""Export the current page as PNG."""
|
|
dpi = args.get("dpi", 72)
|
|
_ensure_dirs()
|
|
path = str(SCREENSHOT_DIR / "current_page.png")
|
|
|
|
if scribus:
|
|
try:
|
|
scribus.savePageAsEPS(path.replace(".png", ".eps"))
|
|
# Fallback: use scribus PDF export + convert, or direct image export
|
|
# Scribus 1.5 has limited direct PNG export; use saveDocAs + external convert
|
|
scribus.saveDocAs(path.replace(".png", ".sla"))
|
|
return {"ok": True, "path": path, "note": "SLA saved; PNG conversion may require external tool"}
|
|
except Exception as e:
|
|
return {"error": f"Screenshot failed: {str(e)}"}
|
|
|
|
return {"ok": True, "path": path, "simulated": True}
|
|
|
|
|
|
def cmd_save_as_sla(args: dict) -> dict:
|
|
"""Save the document as .sla file."""
|
|
space = args.get("space", "default")
|
|
filename = args.get("filename", "design.sla")
|
|
_ensure_dirs()
|
|
save_dir = DESIGNS_DIR / space
|
|
save_dir.mkdir(parents=True, exist_ok=True)
|
|
save_path = str(save_dir / filename)
|
|
|
|
if scribus:
|
|
scribus.saveDocAs(save_path)
|
|
return {"ok": True, "path": save_path}
|
|
|
|
return {"ok": True, "path": save_path, "simulated": True}
|
|
|
|
|
|
def cmd_move_frame(args: dict) -> dict:
|
|
"""Move a frame by relative or absolute coordinates."""
|
|
name = args.get("name", "")
|
|
x = args.get("x", 0)
|
|
y = args.get("y", 0)
|
|
absolute = args.get("absolute", False)
|
|
|
|
if scribus and name:
|
|
if absolute:
|
|
scribus.moveObjectAbs(x, y, name)
|
|
else:
|
|
scribus.moveObject(x, y, name)
|
|
return {"ok": True}
|
|
|
|
return {"ok": True, "simulated": True}
|
|
|
|
|
|
def cmd_delete_frame(args: dict) -> dict:
|
|
"""Delete a frame by name."""
|
|
name = args.get("name", "")
|
|
if scribus and name:
|
|
scribus.deleteObject(name)
|
|
return {"ok": True}
|
|
return {"ok": True, "simulated": True}
|
|
|
|
|
|
def cmd_set_background_color(args: dict) -> dict:
|
|
"""Set the page background color."""
|
|
color = args.get("color", "#ffffff")
|
|
if scribus:
|
|
r, g, b = _parse_color(color)
|
|
color_name = "page_bg"
|
|
scribus.defineColorRGB(color_name, r, g, b)
|
|
# Scribus doesn't have direct page background — create a full-page rect
|
|
w, h = scribus.getPageSize()
|
|
bg = scribus.createRect(0, 0, w, h, "background_rect")
|
|
scribus.setFillColor(color_name, bg)
|
|
scribus.setLineWidth(0, bg)
|
|
scribus.sentToLayer("Background", bg) if False else None
|
|
# Send to back
|
|
try:
|
|
for _ in range(50):
|
|
scribus.moveSelectionToBack()
|
|
except Exception:
|
|
pass
|
|
return {"ok": True, "frame": bg}
|
|
return {"ok": True, "simulated": True}
|
|
|
|
|
|
# ── Helpers ──
|
|
|
|
def _parse_color(color_str: str) -> tuple:
|
|
"""Parse hex color string to (r, g, b) tuple."""
|
|
color_str = color_str.lstrip("#")
|
|
if len(color_str) == 6:
|
|
return (int(color_str[0:2], 16), int(color_str[2:4], 16), int(color_str[4:6], 16))
|
|
return (0, 0, 0)
|
|
|
|
|
|
COMMAND_MAP = {
|
|
"new_document": cmd_new_document,
|
|
"add_text_frame": cmd_add_text_frame,
|
|
"add_image_frame": cmd_add_image_frame,
|
|
"add_shape": cmd_add_shape,
|
|
"get_doc_state": cmd_get_doc_state,
|
|
"screenshot": cmd_screenshot,
|
|
"save_as_sla": cmd_save_as_sla,
|
|
"move_frame": cmd_move_frame,
|
|
"delete_frame": cmd_delete_frame,
|
|
"set_background_color": cmd_set_background_color,
|
|
}
|
|
|
|
|
|
def handle_command(data: dict) -> dict:
|
|
"""Dispatch a command to the appropriate handler."""
|
|
action = data.get("action", "")
|
|
args = data.get("args", {})
|
|
|
|
handler = COMMAND_MAP.get(action)
|
|
if not handler:
|
|
return {"error": f"Unknown action: {action}", "available": list(COMMAND_MAP.keys())}
|
|
|
|
try:
|
|
return handler(args)
|
|
except Exception as e:
|
|
return {"error": f"Command '{action}' failed: {str(e)}", "traceback": traceback.format_exc()}
|
|
|
|
|
|
def run_socket_server():
|
|
"""Listen on Unix socket for commands from the Flask bridge."""
|
|
if os.path.exists(SOCKET_PATH):
|
|
os.remove(SOCKET_PATH)
|
|
|
|
server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
server.bind(SOCKET_PATH)
|
|
os.chmod(SOCKET_PATH, 0o666)
|
|
server.listen(5)
|
|
print(f"[runner] Listening on {SOCKET_PATH}")
|
|
|
|
while True:
|
|
try:
|
|
conn, _ = server.accept()
|
|
data = b""
|
|
while True:
|
|
chunk = conn.recv(4096)
|
|
if not chunk:
|
|
break
|
|
data += chunk
|
|
if b"\n" in data:
|
|
break
|
|
|
|
if data:
|
|
cmd = json.loads(data.decode("utf-8").strip())
|
|
result = handle_command(cmd)
|
|
response = json.dumps(result) + "\n"
|
|
conn.sendall(response.encode("utf-8"))
|
|
conn.close()
|
|
except Exception as e:
|
|
print(f"[runner] Socket error: {e}", file=sys.stderr)
|
|
|
|
|
|
# Always run — Scribus --python-script doesn't set __name__ to "__main__"
|
|
_ensure_dirs()
|
|
print("[runner] Scribus bridge runner starting...")
|
|
# Run socket server in a thread so Scribus event loop can continue
|
|
t = threading.Thread(target=run_socket_server, daemon=True)
|
|
t.start()
|
|
print("[runner] Socket server thread started")
|
|
|
|
# Keep the script alive
|
|
try:
|
|
while True:
|
|
import time
|
|
time.sleep(1)
|
|
except KeyboardInterrupt:
|
|
print("[runner] Shutting down")
|