LONG?

Client/Scripts/test-short.py

@@ -1,3 +1,5 @@
+import gzip
+import json
 import os
 import zlib
 
@@ -8,12 +10,9 @@ MAGIC = b"stealth_pngcomp"
 PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
 USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:126.0) Gecko/20100101 Firefox/126.0"
 REQUEST_TIMEOUT = 45
+LONG = False
 POST_IDS = [
-    "116370378",
-    "137989718",
-    "127319128",
-    "134714737",
-    "121968369",
+    "139395291",
 ]
 
 
@@ -252,6 +251,167 @@ def stream_magic_and_length(session, url, referer):
         raise ValueError("Not enough PNG data to read header bits.")
 
 
+def stream_magic_length_and_payload(session, url, referer):
+    headers = {
+        "Referer": referer,
+        "User-Agent": USER_AGENT,
+        "Accept-Encoding": "identity",
+    }
+    required_bytes = len(MAGIC) + 4
+    required_bits = required_bytes * 8
+    with session.get(url, headers=headers, stream=True, timeout=REQUEST_TIMEOUT) as response:
+        response.raise_for_status()
+        response.raw.decode_content = False
+        stream = response.raw
+
+        signature = read_exact(stream, len(PNG_SIGNATURE))
+        if signature != PNG_SIGNATURE:
+            raise ValueError("Not a PNG file.")
+
+        width = height = None
+        row_bytes = None
+        bpp = None
+        rows_needed_header = None
+        rows_needed_total = None
+        rows_processed = 0
+        prev_row = None
+        scanline_buf = bytearray()
+        out_bytes = bytearray()
+        current_byte = 0
+        bits_in_current = 0
+        decompressor = zlib.decompressobj()
+        length_bits = None
+        total_bytes = None
+
+        while True:
+            length_bytes = read_exact(stream, 4)
+            chunk_length = int.from_bytes(length_bytes, "big")
+            chunk_type = read_exact(stream, 4)
+
+            if chunk_type == b"IHDR":
+                ihdr = read_exact(stream, chunk_length)
+                width, height, bit_depth, color_type, compression, filter_method, interlace = parse_ihdr(ihdr)
+                if compression != 0 or filter_method != 0 or interlace != 0:
+                    raise ValueError("Unsupported PNG compression/filter/interlace.")
+                if bit_depth != 8:
+                    raise ValueError("Unsupported PNG bit depth.")
+                if color_type not in (4, 6):
+                    raise ValueError("PNG does not have an alpha channel.")
+                if width <= 0 or height <= 0:
+                    raise ValueError("Invalid PNG dimensions.")
+                channels = 2 if color_type == 4 else 4
+                bpp = channels
+                row_bytes = width * bpp
+                rows_needed_header = (required_bits + width - 1) // width
+                if rows_needed_header > height:
+                    raise ValueError("PNG too short to read header bits.")
+                prev_row = bytearray(row_bytes)
+                read_exact(stream, 4)
+                continue
+
+            if chunk_type == b"IDAT":
+                if row_bytes is None:
+                    raise ValueError("IDAT before IHDR.")
+                remaining = chunk_length
+                while remaining > 0:
+                    chunk = read_exact(stream, min(16384, remaining))
+                    remaining -= len(chunk)
+                    output = decompressor.decompress(chunk)
+                    if output:
+                        scanline_buf.extend(output)
+
+                    while len(scanline_buf) >= row_bytes + 1:
+                        filter_type = scanline_buf[0]
+                        raw_row = scanline_buf[1:1 + row_bytes]
+                        del scanline_buf[:1 + row_bytes]
+                        row = unfilter_row(filter_type, raw_row, prev_row, bpp)
+                        prev_row = row
+                        rows_processed += 1
+
+                        alpha_offset = bpp - 1
+                        for i in range(alpha_offset, row_bytes, bpp):
+                            bit = row[i] & 1
+                            current_byte = (current_byte << 1) | bit
+                            bits_in_current += 1
+                            if bits_in_current == 8:
+                                out_bytes.append(current_byte)
+                                bits_in_current = 0
+                                current_byte = 0
+
+                        if total_bytes is None and len(out_bytes) >= required_bytes:
+                            magic = bytes(out_bytes[: len(MAGIC)])
+                            length_bits = int.from_bytes(
+                                out_bytes[len(MAGIC):required_bytes],
+                                "big",
+                            )
+                            if magic != MAGIC:
+                                return {
+                                    "magic": magic,
+                                    "length_bits": length_bits,
+                                    "width": width,
+                                    "height": height,
+                                    "rows_needed_header": rows_needed_header,
+                                    "rows_needed_total": None,
+                                    "rows_processed": rows_processed,
+                                    "payload": None,
+                                }
+                            payload_len = length_bits // 8
+                            total_bytes = required_bytes + payload_len
+                            rows_needed_total = (total_bytes * 8 + width - 1) // width
+                            if rows_needed_total > height:
+                                raise ValueError("PNG too short to read payload bits.")
+                            if len(out_bytes) >= total_bytes:
+                                return {
+                                    "magic": magic,
+                                    "length_bits": length_bits,
+                                    "width": width,
+                                    "height": height,
+                                    "rows_needed_header": rows_needed_header,
+                                    "rows_needed_total": rows_needed_total,
+                                    "rows_processed": rows_processed,
+                                    "payload": bytes(out_bytes[required_bytes:total_bytes]),
+                                }
+
+                        if total_bytes is not None and len(out_bytes) >= total_bytes:
+                            return {
+                                "magic": bytes(out_bytes[: len(MAGIC)]),
+                                "length_bits": length_bits,
+                                "width": width,
+                                "height": height,
+                                "rows_needed_header": rows_needed_header,
+                                "rows_needed_total": rows_needed_total,
+                                "rows_processed": rows_processed,
+                                "payload": bytes(out_bytes[required_bytes:total_bytes]),
+                            }
+
+                read_exact(stream, 4)
+                continue
+
+            read_exact(stream, chunk_length)
+            read_exact(stream, 4)
+            if chunk_type == b"IEND":
+                break
+
+        raise ValueError("Not enough PNG data to read payload bits.")
+
+
+def decode_payload(payload):
+    try:
+        json_bytes = gzip.decompress(payload)
+    except Exception:
+        return payload.decode("utf-8", errors="replace"), "raw"
+    try:
+        data = json.loads(json_bytes.decode("utf-8"))
+    except json.JSONDecodeError:
+        return json_bytes.decode("utf-8", errors="replace"), "text"
+    if "Comment" in data and isinstance(data["Comment"], str):
+        try:
+            data["Comment"] = json.loads(data["Comment"])
+        except json.JSONDecodeError:
+            pass
+    return data, "json"
+
+
 def read_post_ids(path):
     with open(path, "r") as handle:
         return [line.strip() for line in handle if line.strip()]
@@ -306,32 +466,59 @@ def main() -> int:
             continue
 
         for page_index, url in png_pages:
-            try:
-                info = stream_magic_and_length(
-                    session,
-                    url,
-                    f"https://www.pixiv.net/artworks/{post_id}",
-                )
-            except Exception as exc:
-                print(f"  Page {page_index}: error {exc}")
-                continue
+            referer = f"https://www.pixiv.net/artworks/{post_id}"
+            if LONG:
+                try:
+                    info = stream_magic_length_and_payload(session, url, referer)
+                except Exception as exc:
+                    print(f"  Page {page_index}: error {exc}")
+                    continue
+
+                magic = info["magic"]
+                if magic != MAGIC:
+                    magic_ascii = magic.decode("ascii", errors="replace")
+                    print(
+                        f"  Page {page_index}: no stealth, "
+                        f"magic={magic_ascii!r} hex={magic.hex()}"
+                    )
+                    continue
 
-            magic = info["magic"]
-            if magic == MAGIC:
                 length_bits = info["length_bits"]
                 length_bytes = length_bits // 8 if length_bits is not None else None
                 print(
                     f"  Page {page_index}: magic ok, "
                     f"len_bits={length_bits}, len_bytes={length_bytes}, "
-                    f"width={info['width']}, rows={info['rows_needed']}"
+                    f"width={info['width']}, rows={info['rows_needed_total']}"
                 )
+                payload = info["payload"] or b""
+                data, data_type = decode_payload(payload)
+                if data_type == "json":
+                    print(json.dumps(data, ensure_ascii=True, sort_keys=True, indent=2))
+                else:
+                    print(data)
             else:
-                magic_ascii = magic.decode("ascii", errors="replace")
-                print(
-                    f"  Page {page_index}: magic mismatch "
-                    f"ascii={magic_ascii!r} hex={magic.hex()}, "
-                    f"width={info['width']}, rows={info['rows_needed']}"
-                )
+                try:
+                    info = stream_magic_and_length(session, url, referer)
+                except Exception as exc:
+                    print(f"  Page {page_index}: error {exc}")
+                    continue
+
+                magic = info["magic"]
+                if magic == MAGIC:
+                    length_bits = info["length_bits"]
+                    length_bytes = length_bits // 8 if length_bits is not None else None
+                    print(
+                        f"  Page {page_index}: magic ok, "
+                        f"len_bits={length_bits}, len_bytes={length_bytes}, "
+                        f"width={info['width']}, rows={info['rows_needed']}"
+                    )
+                else:
+                    magic_ascii = magic.decode("ascii", errors="replace")
+                    print(
+                        f"  Page {page_index}: magic mismatch "
+                        f"ascii={magic_ascii!r} hex={magic.hex()}, "
+                        f"width={info['width']}, rows={info['rows_needed']}"
+                    )
 
     return 0