test-quick-stealth

Client/Scripts/test-short.py

@@ -0,0 +1,340 @@
+import os
+import zlib
+
+import requests
+
+
+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
+POST_IDS = [
+    "116370378",
+    "137989718",
+    "127319128",
+    "134714737",
+    "121968369",
+]
+
+
+def read_dotenv_value(path, key):
+    try:
+        with open(path, "r") as env_file:
+            for line in env_file:
+                line = line.strip()
+                if not line or line.startswith("#") or "=" not in line:
+                    continue
+                k, v = line.split("=", 1)
+                if k == key:
+                    return v
+    except FileNotFoundError:
+        return None
+    return None
+
+
+def get_phpsessid():
+    phpsessid = os.getenv("PHPSESSID")
+    if phpsessid:
+        return phpsessid
+    script_dir = os.path.dirname(os.path.abspath(__file__))
+    env_candidates = [
+        os.path.join(script_dir, "..", ".env"),
+        os.path.join(script_dir, "..", "..", ".env"),
+    ]
+    for env_path in env_candidates:
+        phpsessid = read_dotenv_value(os.path.abspath(env_path), "PHPSESSID")
+        if phpsessid:
+            return phpsessid
+    raise RuntimeError("PHPSESSID is not set in the environment or .env")
+
+
+def build_session(phpsessid):
+    session = requests.Session()
+    session.headers.update({"User-Agent": USER_AGENT, "Referer": "https://www.pixiv.net/"})
+    session.cookies.update({"PHPSESSID": phpsessid})
+    return session
+
+
+def fetch_post_pages(session, post_id):
+    url = f"https://www.pixiv.net/ajax/illust/{post_id}/pages"
+    headers = {"Referer": f"https://www.pixiv.net/artworks/{post_id}"}
+    response = session.get(url, headers=headers, timeout=REQUEST_TIMEOUT)
+    response.raise_for_status()
+    data = response.json()
+    return data.get("body") or []
+
+
+def read_exact(stream, size):
+    chunks = []
+    remaining = size
+    while remaining > 0:
+        data = stream.read(remaining)
+        if not data:
+            raise EOFError("Unexpected end of stream.")
+        chunks.append(data)
+        remaining -= len(data)
+    return b"".join(chunks)
+
+
+def paeth_predictor(a, b, c):
+    p = a + b - c
+    pa = abs(p - a)
+    pb = abs(p - b)
+    pc = abs(p - c)
+    if pa <= pb and pa <= pc:
+        return a
+    if pb <= pc:
+        return b
+    return c
+
+
+def unfilter_row(filter_type, raw, prev, bpp):
+    raw = memoryview(raw)
+    row = bytearray(len(raw))
+    if filter_type == 0:
+        row[:] = raw
+        return row
+    if filter_type == 1:
+        for i in range(len(raw)):
+            left = row[i - bpp] if i >= bpp else 0
+            row[i] = (raw[i] + left) & 0xFF
+        return row
+    if filter_type == 2:
+        for i in range(len(raw)):
+            row[i] = (raw[i] + prev[i]) & 0xFF
+        return row
+    if filter_type == 3:
+        for i in range(len(raw)):
+            left = row[i - bpp] if i >= bpp else 0
+            up = prev[i]
+            row[i] = (raw[i] + ((left + up) // 2)) & 0xFF
+        return row
+    if filter_type == 4:
+        for i in range(len(raw)):
+            left = row[i - bpp] if i >= bpp else 0
+            up = prev[i]
+            up_left = prev[i - bpp] if i >= bpp else 0
+            row[i] = (raw[i] + paeth_predictor(left, up, up_left)) & 0xFF
+        return row
+    raise ValueError(f"Unsupported PNG filter type: {filter_type}")
+
+
+def parse_ihdr(data):
+    if len(data) != 13:
+        raise ValueError("Invalid IHDR chunk length.")
+    width = int.from_bytes(data[0:4], "big")
+    height = int.from_bytes(data[4:8], "big")
+    bit_depth = data[8]
+    color_type = data[9]
+    compression = data[10]
+    filter_method = data[11]
+    interlace = data[12]
+    return width, height, bit_depth, color_type, compression, filter_method, interlace
+
+
+def stream_magic_and_length(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 = None
+        rows_processed = 0
+        prev_row = None
+        scanline_buf = bytearray()
+        out_bytes = bytearray()
+        current_byte = 0
+        bits_in_current = 0
+        decompressor = zlib.decompressobj()
+
+        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 = (required_bits + width - 1) // width
+                if rows_needed > 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 not output:
+                        continue
+                    scanline_buf.extend(output)
+
+                    while rows_processed < rows_needed and 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 len(out_bytes) >= required_bytes:
+                                    return {
+                                        "magic": bytes(out_bytes[: len(MAGIC)]),
+                                        "length_bits": int.from_bytes(
+                                            out_bytes[len(MAGIC):required_bytes],
+                                            "big",
+                                        ),
+                                        "width": width,
+                                        "height": height,
+                                        "rows_needed": rows_needed,
+                                        "rows_processed": rows_processed,
+                                    }
+                    if rows_processed >= rows_needed and len(out_bytes) >= required_bytes:
+                        return {
+                            "magic": bytes(out_bytes[: len(MAGIC)]),
+                            "length_bits": int.from_bytes(
+                                out_bytes[len(MAGIC):required_bytes],
+                                "big",
+                            ),
+                            "width": width,
+                            "height": height,
+                            "rows_needed": rows_needed,
+                            "rows_processed": rows_processed,
+                        }
+
+                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 header bits.")
+
+
+def read_post_ids(path):
+    with open(path, "r") as handle:
+        return [line.strip() for line in handle if line.strip()]
+
+
+def load_post_ids(args):
+    if args:
+        if len(args) == 1 and os.path.isfile(args[0]):
+            return read_post_ids(args[0])
+        return [arg for arg in args if arg.isdigit()]
+    default_path = os.path.join(
+        os.path.dirname(os.path.abspath(__file__)),
+        "..",
+        "txt logs",
+        "test.txt",
+    )
+    if os.path.exists(default_path):
+        return read_post_ids(default_path)
+    return []
+
+
+def main() -> int:
+    try:
+        phpsessid = get_phpsessid()
+    except Exception as exc:
+        print(f"Failed to load PHPSESSID: {exc}")
+        return 1
+
+    post_ids = list(POST_IDS)
+    if not post_ids:
+        print("No post IDs provided.")
+        return 1
+
+    session = build_session(phpsessid)
+
+    for post_id in post_ids:
+        print(f"Post {post_id}")
+        try:
+            pages = fetch_post_pages(session, post_id)
+        except Exception as exc:
+            print(f"  Failed to fetch pages: {exc}")
+            continue
+
+        png_pages = []
+        for idx, page in enumerate(pages):
+            original = page.get("urls", {}).get("original")
+            if original and original.lower().endswith(".png"):
+                png_pages.append((idx + 1, original))
+
+        if not png_pages:
+            print("  No PNG pages found.")
+            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
+
+            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
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())