diff --git a/README.md b/README.md index 1741dcd..5bd41a9 100644 --- a/README.md +++ b/README.md @@ -46,15 +46,19 @@ python3 ppl2psd.py input.ppl -o out.psd --preview out.png # PPL → PSD（文字转可编辑 TypeLayer） python3 ppl2psd.py input.ppl -o out.psd --type-layer +# PPL → PSD（整个矢量项目等比放大 2 倍，画布/图层/文字/阴影 ×2） +python3 ppl2psd.py input.ppl -o out_2x.psd --scale 2 + # PSD → PPL（反向） python3 psd2ppl.py input.psd -o out.ppl ``` 参数： - `-o` / `--output`：输出路径（默认与输入同名） -- `--preview`：同时输出合成预览 PNG +- `--preview`：同时输出合成预览 PNG（`--scale` 非 1 时输出全分辨率预览） - `-j` / `--workers`：并行编码线程数（默认 4） - `--type-layer`：文字转 TypeLayer +- `--scale`：整个矢量项目等比放大倍数（默认 1.0；≥2 时自动降为单进程编码防爆内存） --- diff --git a/ppl2psd.py b/ppl2psd.py index 0943dd0..ac30521 100644 --- a/ppl2psd.py +++ b/ppl2psd.py @@ -28,6 +28,9 @@ from pathlib import Path from typing import List, Optional, Tuple from PIL import Image, ImageDraw, ImageFont +# We render huge 2x canvases (e.g. 21528x10764); disable PIL's decompression +# bomb guard so it does not reject our own output. +Image.MAX_IMAGE_PIXELS = None from tqdm import tqdm try: @@ -44,6 +47,15 @@ try: except ImportError: # pragma: no cover _TYSHA = False +# Global output scale: everything (canvas, layer rects, text size, shadows) +# is multiplied by _SCALE, so the whole vector project is upscaled. +_SCALE = 1.0 + + +def set_scale(s: float) -> None: + global _SCALE + _SCALE = max(0.1, float(s)) + # -------------------------------------------------------------------------- # Fonts # -------------------------------------------------------------------------- @@ -164,9 +176,9 @@ def parse_gradient(spec: str) -> dict: # Layer rendering # -------------------------------------------------------------------------- def bounding_rect(ly: dict) -> Tuple[float, float, float, float]: - """boundingRect is (left, top, right, bottom), not (x, y, w, h).""" + """boundingRect is (left, top, right, bottom), scaled by _SCALE.""" l, t, r, b = (float(v) for v in ly["boundingRect"].split(",")) - return l, t, r, b + return l * _SCALE, t * _SCALE, r * _SCALE, b * _SCALE def crop_to_canvas(l: float, t: float, r: float, b: float, cw: int, ch: int @@ -254,7 +266,7 @@ def render_shape(ly: dict, cw: int, ch: int) -> Optional[Tuple[Image.Image, Tupl img.paste(grad_img, (0, 0), mask) # stroke - thickness = int(float(ly.get("shapeStrokeThickness", "0") or 0)) + thickness = int(float(ly.get("shapeStrokeThickness", "0") or 0) * _SCALE) if thickness > 0: stroke = parse_gradient(ly.get("shapeStrokeColor", "")) if stroke["type"] == "solid": @@ -269,8 +281,8 @@ def render_shape(ly: dict, cw: int, ch: int) -> Optional[Tuple[Image.Image, Tupl img.paste(stroke_img, (0, 0), ring) # shadow (ppl renders a soft offset shadow per layer) - sd = float(ly.get("shadowDistance", "0") or 0) - sb = float(ly.get("shadowBlur", "0") or 0) + sd = float(ly.get("shadowDistance", "0") or 0) * _SCALE + sb = float(ly.get("shadowBlur", "0") or 0) * _SCALE if (sd > 0 or sb > 0) and ly.get("_no_shadow") != "1": img, ox, oy = _add_shadow(img, sd, sb, float(ly.get("shadowAngle", "0") or 0), @@ -325,7 +337,7 @@ def render_text(ly: dict, cw: int, ch: int) -> Optional[Tuple[Image.Image, Tuple size_raw = float(ly.get("textSize", "12") or 12) font_path = _font_path_for(ly) prof = font_profile(font_path) - size = size_raw * prof["k"] + size = size_raw * prof["k"] * _SCALE if font_path: font = ImageFont.truetype(font_path, int(size)) else: @@ -437,8 +449,8 @@ def render_text(ly: dict, cw: int, ch: int) -> Optional[Tuple[Image.Image, Tuple img.paste(gimg, (0, 0), img.getchannel("A")) # soft shadow (used by rasterized output / previews) - sd = float(ly.get("shadowDistance", "0") or 0) - sb = float(ly.get("shadowBlur", "0") or 0) + sd = float(ly.get("shadowDistance", "0") or 0) * _SCALE + sb = float(ly.get("shadowBlur", "0") or 0) * _SCALE if (sd > 0 or sb > 0) and ly.get("_no_shadow") != "1": img, ox, oy = _add_shadow(img, sd, sb, float(ly.get("shadowAngle", "0") or 0), @@ -514,42 +526,32 @@ def channel_blocks(img: Image.Image, workers: int, progress: Optional[tqdm] = None) -> List[Tuple[int, bytes]]: """RLE-encode the planes of a cropped RGB(A) layer (parallel). - RGBA layers yield channel ids [-1, 0, 1, 2], RGB images yield [0, 1, 2].""" - arr = np.asarray(img) if np is not None else None - if arr is not None and arr.ndim == 3: - nch = arr.shape[2] - if nch == 4: - # zero RGB where alpha==0 for better compression - arr = arr.copy() - arr[arr[..., 3] == 0] = 0 - # PSD layer channel ids: -1=alpha mask, 0=R, 1=G, 2=B - planes = [arr[..., 3].tobytes()] + [arr[..., i].tobytes() for i in range(3)] - else: - planes = [arr[..., i].tobytes() for i in range(nch)] - else: - bands = img.getbands() - if bands == ("R", "G", "B", "A"): - planes = [img.getchannel("A").tobytes()] + \ - [img.getchannel(c).tobytes() for c in "RGB"] - else: - planes = [img.getchannel(c).tobytes() for c in bands] + RGBA layers yield channel ids [-1, 0, 1, 2], RGB images yield [0, 1, 2]. + Bands are encoded one at a time and their raw bytes released right away, + so at most two full-image buffers (the raw plane + its RLE result) exist + at once. On 2x canvases the old copy-all-planes approach multiplied the + layer size by ~5 and blew up RAM/ZRAM/SWAP, freezing the phone.""" w = img.width - jobs = [(plane, w) for plane in planes] - if workers > 1 and len(jobs) > 1: + ids = [-1, 0, 1, 2] if img.mode == "RGBA" else [0, 1, 2] + order = ["A", "R", "G", "B"] if img.mode == "RGBA" else ["R", "G", "B"] + if workers > 1 and len(order) > 1: from multiprocessing import Pool with Pool(workers) as pool: enc = [] - for res in pool.imap(_pack_channel_worker, jobs, chunksize=1): - enc.append(res) + for band in order: + plane = img.getchannel(band).tobytes() + enc.append(pool.apply(_pack_channel_worker, ((plane, w),))) + del plane if progress: progress.update(1) else: enc = [] - for j in jobs: - enc.append(_pack_channel_worker(j)) + for band in order: + plane = img.getchannel(band).tobytes() + enc.append(_pack_channel_worker((plane, w))) + del plane if progress: progress.update(1) - ids = [-1, 0, 1, 2] if img.mode == "RGBA" else [0, 1, 2] return list(zip(ids, enc)) @@ -594,7 +596,7 @@ def build_type_layer_tysh(ly: dict, font_ps: str, text_w: Optional[float] = None pass prof = font_profile(_font_path_for(ly)) - size_px = float(ly.get("textSize", "12") or 12) * prof["k"] + size_px = float(ly.get("textSize", "12") or 12) * prof["k"] * _SCALE ssd = eng["StyleRun"]["RunArray"][0]["StyleSheet"]["StyleSheetData"] ssd["FontSize"] = Float(size_px) @@ -710,71 +712,20 @@ def _layer_record(top, left, bottom, right, name, opacity, visible, channels, return bytes(rec) -def build_psd(layers, width: int, height: int, out_path: str, workers: int, - progress: Optional[tqdm] = None) -> None: - """layers: list of (name, opacity, visible, offset, Image[, tysh[, lsct]]) - bottom->top. img=None marks a group divider layer (lsct present); it gets - empty channels (2-byte compression markers only).""" - # encode all channels (parallel) - chan_data = [] - for layer in layers: - img = layer[4] - if img is None: - chan_data.append(None) - else: - ch = channel_blocks(img, workers, progress) - chan_data.append(ch) +def build_psd(records: bytearray, payload_f, merged: Image.Image, layers: list, + width: int, height: int, out_path: str) -> None: + """Assemble the PSD file from pre-encoded pieces. - # layer records + channel payloads - records = bytearray() - payload = bytearray() - for li, layer in enumerate(layers): - name, opacity, visible, (lx, ly), img = layer[:5] - tysh = layer[5] if len(layer) > 5 else None - lsct = layer[6] if len(layer) > 6 else None - if img is None: - # group divider: 4 empty channels (RLE marker only, len 2) - ch = [(cid, b"\x00\x01") for cid in (-1, 0, 1, 2)] - l = t = r = b = 0 - if lsct and lsct[1]: - cr = crop_to_canvas(*lsct[1], width, height) - if cr: - l, t, r, b = cr - records += _layer_record(t, l, b, r, name, opacity, visible, ch, - layer_id=li + 1, - lsct=lsct[0] if lsct else None) - for _, p in ch: - payload += p - continue - ch = chan_data[li] - l, t = lx, ly - r = l + img.width - b = t + img.height - records += _layer_record(t, l, b, r, name, opacity, visible, ch, - layer_id=li + 1, tysh=tysh, lsct=lsct) - for _, p in ch: - payload += p - - # merged preview: composite visible layers over transparent - merged = Image.new("RGBA", (width, height), (0, 0, 0, 0)) - for layer in layers: - name, opacity, visible, (lx, ly), img = layer[:5] - if img is not None and visible and opacity > 0: - _opacity_paste(merged, img, (lx, ly), opacity) - merged_rgb = None # keep the RGBA merged image for the 4-channel output - # Merged image data is written RAW with a single compression field and - # all planes in planar order (R G B A). 4 channels because the document - # has transparency (negative layer count references the merged alpha). - if np is not None: - arr = np.asarray(merged) - merged_block = b"\x00\x00" + b"".join( - arr[..., i].tobytes() for i in range(4)) - else: - merged_block = b"\x00\x00" + b"".join( - merged.getchannel(c).tobytes() for c in "RGBA") - - # assemble file. Header channels = 4 (document has transparency, like PS - # writes when layers have alpha). + The renderer already streamed every layer's RLE payloads into `payload_f` + (temp file) and composited the visible layers onto `merged`, so this only + writes header + records + streamed payloads + merged image. Nothing is + loaded back into RAM.""" + import shutil + payload_f.seek(0, 2) + payload_len = payload_f.tell() + payload_f.seek(0) + # Header: channels=4 (document has transparency, like PS writes when + # layers have alpha) fh = open(out_path, "wb") fh.write(struct.pack(">4sH6sHIIHH", b"8BPS", 1, b"\x00" * 6, 4, height, width, 8, 3)) fh.write(struct.pack(">I", 0)) # color mode data @@ -782,17 +733,23 @@ def build_psd(layers, width: int, height: int, out_path: str, workers: int, # Layer info section: [Length(4)] + [layer count(2) + records + channel data] # Negative layer count: abs = layer count, first alpha channel of the - # merged image holds the merged transparency (PS writes this when the - # document has transparent layers). - layer_info_body = struct.pack(">h", -len(layers)) + bytes(records) + bytes(payload) - layer_info = struct.pack(">I", len(layer_info_body)) + layer_info_body - # layer & mask section: length = layer info + global layer mask info - # (the outer length does NOT include its own 4-byte length field) - lam = layer_info + struct.pack(">I", 0) # global layer mask info - fh.write(struct.pack(">I", len(lam)) + lam) + # merged image holds the merged transparency. + body_len = 2 + len(records) + payload_len + # layer & mask section: length = [4-byte inner length] + body + + # global layer mask info (4); the outer length does NOT include its own + # 4-byte length field + fh.write(struct.pack(">I", 4 + body_len + 4)) + fh.write(struct.pack(">I", body_len)) + fh.write(struct.pack(">h", -len(layers))) + fh.write(bytes(records)) + shutil.copyfileobj(payload_f, fh) # stream, never load payloads into RAM + fh.write(struct.pack(">I", 0)) # global layer mask info - # merged image data (RGB, RAW) - fh.write(merged_block) + # merged image data (RGB, RAW), one plane at a time + fh.write(b"\x00\x00") + for c in "RGBA": + fh.write(merged.getchannel(c).tobytes()) + del merged fh.close() @@ -810,6 +767,10 @@ def _safe_workers(workers: int) -> int: def convert(ppl_path: str, out_path: str, preview_path: Optional[str] = None, workers: int = 4, type_layer: bool = False) -> None: workers = _safe_workers(workers) + if _SCALE >= 2.0: + # 2x canvases are 4x the pixels; parallel channel encoding would + # duplicate layer buffers across processes and exhaust RAM/ZRAM/SWAP + workers = 1 import tempfile with tempfile.TemporaryDirectory() as tmp: with zipfile.ZipFile(ppl_path) as zf: @@ -819,8 +780,8 @@ def convert(ppl_path: str, out_path: str, preview_path: Optional[str] = None, layers, groups = parse_layers(text) mw = re.search(r"^width=(\d+)", text, re.M) mh = re.search(r"^height=(\d+)", text, re.M) - width = int(mw.group(1)) if mw else 0 - height = int(mh.group(1)) if mh else 0 + width = int(float(mw.group(1)) * _SCALE) if mw else 0 + height = int(float(mh.group(1)) * _SCALE) if mh else 0 # --- group mapping --- group_map = {} @@ -866,10 +827,21 @@ def convert(ppl_path: str, out_path: str, preview_path: Optional[str] = None, rendered = [] skipped = 0 + # Streaming render+encode: each layer is rendered, composited onto + # the merged canvas, RLE-encoded into the payload temp file and then + # released. Only ~1 layer + the merged canvas stay in RAM, instead of + # all 142 layers' images (the 2x build previously exhausted RAM and + # swap and got OOM-killed). + merged = Image.new("RGBA", (width, height), (0, 0, 0, 0)) + payload_f = tempfile.TemporaryFile() + records = bytearray() rp = tqdm(total=len(layers), desc="渲染图层", unit="层", file=sys.stderr) + ep = tqdm(total=len(layers) * 4, desc="编码通道", unit="通道", + file=sys.stderr) for i, ly in enumerate(layers): gid = ly.get("groupId") gname = gvis = gb = None + layer_id = len(rendered) + 1 if gid and gid in group_ranges: g = group_map.get(gid, {}) gname = g.get("name", "") or f"group_{gid}" @@ -880,6 +852,16 @@ def convert(ppl_path: str, out_path: str, preview_path: Optional[str] = None, if group_ranges[gid][0] == i: rendered.append((gname, 255, gvis, (0, 0), None, None, (3, gb))) + # group open marker: 4 empty channels + ch = [(cid, b"\x00\x01") for cid in (-1, 0, 1, 2)] + cr = crop_to_canvas(*gb, width, height) if gb else None + l = t = r = b = 0 + if cr: + l, t, r, b = cr + records += _layer_record(t, l, b, r, gname, 255, gvis, + ch, layer_id=layer_id, lsct=3) + for _, p in ch: + payload_f.write(p) img_path = None if ly["_type"] == "bitmap": uri = ly.get("uri", "") @@ -916,30 +898,41 @@ def convert(ppl_path: str, out_path: str, preview_path: Optional[str] = None, except Exception as exc: print(f" warning: text layer {name} -> TySh failed: {exc}", file=sys.stderr) - rendered.append((name, opacity, visible, (lx, ly_), img, tysh)) + if visible and opacity > 0: + _opacity_paste(merged, img, (lx, ly_), opacity) + ch = channel_blocks(img, workers, ep) + records += _layer_record(ly_, lx, ly_ + img.height, lx + img.width, + name, opacity, visible, + ch, layer_id=layer_id, tysh=tysh) + for _, p in ch: + payload_f.write(p) + del ch, img + rendered.append((name, opacity, visible, (lx, ly_), None, tysh, + None)) rp.update(1) + # group close marker (top of the app group) if gid and gid in group_ranges and group_ranges[gid][1] == i: rendered.append((gname, 255, gvis, (0, 0), None, None, (1, None))) + ch = [(cid, b"\x00\x01") for cid in (-1, 0, 1, 2)] + records += _layer_record(0, 0, 0, 0, gname, 255, gvis, + ch, layer_id=layer_id + 1, lsct=1) + for _, p in ch: + payload_f.write(p) rp.close() + ep.close() print(f"渲染完成: {len(rendered)}/{len(layers)} 层 (跳过 {skipped}), " f"画布 {width}x{height}") - ep = tqdm(total=len(rendered) * 4 + 3, desc="编码通道", unit="通道", - file=sys.stderr) - build_psd(rendered, width, height, out_path, workers, ep) - ep.close() + build_psd(records, payload_f, merged, rendered, width, height, out_path) + payload_f.close() print(f"PSD 已写出: {out_path}") if preview_path: - canvas = Image.new("RGBA", (width, height), (0, 0, 0, 0)) - for layer in rendered: - name, opacity, visible, (lx, ly), img = layer[:5] - if img is not None and visible and opacity > 0: - _opacity_paste(canvas, img, (lx, ly), opacity) - # downscale for easy viewing - preview = canvas.convert("RGB") - if preview.width > 2000: + # downscale for easy viewing, unless the project was upscaled + # (then keep the full-res composite so the 2x detail is visible) + preview = merged.convert("RGB") + if _SCALE == 1.0 and preview.width > 2000: scale = 2000 / preview.width preview = preview.resize( (int(preview.width * scale), int(preview.height * scale)), @@ -959,8 +952,13 @@ def main(): ap.add_argument("--type-layer", action="store_true", help="convert text layers to editable TypeLayers (TySh) " "instead of rasterized pixels") + ap.add_argument("--scale", type=float, default=1.0, + help="upscale the whole vector project by this factor " + "(canvas, layer rects, text size, shadows) " + "(default: 1.0)") args = ap.parse_args() + set_scale(args.scale) out = args.output if not out: out = str(Path(args.input).with_suffix(".psd"))