X

Client/Scripts/test-short.py

@@ -1,527 +0,0 @@
-import gzip
-import json
-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
-LONG = False
-POST_IDS = [
-    "139395291",
-]
-
-
-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 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()]
-
-
-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:
-            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
-
-                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_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:
-                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
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())