blog/_posts/2026-11-01-compress.md

---
layout: post
title: 关于压缩的各种探索
tags: [压缩, 探索, Quine, zstd]
---

  压缩……在消除冗余的同时，又创造了新的混乱？<!--more-->    
  
# 对压缩比的追求
  上次我在做[博客的离线模式](/2026/10/01/quine.html)时，为了降低带宽的消耗，我把我的博客压缩包又用7-Zip额外压缩了一遍，从大约16MiB压缩到了4MiB，效果感觉挺不错。另外我又试了试几个常见的压缩算法，比如Zip、zstd、BZip2等，全部使用最高等级压缩，但压缩效果都不如使用LZMA2的7-Zip好，速度甚至还比7-Zip慢，看来LZMA2还是相当强啊。   
  所以我在想，LZMA2是不是最强的压缩算法呢？虽然不同压缩算法可能对不同类型的数据效果不一样，但目前来看对我的数据来说，LZMA2看起来是最强的啊……不过这些只是常见算法，如果是用那些因为压缩速度慢而不太常见的算法，会不会比LZMA2更强呢？   
## ZPAQ
  我搜了一下，PAQ系列算法似乎是公认最强的，通过类似机器学习的算法花大量的时间达到更高的压缩比。其中比较知名的就是[ZPAQ](https://github.com/zpaq/zpaq)，这个程序已经有10年不更新了，当然对于算法来说这点倒也不是什么问题，毕竟LZMA2也是十几年前的算法。最终我试着用它压缩我的博客压缩包，用了一分钟左右的时间，压缩到了3.5MiB，比7-Zip压缩的少了大约10%，感觉效果还不错。   
## PAQ8PX
  当然ZPAQ只是PAQ系列中的其中一个算法，有些人为了追求更高的压缩比，依然在对其进行改进，据我所知，目前仍在更新、压缩效果也非常出色的应该是[PAQ8PX](https://github.com/hxim/paq8px)，我试着用v217版本压缩了一下，最终把压缩包压到了3.1MiB，真的是非常强。而代价也是极其高，首先压缩花了一个半小时，而且在这个过程中需要占用30GiB内存，更关键的问题是在解压时同样需要消耗这么多时间和资源，所以说这个算法真的是只能追求压缩比，基本上没办法正常使用……   
## NNCP
  除了PAQ系列之外，FFMPEG的作者[Fabrice Bellard](https://bellard.org)也做过一个使用机器学习的压缩算法，叫做[NNCP](https://bellard.org/nncp/)，使用了和现代LLM同款的Transformer模型。于是我编译了一下，试着在RTX4090上压缩我的压缩包，也差不多用了一个半小时，结果压缩比还没LZMA2高😂，给我压到了4.1MiB，也许是因为我的博客里有些图片，而NNCP貌似是专攻英文文本的，所以效果差了不少……又一次证明了LZMA2真的很强啊～   
## cmix
  后来我又听说还有一个叫做[cmix](https://github.com/byronknoll/cmix)的压缩算法，一样是基于机器学习，据说效果好像还不错，只不过在我实际用v21版本测试的时候不知道为什么总是压到后面的时候就会崩溃……这个项目也有2年没更新了，可能是代码有什么BUG。而且和PAQ8PX一样内存占用很高，用时非常长，但就从崩溃的位置来看，它压缩一半的结果还是比PAQ8PX更大，看来这个压缩算法的地位有点尴尬啊……不知道它是不是又是针对文本特化过的算法，但目前的结果来看感觉不如PAQ8PX。   
## 总结
  最后PAQ8PX在这次测试中获得了最终的冠军🏆，我的博客压缩包里面包含的数据分布也算是有一定代表性吧，有一定的图片，大量的纯文本网页，和不少网站的结构应该很相似，所以这个压缩结果可能对不少数据都有一定的参考性。不过无论是PAQ8PX、NNCP还是cmix，它们压缩和解压需要的资源实在是太多了，不太适合一般情况下使用，至于ZPAQ虽然也是那种压缩和解压消耗相同的资源的算法，但作为十年前的东西来说，它在现在的速度还算是能在接受范围内，所以说如果想把不算太大的文件压到尽可能小，用ZPAQ看起来是最合适的选择，至于绝大多数的文件，还是用使用LZMA2算法的7-Zip吧～   

# 关于自包含压缩包的继续研究
## zstd Quine
  在上次我做博客压缩包的时候，AI告诉我LZMA2和zstd都可以做能包含大文件的Quine，但我当时却以“zstd感觉不是很知名”的理由选择了LZMA2。然而实际上zstd甚至都有专门的[RFC 8878](https://datatracker.ietf.org/doc/html/rfc8878)，甚至有不少网站都已经选择了zstd作为传输的压缩算法……既然如此，那我就让AI再简单研究一下zstd Quine吧，搜了一下以前也从来没人做过。   
  AI看了一眼规范之后，说这个比7z Quine制作起来在结构上会更简单，但在计算校验值上会更困难，简单是因为zstd支持一种叫做“可跳过帧”的东西，有这个东西就能在构建的时候有更多的自由度，不过zstd用的是一种叫做“有限状态熵”的编码方式，这个要重新研究一下。至于困难是因为zstd默认使用XXH64来校验，相比于7z格式默认使用的CRC32来说，它是非线性的，因此很难像CRC32那样靠几个公式轻松计算出来。不过有一点好处是XXH64的计算速度比CRC32快不少，而且zstd只校验XXH64结果的低32位，因此即使是通过爆破方式，也用不了多长时间就能计算出来。当然这个值和7-Zip一样都是可选的，其实不算也不影响什么，但既然能算的话就算一下呗～   
  然后AI就开始想办法生成了，用了一个小时就做出来了，最开始它是直接做的`.zst`格式，但libarchive又开始出幺蛾子，显示`Input requires too many filters for decoding`错误，我试了一下[Gzip Quine](http://swtch.com/r.gz)用libarchive也有相同的问题……实在是令人难受，所以我又要求它改成`.tar.zst`格式，这样直接用Windows资源管理器也能直接打开，还增加了和我之前7z Quine一样的文件夹打包功能，说不定以后会用得上呢。   
  顺便一说，XXH64虽然说是非线性的，但作为非密码学哈希，倒也可以在不爆破的情况下构造值，于是以下就是最终的生成代码：   
<details markdown="1">
<summary markdown="span">
Show Code
</summary>

```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Pack files/directories and a byte-identical copy of this .tar.zst archive.

Example: python3 blogquine_tarzstd.py ./site ./site.tar.zst
One directory packs its contents; multiple inputs retain their basenames.
"""
from __future__ import annotations

import argparse
import hashlib
import os
from pathlib import Path, PurePosixPath
import stat
import sys
import tempfile
import ctypes
import ctypes.util
import io
import shutil
import subprocess
import tarfile
from dataclasses import dataclass
from functools import lru_cache

BLOCK = 65536
MASK = (1 << 64) - 1
P1, P2 = 0x9E3779B185EBCA87, 0xC2B2AE3D27D4EB4F
P3, P4 = 0x165667B19E3779F9, 0x85EBCA77C2B2AE63
I1, I2 = pow(P1, -1, 1 << 64), pow(P2, -1, 1 << 64)
LL_BASE = tuple(range(16)) + (16,18,20,22,24,28,32,40,48,64,128,256,512,1024,2048,4096,8192,16384,32768,65536)
LL_BITS = (0,)*16 + (1,1,1,1,2,2,3,3,4,6,7,8,9,10,11,12,13,14,15,16)
ML_BASE = tuple(range(3,35)) + (35,37,39,41,43,47,51,59,67,83,99,131,259,515,1027,2051,4099,8195,16387,32771,65539)
ML_BITS = (0,)*32 + (1,1,1,1,2,2,3,3,4,4,5,7,8,9,10,11,12,13,14,15,16)


def _rol(x: int, n: int) -> int:
    return ((x << n) | (x >> (64 - n))) & MASK


def _ror(x: int, n: int) -> int:
    return ((x >> n) | (x << (64 - n))) & MASK


def _round(v: int, w: int) -> int:
    return _rol((v + w * P2) & MASK, 31) * P1 & MASK


def checksum_for_size(size: int) -> int:
    """XXH64 digest when all four terminal accumulators are zero."""
    if size < 32 or size % 32:
        raise ValueError("decoded size must be a positive multiple of 32")
    h = 0
    for _ in range(4):
        h = (h * P1 + P4) & MASK
    h = (h + size) & MASK
    h ^= h >> 33
    h = h * P2 & MASK
    h ^= h >> 29
    h = h * P3 & MASK
    return h ^ (h >> 32)


def force_words(data: bytes | bytearray, offset: int) -> bytes:
    """Force the four final lanes to zero using four unique aligned words."""
    if len(data) % 32 or offset % 8 or not 0 <= offset <= len(data)-32:
        raise ValueError("invalid checksum patch position")
    before = [(P1+P2)&MASK, P2, 0, (-P1)&MASK]
    after = [0, 0, 0, 0]
    view = memoryview(data)
    for p in range(0, offset, 8):
        lane = (p//8) % 4
        before[lane] = _round(before[lane], int.from_bytes(view[p:p+8], "little"))
    for p in range(len(data)-8, offset+31, -8):
        lane = (p//8) % 4
        w = int.from_bytes(view[p:p+8], "little")
        after[lane] = (_ror(after[lane]*I1 & MASK, 31) - w*P2) & MASK
    result = bytearray()
    for p in range(offset, offset+32, 8):
        lane = (p//8) % 4
        w = ((_ror(after[lane]*I1 & MASK, 31) - before[lane])*I2) & MASK
        result += w.to_bytes(8, "little")
    return bytes(result)


def _length_code(value, bases, widths):
    for symbol, (base, width) in enumerate(zip(bases, widths)):
        if base <= value < base + (1 << width):
            return symbol, value-base, width
    raise ValueError(f"unencodable Zstd length {value}")


def _literal_header(n):
    if n < 32:
        return bytes((n << 3,))
    if n < 4096:
        return ((n << 4)|4).to_bytes(2,"little")
    return ((n << 4)|12).to_bytes(3,"little")


def _sequence(ll, ml, distance):
    if distance < 1:
        raise ValueError("back reference must be positive")
    lc, lv, ln = _length_code(ll, LL_BASE, LL_BITS)
    mc, mv, mn = _length_code(ml, ML_BASE, ML_BITS)
    oc = (distance+3).bit_length()-1
    if oc > 31:
        raise ValueError("input needs a Zstd history window larger than 2 GiB")
    ov = distance+3-(1 << oc)
    bits = lv | (mv << ln) | (ov << (ln+mn)) | (1 << (ln+mn+oc))
    stream = bits.to_bytes((ln+mn+oc+8)//8, "little")
    return bytes((1, 0x54, lc, oc, mc)) + stream


def _copy_block(literals, length, distance):
    if len(literals)+length > 131072:
        raise ValueError("Zstd decoded block exceeds 128 KiB")
    seq = _sequence(len(literals), length, distance)
    body = _literal_header(len(literals)) + literals + seq
    return ((len(body)<<3)|4).to_bytes(3,"little") + body, seq


def _raw_header(size):
    return (size << 3).to_bytes(3, "little")


def _parts(total):
    """Nonempty pieces with length >=3 and bounded decoded block size."""
    out = []
    while total > BLOCK:
        n = BLOCK
        if total-n < 3:
            n -= 3-(total-n)
        out.append(n)
        total -= n
    if total:
        if total < 3:
            raise ValueError("internal block split too short")
        out.append(total)
    return out


def _header(name, size, mtime):
    info = tarfile.TarInfo(name)
    info.size, info.mode, info.mtime = size, 0o644, mtime
    info.uid = info.gid = 0
    return info.tobuf(format=tarfile.PAX_FORMAT, encoding="utf-8", errors="surrogateescape")


def _window_descriptor(required):
    if required > (1 << 31):
        raise ValueError("input needs a history window larger than the supported 2 GiB")
    for wd in range(256):
        base = 1 << (10 + (wd >> 3))
        size = base + (base >> 3)*(wd & 7)
        if size >= required:
            return wd, size
    raise ValueError("input exceeds the Zstandard frame window format")


@lru_cache(maxsize=1)
def _libzstd():
    path = ctypes.util.find_library("zstd")
    if not path:
        raise RuntimeError("libzstd is required (install the Zstandard library)")
    lib = ctypes.CDLL(path)
    lib.ZSTD_decompress.argtypes = [ctypes.c_void_p,ctypes.c_size_t,ctypes.c_void_p,ctypes.c_size_t]
    lib.ZSTD_decompress.restype = ctypes.c_size_t
    lib.ZSTD_isError.argtypes = [ctypes.c_size_t]
    lib.ZSTD_isError.restype = ctypes.c_uint
    lib.ZSTD_getErrorName.argtypes = [ctypes.c_size_t]
    lib.ZSTD_getErrorName.restype = ctypes.c_char_p
    lib.ZSTD_compressBound.argtypes = [ctypes.c_size_t]
    lib.ZSTD_compressBound.restype = ctypes.c_size_t
    lib.ZSTD_compress.argtypes = [ctypes.c_void_p,ctypes.c_size_t,ctypes.c_void_p,ctypes.c_size_t,ctypes.c_int]
    lib.ZSTD_compress.restype = ctypes.c_size_t
    return lib


def decompress(data: bytes, decoded_size: int) -> bytes:
    """Decode and verify the one frame through the system reference library."""
    try:
        lib = _libzstd()
    except RuntimeError:
        try:
            import zstandard
        except ImportError:
            command = shutil.which("zstd")
            if not command:
                raise RuntimeError("verification needs libzstd, Python zstandard, or the zstd executable") from None
            result = subprocess.run([command,"-q","-d","-c","--long=31"],input=data,capture_output=True)
            if result.returncode:
                raise RuntimeError("Zstandard verification: "+result.stderr.decode(errors="replace").strip())
            raw = result.stdout
        else:
            decoder = zstandard.ZstdDecompressor(max_window_size=1 << 31)
            with decoder.stream_reader(io.BytesIO(data),read_across_frames=True) as reader:
                raw = reader.read(decoded_size+1)
                if reader.read(1):
                    raise RuntimeError("decoded data exceeds the declared size")
        if len(raw) != decoded_size:
            raise RuntimeError(f"decoded size mismatch: {len(raw)} != {decoded_size}")
        return raw
    dest = ctypes.create_string_buffer(decoded_size)
    n = lib.ZSTD_decompress(dest, decoded_size, data, len(data))
    if lib.ZSTD_isError(n):
        raise RuntimeError("Zstandard verification: " + lib.ZSTD_getErrorName(n).decode())
    if n != decoded_size:
        raise RuntimeError(f"decoded size mismatch: {n} != {decoded_size}")
    return dest.raw


@dataclass
class _Chunk:
    kind: str
    f: int
    size: int
    decoded: int
    content: bytes = b""
    source: int = 0


def _build_quine(prefix_tar: bytes, quine_name: str, *, mtime=0, compressed=None):
    """Return (archive, info, decoded TAR), adding only a self member.

    The leading skippable frame is the quotation template. No seed TAR member
    is created. The baseline stores source payloads once as raw Zstd literals.
    """
    template_size = 512
    self_header = _header(quine_name, 1, mtime)
    q = len(prefix_tar)+len(self_header)
    chunks = []
    f, o = template_size+6, 0
    # Native prefix blocks may use a wider history than the quotation replay.
    # Also, a block may regenerate at most min(window_size, 128 KiB).
    max_distance = max(1,len(prefix_tar) if compressed is not None else 1)
    raw_start = len(prefix_tar) if compressed is not None else 0
    prefix_pieces = _parts(q-raw_start)
    if compressed is not None:
        chunks.append(_Chunk("code",f,len(compressed),len(prefix_tar),compressed))
        f += len(compressed)
        o += len(prefix_tar)
    template_pieces = _parts(template_size)
    # First output the actual TAR prefix, then the complete skippable template
    # as the beginning of the embedded archive itself.
    for size in prefix_pieces:
        chunks.append(_Chunk("initial", f, 3+size, size, source=o))
        f += 3+size
        o += size
    source = 0
    for size in template_pieces:
        chunks.append(_Chunk("template",f,3+size,size,source=source))
        f += 3+size
        o += size
        source += size
    # Emit the physical frame header, then quote every preceding RAW block.
    # The tiny header literals are fixed structural bytes, not new payload copies.
    chunks.append(_Chunk("framecopy",f,9,6))
    f += 9
    o += 6
    if compressed is not None:
        at = 0
        while at < len(compressed):
            literal = compressed[at:at+BLOCK]
            code = _raw_header(len(literal))+literal
            chunks.append(_Chunk("code",f,len(code),len(literal),code))
            f += len(code)
            o += len(literal)
            at += len(literal)
    for source_start,pieces in ((raw_start,prefix_pieces),(q,template_pieces)):
        source = source_start
        for size in pieces:
            literal = _raw_header(size)
            distance = o+3-source
            code,_ = _copy_block(literal,size,distance)
            chunks.append(_Chunk("code",f,len(code),3+size,code))
            f += len(code)
            o += 3+size
            source += size
            max_distance = max(max_distance,distance)
    # A quote/replay gear may itself use multiple RAW chunks. Its quoted source
    # ends just before the FIRST new RAW payload, so no circular data is read.
    for _ in range(64):
        gap = f-(o-q)
        if gap <= 0:
            raise RuntimeError("invalid forward quotation gap")
        start_o = o
        group = _parts(gap+3)
        source = 0
        for size in group:
            chunks.append(_Chunk("gear",f,3+size,size,source=o-q))
            f += 3+size
            o += size
        for index,size in enumerate(group):
            literal = _raw_header(size) if index else b""
            distance = o+len(literal)-(start_o+source)
            code,_ = _copy_block(literal,size,distance)
            chunks.append(_Chunk("code",f,len(code),len(literal)+size,code))
            f += len(code)
            o += len(literal)+size
            source += size
            max_distance = max(max_distance,distance)
        if f-(o-q) <= 24:
            break
    else:
        raise RuntimeError("quotation gear normalization did not converge")
    plant_pos,tail_pos = q+32,q+256
    choice = None
    for empty in range(32):
        jf = f+3*empty
        gap = jf-(o-q)
        distance = o-plant_pos
        for length in range(max(3,gap+5),min(220,gap+40)):
            jump,_ = _copy_block(b"",length,distance)
            if length != gap+len(jump)+5:
                continue
            patch_f = jf+len(jump)+5
            if patch_f % 8:
                continue
            xo = o+length
            match = 15
            for _ in range(10):
                xcode,sx = _copy_block(bytes(32),match,xo+32-tail_pos)
                new_match = len(sx)+8
                if new_match == match:
                    break
                match = new_match
            else:
                continue
            total = jf+len(jump)+len(xcode)+8
            choice = (total,empty,length,jump,xcode,sx,patch_f,xo)
            break
        if choice:
            break
    if choice is None:
        raise RuntimeError("no aligned terminal block layout")
    total,empty,length,jump,xcode,sx,patch_f,xo = choice
    for _ in range(empty):
        chunks.append(_Chunk("code",f,3,0,bytes(3)))
        f += 3
    jump_output = o
    chunks.append(_Chunk("code",f,len(jump),length,jump))
    f += len(jump)
    o += length
    chunks.append(_Chunk("code",f,len(xcode),32+len(sx)+8,xcode))
    f += len(xcode)
    o += 32+len(sx)+8
    padding = (-total)%512+1024
    decoded_size = q+total+padding
    if o != q+total or f+8 != total:
        raise RuntimeError("terminal self-copy equation did not close")
    checksum = checksum_for_size(decoded_size)
    checksum_bytes = (checksum & 0xffffffff).to_bytes(4,"little")
    rle = ((padding << 3)|3).to_bytes(3,"little") + b"\0"
    max_distance = max(max_distance,jump_output-plant_pos,xo+32-tail_pos)
    wd,window_size = _window_descriptor(max_distance)
    frame_header = bytes.fromhex("28b52ffd04") + bytes((wd,))
    out = bytearray(total)
    out[:8] = bytes.fromhex("502a4d18")+(template_size-8).to_bytes(4,"little")
    out[template_size:template_size+6] = frame_header
    for chunk in chunks:
        if chunk.kind == "code":
            out[chunk.f:chunk.f+chunk.size] = chunk.content
        elif chunk.kind == "framecopy":
            out[chunk.f:chunk.f+chunk.size] = _raw_header(6)+frame_header
        else:
            out[chunk.f:chunk.f+3] = _raw_header(chunk.decoded)
    out[-8:-4],out[-4:] = rle,checksum_bytes
    template = bytearray(out[:template_size])
    plant = out[jump_output-q:jump_output-q+length]
    if len(plant) != length or 32+length > 256:
        raise RuntimeError("skippable jump template overflow")
    template[32:32+length] = plant
    tail = sx+rle+checksum_bytes
    template[256:256+len(tail)] = tail
    note = b"\nZstandard quine template; checksum solved by lane inversion.\n"
    template[320:] = (note*5)[:192]
    out[:template_size] = template
    self_header = _header(quine_name,total,mtime)
    if len(prefix_tar)+len(self_header) != q:
        raise ValueError("self header changed serialized length; input too large for this layout")
    tar_prefix = prefix_tar+self_header
    for chunk in chunks:
        if chunk.kind == "initial":
            out[chunk.f+3:chunk.f+chunk.size] = tar_prefix[chunk.source:chunk.source+chunk.decoded]
        elif chunk.kind == "template":
            out[chunk.f+3:chunk.f+chunk.size] = template[chunk.source:chunk.source+chunk.decoded]
        elif chunk.kind == "gear":
            out[chunk.f+3:chunk.f+chunk.size] = out[chunk.source:chunk.source+chunk.decoded]
    raw = bytearray(decoded_size)
    raw[:q] = tar_prefix
    raw[q:q+total] = out
    patch = force_words(raw,q+patch_f)
    out[patch_f:patch_f+32] = patch
    raw[q+patch_f:q+patch_f+32] = patch
    archive = bytes(out)
    actual = decompress(archive,decoded_size)
    if actual != raw:
        mismatch = next((p for p,(a,b) in enumerate(zip(actual,raw)) if a != b),min(len(actual),len(raw)))
        raise RuntimeError(f"self reproduction mismatch at decoded offset {mismatch}")
    info = dict(quine_name=quine_name,decoded_size=decoded_size,quine_offset=q,
                frame_count=1,skippable_frames=1,compressed_size=total,
                checksum=f"{checksum & 0xffffffff:08x}",xxh64=f"{checksum:016x}",
                window_size=window_size,patch_offset=patch_f,block_count=len(chunks)+1,
                prefix_size=len(prefix_tar),template_size=template_size,
                payload_mode="compressed" if compressed is not None else "store",
                compressed_prefix_size=len(compressed) if compressed is not None else None)
    return archive,info,actual


def _compress_blocks(data: bytes, level: int) -> bytes:
    """Independent source-prefix blocks, ready to execute after our frame header."""
    try:
        lib = _libzstd()
    except RuntimeError:
        try:
            import zstandard
        except ImportError:
            command = shutil.which("zstd")
            if not command:
                raise RuntimeError("compression needs libzstd, Python zstandard, or the zstd executable") from None
            option = f"--fast={-level}" if level < 0 else f"-{level or 3}"
            result = subprocess.run([command,"-q","--no-check","--ultra",option,"-c"],input=data,capture_output=True)
            if result.returncode:
                raise RuntimeError("Zstandard prefix compression: "+result.stderr.decode(errors="replace").strip())
            frame = bytearray(result.stdout)
        else:
            frame = bytearray(zstandard.ZstdCompressor(level=level,write_checksum=False).compress(data))
    else:
        capacity = lib.ZSTD_compressBound(len(data))
        dest = ctypes.create_string_buffer(capacity)
        result = lib.ZSTD_compress(dest,capacity,data,len(data),level)
        if lib.ZSTD_isError(result):
            raise RuntimeError("Zstandard prefix compression: "+lib.ZSTD_getErrorName(result).decode())
        frame = bytearray(dest.raw[:result])
    descriptor = frame[4]
    single = bool(descriptor & 32)
    dictionary_size = (0,1,2,4)[descriptor & 3]
    fcs_flag = descriptor >> 6
    content_size = (1 if single else 0,2,4,8)[fcs_flag]
    header_size = 5+(0 if single else 1)+dictionary_size+content_size
    at = header_size
    while True:
        if at+3 > len(frame):
            raise RuntimeError("truncated native prefix frame")
        h = int.from_bytes(frame[at:at+3],"little")
        last,kind,size = h & 1,(h >> 1) & 3,h >> 3
        if kind == 3:
            raise RuntimeError("invalid native prefix block")
        frame[at] &= 254
        at += 3+(1 if kind == 1 else size)
        if last:
            break
    if at+(4 if descriptor & 4 else 0) != len(frame):
        raise RuntimeError("unexpected native prefix frame suffix")
    return bytes(frame[header_size:at])


def build_quine(prefix_tar: bytes, quine_name: str, *, mtime=0, level=9,
                verify_decoded=None):
    """Build a seedless single-data-frame TAR quine.

    level=None stores the source prefix once verbatim. An integer enables native
    prefix compression when its three encoded copies plus overhead are smaller
    than the raw prefix. This avoids expanding incompressible payloads. Both
    paths use the same algebraic checksum construction and contain no seed file.
    After standard decoding and self-copy verification, an optional callback
    receives (archive, decoded_tar) for additional checks without decoding again.
    """
    if len(prefix_tar) % 512:
        raise ValueError("prefix_tar must end on a 512-byte TAR boundary")
    if not quine_name:
        raise ValueError("self member name must be nonempty")
    if level is not None and (not isinstance(level,int) or not -131072 <= level <= 22):
        raise ValueError("compression level must be between -131072 and 22")
    compressed = None
    if level is not None and prefix_tar:
        candidate = _compress_blocks(prefix_tar,level)
        if 3*len(candidate)+2048 < len(prefix_tar):
            compressed = candidate
    result,info,decoded = _build_quine(prefix_tar,quine_name,mtime=mtime,compressed=compressed)
    if verify_decoded is not None:
        verify_decoded(result,decoded)
    info["compression_level"] = level
    return result,info


@dataclass
class Entry:
    name: str
    kind: str
    mode: int
    path: Path | None = None
    link: str = ""
    size: int = 0
    digest: bytes = b""


def archive_name(value: str, *, empty: bool = False) -> str:
    """Validate a relative POSIX member path without silently truncating it."""
    if "\0" in value or value.startswith("/") or ".." in value.split("/"):
        raise ValueError(f"unsafe archive path: {value!r}")
    if len(value) >= 2 and value[1] == ":":
        raise ValueError(f"archive path must be relative: {value!r}")
    name = str(PurePosixPath(value))
    if name == ".":
        if empty:
            return ""
        raise ValueError("archive member name cannot be empty")
    return name


def join_name(prefix: str, name: str) -> str:
    return archive_name(f"{prefix}/{name}" if prefix else name)


def collect_entries(inputs: list[Path], output: Path, prefix: str,
                    reserved: str) -> tuple[list[Entry], bool]:
    entries: list[Entry] = []
    names: dict[str, Entry] = {}
    excluded = False
    output_absolute = os.path.abspath(output)
    if any(os.path.abspath(path) == output_absolute for path in inputs):
        raise ValueError("output cannot also be an explicit input")

    def add(entry: Entry) -> None:
        entry.name = archive_name(entry.name)
        old = names.get(entry.name)
        if old is not None:
            if old.kind == entry.kind == "directory" and old.path is None:
                old.path, old.mode = entry.path, entry.mode
                return
            raise ValueError(f"duplicate archive path: {entry.name}")
        names[entry.name] = entry
        entries.append(entry)

    def parents(name: str) -> None:
        parts = PurePosixPath(name).parts
        for end in range(1, len(parts)):
            parent = "/".join(parts[:end])
            existing = names.get(parent)
            if existing is not None:
                if existing.kind != "directory":
                    raise ValueError(f"archive parent is not a directory: {parent}")
            else:
                add(Entry(parent, "directory", 0o755))

    def walk(path: Path, name: str) -> None:
        nonlocal excluded
        if os.path.abspath(path) == output_absolute:
            excluded = True
            return
        st = path.lstat()
        parents(name)
        mode = stat.S_IMODE(st.st_mode)
        if stat.S_ISLNK(st.st_mode):
            add(Entry(name, "symlink", mode, path, os.readlink(path)))
        elif stat.S_ISDIR(st.st_mode):
            add(Entry(name, "directory", mode, path))
            for child in sorted(path.iterdir(), key=lambda p: p.name):
                walk(child, join_name(name, child.name))
        elif stat.S_ISREG(st.st_mode):
            add(Entry(name, "file", mode, path))
        else:
            raise ValueError(f"unsupported special file (not read): {path}")

    # Explicit parents also cover empty sources and custom self locations.
    parents(reserved)
    if prefix:
        parents(prefix + "/placeholder")
    for path in inputs:
        st = path.lstat()  # Fail for nonexistent inputs, including broken parents.
        if len(inputs) == 1 and stat.S_ISDIR(st.st_mode):
            for child in sorted(path.iterdir(), key=lambda p: p.name):
                walk(child, join_name(prefix, child.name))
        else:
            walk(path, join_name(prefix, path.name))
    if reserved in names:
        raise ValueError(f"source collides with reserved archive member: {reserved}")
    for name in names:
        if name.startswith(reserved + "/"):
            raise ValueError(f"reserved file would replace a directory: {reserved}")
    return entries, excluded


def encode_entries(entries: list[Entry], mtime: int) -> bytes:
    out = bytearray()
    for entry in entries:
        member = tarfile.TarInfo(entry.name)
        member.mode, member.mtime = entry.mode, mtime
        data = b""
        if entry.kind == "directory":
            member.type = tarfile.DIRTYPE
        elif entry.kind == "symlink":
            member.type, member.linkname = tarfile.SYMTYPE, entry.link
        else:
            flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0)
            assert entry.path is not None
            with os.fdopen(os.open(entry.path, flags), "rb") as source:
                before = os.fstat(source.fileno())
                if not stat.S_ISREG(before.st_mode):
                    raise ValueError(f"input is no longer a regular file: {entry.path}")
                data = source.read()
                after = os.fstat(source.fileno())
            if (len(data) != before.st_size or before.st_size != after.st_size or
                    before.st_mtime_ns != after.st_mtime_ns):
                raise ValueError(f"input changed while reading: {entry.path}")
            entry.size, entry.digest = len(data), hashlib.sha256(data).digest()
            member.size = entry.size
        out.extend(member.tobuf(format=tarfile.PAX_FORMAT, encoding="utf-8",
                                errors="surrogateescape"))
        out.extend(data)
        out.extend(bytes((-len(data)) % 512))
    return bytes(out)


def verify_archive(archive: bytes, raw: bytes, entries: list[Entry],
                   quine_name: str) -> None:
    """Check members using the core's already checksum-verified decoded bytes."""
    if len(raw) % 512 or raw[-1024:] != bytes(1024):
        raise ValueError("TAR is missing its complete zero end records")
    with tarfile.open(fileobj=io.BytesIO(raw), mode="r:") as tar:
        members = tar.getmembers()
        by_name = {member.name.rstrip("/"): member for member in members}
        if len(by_name) != len(members):
            raise ValueError("duplicate member in generated TAR")
        expected = {entry.name for entry in entries} | {quine_name}
        if set(by_name) != expected:
            raise ValueError("TAR member list does not match the input snapshot")
        member = by_name[quine_name]
        if not member.isfile() or tar.extractfile(member).read() != archive:
            raise ValueError("Quine member is not byte-identical to the archive")
        for entry in entries:
            member = by_name[entry.name]
            if entry.kind == "directory":
                ok = member.isdir()
            elif entry.kind == "symlink":
                ok = member.issym() and member.linkname == entry.link
            else:
                ok = (member.isfile() and member.size == entry.size and
                      hashlib.sha256(tar.extractfile(member).read()).digest() == entry.digest)
            if not ok or stat.S_IMODE(member.mode) != entry.mode:
                raise ValueError(f"input entry did not survive packing: {entry.name}")


def atomic_write(path: Path, data: bytes, force: bool) -> None:
    if path.exists() and not force:
        raise FileExistsError(f"output exists; use --force to replace it: {path}")
    temporary = None
    try:
        with tempfile.NamedTemporaryFile(prefix=".zstd-quine-", dir=path.parent,
                                         delete=False) as output:
            temporary = Path(output.name)
            output.write(data)
            output.flush()
            os.fsync(output.fileno())
        if force:
            os.replace(temporary, path)
        else:
            # link is atomic and cannot overwrite a file created after the check.
            os.link(temporary, path)
            temporary.unlink()
        temporary = None
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("inputs", metavar="INPUT", nargs="+", type=Path)
    parser.add_argument("output", metavar="OUTPUT", type=Path)
    parser.add_argument("--quine-name", help="self member name (default: output basename)")
    parser.add_argument("--src-dir", default="", help="archive prefix for source entries")
    parser.add_argument("--quine-dir", default="", help="archive prefix for the self member")
    parser.add_argument("--mtime", type=int, default=0, help="fixed TAR timestamp (default: 0)")
    compression = parser.add_mutually_exclusive_group()
    compression.add_argument("--level", type=int, default=9,
                             help="Zstd source compression level (default: 9; auto-falls back to storage)")
    compression.add_argument("--store", dest="level", action="store_const", const=None,
                             help="store the source prefix without trying compression")
    parser.add_argument("--force", action="store_true", help="replace an existing output after verification")
    args = parser.parse_args(argv)
    try:
        if args.output.is_symlink():
            raise ValueError("output must not be a symbolic link")
        if args.output.exists() and not args.output.is_file():
            raise ValueError("output must be a regular file path")
        if args.output.exists() and not args.force:
            raise FileExistsError(f"output exists; use --force to replace it: {args.output}")
        if not args.output.parent.is_dir():
            raise ValueError(f"output directory does not exist: {args.output.parent}")
        prefix = archive_name(args.src_dir, empty=True)
        quine_dir = archive_name(args.quine_dir, empty=True)
        quine = join_name(quine_dir, archive_name(args.quine_name or args.output.name))
        entries, excluded = collect_entries(args.inputs, args.output, prefix, quine)
        prefix_tar = encode_entries(entries, args.mtime)
        archive, info = build_quine(
            prefix_tar, quine, mtime=args.mtime, level=args.level,
            verify_decoded=lambda data, raw: verify_archive(data, raw, entries, quine),
        )
        atomic_write(args.output, archive, args.force)
        files = sum(entry.kind == "file" for entry in entries)
        dirs = sum(entry.kind == "directory" for entry in entries)
        links = sum(entry.kind == "symlink" for entry in entries)
        print(f"[tar.zst quine] {files} files, {dirs} directories, {links} symlinks")
        if excluded:
            print("[tar.zst quine] excluded the output path from source traversal")
        print(f"[tar.zst quine] wrote {args.output}: {len(archive)} bytes")
        print(f"[layout] {info['frame_count']} checked data frame; payload mode: {info['payload_mode']}; "
              f"window: {info['window_size']} bytes; XXH64: {info['checksum']}")
        print("[verify] standard Zstd checksum, TAR contents and exact self: PASS")
        print(f"[self] {quine}")
        return 0
    except (OSError, ValueError, RuntimeError, tarfile.TarError) as error:
        print(f"error: {error}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())
```
</details>

  现在AI写的代码真的是越来越复杂了，真的是根本看不懂啊……但就是这样，很神奇的就把问题解决了。总之，zstd Quine也轻松实现了，也算是给自包含压缩包再提供一种新的格式吧。   

## ZIP Quine
  后来闲的没事又让AI继续研究有没有可能做出大文件的ZIP Quine，结果它研究了一段时间还真让它研究出来了，当然[Ruben Van Mello](https://github.com/ruvmello)所说的窗口大小的问题依然存在，只不过AI用了某种方式绕过去了，而代价则是会存在成员数据有重叠的部分。比如在Python的标准库zipfile中，在[这里](https://github.com/python/cpython/blob/3fb88e94db547e9dbed97de68da87bcf9335a53a/Lib/zipfile/__init__.py#L2276)有检查自身成员压缩数据是不是越过了后续成员的局部头，导致我这种方法生成的ZIP Quine无法被解压，然而7-Zip和libarchive都能正常解压……总之这种方式生成的有一点点兼容性问题，但问题不是很严重，那么以下就是这种生成器的代码：   
<details markdown="1">
<summary markdown="span">
Show Code
</summary>

```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Pack files/directories plus an exact copy of this ZIP archive.

Example: python zip_quine.py ./site ./site.zip
One directory packs its contents; multiple inputs retain their basenames.
Python 3.10+; standard library only. Extract with 7-Zip or libarchive.
The archive deliberately shares physical bytes between overlapping members.
"""
from __future__ import annotations
import argparse
import io
import os
import stat
import struct
import sys
import tempfile
import time
import zipfile
import zlib
from pathlib import Path, PurePosixPath

LBASE = list(range(3, 11)) + [11,13,15,17,19,23,27,31,35,43,51,59,67,83,99,115,131,163,195,227,258]
LBITS = [0]*8 + [1]*4 + [2]*4 + [3]*4 + [4]*4 + [5]*4 + [0]
DBASE = [1,2,3,4,5,7,9,13,17,25,33,49,65,97,129,193,257,385,513,769,1025,1537,2049,3073,4097,6145,8193,12289,16385,24577]
DBITS = [0]*4 + [v for n in range(1,14) for v in [n,n]]

CL_ORDER = [16, 17, 18, 0, 8, 7, 9, 6, 10, 5, 11, 4, 12, 3, 13, 2, 14, 1]
CL_LEN = {0: 1, 1: 3, 17: 3, 18: 2}
# CL codes in stream order (MSB of the Huffman code first):
#   0 -> "0", 18 -> "10", 1 -> "110", 17 -> "111"
CL_CODE = {0: (0b0, 1), 18: (0b1, 2), 1: (0b011, 3), 17: (0b111, 3)}

# (cell length, zero-length prefix, code lengths per unit copy).
# Exhaustively found for the fixed distance-8 match used by unit_bytes().
CONNECTORS = [
    (576, 9, 129), (639, 9, 129), (927, 33, 118), (32872, 1, 128),
    (33772, 49, 118), (39936, 17, 120), (39951, 17, 120), (39953, 9, 124),
    (39999, 17, 120), (40732, 41, 108), (40944, 17, 120), (45559, 33, 124),
    (45560, 33, 120), (49300, 1, 134), (49934, 57, 107), (50078, 49, 107),
    (57356, 49, 118), (57373, 1, 129), (57374, 25, 117), (57572, 1, 133),
    (57599, 41, 118), (57854, 41, 117), (58126, 57, 107), (58319, 49, 107),
    (59264, 25, 117), (59390, 25, 117), (60536, 41, 120), (60540, 41, 118),
    (61540, 9, 129), (61903, 49, 107), (63495, 25, 118), (63544, 33, 118),
    (63608, 33, 117), (63743, 41, 118), (63943, 41, 110), (63984, 25, 118),
    (64423, 25, 124), (64551, 25, 120), (65048, 49, 117), (65311, 33, 118),
    (65420, 49, 118), (65436, 25, 120),
]


class Bits:
    def __init__(self):
        self.value = 0
        self.count = 0

    def put(self, value, width):
        assert 0 <= value < (1 << width), (value, width)
        self.value |= value << self.count
        self.count += width

    def align(self):
        self.put(0, -self.count % 8)

    def bytes(self):
        return self.value.to_bytes((self.count + 7) // 8, 'little')


def reverse(value, width):
    return int(f'{value:0{width}b}'[::-1], 2)


def symbol(bits, s):
    if s <= 143:
        code, width = 0x30+s, 8
    elif s <= 255:
        code, width = 0x190+s-144, 9
    elif s <= 279:
        code, width = s-256, 7
    else:
        code, width = 0xc0+s-280, 8
    bits.put(reverse(code, width), width)


def number(value, bases, widths):
    for i, (base, width) in enumerate(zip(bases, widths)):
        if base <= value < base + (1 << width):
            return i, value - base, width
    raise ValueError(value)


def copy_code(length, distance):
    assert 3 <= length and 1 <= distance <= 32768
    bits = Bits()
    bits.put(2, 3)
    remaining = length
    while remaining:
        size = min(258, remaining)
        if 0 < remaining - size < 3:
            size -= 3 - (remaining - size)
        slot, extra, width = number(size, LBASE, LBITS)
        symbol(bits, 257 + slot)
        bits.put(extra, width)
        slot, extra, width = number(distance, DBASE, DBITS)
        bits.put(reverse(slot, 5), 5)
        bits.put(extra, width)
        remaining -= size
    symbol(bits, 256)
    bits.put(0, 3)
    bits.align()
    bits.put(0, 16)
    bits.put(0xffff, 16)
    return bits.bytes()


def raw_header(size):
    assert 0 <= size <= 65535
    return b'\0' + struct.pack('<HH', size, size ^ 0xffff)


def crc(data):
    return zlib.crc32(data)


# --------------------------------------------------------------- connector
def unit_bytes(lc, final=False):
    code = b'\x82\xd1\x80\0' if final else b'\x82\xd1\0\0'
    return code + struct.pack('<HH', lc, lc ^ 0xffff)


def connector_tail(unit, m):
    w = Bits()
    w.put(4, 3)          # BFINAL=0, BTYPE=10
    w.put(0, 5)          # HLIT=0 -> 257 literal lengths
    w.put(29, 5)         # HDIST=29 -> 30 distance codes
    w.put(14, 4)         # HCLEN=14 -> 18 CL lengths
    for s in CL_ORDER:
        w.put(CL_LEN.get(s, 0), 3)
    assert w.count == 71
    w.put(0, m)          # m zero lengths (m == 1 mod 8)
    assert w.count % 8 == 0
    for b in unit:
        w.put(b, 8)
    return w.bytes()


def payload_head(m, c1, final):
    window_end = m + 2 * c1
    first_free = max(257, window_end)
    count = 287 - first_free
    if count < 2:
        raise ValueError('no free distance codes')
    w = Bits()
    for i in range(count):
        w.put(*(CL_CODE[1] if i < 2 else CL_CODE[0]))
    w.put(1, 1)                      # EOB (literal code 1)
    w.put(1 if final else 0, 3)      # stored block header
    w.align()
    return w.bytes()


def geometry(lc, m, c1):
    head = payload_head(m, c1, False)
    return dict(lc=lc, m=m, unit=unit_bytes(lc), p_head=len(head)+4,
                k=8+(71+m)//8, head=head,
                head_final=payload_head(m, c1, True))


UINT32 = 0xffffffff
UTF8_FLAG = 0x800


def _stamp(epoch):
    if epoch <= 315532800:
        return 0, 33
    t = time.gmtime(min(epoch, 4354819198))
    return ((t.tm_hour << 11) | (t.tm_min << 5) | (t.tm_sec // 2),
            ((t.tm_year - 1980) << 9) | (t.tm_mon << 5) | t.tm_mday)


def _extra_padding(size):
    if not size:
        return b''
    if size < 4:
        raise ValueError('extra padding needs at least four bytes')
    return struct.pack('<HH', 0xca05, size - 4) + bytes(size - 4)


def _local(e, *, extra=b''):
    tm, dt = _stamp(e['mtime'])
    return (struct.pack('<IHHHHHIIIHH', 0x04034b50, 20, UTF8_FLAG,
                        e['method'], tm, dt, e['crc'], e['csize'],
                        e['size'], len(e['encoded']), len(extra))
            + e['encoded'] + extra)


def _central(e, *, extra=b'', extra_size=None):
    tm, dt = _stamp(e['mtime'])
    attrs = ((e['mode'] & 0xffff) << 16) | (0x10 if e['is_dir'] else 0)
    return (struct.pack('<IHHHHHHIIIHHHHHII', 0x02014b50,
                        (3 << 8) | 20, 20, UTF8_FLAG, e['method'], tm, dt,
                        e['crc'], e['csize'], e['size'], len(e['encoded']),
                        len(extra) if extra_size is None else extra_size,
                        0, 0, 0, attrs, e['local_offset'])
            + e['encoded'] + extra)


def _entry(item, mtime):
    e = dict(item)
    e['is_dir'] = bool(e.get('is_dir', False))
    name = archive_name(e['name']) + ('/' if e['is_dir'] else '')
    encoded = name.encode('utf8')
    if len(encoded) > 60000:
        raise ValueError(f'UTF-8 archive path exceeds 60000 bytes: {name!r}')
    data = bytes(e.get('data', b''))
    if e['is_dir'] and data:
        raise ValueError('directory entries cannot have data')
    e.update(name=name, encoded=encoded, data=data, size=len(data),
             mtime=e.get('mtime', mtime), crc=crc(data), method=0,
             mode=e.get('mode', (stat.S_IFDIR | 0o755) if e['is_dir']
                        else (stat.S_IFREG | 0o644)), csize=len(data))
    if len(data) >= UINT32:
        raise ValueError('ZIP64 is not implemented: a member is too large')
    return e


def _large_plan(e):
    best = None
    base = 30 + len(e['encoded'])
    for values in CONNECTORS:
        g = geometry(*values)
        lc, ph, k = g['lc'], g['p_head'], g['k']
        first_cap, middle_cap, last_cap = lc - base - 5 - k, lc - ph - k, lc - ph - 8
        if min(first_cap, middle_cap, last_cap) < 4:
            continue
        count = 2 + max(0, (e['size'] - first_cap - last_cap + middle_cap - 1) // middle_cap)
        left = e['size'] - (count - 2) * middle_cap
        first = min(first_cap, left - 4)
        if first < 1:
            continue
        extra = first_cap - first
        if 0 < extra < 4:
            first -= 4 - extra
            extra = 4
        last = left - first
        if not 1 <= last <= last_cap:
            continue
        physical = count * lc + (count - 1) * 8
        candidate = dict(g, count=count, first=first, middle=middle_cap,
                         last=last, extra=extra, physical=physical,
                         csize=e['size'] + 5 + (count - 1) * (k + 8 + ph))
        if best is None or physical < best['physical']:
            best = candidate
    if best is None:
        raise ValueError(f'cannot fit payload layout for {e["name"]!r}')
    return best


def _solve_patch(baseline):
    # The final four bytes are a unique free field in the archive comment.
    # CRC differences do not depend on the prefix bytes, so this needs only
    # two full-archive CRC scans and 32 tiny four-byte computations.
    zero = crc(bytes(4))
    cols = [crc((1 << i).to_bytes(4, 'little')) ^ zero for i in range(32)]
    rows = [sum(((value >> bit) & 1) << col for col, value in enumerate(cols))
            | (((baseline >> bit) & 1) << 32) for bit in range(32)]
    for col in range(32):
        pivot = next((r for r in range(col, 32) if rows[r] >> col & 1), None)
        if pivot is None:
            raise RuntimeError('CRC patch matrix unexpectedly singular')
        rows[col], rows[pivot] = rows[pivot], rows[col]
        for r in range(32):
            if r != col and rows[r] >> col & 1:
                rows[r] ^= rows[col]
    return sum(((rows[i] >> 32) & 1) << i for i in range(32)).to_bytes(4, 'little')


def build(entries, self_name, *, mtime=0):
    """Return ZIP bytes containing the inputs and a byte-identical self member.

    Each entry has name, data, is_dir, mode and optional mtime. Unix type bits
    in mode are preserved, including symlinks whose data is their target text.
    """
    items = [_entry(e, mtime) for e in entries]
    me = _entry(dict(name=self_name, data=b'', mode=stat.S_IFREG | 0o644), mtime)
    if len(me['encoded']) > 32000:
        raise ValueError('self member name is too long for the bootstrap window')
    names = [me['name'].rstrip('/').casefold()] + [e['name'].rstrip('/').casefold() for e in items]
    if len(set(names)) != len(names):
        raise ValueError('duplicate or case-colliding member names')
    if len(items) + 1 >= 65535:
        raise ValueError('ZIP64 is not implemented: too many members')
    me.update(method=8, crc=0, local_offset=0)
    h = 30 + len(me['encoded']) + 260
    p0 = h + 5
    plant_at = 30 + len(me['encoded']) + 12
    first = copy_code(p0, p0)
    jump = None
    for length in range(3, 4096):
        code = copy_code(length, 2 * p0 - plant_at)
        if length == len(first) + len(code) + 5:
            jump = code
            break
    if jump is None:
        raise ValueError('no bootstrap layout for this self name')
    if len(first) + len(jump) + 5 > 248:
        raise ValueError('self member name needs a bootstrap template larger than 248 bytes; use a shorter name')
    prefix_size = h + 5 + p0 + len(first) + len(jump) + 5
    cells, pending, pending_entries = [], bytearray(), []

    def flush_small():
        nonlocal pending, pending_entries
        if pending_entries:
            for e, relative in pending_entries:
                e['cell_index'], e['cell_relative'] = len(cells), relative
            cells.append(bytearray(pending) + bytes(8))
            pending, pending_entries = bytearray(), []

    for e in items:
        record_size = 30 + len(e['encoded']) + len(e['data'])
        if record_size <= 65527:
            if len(pending) + record_size > 65527:
                flush_small()
            pending_entries.append((e, len(pending)))
            pending += _local(e) + e['data']
            continue
        flush_small()
        g = _large_plan(e)
        e.update(method=8, csize=g['csize'], cell_index=len(cells), cell_relative=0)
        tail, at = connector_tail(g['unit'], g['m']), g['first']
        cell = (_local(e, extra=_extra_padding(g['extra']))
                + b'\0' + struct.pack('<HH', at, at ^ 65535)
                + e['data'][:at] + tail)
        if len(cell) != g['lc']:
            raise RuntimeError('first payload cell length mismatch')
        cells.append(bytearray(cell))
        for index in range(1, g['count']):
            last = index == g['count'] - 1
            size = g['last'] if last else g['middle']
            cell = ((g['head_final'] if last else g['head'])
                    + struct.pack('<HH', size, size ^ 65535)
                    + e['data'][at:at + size])
            at += size
            cell += bytes(g['lc'] - len(cell)) if last else tail
            if len(cell) != g['lc']:
                raise RuntimeError('payload cell length mismatch')
            cells.append(bytearray(cell))
        if at != e['size']:
            raise RuntimeError('payload split length mismatch')
    flush_small()
    offsets, position = [], prefix_size
    for cell in cells:
        offsets.append(position)
        position += len(cell) + 8
    for e in items:
        e['local_offset'] = offsets[e['cell_index']] + e['cell_relative']

    # Directory cell boundaries are hidden inside a valid custom extra TLV.
    # Its data is the 8-byte seed in the cell plus the next actual 8-byte unit.
    all_entries = [me] + items
    batches, current, size = [], [], 0
    for e in all_entries:
        record_size = 46 + len(e['encoded'])
        if size + record_size > 65000 and current:
            batches.append(current)
            current, size = [], 0
        current.append(e)
        size += record_size
    batches.append(current)
    cd_start = position if cells else prefix_size
    cd_size = sum(46 + len(e['encoded']) for e in all_entries) + 20 * (len(batches) - 1)
    directory_sizes = [sum(46 + len(e['encoded']) for e in batch)
                       + (12 if i < len(batches) - 1 else 26)
                       for i, batch in enumerate(batches)]
    total = (prefix_size + sum(map(len, cells)) + sum(directory_sizes)
             + 8 * (len(cells) + len(batches) - 1))
    if max(total, cd_start, cd_size) >= UINT32:
        raise ValueError('ZIP64 is not implemented: archive exceeds ZIP32 limits')
    me.update(size=total, csize=total - h)
    eocd = struct.pack('<IHHHHIIH', 0x06054b50, 0, 0, len(all_entries),
                       len(all_entries), cd_size, cd_start, 4)
    for i, batch in enumerate(batches):
        buf = bytearray()
        for j, e in enumerate(batch):
            if i < len(batches) - 1 and j == len(batch) - 1:
                partial = struct.pack('<HH', 0xca04, 16) + bytes(8)
                buf += _central(e, extra=partial, extra_size=20)
            else:
                buf += _central(e)
        if i == len(batches) - 1:
            buf += eocd + bytes(4)
        if len(buf) != directory_sizes[i]:
            raise RuntimeError('directory cell length mismatch')
        cells.append(buf)
    if any(not 1 <= len(cell) <= 65535 for cell in cells):
        raise ValueError('cell exceeds Deflate stored-block bounds')
    units = [unit_bytes(len(cell), final=i == len(cells)-1)
             for i, cell in enumerate(cells)]
    for i in range(len(cells) - 1):
        cells[i][-8:] = units[i + 1]
    extra = bytearray(256)
    extra[:8] = b'VACUUM04'
    first_raw = raw_header(len(cells[0]))
    if len(cells) == 1:
        first_raw = b'\x01' + first_raw[1:]
    plant = first + jump + first_raw
    extra[8:8 + len(plant)] = plant
    header = _local(me, extra=struct.pack('<HH', 0xca03, len(extra)) + extra)
    if len(header) != h:
        raise RuntimeError('self header length mismatch')
    archive = bytearray(header + raw_header(p0) + header + raw_header(p0)
                        + first + jump + first_raw)
    for i, cell in enumerate(cells):
        if i:
            archive += units[i]
        archive += cell
    if len(archive) != total:
        raise RuntimeError('archive length mismatch')
    patch = _solve_patch(crc(archive))
    archive[-4:] = patch
    if crc(archive) != 0:
        raise RuntimeError('archive CRC patch failed')
    return bytes(archive)


def archive_name(value, *, empty=False):
    value = str(value).replace('\\', '/')
    if value.startswith('/') or '\0' in value:
        raise ValueError(f'invalid archive path: {value!r}')
    parts = value.split('/')
    if any(part == '..' or ':' in part for part in parts):
        raise ValueError(f'archive paths must be relative, without .. or drive names: {value!r}')
    parts = [part for part in parts if part not in ('', '.')]
    name = '/'.join(parts)
    if not name and not empty:
        raise ValueError('archive member name must not be empty')
    if len(name.encode('utf-8')) > 65535:
        raise ValueError('UTF-8 archive path exceeds the ZIP filename limit')
    return name


def join_name(prefix, name):
    return archive_name(f'{prefix}/{name}' if prefix else name)


def path_key(path):
    return os.path.normcase(str(Path(path).resolve()))


def read_snapshot(path):
    flags = os.O_RDONLY | getattr(os, 'O_BINARY', 0) | getattr(os, 'O_NOFOLLOW', 0)
    with os.fdopen(os.open(path, flags), 'rb') as source:
        before = os.fstat(source.fileno())
        if not stat.S_ISREG(before.st_mode):
            raise ValueError(f'input is no longer a regular file: {path}')
        if before.st_size >= 0xffffffff:
            raise ValueError(f'file exceeds this generator\'s ZIP32 size limit: {path}')
        data = source.read()
        after = os.fstat(source.fileno())
    if (len(data) != before.st_size or before.st_size != after.st_size
            or before.st_mtime_ns != after.st_mtime_ns):
        raise ValueError(f'input changed while reading: {path}')
    return data


def collect_entries(inputs, output, prefix, reserved, mtime=0):
    entries, names, case_names = [], {}, {}
    excluded = False
    output_key = path_key(output)
    if any(path_key(path) == output_key for path in inputs):
        raise ValueError('output cannot also be an explicit input')

    def add(name, kind, mode, path=None):
        name = archive_name(name)
        old = names.get(name)
        if old is not None:
            if kind == 'directory' and old['is_dir'] and old.get('_path') is None:
                old['_path'], old['mode'] = path, mode
                return
            raise ValueError(f'duplicate archive path: {name}')
        if name.casefold() in case_names:
            raise ValueError(f'archive paths collide on case-insensitive systems: {name}')
        entry = dict(name=name, data=b'', mode=mode, mtime=mtime,
                     is_dir=kind == 'directory', kind=kind, _path=path)
        entries.append(entry)
        names[name] = entry
        case_names[name.casefold()] = name

    def parents(name):
        parts = PurePosixPath(name).parts
        for end in range(1, len(parts)):
            parent = '/'.join(parts[:end])
            existing = names.get(parent)
            if existing is not None:
                if not existing['is_dir']:
                    raise ValueError(f'archive parent is not a directory: {parent}')
            else:
                add(parent, 'directory', stat.S_IFDIR | 0o755)

    def walk(path, name):
        nonlocal excluded
        if path_key(path) == output_key:
            excluded = True
            return
        st = path.lstat()
        parents(name)
        if stat.S_ISLNK(st.st_mode):
            add(name, 'symlink', stat.S_IFLNK | stat.S_IMODE(st.st_mode), path)
        elif stat.S_ISDIR(st.st_mode):
            if getattr(path, 'is_junction', lambda: False)():
                raise ValueError(f'directory junctions are not followed: {path}')
            add(name, 'directory', stat.S_IFDIR | stat.S_IMODE(st.st_mode), path)
            for child in sorted(path.iterdir(), key=lambda p: p.name):
                walk(child, join_name(name, child.name))
        elif stat.S_ISREG(st.st_mode):
            add(name, 'file', stat.S_IFREG | stat.S_IMODE(st.st_mode), path)
        else:
            raise ValueError(f'unsupported special file: {path}')

    parents(reserved)
    if prefix:
        parents(prefix + '/placeholder')
    for path in inputs:
        st = path.lstat()
        if len(inputs) == 1 and stat.S_ISDIR(st.st_mode):
            for child in sorted(path.iterdir(), key=lambda p: p.name):
                walk(child, join_name(prefix, child.name))
        else:
            walk(path, join_name(prefix, path.name or path.resolve().name))
    if reserved.casefold() in case_names:
        raise ValueError(f'source collides with reserved archive member: {reserved}')
    if any(name.casefold().startswith(reserved.casefold()+'/') for name in names):
        raise ValueError(f'reserved self member would replace a directory: {reserved}')
    for entry in entries:
        path = entry.pop('_path')
        if entry['kind'] == 'file':
            entry['data'] = read_snapshot(path)
        elif entry['kind'] == 'symlink':
            entry['data'] = os.readlink(path).encode('utf-8')
    return entries, excluded


def verify_archive(archive, entries, self_name):
    expected = {entry['name']+('/' if entry['is_dir'] else ''):entry['data'] for entry in entries}
    expected[self_name] = archive
    with zipfile.ZipFile(io.BytesIO(archive)) as reader:
        infos = reader.infolist()
        if len(infos) != len(expected) or {i.filename for i in infos} != set(expected):
            raise ValueError('generated member list does not match the input snapshot')
        for member in infos:
            local = struct.unpack_from('<IHHHHHIIIHH', archive, member.header_offset)
            if (local[0] != 0x04034b50 or local[3] != member.compress_type
                    or local[6:9] != (member.CRC, member.compress_size, member.file_size)):
                raise ValueError(f'inconsistent ZIP local header: {member.filename}')
            at = member.header_offset + 30 + local[-2] + local[-1]
            wire = memoryview(archive)[at:at+member.compress_size]
            original = memoryview(expected[member.filename])
            if member.file_size != len(original):
                raise ValueError(f'wrong member size: {member.filename}')
            checksum, position = 0, 0
            def check_chunk(decoded):
                nonlocal checksum, position
                if decoded != original[position:position+len(decoded)]:
                    raise ValueError(f'member data mismatch: {member.filename}')
                checksum = zlib.crc32(decoded, checksum)
                position += len(decoded)
            inflater = zlib.decompressobj(-15) if member.compress_type == 8 else None
            if member.compress_type not in (0, 8):
                raise ValueError(f'unexpected ZIP compression method: {member.filename}')
            for start in range(0, len(wire), 1 << 20):
                pending = wire[start:start+(1 << 20)]
                while len(pending):
                    decoded = inflater.decompress(pending, 1 << 20) if inflater else pending
                    check_chunk(decoded)
                    pending = inflater.unconsumed_tail if inflater else b''
            if inflater:
                while True:
                    decoded = inflater.decompress(b'', 1 << 20)
                    if not decoded:
                        break
                    check_chunk(decoded)
            if inflater and (not inflater.eof or inflater.unused_data or inflater.unconsumed_tail):
                raise ValueError(f'incomplete or trailing Deflate data: {member.filename}')
            if position != len(original) or checksum != member.CRC:
                raise ValueError(f'member length or CRC mismatch: {member.filename}')


def atomic_write(path, data, force=False):
    if path.exists() and not force:
        raise FileExistsError(f'output exists; use --force to replace it: {path}')
    temporary = None
    try:
        with tempfile.NamedTemporaryFile(prefix='.zip-quine-', dir=path.parent, delete=False) as output:
            temporary = Path(output.name)
            output.write(data)
            output.flush()
            os.fsync(output.fileno())
        if force:
            os.replace(temporary, path)
        else:
            os.link(temporary, path)
            temporary.unlink()
        temporary = None
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('inputs', metavar='INPUT', nargs='+', type=Path)
    parser.add_argument('output', metavar='OUTPUT', type=Path)
    parser.add_argument('--quine-name', help='self member name (default: output basename)')
    parser.add_argument('--src-dir', default='', help='archive prefix for source entries')
    parser.add_argument('--quine-dir', default='', help='archive prefix for the self member')
    parser.add_argument('--mtime', type=int, default=0, help='fixed timestamp; ZIP dates clamp to 1980..2107 (default: 0)')
    parser.add_argument('--force', action='store_true', help='replace an existing output after verification')
    args = parser.parse_args(argv)
    try:
        if args.output.is_symlink():
            raise ValueError('output must not be a symbolic link')
        if args.output.exists() and not args.output.is_file():
            raise ValueError('output must be a regular file path')
        if args.output.exists() and not args.force:
            raise FileExistsError(f'output exists; use --force to replace it: {args.output}')
        if not args.output.parent.is_dir():
            raise ValueError(f'output directory does not exist: {args.output.parent}')
        prefix = archive_name(args.src_dir, empty=True)
        self_dir = archive_name(args.quine_dir, empty=True)
        self_name = join_name(self_dir, archive_name(args.quine_name or args.output.name))
        entries, excluded = collect_entries(args.inputs, args.output, prefix, self_name, args.mtime)
        archive = build(entries, self_name, mtime=args.mtime)
        verify_archive(archive, entries, self_name)
        atomic_write(args.output, archive, args.force)
        files = sum(e['kind'] == 'file' for e in entries)
        dirs = sum(e['is_dir'] for e in entries)
        links = sum(e['kind'] == 'symlink' for e in entries)
        print(f'[ZIP quine] {files} files, {dirs} directories, {links} symlinks')
        if excluded:
            print('[ZIP quine] excluded the output path from source traversal')
        print(f'[ZIP quine] wrote {args.output}: {len(archive)} bytes')
        print('[verify] member contents, CRC32 and exact self: PASS')
        print(f'[self] {self_name}')
        return 0
    except (OSError, ValueError, RuntimeError, OverflowError, zipfile.BadZipFile,
            zlib.error, struct.error) as error:
        print(f'error: {error}', file=sys.stderr)
        return 1


if __name__ == '__main__':
    raise SystemExit(main())
```
</details>

## 更多格式——RAR Quine
  其实大多数常用的压缩算法都和ZIP差不多，核心思想基本上都是LZ77那套用距离和长度来表示重复内容，然后再用各种熵编码来尽可能用少的比特存储数据……所以说RAR Quine理论上也可以存在，毕竟原理都差不多。同样，我也试着让AI做了一下，依旧轻轻松松就做出来了。不过毕竟RAR不是开源软件，生成的代码我就不放出来了，这里有一份生成后的RAR Quine：[quine.rar](/assets/quine.rar)，如果有人有兴趣，也可以研究一下这个档案。   

# 感想
  我明明对压缩的原理也只是了解点皮毛，但是有了AI之后能做的事情真的是太可怕了，各种只能在想象中的东西都被轻松实现……真的是越来越离不开AI了，再看看最近的AI又发现了大量基础设施的高危漏洞，很明显，人被AI淘汰的时代可能越来越近了，到时候……还有继续记录的意义吗？