2026-11-01 00:00:00
压缩……在消除冗余的同时，又创造了新的混乱？
上次我在做博客的离线模式时，为了降低带宽的消耗，我把我的博客压缩包又用7-Zip额外压缩了一遍，从大约16MiB压缩到了4MiB，效果感觉挺不错。另外我又试了试几个常见的压缩算法，比如Zip、zstd、BZip2等，全部使用最高等级压缩，但压缩效果都不如使用LZMA2的7-Zip好，速度甚至还比7-Zip慢，看来LZMA2还是相当强啊。
所以我在想，LZMA2是不是最强的压缩算法呢？虽然不同压缩算法可能对不同类型的数据效果不一样，但目前来看对我的数据来说，LZMA2看起来是最强的啊……不过这些只是常见算法，如果是用那些因为压缩速度慢而不太常见的算法，会不会比LZMA2更强呢？
我搜了一下，PAQ系列算法似乎是公认最强的，通过类似机器学习的算法花大量的时间达到更高的压缩比。其中比较知名的就是ZPAQ，这个程序已经有10年不更新了，当然对于算法来说这点倒也不是什么问题，毕竟LZMA2也是十几年前的算法。最终我试着用它压缩我的博客压缩包，用了一分钟左右的时间，压缩到了3.5MiB，比7-Zip压缩的少了大约10%，感觉效果还不错。
当然ZPAQ只是PAQ系列中的其中一个算法，有些人为了追求更高的压缩比，依然在对其进行改进，据我所知，目前仍在更新、压缩效果也非常出色的应该是PAQ8PX，我试着用v217版本压缩了一下，最终把压缩包压到了3.1MiB，真的是非常强。而代价也是极其高，首先压缩花了一个半小时，而且在这个过程中需要占用30GiB内存，更关键的问题是在解压时同样需要消耗这么多时间和资源，所以说这个算法真的是只能追求压缩比，基本上没办法正常使用……
除了PAQ系列之外，FFMPEG的作者Fabrice Bellard也做过一个使用机器学习的压缩算法，叫做NNCP，使用了和现代LLM同款的Transformer模型。于是我编译了一下，试着在RTX4090上压缩我的压缩包，也差不多用了一个半小时，结果压缩比还没LZMA2高😂，给我压到了4.1MiB，也许是因为我的博客里有些图片，而NNCP貌似是专攻英文文本的，所以效果差了不少……又一次证明了LZMA2真的很强啊～
后来我又听说还有一个叫做cmix的压缩算法，一样是基于机器学习，据说效果好像还不错，只不过在我实际用v21版本测试的时候不知道为什么总是压到后面的时候就会崩溃……这个项目也有2年没更新了，可能是代码有什么BUG。而且和PAQ8PX一样内存占用很高，用时非常长，但就从崩溃的位置来看，它压缩一半的结果还是比PAQ8PX更大，看来这个压缩算法的地位有点尴尬啊……不知道它是不是又是针对文本特化过的算法，但目前的结果来看感觉不如PAQ8PX。
最后PAQ8PX在这次测试中获得了最终的冠军🏆，我的博客压缩包里面包含的数据分布也算是有一定代表性吧，有一定的图片，大量的纯文本网页，和不少网站的结构应该很相似，所以这个压缩结果可能对不少数据都有一定的参考性。不过无论是PAQ8PX、NNCP还是cmix，它们压缩和解压需要的资源实在是太多了，不太适合一般情况下使用，至于ZPAQ虽然也是那种压缩和解压消耗相同的资源的算法，但作为十年前的东西来说，它在现在的速度还算是能在接受范围内，所以说如果想把不算太大的文件压到尽可能小，用ZPAQ看起来是最合适的选择，至于绝大多数的文件，还是用使用LZMA2算法的7-Zip吧～
在上次我做博客压缩包的时候，AI告诉我LZMA2和zstd都可以做能包含大文件的Quine，但我当时却以“zstd感觉不是很知名”的理由选择了LZMA2。然而实际上zstd甚至都有专门的RFC 8878，甚至有不少网站都已经选择了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用libarchive也有相同的问题……实在是令人难受，所以我又要求它改成.tar.zst格式，这样直接用Windows资源管理器也能直接打开，还增加了和我之前7z Quine一样的文件夹打包功能，说不定以后会用得上呢。
顺便一说，XXH64虽然说是非线性的，但作为非密码学哈希，倒也可以在不爆破的情况下构造值，于是以下就是最终的生成代码：
#!/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())
现在AI写的代码真的是越来越复杂了，真的是根本看不懂啊……但就是这样，很神奇的就把问题解决了。总之，zstd Quine也轻松实现了，也算是给自包含压缩包再提供一种新的格式吧。
后来闲的没事又让AI继续研究有没有可能做出大文件的ZIP Quine，结果它研究了一段时间还真让它研究出来了，当然Ruben Van Mello所说的窗口大小的问题依然存在，只不过AI用了某种方式绕过去了，而代价则是会存在成员数据有重叠的部分。比如在Python的标准库zipfile中，在这里有检查自身成员压缩数据是不是越过了后续成员的局部头，导致我这种方法生成的ZIP Quine无法被解压，然而7-Zip和libarchive都能正常解压……总之这种方式生成的有一点点兼容性问题，但问题不是很严重，那么以下就是这种生成器的代码：
#!/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())
其实大多数常用的压缩算法都和ZIP差不多，核心思想基本上都是LZ77那套用距离和长度来表示重复内容，然后再用各种熵编码来尽可能用少的比特存储数据……所以说RAR Quine理论上也可以存在，毕竟原理都差不多。同样，我也试着让AI做了一下，依旧轻轻松松就做出来了。不过毕竟RAR不是开源软件，生成的代码我就不放出来了，这里有一份生成后的RAR Quine：quine.rar，如果有人有兴趣，也可以研究一下这个档案。
我明明对压缩的原理也只是了解点皮毛，但是有了AI之后能做的事情真的是太可怕了，各种只能在想象中的东西都被轻松实现……真的是越来越离不开AI了，再看看最近的AI又发现了大量基础设施的高危漏洞，很明显，人被AI淘汰的时代可能越来越近了，到时候……还有继续记录的意义吗？
2026-10-01 00:00:00
让AI做出真正的“完整”将不再是难事……
在上次用AI制作了各种东西之后，我已经完全理解了AI确实是无所不能的。既然如此，那就让它帮我解决曾经未能解决的事情吧？
去年，我为了让下载全站压缩包的按钮不断链，让这个压缩包也包含它本身而研究了ZIP Quine，但限于DEFLATE的回溯窗口大小没能做到……但那是人做的东西，人还是太弱小了，现在换AI来试试，也许一切将变得不一样？
首先，我把Ruben Van Mello写的那篇论文《A Generator for Recursive Zip Files》以及生成器zip-quine-generator发给了AI，问它这个限制是不是真的，是不是真的无法创造出超过32KiB的ZIP Quine？没过多久它分析完了，告诉我确实存在这样的问题，但并不是无法解决的，最简单的办法就是换个回溯窗口更大的压缩算法。我看了一下它给我列出的几个算法，看起来LZMA2和zstd比较符合要求。不过zstd感觉不是很知名，一般的解压软件应该处理不了，所以我就选择让它基于LZMA2来制作了。
不过LZMA2只是算法，还得选个容器，考虑到7z算是最出名，而且LZMA2本来就是7-Zip的作者开发的，所以就直接让它做7z版本的了。另外因为Nate Choe做了通过扩展欧几里得求逆元来计算CRC32的PR，所以我也告诉它要用这种方法在多项式时间内解出CRC32的值，而不是爆破。就这样AI花了两个小时左右成功把代码写出来了，效果非常完美，另外它还偷懒地将计算CRC32的算法改成了高斯消元法😆，因为7z格式有3个相互影响的CRC32值，用扩展欧几里得会更复杂一点，不过对7z来说不写入文件的CRC32也不影响，只是会不校验罢了，至少它还是按我要求做了……但其实这个问题也并不是完全不能用扩展欧几里得算法，后来我又强烈要求了一下，AI还是给我写出来了。以下是代码片段，有兴趣的人可以参考一下（不过这个看起来写的就是很复杂，所以我自己只用了高斯消元法）：
RU_POLY = 0x104C11DB7 # x^32+x^26+x^23+...+1（含 x^32 首项）
def _find_probe(p):
ret = 0
for i in range(64):
if (1 << i) & p:
ret = i
return ret
def _mul_raw(p1, p2):
"""多项式乘法（无模）。"""
ret = 0
probe = _find_probe(p1)
for i in range(64):
if p2 & (1 << i):
assert probe + i < 64, "多项式乘法溢出"
ret ^= p1 << i
return ret
def _poly_divmod(dividend, divisor):
"""多项式长除法，返回 (商, 余)。"""
probe = _find_probe(divisor)
probe_bit = 1 << probe
quot = 0
rem = dividend
for i in range(63 - probe, -1, -1):
if rem & (probe_bit << i):
quot |= 1 << i
rem ^= divisor << i
return quot, rem
def _mul(p1, p2, mod):
"""p1*p2 % mod。"""
return _poly_divmod(_mul_raw(p1, p2), mod)[1]
def _xgcd(p1, p2):
"""扩展欧几里得：返回 (k1,k2,gcd)，满足 p1*k1+p2*k2=gcd。"""
if _find_probe(p1) < _find_probe(p2):
k1, k2, g = _xgcd(p2, p1)
return k2, k1, g
if p2 == 0:
return p1, 0, p1
q, r = _poly_divmod(p1, p2)
c1, c2, g = _xgcd(p2, r)
return c2, c1 ^ _mul_raw(c2, q), g
def _minv(p, mod):
"""p 在环上的乘法逆元；不存在时返回 0。"""
k1, _, g = _xgcd(p, mod)
if g != 1:
return 0
return _poly_divmod(k1, mod)[1]
def _bitrev32(x):
"""32 位比特反转。CRC 的输出循环冗余（binascii 反射）↔ 本环（非反射）互为 bit-reverse。"""
return int(format(x & 0xFFFFFFFF, "032b")[::-1], 2)
def _ring_pow_x(k):
"""环内 x^k mod P（P=RU_POLY），二进制幂 O(log k)。变量列就是单个环幂。"""
res, base, mod, e = 1, 2, RU_POLY, k
while e:
if e & 1:
res = _mul(res, base, mod)
base = _mul(base, base, mod)
e >>= 1
return res
def solve_crc_system(files):
"""解析解 CRC 定点系统
- 常数列 ：对“变量清零”的数据做一次 binascii.crc32，再 bit-reverse 32 位
（binascii 反射 CRC 与本环非反射 CRC 互为 bit-reverse，见 _bitrev32）。
- 变量列 ：单个环幂 x^(32 + 8*(len-pos-4))，二进制幂 O(log N) 一步算出；
同一变量出现在多处时按位异或累加。
各自 O(#rows) 次 CRC + 少量环幂即可建出整矩阵，Gauss 消元不变。
"""
n = len(files)
mod = RU_POLY
matrix = [[0] * (n + 1) for _ in range(n)]
for file in range(n):
data = files[file][0]
offsets = files[file][1]
N = len(data)
zero = bytearray(data)
for pos in offsets: # 未知量占 4 字节，清零后求常数项
zero[pos:pos + 4] = b"\x00\x00\x00\x00"
matrix[file][n] = _bitrev32(crc32(bytes(zero)))
for pos, vid in offsets.items():
rest = N - pos - 4 # 变量被 4 字节（x^32）+ 其后字节（x^8/个）推到底
matrix[file][vid] ^= _ring_pow_x(32 + 8 * rest)
matrix[file][file] ^= 1
for sr in range(n):
ex = sr
while ex < n and matrix[ex][sr] == 0:
ex += 1
if ex == n:
raise RuntimeError("CRC 多项式系统奇异 @row%d" % sr)
if ex != sr:
matrix[sr], matrix[ex] = matrix[ex], matrix[sr]
inv = _minv(matrix[sr][sr], mod)
for er in range(sr + 1, n):
mult = _mul(inv, matrix[er][sr], mod)
for ec in range(n + 1):
matrix[er][ec] ^= _mul(mult, matrix[sr][ec], mod)
res = [0] * n
for row in range(n - 1, -1, -1):
v = matrix[row][n]
for col in range(row + 1, n):
v ^= _mul(res[col], matrix[row][col], mod)
res[row] = _mul(v, _minv(matrix[row][row], mod), mod)
return [_impl_val(res[i]) for i in range(n)]
def _impl_val(math_val):
imp = 0
for bit in range(32):
if math_val & (1 << bit):
imp |= 1 << (31 - bit)
return imp
class BlogQuine:
def solve_crc(self, F, lay, seed):
T, h, d = lay["T"], lay["h"], lay["d"]
k, n_sub, crc_base = lay["k"], lay["n_sub"], lay["crc_base"]
hoff = lay["total"] - h
vout_f = lay["total"] - 2 * T
hcopy = vout_f + (T - h)
# 已知 CRC：seed + 各文件（quine 的是未知量 D）
known = [crc32(seed)] + [crc32(data) for _, data in self.files]
assert len(known) == n_sub - 1
for i, e in enumerate(known):
struct.pack_into("<I", F, hoff + crc_base + 4 * i, e)
struct.pack_into("<I", F, hcopy + crc_base + 4 * i, e)
# 未知量 D N S：D=整个文件、N=头部 F[hoff:hoff+h]、S=签名 F[12:32]
dpos = crc_base + 4 * (n_sub - 1)
ncopy = 32 + 3 * k + d # C1k 载荷里 F[0:35) 副本起点
groups = [[hoff + dpos, hcopy + dpos], # D 出现两处
[28, ncopy + 28], # N
[8, ncopy + 8]] # S
# 构造 3 个“文件”喂给解析解系统：
# 文件0 = 整个 F，三个未知量各自出现在 groups 两处
# 文件1 = 头部块，只有 D 出现在相对 dpos（另一处 hcopy+dpos 在头外）
# 文件2 = 签名块，只有 N 出现在相对 16（28-12）
whole = bytes(F)
hdr = bytes(F[hoff:hoff + h])
sig = bytes(F[12:32])
f0_off = {hoff + dpos: 0, hcopy + dpos: 0,
28: 1, ncopy + 28: 1,
8: 2, ncopy + 8: 2}
f1_off = {dpos: 0}
f2_off = {28 - 12: 1}
vals = solve_crc_system([(whole, f0_off), (hdr, f1_off), (sig, f2_off)])
for g in range(3):
for p in groups[g]:
F[p:p + 4] = struct.pack("<I", vals[g])
real = (crc32(F), crc32(F[hoff:hoff + h]), crc32(F[12:32]))
assert real == tuple(vals), "CRC 定点失败: %s != %s" % (real, vals)
return vals
于是我第一时间就把原来的TGZ压缩命令换掉，换成了AI给我写的blogquine.py，现在就可以通过这里下载到“完整”包含我博客所有内容的压缩包了。
不过唯一的问题就是这样做出来的压缩包并没有压缩😂，相当于给做成了普通的归档了。当然我的博客本身倒是不大，没压缩也多不了多少空间，但相比于能做出“完整”的效果来说，这点浪费的空间也是小问题了。
既然现在的AI什么都能做了，那上次我因为技术不足仅仅只是用Service Worker做了个代理，现在也可以升级成下载全站之后让全站离线的功能吧？不过之前做的压缩包并没有压缩，导致整个压缩包有接近20MiB的大小，后来我仔细想了一下我的需求，我只是想让下载全站压缩包的按钮不断链，并没有说解压后的文件一定就必须是它本身，所以我只要用相同的文件名再压缩一次应该就可以了，这样既能满足我的想法，又可以让其他人下载的时候不会花费太多的时间。
在我直接压缩之后，整个压缩包占用大概9MiB的空间，其实我已经很满意了，不过压之前是18MiB，也就是说压缩率是50%……虽然也不是不能接受，但我在想这是不是合理的结果呢？我让AI分析了一下50%的压缩率正不正常，它看了一眼之后告诉我文本部分的10MiB压缩完之后有1MiB，图片因为压不动所以直接占用了8MiB😅，这时候我才意识到原来我一直存着一堆占着大量空间的垃圾，而且还是用的PNG/JPG这种压缩率很低的格式……考虑到兼容性和压缩效果，我把占用比较大的图片转换为了WebP格式，最终整个压缩包就只占用了4MiB的空间，一口气又节省了50%的空间。
最后做完对压缩包的处理之后，我直接让AI给我基于那个压缩包制作离线模式，它轻轻松松就做出来了，最终使用的是libarchive库，在启用离线模式之后整个网站的速度简直就是0延迟，点击就能瞬间跳转，效果算是相当不错了。
做完博客压缩包之后我突发奇想，虽然东西是AI做的，但也是在我的手中做出了世界上第一个LZMA2 Quine，那么如果我向全世界第一个制作ZIP Quine的人展示一下我的作品会怎么样呢？目前可以知道，世界上第一个制作ZIP Quine的人是Erling Ellingsen，所以我向他发了个帖子展示了一下。大佬不愧是大佬，一眼就能看出理论上能压缩的7z Quine是可以做出来的，对我来说，我是完全不知道7-Zip居然还支持链式编解码，能可以通过“double lzma2”的方式做到既可以自包含、又能真正被压缩的压缩包。既然大佬说理论上可以，那我自然是相信的，于是就让AI给我使劲折腾想办法把它做出来。在折腾了一晚上之后AI终于做出来了，真是强得可怕：
# ============================================================
# 双层 LZMA2 的 7z quine（"double lzma2"）
#
# F = S32 ‖ P ‖ H，folder 的 coder 链 = [LZMA2(外层), LZMA2(内层)]，
# 解码管线：P --外层--> I --内层--> U。
#
# I = Cseg ‖ LBseg ‖ Mseg ‖ 0x00
# Cseg : 真实 LZMA2 对 content 的压缩流（内容真正的压缩数据 C）
# LBseg: 内层 store 链，载荷 = seed（seed 内含一份 C 的参照副本）
# Mseg : 内层 match 链，输出 quine 区（= F 的逐字节副本）
# U = content ‖ seed ‖ F
#
# 外层把 I 再压一遍：Cseg 用 store 链原样承载；seed 里的 C 参照副本与 Cseg
# 逐字节相同，被外层的 rep0 回溯匹配消掉，于是成品里只剩一份压缩数据。
# quine 区回引 P 的字节时，源全在 seed 里（C 副本 / 头样本表 / 机械串）。
# ============================================================
def lzma_chunk(tokens, out_pos):
"""编一个 LZMA chunk（0xC0: state+props reset，无 dict reset，无 end marker）。
match token 的编码字节只取决于 (dist,len,pos)，与历史内容无关（不维护输出历史）。
解压长度用 21 位（控制字节低 5 位存高 5 位），故单 chunk 输出上限 2 MiB。"""
enc = LzmaEncoder()
enc.pos = out_pos
total = 0
for t in tokens:
if t[0] == "m":
enc.match(t[1], t[2])
total += t[2]
else:
enc.rep_match(t[1])
total += t[1]
data = enc.finish()
u = total - 1
c = len(data) - 1
assert 0 <= u < (1 << 21) and 0 <= c < 65536, (total, len(data))
hdr = bytes([0xC0 | ((u >> 16) & 0x1F)]) + (u & 0xFFFF).to_bytes(2, "big") + \
(c & 0xFFFF).to_bytes(2, "big") + bytes([PROPS_BYTE])
return hdr + data, total
LZ2_MAXU = 1 << 21 # LZMA2 单个 LZMA chunk 的解压上限
def lzma2_raw(data, dict_size, preset=6):
"""FORMAT_RAW + LZMA2 压缩；去掉结尾 0x00 便于后面继续追加 chunk。"""
c = lzma.LZMACompressor(
format=lzma.FORMAT_RAW,
filters=[{"id": lzma.FILTER_LZMA2, "dict_size": dict_size, "preset": preset}])
out = c.compress(data) + c.flush()
assert out and out[-1] == 0x00, "LZMA2 流应以 0x00 结尾"
return out[:-1]
def split_len(n, cap):
"""按 cap 上限切分 n 字节。末片若为 1 字节则从前一片借 1（match 最短为 2）。"""
Ls, rem = [], n
while rem > cap:
Ls.append(cap)
rem -= cap
if rem == 1 and Ls:
Ls[-1] -= 1
rem = 2
Ls.append(rem)
return Ls
class DoubleQuine(BlogQuine):
"""双层 LZMA2 版本：folder = [LZMA2(外层), LZMA2(内层)]。"""
def entries(self):
"""U 的顺序是 内容文件… ‖ seed ‖ quine，条目顺序必须与之对应。"""
ent = [{"name": r, "dir": isd} for r, isd in self.walk_entries]
ent.append({"name": self.seed_name, "dir": False})
ent.append({"name": self.quine_name, "dir": False})
return ent
# ---------- 头部：双 coder + BindPair ----------
def build_header(self, n, total, d, entries, sub_sizes, size_I):
n_sub = len(sub_sizes)
N = len(entries)
h = bytearray()
h += b"\x01\x04" # kHeader, kMainStreamsInfo
h += b"\x06" + varint(0) + varint(1) + b"\x09" + varint(n) + b"\x00"
h += b"\x07" # kUnpackInfo
h += b"\x0b" + varint(1) + b"\x00" # kFolder: 1, local
h += varint(2) # NumCoders = 2
h += b"\x21\x21" + varint(1) + bytes([self.dict_prop_out]) # coder0 = 外层
h += b"\x21\x21" + varint(1) + bytes([self.dict_prop_in]) # coder1 = 内层
h += varint(1) + varint(0) # BindPair = (InIndex=1, OutIndex=0)
h += b"\x0c" + varint(size_I) + varint(d + total) # 两个 coder 的输出大小
h += b"\x00"
h += b"\x08" # SubStreamsInfo
h += b"\x0d" + varint(n_sub)
h += b"\x09" + b"".join(varint(s) for s in sub_sizes[:-1])
h += b"\x0a" + b"\x01"
crc_base = len(h)
h += b"\x00" * (4 * n_sub)
h += b"\x00\x00"
h += b"\x05" + varint(N) # FilesInfo
bits = bytearray((N + 7) // 8)
for i, e in enumerate(entries):
if e["dir"]:
bits[i // 8] |= 1 << (7 - i % 8)
h += b"\x0e" + varint(len(bits)) + bytes(bits)
mt = b"\x01\x00" + struct.pack("<Q", self.mtime_ft) * N
h += b"\x14" + varint(len(mt)) + mt
names = bytearray(b"\x00")
for e in entries:
names += e["name"].encode("utf-16-le") + b"\x00\x00"
h += b"\x11" + varint(len(names)) + bytes(names)
attrs = b"".join(struct.pack("<I", 0x10 if e["dir"] else 0x20)
for e in entries)
h += b"\x15" + varint(2 + len(attrs)) + b"\x01\x00" + attrs
h += b"\x00\x00"
return bytes(h), crc_base
# ---------- 布局：三层坐标的不动点迭代 ----------
def layout(self):
content = self.content
dc = len(content)
dict_c = max(1 << 20, 1 << max(20, dc.bit_length()))
C = lzma2_raw(content, dict_c)
self._C = C # 内容压缩只做一次，finalize 复用
lc = len(C)
# 迭代变量：Y = plant 长度, H = 头部长度, M = Mseg 字节数
Y, H, M = 96, 256, 256
seen = set()
for _ in range(200):
ds = lc + 32 + H + Y # seed = C副本‖S32‖H‖plant
d = dc + ds
Ls_LB = split_len(ds, CHUNK)
lLB = 3 * len(Ls_LB) + ds
off_in = lc + lLB
M1 = M + 1 # Mseg ‖ 内层流终止符 0x00
Ls_M = split_len(M1, CHUNK)
Li = off_in + M1
self.dict_prop_out, _ = dict_prop_for(max(Li, 1 << 20))
self.dict_prop_in, _ = dict_prop_for(d + 32 + H + 8192)
st = self._build_streams(C, lc, ds, d, Ls_LB, lLB, off_in, Ls_M,
M1, H)
if st is None:
Y += 8
continue
mseg = self._build_mseg(st, d, lc, ds, H, M)
if mseg is None:
Y += 8
continue
M_new = len(mseg)
Y_new = len(st["hdr_bytes"]) + len(st["mech"]) + M_new + 2
total = st["n"] + 32 + H
file_sizes = [len(x) for _, x in self.files]
sub_sizes = file_sizes + [ds] + [total]
entries = self.entries()
hdr, crc_base = self.build_header(st["n"], total, d, entries,
sub_sizes, Li)
H_new = len(hdr)
key = (Y_new, H_new, M_new)
if key in seen: # 再次见到同一组值 = 已收敛
Y, H, M = Y_new, H_new, M_new
break
seen.add(key)
Y, H, M = Y_new, H_new, M_new
else:
raise RuntimeError("double layout 不收敛")
return {"Y": Y, "H": H, "M": M}
# ---------- 定稿：用收敛后的 (Y, H, M) 重算全部结构 ----------
def finalize(self, Y, H, M):
content = self.content
dc = len(content)
C = self._C # layout 里已压缩过
lc = len(C)
ds = lc + 32 + H + Y
d = dc + ds
Ls_LB = split_len(ds, CHUNK)
lLB = 3 * len(Ls_LB) + ds
off_in = lc + lLB
M1 = M + 1
Ls_M = split_len(M1, CHUNK)
Li = off_in + M1
self.dict_prop_out, ds_out = dict_prop_for(max(Li, 1 << 20))
self.dict_prop_in, ds_in = dict_prop_for(d + 32 + H + 8192)
st = self._build_streams(C, lc, ds, d, Ls_LB, lLB, off_in, Ls_M, M1, H)
mseg = self._build_mseg(st, d, lc, ds, H, M)
assert len(mseg) == M, "Mseg %d != %d" % (len(mseg), M)
total = st["n"] + 32 + H
file_sizes = [len(x) for _, x in self.files]
sub_sizes = file_sizes + [ds] + [total]
entries = self.entries()
hdr, crc_base = self.build_header(st["n"], total, d, entries,
sub_sizes, Li)
assert len(hdr) == H, "header %d != %d" % (len(hdr), H)
plant = st["hdr_bytes"] + bytes(st["mech"]) + mseg + b"\x00\x00"
assert len(plant) == Y, "plant %d != %d" % (len(plant), Y)
return {"C": C, "lc": lc, "dc": dc, "ds": ds, "d": d, "Li": Li,
"ochunks": st["ochunks"], "hdr_tab": st["hdr_tab"],
"mseg": mseg, "M": M, "H": H, "n": st["n"], "total": total,
"header": hdr, "crc_base": crc_base, "plant": plant,
"dict_size_in": ds_in, "dict_size_out": ds_out,
"n_sub": len(sub_sizes)}
# ---------- 外层流 P 的构造 ----------
def _build_streams(self, C, lc, ds, d, Ls_LB, lLB, off_in, Ls_M, M1, H):
hdr_tab, hdr_idx = [], {}
def hidx(b):
assert len(b) == 3 and b[0] in (0x01, 0x02), b.hex()
if b not in hdr_idx:
hdr_idx[b] = len(hdr_tab)
hdr_tab.append(b)
return hdr_idx[b]
ochunks = []
f, io = 32, 0
mech = bytearray()
def put_lzma(tokens, ioff):
blk, t = lzma_chunk(tokens, ioff)
ochunks.append({"kind": "lzma", "foff": f, "size": len(blk),
"ioff": ioff, "olen": t, "bytes": blk,
"mech": len(mech)})
mech.extend(blk)
return len(blk), t
def put_store(ioff, olen, hi, src):
ochunks.append({"kind": "store", "foff": f, "size": 3 + olen,
"ioff": ioff, "olen": olen, "hdr": hi, "src": src})
return 3 + olen
# 段 1：store 链承载 Cseg（I[0:lc] = C 的字节；源 = seed 的 C 参照副本）
Ls_C = split_len(lc, CHUNK)
c_off = 0
for L in Ls_C:
f += put_store(io, L, hidx(store_hdr(L, first=(c_off == 0))),
("seed", c_off))
io += L
c_off += L
assert io == lc
# 段 2：LBseg（I[lc : lc+lLB]）—— 内层 store 链，载荷 = seed
lb_off = lc
seed_off = 0
for j, L in enumerate(Ls_LB):
lb_hdr = store_hdr(L, first=False)
if j == 0 or L != CHUNK:
# 没有可回引的相同样本，用外层 store chunk 原样携带这 3 字节
f += put_store(lb_off, 3, hidx(store_hdr(3, first=False)),
("tab", hidx(lb_hdr)))
io += 3
else:
# 与 chunk0 的头逐字节相同，直接从 I[lc:lc+3] 复制
n, t = put_lzma([("m", lb_off - lc, 3)], lb_off)
assert t == 3
f += n
io += 3
# 载荷：C 参照副本区走 rep0 回溯匹配（去重发生处），其余走 store
pay = lb_off + 3
i = 0
while i < L:
if seed_off + i < lc:
a = min(L - i, lc - (seed_off + i), LZ2_MAXU)
dist = (pay + i) - (seed_off + i)
n, t = put_lzma(matches_for(dist, a), pay + i)
assert t == a
f += n
io += a
i += a
else:
a = min(L - i, LZ2_MAXU)
f += put_store(pay + i, a, hidx(store_hdr(a, first=False)),
("seed", seed_off + i))
io += a
i += a
lb_off += 3 + L
seed_off += L
assert lb_off == lc + lLB and io == lc + lLB
# 段 3：store 链承载 Mseg ‖ 0x00（0x00 是内层流的终止符）
m_off = 0
for L in Ls_M:
f += put_store(off_in + m_off, L,
hidx(store_hdr(L, first=False)), ("mseg", m_off))
io += L
m_off += L
# 外层流自己的终止符 0x00
ochunks.append({"kind": "term", "foff": f, "size": 1, "ioff": io,
"olen": 0, "bytes": b"\x00"})
f += 1
return {"ochunks": ochunks, "mech": mech, "hdr_tab": hdr_tab,
"hdr_bytes": b"".join(hdr_tab), "n": f - 32}
# ---------- 内层 Mseg：把 quine 区的每个 F 区段映射成 match 链 ----------
def _build_mseg(self, st, d, lc, ds, H, M):
dc = d - ds
s_S32 = dc + lc # seed 里 S32 副本
s_H = dc + lc + 32 # seed 里 H 副本
s_tab = dc + lc + 32 + H # seed 里头样本表
s_mech = s_tab + len(st["hdr_bytes"]) # seed 里机械串
s_mseg = s_mech + len(st["mech"]) # seed 里 Mseg 副本
total = st["n"] + 32 + H
hoff = 32 + st["n"]
segs = [(0, 32, s_S32)] # F[0:32) 签名头
for c in st["ochunks"]:
if c["kind"] == "store":
segs.append((c["foff"], 3, s_tab + 3 * c["hdr"]))
src = c["src"]
if src[0] == "seed":
u = dc + src[1]
elif src[0] == "tab":
u = s_tab + 3 * src[1]
else: # ("mseg", off)
u = s_mseg + src[1]
segs.append((c["foff"] + 3, c["olen"], u))
elif c["kind"] == "lzma":
segs.append((c["foff"], c["size"], s_mech + c["mech"]))
else: # term：并入前一个载荷段
pf, pl, pu = segs[-1]
segs[-1] = (pf, pl + 1, pu)
segs.append((hoff, H, s_H)) # F[hoff:] 头部
pos = 0
for (fo, ln, u) in segs:
assert fo == pos, "段不连续: %d != %d" % (fo, pos)
assert ln >= 2, "段太短 %d @%d" % (ln, fo)
pos += ln
assert pos == total, "段总长 %d != total %d" % (pos, total)
# 生成 match 链，按 2 MiB 上限切成多个 0xC0 chunk
mseg = bytearray()
out_pos = d
cur, cur_len = [], 0
chunks = []
for (fo, ln, u) in segs:
dist = (d + fo) - u
assert 0 < dist, "dist=%d @%d" % (dist, fo)
toks = matches_for(dist, ln)
if cur_len + ln > LZ2_MAXU - MAX_MATCH:
chunks.append((cur, out_pos, cur_len))
out_pos += cur_len
cur, cur_len = [], 0
cur.extend(toks)
cur_len += ln
if cur:
chunks.append((cur, out_pos, cur_len))
out_pos += cur_len
assert out_pos == d + total, "Mseg 输出 %d != %d" % (out_pos, d + total)
for (toks, op, cl) in chunks:
blk, t = lzma_chunk(toks, op)
assert t == cl
mseg.extend(blk)
return bytes(mseg)
# ---------- F 中承载 seed[soff] 的那个字节的偏移 ----------
def _f_off_of_seed(self, lay, soff):
"""返回 (F 偏移, 该 store 载荷内从 soff 起还剩多少字节)。
seed_head 超过 64 KiB 时会被切成多片，所以调用方要循环推进。"""
for c in lay["ochunks"]:
if c["kind"] == "store" and c["src"][0] == "seed":
s0 = c["src"][1]
if s0 <= soff < s0 + c["olen"]:
return c["foff"] + 3 + (soff - s0), s0 + c["olen"] - soff
raise RuntimeError("seed 偏移 %d 未被 store 载荷覆盖" % soff)
def _f_pos_list(self, lay, soff, length):
"""seed[soff:soff+length] 在 F 中对应的字节位置列表（可能跨多片）。"""
res = []
while length > 0:
p, avail = self._f_off_of_seed(lay, soff)
a = min(length, avail)
res.extend(range(p, p + a))
soff += a
length -= a
return res
# ---------- 装配 ----------
def assemble(self, lay):
total, H = lay["total"], lay["H"]
hoff = total - H
C, plant = lay["C"], lay["plant"]
F = bytearray(total)
# pass A: 签名头 + 所有非 store 载荷字节
F[0:6] = SIG
F[6:8] = VER
struct.pack_into("<Q", F, 12, lay["n"])
struct.pack_into("<Q", F, 20, H)
for c in lay["ochunks"]:
if c["kind"] == "store":
F[c["foff"]:c["foff"] + 3] = lay["hdr_tab"][c["hdr"]]
else:
F[c["foff"]:c["foff"] + c["size"]] = c["bytes"]
F[hoff:hoff + H] = lay["header"]
# pass B: seed = C副本 ‖ S32副本 ‖ H副本 ‖ plant
seed = bytearray()
seed += C
seed += bytes(F[0:32])
seed += bytes(F[hoff:hoff + H])
seed += plant
assert len(seed) == lay["ds"], "seed %d != %d" % (len(seed), lay["ds"])
# pass C: 各 store 的载荷
msegz = lay["mseg"] + b"\x00"
for c in lay["ochunks"]:
if c["kind"] != "store":
continue
p = c["foff"] + 3
src, L = c["src"], c["olen"]
if src[0] == "seed":
F[p:p + L] = seed[src[1]:src[1] + L]
elif src[0] == "tab":
F[p:p + 3] = lay["hdr_tab"][src[1]]
else:
F[p:p + L] = msegz[src[1]:src[1] + L]
return F, bytes(seed)
# ---------- CRC 定点（128x128 GF(2)：seed/quine/Next/Start 四个 CRC） ----------
def solve_crc(self, F, lay, C, plant):
H, hoff = lay["H"], lay["total"] - lay["H"]
n_sub, crc_base = lay["n_sub"], lay["crc_base"]
lc = lay["lc"]
# seed 里 S32 / H 副本在 F 中的镜像（可能跨多片，逐字节列出）
scopy0 = self._f_pos_list(lay, lc, 32)
Fv = memoryview(F)
def seed_crc():
v = crc32(C)
v = crc32(Fv[0:32], v) # S32 副本
v = crc32(Fv[hoff:hoff + H], v) # H 副本
return crc32(plant, v)
def targets():
return (seed_crc(), crc32(Fv), crc32(Fv[hoff:hoff + H]),
crc32(Fv[12:32]))
# 已知：各内容文件的 CRC，写本体 + seed 里 H 副本的镜像两处
known = [crc32(data) for _, data in self.files]
for i, e in enumerate(known):
slot = crc_base + 4 * i
b = struct.pack("<I", e)
for k in range(4):
F[hoff + slot + k] = b[k]
mir = self._f_pos_list(lay, lc + 32 + slot, 4)
assert len(mir) == 4
for k in range(4):
F[mir[k]] = b[k]
# 未知量：D_seed(seed 的 CRC)、D(quine 的 CRC)、N、S
# 每个未知量 4 字节，每字节有"本体 + seed 镜像"两个落点
qs = crc_base + 4 * len(known) # seed 子流的 CRC 槽
q = crc_base + 4 * (n_sub - 1) # quine 子流的 CRC 槽
groups = []
for slot in (qs, q):
groups.append([list(range(hoff + slot, hoff + slot + 4)),
self._f_pos_list(lay, lc + 32 + slot, 4)])
groups.append([list(range(28, 32)), scopy0[28:32]]) # NextHeaderCRC
groups.append([list(range(8, 12)), scopy0[8:12]]) # StartHeaderCRC
nb = 32 * len(groups)
base = targets()
basevec = 0
for i, v in enumerate(base):
basevec |= v << (32 * i)
cols = []
for g in range(len(groups)):
own, mir = groups[g]
for kb in range(32):
bi, bb = kb // 8, kb % 8
poss = [own[bi], mir[bi]]
for p in poss:
F[p] ^= (1 << bb)
t = targets()
for p in poss:
F[p] ^= (1 << bb)
delta = 0
for i, v in enumerate(t):
delta |= (v ^ base[i]) << (32 * i)
cols.append(delta)
eqs = []
for i in range(nb):
coeff = 0
for j in range(nb):
if (cols[j] >> i) & 1:
coeff |= 1 << j
coeff ^= 1 << i
eqs.append([coeff, (basevec >> i) & 1])
for j in range(nb):
p = next((i for i in range(j, nb) if (eqs[i][0] >> j) & 1), None)
if p is None:
raise RuntimeError("GF(2) 奇异 @bit%d" % j)
eqs[j], eqs[p] = eqs[p], eqs[j]
for i in range(nb):
if i != j and ((eqs[i][0] >> j) & 1):
eqs[i][0] ^= eqs[j][0]
eqs[i][1] ^= eqs[j][1]
x = 0
for i in range(nb):
assert eqs[i][0] == (1 << i)
if eqs[i][1]:
x |= 1 << i
vals = [(x >> (32 * i)) & 0xFFFFFFFF for i in range(len(groups))]
for g in range(len(groups)):
own, mir = groups[g]
b = struct.pack("<I", vals[g])
for k in range(4):
F[own[k]] = b[k]
F[mir[k]] = b[k]
t = targets()
assert t == tuple(vals), "CRC 定点失败: %s != %s" % (t, vals)
return vals
def build(self):
l0 = self.layout()
lay = self.finalize(l0["Y"], l0["H"], l0["M"])
F, seed = self.assemble(lay)
crcs = self.solve_crc(F, lay, lay["C"], lay["plant"])
# CRC 定点会改写 H / S32，而 seed 是它们的镜像，定点后需重建
hoff = lay["total"] - lay["H"]
seed = lay["C"] + bytes(F[0:32]) + bytes(F[hoff:hoff + lay["H"]]) \
+ lay["plant"]
return bytes(F), lay, seed, crcs
def verify_double(F, lay, files, seed, content):
"""串联两个原始 LZMA2 解码器还原 U，逐段校验。"""
d, total = lay["d"], lay["total"]
dec_o = lzma.LZMADecompressor(
format=lzma.FORMAT_RAW,
filters=[{"id": lzma.FILTER_LZMA2, "dict_size": lay["dict_size_out"]}])
I = dec_o.decompress(F[32:32 + lay["n"]])
ok = True
if len(I) != lay["Li"]:
print("[verify] 外层输出 %d != I %d" % (len(I), lay["Li"]))
return False
dec_i = lzma.LZMADecompressor(
format=lzma.FORMAT_RAW,
filters=[{"id": lzma.FILTER_LZMA2, "dict_size": lay["dict_size_in"]}])
U = dec_i.decompress(I)
exp = d + total
if len(U) != exp:
print("[verify] 内层输出 %d != %d" % (len(U), exp))
return False
dc, ds = lay["dc"], lay["ds"]
if U[:dc] != content:
print("[verify] 内容不匹配")
ok = False
pos = 0
for rel, data in files: # 内容文件排在 U 的最前面
if U[pos:pos + len(data)] != data:
print("[verify] 文件不匹配:", rel)
ok = False
pos += len(data)
assert pos == dc
if U[dc:dc + ds] != seed:
print("[verify] seed 不匹配")
ok = False
if U[d:] != F:
print("[verify] quine 自复制不匹配")
ok = False
print("[verify] 自解码: |I|=%d, |U|=%d, seed/文件/quine 全部匹配: %s"
% (len(I), len(U), ok))
return ok
仅仅是大佬的一句话，AI就能做出来，实在是太可怕了。最终我试了一下，官方的7-Zip可以正常解压，但是Windows资源管理器不行……Windows资源管理器使用的也是libarchive，似乎是因为libarchive只支持2个解码器，并且第二个解码器只能是像BCJ这样的过滤器……明明7-Zip官方是可以解压的，不知道如果把这个问题当作BUG报告给libarchive他们会不会解决？但目前存在兼容性问题的话我博客压缩包就先不用这个方案了吧🥲……
在做完7z格式的压缩包之后，我发现了一个问题，虽然7z确实很流行，但是在Linux下解压起来有点麻烦，7-Zip历史上主要面向Windows，Linux上长期以来更多依赖p7zip等第三方移植，因此生态集成度不如tar.xz，想要解压7z文件还得额外安装。
不过Linux下也有个支持LZMA2算法的压缩软件，那就是前些年出过后门的XZ Utils，配合tar就可以做出TXZ（tar.xz）文件，甚至用我博客终端中的BusyBox也能解压。我想了一下反正有AI，干脆一句话让AI帮我把blogquine.py改成tar.xz格式的好了，结果倒也没费多少功夫，AI就这样写出来了：
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import argparse
import binascii
import io
import lzma
import os
import struct
import sys
import tarfile
import time
import ctypes
import ctypes.util
# ============================================================
# Part 1: 最小 LZMA1 range coder（只做编码）
# ============================================================
kNumBitModelTotalBits = 11
kBitModelTotal = 1 << kNumBitModelTotalBits # 2048
kNumMoveBits = 5
kTopValue = 1 << 24 # 0x1000000
kNumStates = 12
kNumLitStates = 7
kNumPosBitsMax = 4
kNumLenToPosStates = 4
kNumAlignBits = 4
kEndPosModelIndex = 14
kNumFullDistances = 1 << (kEndPosModelIndex >> 1) # 128
kMatchMinLen = 2
kNumLowLenBits = 3
kNumMidLenBits = 3
kNumHighLenBits = 8
kNumLowLenSymbols = 1 << kNumLowLenBits # 8
kNumMidLenSymbols = 1 << kNumMidLenBits # 8
kNumPosSlotBits = 6
PROB_INIT = kBitModelTotal >> 1 # 1024
class RangeEncoder:
"""LZMA 的区间编码器。low 是 64 位（要容纳进位），range 是 32 位。"""
def __init__(self):
self.low = 0
self.range = 0xFFFFFFFF
self.cache = 0
self.cache_size = 1 # 初值 1 -> 第一个输出字节恒为 0x00
self.buf = bytearray()
def _shift_low(self):
if (self.low & 0xFFFFFFFF) < 0xFF000000 or (self.low >> 32) != 0:
temp = self.cache
while True:
self.buf.append((temp + (self.low >> 32)) & 0xFF)
temp = 0xFF
self.cache_size -= 1
if self.cache_size == 0:
break
self.cache = (self.low >> 24) & 0xFF
self.cache_size += 1
self.low = ((self.low & 0xFFFFFFFF) << 8) & 0xFFFFFFFF
def encode_bit(self, probs, idx, bit):
p = probs[idx]
bound = (self.range >> kNumBitModelTotalBits) * p
if bit == 0:
self.range = bound
probs[idx] = p + ((kBitModelTotal - p) >> kNumMoveBits)
else:
self.low += bound
self.range -= bound
probs[idx] = p - (p >> kNumMoveBits)
while self.range < kTopValue:
self.range = (self.range << 8) & 0xFFFFFFFF
self._shift_low()
def encode_direct_bits(self, value, num_bits):
for i in range(num_bits - 1, -1, -1):
self.range >>= 1
if (value >> i) & 1:
self.low += self.range
while self.range < kTopValue:
self.range = (self.range << 8) & 0xFFFFFFFF
self._shift_low()
def bittree_encode(self, probs, off, num_bits, symbol):
m = 1
for i in range(num_bits - 1, -1, -1):
bit = (symbol >> i) & 1
self.encode_bit(probs, off + m, bit)
m = (m << 1) | bit
def bittree_reverse_encode(self, probs, off, num_bits, symbol):
m = 1
for i in range(num_bits):
bit = symbol & 1
symbol >>= 1
self.encode_bit(probs, off + m, bit)
m = (m << 1) | bit
def finish(self):
for _ in range(5):
self._shift_low()
return bytes(self.buf)
def _pos_slot_and_bits(d):
if d < 4:
return d, 0, 0
for slot in range(4, 64):
n = (slot >> 1) - 1
base = (2 | (slot & 1)) << n
if d < base + (1 << n):
return slot, n, d - base
raise ValueError("distance too large: %d" % d)
# pb 必须与写进每个 LZMA2 chunk 头的 PROPS_BYTE(0x5D) 一致。
# lc/lp 只影响 literal 编码的概率表索引，而本编码器不实现 literal 路径，故无需常量。
PB = 2
class LzmaEncoder:
"""只实现 quine 用到的两条路径：match 与 rep0-match。
literal / rep-g1 / rep-g2 三条分支从未被构造，因此它们对应的概率表
（p_lit、p_is_rep_g1、p_is_rep_g2）以及 prev_byte 都不必存在。
reps 同理：编码器从不回读（rep0 的距离由解码器自行维护），故不保存。
"""
def __init__(self):
self.pos_mask = (1 << PB) - 1
self.rc = RangeEncoder()
self.pos = 0
self.state = 0
n = PROB_INIT
self.p_is_match = [n] * (kNumStates << kNumPosBitsMax)
self.p_is_rep = [n] * kNumStates
self.p_is_rep_g0 = [n] * kNumStates
self.p_rep0_long = [n] * (kNumStates << kNumPosBitsMax)
self.p_pos_slot = [n] * (kNumLenToPosStates << kNumPosSlotBits)
self.p_spec_pos = [n] * (kNumFullDistances - kEndPosModelIndex)
self.p_align = [n] * (1 << kNumAlignBits)
self.p_len_choice = [n] * 2
self.p_len_low = [n] * (16 * kNumLowLenSymbols)
self.p_len_mid = [n] * (16 * kNumMidLenSymbols)
self.p_len_high = [n] * (1 << kNumHighLenBits)
self.p_rep_len_choice = [n] * 2
self.p_rep_len_low = [n] * (16 * kNumLowLenSymbols)
self.p_rep_len_mid = [n] * (16 * kNumMidLenSymbols)
self.p_rep_len_high = [n] * (1 << kNumHighLenBits)
@property
def pos_state(self):
return self.pos & self.pos_mask
def _encode_len(self, choice, low, mid, high, length):
ps = self.pos_state
l = length - kMatchMinLen
if l < kNumLowLenSymbols:
self.rc.encode_bit(choice, 0, 0)
self.rc.bittree_encode(low, ps << kNumLowLenBits, kNumLowLenBits, l)
else:
self.rc.encode_bit(choice, 0, 1)
l -= kNumLowLenSymbols
if l < kNumMidLenSymbols:
self.rc.encode_bit(choice, 1, 0)
self.rc.bittree_encode(mid, ps << kNumMidLenBits, kNumMidLenBits, l)
else:
self.rc.encode_bit(choice, 1, 1)
self.rc.bittree_encode(high, 0, kNumHighLenBits, l - kNumMidLenSymbols)
def _write_dist(self, dist, lts):
d = dist - 1
slot, n, low_bits = _pos_slot_and_bits(d)
self.rc.bittree_encode(self.p_pos_slot, lts << kNumPosSlotBits,
kNumPosSlotBits, slot)
if slot >= 4:
if slot < kEndPosModelIndex:
base = (2 | (slot & 1)) << n
off = base - slot - 1
self.rc.bittree_reverse_encode(self.p_spec_pos, off, n, low_bits)
else:
self.rc.encode_direct_bits(low_bits >> kNumAlignBits, n - kNumAlignBits)
self.rc.bittree_reverse_encode(self.p_align, 0, kNumAlignBits,
low_bits & ((1 << kNumAlignBits) - 1))
def match(self, dist, length):
assert kMatchMinLen <= length <= 273, length
ps = self.pos_state
self.rc.encode_bit(self.p_is_match, (self.state << kNumPosBitsMax) + ps, 1)
self.rc.encode_bit(self.p_is_rep, self.state, 0)
lts = min(length - kMatchMinLen, kNumLenToPosStates - 1)
self._encode_len(self.p_len_choice,
self.p_len_low, self.p_len_mid, self.p_len_high, length)
self._write_dist(dist, lts)
self.state = 7 if self.state < kNumLitStates else 10
self.pos += length
def rep_match(self, length):
assert kMatchMinLen <= length <= 273, length
ps = self.pos_state
self.rc.encode_bit(self.p_is_match, (self.state << kNumPosBitsMax) + ps, 1)
self.rc.encode_bit(self.p_is_rep, self.state, 1)
self.rc.encode_bit(self.p_is_rep_g0, self.state, 0)
self.rc.encode_bit(self.p_rep0_long, (self.state << kNumPosBitsMax) + ps, 1)
self._encode_len(self.p_rep_len_choice,
self.p_rep_len_low, self.p_rep_len_mid, self.p_rep_len_high, length)
self.state = 8 if self.state < kNumLitStates else 11
self.pos += length
def finish(self):
return self.rc.finish()
# ============================================================
# Part 2: tar + xz 格式原语
# ============================================================
MAX_MATCH = 273 # LZMA1 单个 match 长度上限
CHUNK = 65536 # LZMA2 单 chunk 解压上限
PROPS_BYTE = 0x5D # lc=3, lp=0, pb=2
SEED_NOTE = ("\nMayx's Blog!").encode("utf-8")
# xz 格式常量
XZ_MAGIC = b"\xFD7zXZ\x00"
XZ_FOOTER_MAGIC = b"YZ"
XZ_CHECK_CRC64 = 0x04
LZMA2_FILTER_ID = 0x21
CHECK_SIZE = 8 # CRC64 占 8 字节
_XZ_FLAGS = bytes([0x00, XZ_CHECK_CRC64]) # Stream Flags: reserved + check_type
# tar 格式常量
TAR_BLOCK = 512
def round_up_512(n):
"""长度按 tar 块 512 字节向上取整。"""
return (n + TAR_BLOCK - 1) // TAR_BLOCK * TAR_BLOCK
def crc32(b):
return binascii.crc32(b) & 0xFFFFFFFF
# ---------- CRC64 (ECMA-182, reflected) ----------
_CRC64_POLY = 0xC96C5795D7870F42 # reflected ECMA-182
_CRC64_INIT = 0xFFFFFFFFFFFFFFFF
_CRC64_XOROUT = 0xFFFFFFFFFFFFFFFF
_CRC64_NON_REFLECTED = 0x142F0E1EBA9EA3693 # 非反射多项式 (65 bit)
_CRC64_MOD_BIT = 1 << 64 # x^64 对应的 bit
_liblzma = None
try:
_lzma_name = ctypes.util.find_library('lzma')
if _lzma_name:
_liblzma = ctypes.CDLL(_lzma_name)
_liblzma.lzma_crc64.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.c_uint64]
_liblzma.lzma_crc64.restype = ctypes.c_uint64
# 验证：CRC64("123456789") 应为 0x995dc9bbdf1939fa
if _liblzma.lzma_crc64(b'123456789', 9, 0) != 0x995dc9bbdf1939fa:
_liblzma = None
except (OSError, AttributeError):
_liblzma = None
def _make_crc64_table():
table = []
for i in range(256):
crc = i
for _ in range(8):
crc = (crc >> 1) ^ _CRC64_POLY if (crc & 1) else crc >> 1
table.append(crc)
return table
_CRC64_TABLE = _make_crc64_table()
def _crc64_pure(data):
"""纯 Python CRC64（ECMA-182 反射，init/xorout = 0xFFFF...FFFF）。"""
crc = _CRC64_INIT
for byte in data:
crc = (crc >> 8) ^ _CRC64_TABLE[(crc ^ byte) & 0xFF]
return crc ^ _CRC64_XOROUT
def _crc64_lib(data):
"""liblzma CRC64（init=0，内部处理 init/xorout）。"""
return _liblzma.lzma_crc64(data, len(data), 0)
crc64 = _crc64_lib if _liblzma else _crc64_pure
# ---------- GF(2^64) 多项式运算 ----------
# 用于 CRC64 自引用求解。CRC64 的线性贡献可表示为
# bit_reverse(contribution(D)) = D_poly * P (mod G_non)
# 其中 D_poly = bit_reverse(D)，P 是位置决定的多项式。
# 方程 D_poly * (1 ^ P1 ^ P2) = bit_reverse(CRC64(W)) 通过多项式逆元求解。
def _bit_reverse64(v):
r = 0
for _ in range(64):
r = (r << 1) | (v & 1)
v >>= 1
return r
def _poly_mul_mod(a, b):
"""GF(2) 多项式乘法 mod G（非反射）。"""
result = 0
while b:
if b & 1:
result ^= a
b >>= 1
a <<= 1
if a & _CRC64_MOD_BIT:
a ^= _CRC64_NON_REFLECTED
return result
def _poly_pow(base, exp):
"""多项式幂：base^exp mod G。"""
result = 1
while exp:
if exp & 1:
result = _poly_mul_mod(result, base)
base = _poly_mul_mod(base, base)
exp >>= 1
return result
def _poly_minv(p):
"""多项式逆元 p^(-1) mod G，用扩展欧几里得算法。"""
if p == 0:
return 0
old_r, r = p, _CRC64_NON_REFLECTED
old_s, s = 1, 0
while r:
# poly divmod(old_r, r) -> (q, rem)
probe_r = r.bit_length() - 1
rem = old_r
q = 0
while rem.bit_length() - 1 >= probe_r and rem:
shift = rem.bit_length() - 1 - probe_r
q |= 1 << shift
rem ^= r << shift
old_r, r = r, rem
# s = old_s ^ q*s (raw 多项式乘法)
prod = 0
a, b = q, s
while b:
if b & 1:
prod ^= a
b >>= 1
a <<= 1
old_s, s = s, old_s ^ prod
if old_r != 1:
return 0
# old_s mod G
probe_m = _CRC64_NON_REFLECTED.bit_length() - 1
rem = old_s
while rem.bit_length() - 1 >= probe_m and rem:
shift = rem.bit_length() - 1 - probe_m
rem ^= _CRC64_NON_REFLECTED << shift
return rem
def store_hdr(payload_len, first=False):
"""LZMA2 uncompressed chunk 头（3 字节，大端 size-1）。"""
assert 1 <= payload_len <= 65536
return bytes([0x01 if first else 0x02]) + (payload_len - 1).to_bytes(2, "big")
def matches_for(dist, total):
"""dist 固定、总长 total 的 token 序列：首个 match + 后续 rep0。"""
assert total >= 2
toks = []
first = min(MAX_MATCH, total)
if total - first == 1:
first -= 1
toks.append(("m", dist, first))
total -= first
while total > 0:
l = min(MAX_MATCH, total)
if total - l == 1:
l -= 1
toks.append(("r0", l))
total -= l
return toks
def _add_gear(chunks, o, f, q):
"""追加一组标准 gear：store 载荷 x = slip+3，紧跟一个 lzma 把那 x 字节复制一遍。
lzma chunk 解压出的字节数正好等于 x，所以齿轮转完之后 slip (f - (o-q)) 精确
归约为 len(cb)（很小的正数，x 被 CHUNK 截断时则按 CHUNK 缩减）。
_layout_iterate 反复套用把 slip 压到 <= 16，_trim_layouts 用它做相位微调。
返回追加后的 (o, f)。
"""
x = min(f - (o - q) + 3, CHUNK)
chunks.append({"kind": "store", "foff": f, "size": 3 + x,
"ooff": o, "olen": x})
o += x
f += 3 + x
cb, cu = lzma_chunk(matches_for(x, x), o)
chunks.append({"kind": "lzma", "foff": f, "size": len(cb),
"ooff": o, "olen": cu, "bytes": cb})
return o + cu, f + len(cb)
def _forward_vac(F, vac, q):
"""vac_store：把紧随其后的等长字节块前移覆盖自身。"""
fo = vac["ooff"] - q
olen = vac["olen"]
F[fo:fo + olen] = F[fo + olen:fo + 2 * olen]
def dict_prop_for(maxdist):
"""选最小的 LZMA2 dict prop 使字典 >= maxdist。"""
for p in range(41):
if (2 | (p & 1)) << (p // 2 + 11) >= maxdist:
return p, (2 | (p & 1)) << (p // 2 + 11)
raise ValueError("distance too large")
def lzma_chunk(tokens, out_pos):
"""编一个 LZMA chunk（0xC0: state+props reset，无 dict reset，无 end marker）。"""
enc = LzmaEncoder()
enc.pos = out_pos
total = 0
for t in tokens:
if t[0] == "m":
enc.match(t[1], t[2])
total += t[2]
else:
enc.rep_match(t[1])
total += t[1]
data = enc.finish()
assert total - 1 < 65536 and len(data) - 1 < 65536
hdr = bytes([0xC0]) + (total - 1).to_bytes(2, "big") + \
(len(data) - 1).to_bytes(2, "big") + bytes([PROPS_BYTE])
return hdr + data, total
# ---------- tar 原语 ----------
def tar_header(name, size, mtime, mode=0o644, typeflag=b'0', uid=0, gid=0):
"""构建 512 字节 POSIX ustar tar header。
注意：tar 数值字段包含 null 终止符，slice 范围必须精确匹配字段宽度。
mode/uid/gid 各 8 字节（7 octal + NUL），size/mtime 各 12 字节（11 octal + NUL）。
"""
h = bytearray(512)
name_bytes = name.encode('utf-8')[:100]
h[0:len(name_bytes)] = name_bytes
h[100:108] = b'%07o\x00' % mode # 8 bytes (field 100-107)
h[108:116] = b'%07o\x00' % uid # 8 bytes (field 108-115)
h[116:124] = b'%07o\x00' % gid # 8 bytes (field 116-123)
h[124:136] = b'%011o\x00' % size # 12 bytes (field 124-135)
h[136:148] = b'%011o\x00' % mtime # 12 bytes (field 136-147)
h[148:156] = b' ' # checksum placeholder (8 spaces)
h[156:157] = typeflag
h[257:263] = b'ustar\x00'
h[263:265] = b'00'
assert len(h) == 512, "header length %d != 512" % len(h)
# 计算校验和：所有字节之和，checksum 字段视为空格
chksum = sum(h) & 0o7777777
h[148:156] = b'%06o\x00 ' % chksum # 6 octal + NUL + space = 8 bytes
assert len(h) == 512, "header length %d != 512 after checksum" % len(h)
return bytes(h)
# ---------- xz 原语 ----------
def xz_varint(v):
"""xz 变长整数编码：每字节 7 位数据（LSB first），MSB 为续位。"""
b = []
while v >= 0x80:
b.append((v & 0x7F) | 0x80)
v >>= 7
b.append(v & 0x7F)
return bytes(b)
def xz_stream_header():
"""12 字节 xz stream header (CHECK_CRC64)。
Stream Flags = [reserved(0x00), check_type(CRC64=0x04)]。
"""
flags = _XZ_FLAGS
crc = crc32(flags)
return XZ_MAGIC + flags + struct.pack('<I', crc)
def xz_block_header(dict_size):
"""xz block header（单 LZMA2 filter）。
返回 (header_bytes, 实际字典大小)；header 长度即 block header 总长度。
"""
prop, real_dict = dict_prop_for(dict_size)
# filter flags: filter_id(1) + props_size(1) + props(1=prop byte)
filter_flags = bytes([LZMA2_FILTER_ID, 1, prop])
# block header content: flags(1, 0x00=1 filter) + filter_flags
content = bytes([0x00]) + filter_flags
# 填充到 (1 + len(content) + 4) 是 4 的倍数
total = 1 + len(content) + 4
pad = (4 - total % 4) % 4
content += b'\x00' * pad
bh = 1 + len(content) + 4
assert bh % 4 == 0, "bh not 4-aligned: %d" % bh
crc = crc32(bytes([bh // 4 - 1]) + content)
header = bytes([bh // 4 - 1]) + content + struct.pack('<I', crc)
assert len(header) == bh
return header, real_dict
def xz_index(records):
"""xz index。records = [(unpadded_size, uncompressed_size), ...]。"""
data = bytes([0x00]) # index indicator
data += xz_varint(len(records))
for unpadded, uncompressed in records:
data += xz_varint(unpadded)
data += xz_varint(uncompressed)
# 填充到 (len(data) + 4) 是 4 的倍数
pad = (4 - (len(data) + 4) % 4) % 4
data += b'\x00' * pad
return data + struct.pack('<I', crc32(data))
def xz_stream_footer(backward_size):
"""12 字节 xz stream footer (CHECK_CRC64)。
backward_size = (index_size / 4) - 1。
Stream Flags = [reserved(0x00), check_type(CRC64=0x04)]。
CRC32 覆盖 Backward Size + Stream Flags（6 字节，不含 CRC32 和 Footer Magic）。
"""
flags = _XZ_FLAGS
crc_data = struct.pack('<I', backward_size) + flags # 6 bytes, no magic
crc = crc32(crc_data)
return struct.pack('<I', crc) + crc_data + XZ_FOOTER_MAGIC
# ============================================================
# Part 3: 多文件 tar.xz quine 构造器
# ============================================================
class TarXzQuine:
def __init__(self, root, quine_name="quine.tar.xz", seed_name=".this_is_mayx_blog",
src_dir=""):
self.quine_name = quine_name
self.seed_name = seed_name
self.src_dir = src_dir.strip("/")
self.mtime = int(time.time())
# walk 源目录
self.walk_entries = []
self.files = []
if self.src_dir:
parts = self.src_dir.split("/")
for i in range(1, len(parts) + 1):
self.walk_entries.append(("/".join(parts[:i]), True))
for dirpath, dirnames, filenames in os.walk(root):
dirnames.sort()
for dn in dirnames:
rel = os.path.relpath(os.path.join(dirpath, dn), root).replace(os.sep, "/")
self.walk_entries.append((self._prefixed(rel), True))
for fn in sorted(filenames):
p = os.path.join(dirpath, fn)
rel = os.path.relpath(p, root).replace(os.sep, "/")
with open(p, "rb") as fp:
data = fp.read()
self.walk_entries.append((self._prefixed(rel), False))
self.files.append((self._prefixed(rel), data))
# rel -> data 映射，避免下面三处按 rel 回头线性搜索整个 files 列表
self.file_map = dict(self.files)
self.dirs = [r for r, isd in self.walk_entries if isd]
self.content = b"".join(data for _, data in self.files)
def _prefixed(self, rel):
return self.src_dir + "/" + rel if self.src_dir else rel
def tar_entries(self, d_seed, xz_size):
"""计算 tar 归档中每个条目的 (header, data_size, is_dir)。
顺序：seed, walk_entries (interleaved files+dirs), quine。
"""
entries = []
# seed 文件
seed_hdr = tar_header(self.seed_name, d_seed, self.mtime)
entries.append({"name": self.seed_name, "hdr": seed_hdr,
"data_size": d_seed, "is_dir": False})
# walk 顺序的文件和目录
for rel, isd in self.walk_entries:
if isd:
hdr = tar_header(rel, 0, self.mtime, mode=0o755, typeflag=b'5')
entries.append({"name": rel, "hdr": hdr,
"data_size": 0, "is_dir": True})
else:
# file_map 与 walk_entries 在 __init__ 里同步构造，这里必有
data = self.file_map[rel]
entries.append({"name": rel, "hdr": tar_header(rel, len(data), self.mtime),
"data_size": len(data), "is_dir": False})
# quine 文件（放在最后）
qhdr = tar_header(self.quine_name, xz_size, self.mtime)
entries.append({"name": self.quine_name, "hdr": qhdr,
"data_size": xz_size, "is_dir": False})
return entries
# ---------- 结构计算 ----------
def _compute_q(self, d_seed):
"""计算 xz_bytes 在 tar 中的偏移 q。
q = seed + files + dirs 的 tar 占用 + quine tar header。
q 不依赖 xz_size（quine 的 data 在 q 之后）。"""
q = 0
# seed 文件
q += TAR_BLOCK # seed tar header
q += round_up_512(d_seed) # seed data + pad
# walk 顺序的文件和目录
for rel, isd in self.walk_entries:
q += TAR_BLOCK # tar header
if not isd:
q += round_up_512(len(self.file_map[rel]))
# quine tar header
q += TAR_BLOCK
return q
def _trim_layouts(self, base, o_g, f_g, q, w_plant_off, d_seed):
"""枚举 trim 配置，产出一批「相位」不同的候选布局。
trim = [可选的裸 store chunk，载荷长度 x0] + 一组标准 gear
（store 载荷 x = s+3，紧接着一个 lzma 复制这 x 字节）。
裸 store 把 slip 抬高 3、把 f 抬高 3+x0；随后这组 gear 把 slip 重新归一到
len(cb)，jump 再归一到 -3。净效果是 (o_pre, f_pre) 被整体平移
δ ≈ s_g + 6 + len(cb) + 3 + x0，而 δ 决定了自洽方程的模 4 相位。
旧实现 δ 恒为 0，相位一旦不对就永远无解 —— 而 d_seed += 1 根本不改变
布局（q 只在跨 512 字节块时才变），所以那 80 次重试是纯粹的无效空转。
注意：不能只插一个裸的 lzma 复制块。lzma 复制块的输出必须等于
F[o-q : o-q+olen]，而这只能靠前面那个「载荷 = F[o-q : o-q+x] 且 x = s+3」
的 store chunk 把字节搬到正确位置来实现；裸插复制块解压出来和 F 对不上。
返回 [(chunks, o_before_jump, f_before_jump, y, jb), ...]，按文件体积升序。
"""
s_g = f_g - (o_g - q)
out = []
seen = set()
for x0 in range(0, s_g + 4):
chunks = list(base)
o, f = o_g, f_g
if x0:
# 裸 store：载荷 = F[o-q : o-q+x0]，要求 x0 <= s+3 以免读到尚未定稿的字节
chunks.append({"kind": "store", "foff": f, "size": 3 + x0,
"ooff": o, "olen": x0})
o += x0
f += 3 + x0
o, f = _add_gear(chunks, o, f, q) # 一组标准 gear
# jump 不动点：y = s + 3 + len(jb)
s = f - (o - q)
y = max(2, s + 17)
jb = ju = None
for _ in range(64):
try:
jb, ju = lzma_chunk([("m", o - w_plant_off, y)], o)
except AssertionError:
jb = None
break
y2 = s + 3 + len(jb)
if y2 == y:
break
if y2 < 2 or y2 > 273:
jb = None
break
y = y2
if jb is None or ju != y:
continue
if y + w_plant_off - TAR_BLOCK > d_seed:
continue
key = (o + y, f + len(jb))
if key in seen:
continue
seen.add(key)
out.append((chunks, o, f, y, jb))
out.sort(key=lambda t: t[2] + len(t[4]))
return out
def _solve_gadget(self, chunks_pre, o_j, f_j, y, jb, q, bh, FEW,
block_hdr, dict_size, hdrA, hdrB, d_seed, k):
"""在一个给定相位的布局上求解 gadget（T / gb / suffix / tz / k）。
关键简化：tz 只依赖 u = gb + suffix 与 k；T = u + tz + 4 也只依赖 (u, k)；
而 gb 又由 T 唯一确定（len(lzma_chunk(matches_for(T, T))))。
因此只需一维枚举 u 即可遍历全部解，既不漏解也远快于原来的三重暴力。
"""
o_pre = o_j + y
f_pre = f_j + len(jb)
if f_pre != (o_pre - q) - 3: # jump 应把 slip 归一到 -3
return None
chunks = list(chunks_pre)
chunks.append({"kind": "lzma", "foff": f_j, "size": len(jb),
"ooff": o_j, "olen": y, "bytes": jb, "jump": True})
best = None
for u in range(8, 512):
# 512k 必须落在 [f_pre + 2u + 2059, f_pre + 2u + 3595)
lo = f_pre + 2 * u + 2059
hi = f_pre + 2 * u + 3595
for k_cand in range(max(1, (lo - 512) // 512), (hi + 512) // 512 + 2):
num = k_cand * 512 - (o_pre + 2 * u + 8 - q - 1024)
if num % 3 != 0:
continue
tz_cand = num // 3
if tz_cand < 1024:
continue
T = u + tz_cand + 4
if T < 2:
continue
gb, gu = lzma_chunk(matches_for(T, T), o_pre + T)
gb_len = len(gb)
suffix = u - gb_len
if suffix < 1:
continue
# 精确自检（不再用带 9 字节偏差的预筛公式）
f_final = f_pre + 3 + T + gb_len + 3 + tz_cand + 1
n = f_final - (12 + bh)
bp = (4 - (12 + bh + n) % 4) % 4
unpadded = bh + n + CHECK_SIZE
idx = xz_index([(unpadded, q + k_cand * 512 + 1024)])
ix = len(idx)
suffix_actual = bp + CHECK_SIZE + ix + 12
if suffix_actual != suffix:
continue
xz_size = 12 + bh + n + bp + CHECK_SIZE + ix + 12
xz_pad_actual = round_up_512(xz_size)
if xz_pad_actual != k_cand * 512:
continue
total_w_actual = q + xz_pad_actual + 1024
o_final = o_pre + 2 * T + tz_cand
if o_final != total_w_actual:
continue
if best is not None and xz_size >= best["xz_size"]:
continue # 已找到更小的解，跳过
backward_size = ix // 4 - 1
final = list(chunks)
final.append({"kind": "store", "foff": f_pre, "size": 3 + T,
"ooff": o_pre, "olen": T, "vac": True})
final.append({"kind": "lzma", "foff": f_pre + 3 + T,
"size": gb_len, "ooff": o_pre + T, "olen": gu,
"bytes": gb})
final.append({"kind": "store",
"foff": f_pre + 3 + T + gb_len,
"size": 3 + tz_cand, "ooff": o_pre + 2 * T,
"olen": tz_cand, "trailing": True})
final.append({"kind": "term",
"foff": f_pre + 3 + T + gb_len + 3 + tz_cand,
"size": 1, "ooff": o_final, "olen": 0,
"bytes": b"\x00"})
best = {
"q": q, "d_seed": d_seed, "k": k,
"chunks": final, "block_hdr": block_hdr, "bh": bh,
"n": n, "xz_size": xz_size,
"T": T, "suffix": suffix_actual,
"bp": bp, "ix": ix,
"dict_size": dict_size,
"FEW": FEW, "hdrA": hdrA,
"hdrB": hdrB, "jump_y": y, "index": idx,
"stream_hdr": xz_stream_header(),
"footer": xz_stream_footer(backward_size),
"entries": self.tar_entries(d_seed, xz_size),
}
return best
def _layout_iterate(self, d_seed, maxdist_hint):
"""核心布局。
阶段 1a：C1 / repro / gears（只依赖 q，所有 trim 配置共用，只算一次）
阶段 1b：插入 trim 微调块，平移 (o_pre, f_pre) 改变模 4 相位
阶段 2 ：按 u = gb + suffix 一维搜索自洽解
"""
block_hdr, dict_size = xz_block_header(maxdist_hint)
bh = len(block_hdr)
FEW = 12 + bh + 3
w_hdrA_off = TAR_BLOCK + 12 + bh
w_hdrB_off = TAR_BLOCK + 12 + bh + 3
w_plant_off = TAR_BLOCK + 12 + bh + 6
q = self._compute_q(d_seed)
# ---------- 阶段 1a：C1 + repro + gears ----------
base = []
o, f = 0, 12 + bh
Ls = []
rem = q + FEW
while rem > CHUNK:
Ls.append(CHUNK)
rem -= CHUNK
Ls.append(rem)
assert Ls[-1] >= 2, "末 C1 chunk 载荷 %d 太小" % Ls[-1]
k = len(Ls)
hdrA = store_hdr(CHUNK, first=False)
hdrB = store_hdr(Ls[-1], first=False)
S = 0
for j, L in enumerate(Ls):
base.append({"kind": "store", "foff": f, "size": 3 + L,
"ooff": o, "olen": L, "first": j == 0,
"c1": True, "S": S})
o += L
f += 3 + L
S += L
assert o == q + FEW
S = 0
for j, L in enumerate(Ls):
if j > 0:
srcpos = w_hdrA_off if (j < k - 1 or L == CHUNK) else w_hdrB_off
dist_h = (q + FEW + S + 3 * (j - 1)) - srcpos
hb, hu = lzma_chunk([("m", dist_h, 3)], o)
base.append({"kind": "lzma", "foff": f, "size": len(hb),
"ooff": o, "olen": hu, "bytes": hb})
o += hu
f += len(hb)
cb, cu = lzma_chunk(matches_for(q + FEW + 3 * j, L), o)
base.append({"kind": "lzma", "foff": f, "size": len(cb),
"ooff": o, "olen": cu, "bytes": cb})
o += cu
f += len(cb)
S += L
# gears：把 slip 压到 <= 16
while f - (o - q) > 16:
o, f = _add_gear(base, o, f, q)
o_g, f_g = o, f
# ---------- 阶段 1b + 阶段 2 ----------
# 相位按文件体积升序枚举；第一个成功的相位通常已经足够好，但不同相位的
# (u, tz) 不同，xz_size 仍可能相差几百字节，所以再多看几个取最小的。
tried = 0
best = None
since_hit = 0
for chunks_pre, o_j, f_j, y, jb in self._trim_layouts(
base, o_g, f_g, q, w_plant_off, d_seed):
lay = self._solve_gadget(chunks_pre, o_j, f_j, y, jb, q, bh, FEW,
block_hdr, dict_size, hdrA, hdrB,
d_seed, k)
tried += 1
if lay is not None and (best is None
or lay["xz_size"] < best["xz_size"]):
best = lay
if best is not None:
since_hit += 1
if since_hit >= 10:
break
elif tried >= 80: # 一直无解，放弃后续相位
break
if best is not None:
return best
raise RuntimeError("layout 不收敛: 无自洽解 (q=%d, f_pre=%d, o_pre=%d)"
% (q, f_g, o_g))
# ---------- 装配 ----------
def assemble(self, lay):
"""构建 xz 文件 F 和 seed。"""
F = bytearray(lay["xz_size"])
q = lay["q"]
d_seed = lay["d_seed"]
bh = lay["bh"]
FEW = lay["FEW"]
# bytearray 已全零初始化：block padding、CRC64 占位、trailing-zero 载荷
# 都不必再显式写 0（后面几处赋值只会写到各自独立的区间，不会交叉覆盖）。
# pass A: 写 xz stream header + block header + 所有 chunk bytes + suffix
F[0:12] = lay["stream_hdr"]
F[12:12+bh] = lay["block_hdr"]
jump_c = None
for c in lay["chunks"]:
if c["kind"] == "store":
F[c["foff"]:c["foff"] + 3] = store_hdr(c["olen"],
first=c.get("first", False))
else:
F[c["foff"]:c["foff"] + c["size"]] = c["bytes"]
if c.get("jump"):
jump_c = c
# block padding + CRC64(pass D 填) + index + footer
bp_off = 12 + bh + lay["n"]
ix_off = bp_off + lay["bp"] + CHECK_SIZE
F[ix_off:ix_off + lay["ix"]] = lay["index"]
sf_off = ix_off + lay["ix"]
F[sf_off:sf_off + 12] = lay["footer"]
assert jump_c is not None
# pass B: 构建 seed
y = lay["jump_y"]
plant = bytes(F[jump_c["ooff"] - q: jump_c["ooff"] - q + y])
seed = bytearray(d_seed)
seed[0:12] = lay["stream_hdr"]
seed[12:12+bh] = lay["block_hdr"]
seed[12+bh:12+bh+3] = lay["hdrA"]
seed[12+bh+3:12+bh+6] = lay["hdrB"]
seed[12+bh+6:12+bh+6+y] = plant
pad = (SEED_NOTE * (d_seed // len(SEED_NOTE) + 2))[:d_seed - 12 - bh - 6 - y]
seed[12+bh+6+y:] = pad
# 构建 tar 前缀 T（含 seed data）
T = self._build_tar_prefix_with_seed(lay, seed)
F_prefix = bytes(F[0:FEW])
P = T + F_prefix
assert len(P) == q + FEW, "P 长度 %d != q+FEW %d" % (len(P), q + FEW)
# pass C: 填充 store payloads
for c in lay["chunks"]:
if c["kind"] != "store":
continue
ooff, olen = c["ooff"], c["olen"]
if c.get("c1"):
F[c["foff"] + 3:c["foff"] + 3 + olen] = P[c["S"]:c["S"] + olen]
elif c.get("trailing"):
# 载荷是尾部零：F 全零初始化，且唯一会写到 chunk 之外的 vac_store
# 目标是 [f_pre+3, f_pre+3+T)（f_pre == (o_pre-q)-3 已由 _solve_gadget
# 断言），与本区块不重叠，因此无需写入。
continue
elif c.get("vac"):
_forward_vac(F, c, q)
else:
# 其余 store 的载荷 = 输出流自身对应位置的字节（自引用复制）
fo = ooff - q
F[c["foff"] + 3:c["foff"] + 3 + olen] = F[fo:fo + olen]
# pass D: 求解 CRC64 自引用
self._solve_crc64(F, lay, T)
return bytes(F), bytes(seed)
def _build_tar_prefix_with_seed(self, lay, seed):
"""构建 tar 前缀 T（含 seed data）。"""
entries = lay["entries"]
T = bytearray()
for e in entries[:-1]:
T += e["hdr"]
if not e["is_dir"] and e["data_size"] > 0:
data = seed if e["name"] == self.seed_name else self.file_map[e["name"]]
T += data
T += b'\x00' * (round_up_512(e["data_size"]) - e["data_size"])
T += entries[-1]["hdr"] # quine tar header
return bytes(T)
def _solve_crc64(self, F, lay, T_prefix):
"""求解 CRC64 自引用。
CRC64 值 D = CRC64(W)，W = T + F + 尾部零。
D 在 W 中出现两次：F 的 CRC64 字段 + vac_store 副本。
解法（多项式逆元）：
CRC64 的线性贡献可表示为 GF(2^64) 多项式乘法：
bit_reverse(contribution(D)) = D_poly * P (mod G)
其中 D_poly = bit_reverse(D)，P = x^(64+trailing) mod G。
方程 D_poly * (1 ^ P1 ^ P2) = bit_reverse(CRC64(W)) (mod G)
用扩展欧几里得求 (1 ^ P1 ^ P2) 的逆元，一步求解。
"""
q = lay["q"]
bh = lay["bh"]
n = lay["n"]
bp = lay["bp"]
xz_size = lay["xz_size"]
T = lay["T"]
# CRC64 位置 1：在 F 中（block padding 后）
crc_f_off = 12 + bh + n + bp
crc_w1 = q + crc_f_off
# CRC64 位置 2：vac_store 副本
crc_f_copy = crc_f_off - T
crc_w2 = q + crc_f_copy
# 构造 W = T + F + 尾部零（先把两处 CRC64 置 0）
xz_pad = round_up_512(xz_size)
tz = xz_pad - xz_size + 1024
W = bytearray(T_prefix) + F + b'\x00' * tz
W[crc_w1:crc_w1 + CHECK_SIZE] = b'\x00' * CHECK_SIZE
W[crc_w2:crc_w2 + CHECK_SIZE] = b'\x00' * CHECK_SIZE
# CRC64(W) 的多项式表示
crc_poly = _bit_reverse64(crc64(bytes(W)))
# P1 = x^(64 + 位置1之后的bit数) mod G
trailing1 = (len(W) - crc_w1 - 8) * 8
P1 = _poly_pow(2, 64 + trailing1)
# P2 = x^(64 + 位置2之后的bit数) mod G
trailing2 = (len(W) - crc_w2 - 8) * 8
P2 = _poly_pow(2, 64 + trailing2)
# D_poly = crc_poly * inv(1 ^ P1 ^ P2)
coeff = 1 ^ P1 ^ P2
inv = _poly_minv(coeff)
if inv == 0:
raise RuntimeError("CRC64 多项式无逆元")
D_poly = _poly_mul_mod(crc_poly, inv)
D = _bit_reverse64(D_poly)
# 写入 D 到 F 的 CRC64 位置
struct.pack_into('<Q', F, crc_f_off, D)
# 重新执行 vac_store 副本
_forward_vac(F, next(c for c in lay["chunks"] if c.get("vac")), q)
# 验证
W2 = bytearray(T_prefix) + F + b'\x00' * tz
assert crc64(bytes(W2)) == D, "CRC64 求解验证失败"
def build(self, maxdist_hint=None):
if maxdist_hint is None:
maxdist_hint = 2 * (len(self.content) + 128) + (1 << 21)
last_err = None
# d_seed 在同一 512 字节块内不改变 q（seed 数据按 512 对齐），所以旧的
# "d_seed += 1" 重试完全不会改变布局，只是空转 80 次。真正需要 d_seed
# 变化的情况只有两种：seed 装不下 jump 种植串，或需要跨块改变 q 作为兜底。
for d_seed in (64, 96, 128, 160, 192, 256, 320, 384, 448,
576, 704, 832, 960, 1088, 1600, 2112, 2624, 3648):
try:
lay = self._layout_iterate(d_seed, maxdist_hint)
F, seed = self.assemble(lay)
return F, lay, seed
except (RuntimeError, AssertionError, ValueError) as e:
last_err = e
raise RuntimeError("build 多次失败: %s" % last_err)
# ============================================================
# Part 4: 验证 + main
# ============================================================
def verify(F, lay, files, quine_name):
"""验证：解压 xz → 解析 tar → 检查文件 + quine 自复制。"""
# 1. 解压 xz
try:
dec = lzma.LZMADecompressor(format=lzma.FORMAT_XZ)
tar_data = dec.decompress(F)
except Exception as e:
print("[verify] xz 解压失败: %s" % e)
return False
# 2. 检查 quine 自复制
if tar_data[lay["q"]:lay["q"] + lay["xz_size"]] != F:
print("[verify] quine 自复制不匹配")
a = tar_data[lay["q"]:lay["q"] + lay["xz_size"]]
for i in range(min(len(a), len(F))):
if a[i] != F[i]:
print("[verify] 首个不匹配在偏移 %d: tar=%02x vs xz=%02x" % (i, a[i], F[i]))
break
return False
# 3. 解析 tar
try:
tf = tarfile.open(fileobj=io.BytesIO(tar_data))
members = tf.getmembers()
except Exception as e:
print("[verify] tar 解析失败: %s" % e)
return False
print("[verify] tar 条目: %s" % [m.name for m in members])
# 4. 检查源文件
ok = True
for rel, data in files:
try:
f = tf.extractfile(rel)
if f is None:
print("[verify] 文件 %s 是目录或不存在" % rel)
ok = False
continue
extracted = f.read()
if extracted != data:
print("[verify] 文件 %s 内容不匹配" % rel)
ok = False
except KeyError:
print("[verify] 文件 %s 不在 tar 中" % rel)
ok = False
# 5. 检查 quine 文件
try:
qf = tf.extractfile(quine_name)
if qf:
qdata = qf.read()
if qdata != F:
print("[verify] quine 文件内容不匹配 (%d vs %d)" % (len(qdata), len(F)))
ok = False
else:
print("[verify] quine 文件匹配!")
except KeyError:
print("[verify] quine 文件不在 tar 中")
ok = False
print("[verify] 结果: %s" % ("通过" if ok else "失败"))
return ok
def main():
ap = argparse.ArgumentParser(description="多文件 tar.xz quine 生成器")
ap.add_argument("srcdir")
ap.add_argument("output")
ap.add_argument("--quine-name", default=None,
help="归档内 quine 文件名（默认 = output 的文件名）")
ap.add_argument("--seed-name", default=".this_is_mayx_blog")
ap.add_argument("--src-dir", default="",
help="源文件在归档内的前缀目录")
ap.add_argument("--quine-dir", default="",
help="quine 在归档内所在目录")
args = ap.parse_args()
quine_name = args.quine_name or os.path.basename(args.output)
if args.quine_dir:
quine_dir = args.quine_dir.strip("/")
quine_name = quine_dir + "/" + quine_name
bq = TarXzQuine(args.srcdir, quine_name=quine_name, seed_name=args.seed_name,
src_dir=args.src_dir)
print("[blogquine-tarxz] %d 个文件, %d 个目录, 内容 %d 字节"
% (len(bq.files), len(bq.dirs), len(bq.content)))
F, lay, seed = bq.build()
with open(args.output, "wb") as fp:
fp.write(F)
print("[blogquine-tarxz] written %s: %d 字节 (q=%d, k=%d chunks, n=%d, xz_size=%d)"
% (args.output, len(F), lay["q"], lay["k"], lay["n"], lay["xz_size"]))
print("[blogquine-tarxz] T=%d, suffix=%d (bp=%d, ix=%d), dict_size=%d"
% (lay["T"], lay["suffix"], lay["bp"], lay["ix"], lay["dict_size"]))
ok = verify(F, lay, bq.files, quine_name)
sys.exit(0 if ok else 1)
if __name__ == "__main__":
main()
不过对我来说，相比于TXZ格式我还是更喜欢7z一点，一是因为LZMA2本来就是7-Zip的作者发明的，XZ感觉像是摘桃子的，二是TXZ这个名字听起来有点怪，感觉不像压缩包，三是XZ Utils出过后门，尽管整个Linux社区都在使用XZ，但是我还是稍微有点偏见，所以这份TXZ的生成器我就粘贴出来给需要的人吧，我博客用7z格式就好了。
以前总是有人说AI没有创新能力，只是对曾经在网络上出现的东西进行重组，但人何尝不是这样呢？像这次制作的7z/TXZ Quine生成器在整个网络上没有任何公开信息，当然我也知道这并不是理论上的创新，但谁说组合创新不是创新呢？再看看最近OpenAI又解决了一大堆数学难题，完全可以相信AI是真的拥有智能，所以我相信总有一天AI将能完成人类能做的所有事情，人类将不再需要额外的思考，只需要做自己想做的事情吧。
2026-09-01 00:00:00
AI is Everything!
在两个月前，我体验了现在的Vibe Coding，不过那时候我其实还是没有完全信任AI，认为它的能力可能还不够强，大多数内容依然需要人来做。
但是自从上次做了终端版博客之后，我意识到，它的能力也许比我想象的更强，可能真的能做到超越99%的开发者了，因此在最近一段时间，我一直使用AI做我想做的各种东西，结果就是一次又一次地让我感到难以置信，现在的AI真的实在是太强了😨。而且我做的这一切并不是用最先进的那批AI，仅仅是国内被大家不怎么看好的免费模型——腾讯混元Hy3，那么我很难想象，如果我能用像GPT 5.6 Sol或者Fable 5那样当下最先进的AI，我做的这些东西得有多完美啊……
在上次做完终端版博客之后，我就在想，用Linux终端来呈现博客的形式真的是个好选择吗？毕竟Linux在浏览器上并不是原生的，尽管用了WASM Linux可以做到WASM原生运行Linux，但这还是额外添加的东西。于是我就想，JavaScript应该也能作为终端吧？就像Node.js一样，REPL也可以作为终端使用，正好那段时间DeepSeek-V4-Flash正式版出了，就让它直接写了一个Demo出来试试看，可以点开这里体验。
写出来效果还挺不错的，和之前做的终端版博客的主要区别就是我写的是函数，而不是命令，不管怎么说也算是“真的”终端吧，唯一体验不太好的可能就是函数需要带个括号才能运行，稍微有点麻烦。
但后来我想了想，明明开发者工具中就有个控制台，根本没必要搞什么REPL终端啊，直接把要用的功能提出来作为函数，在控制台中运行效果比那个终端的效果要好得多啊，而且这样就可以直接集成到我的博客中，不需要单独搞个页面，感觉体验更好一些。
既然有了函数，我觉得只给人类用就太浪费了，前段时间看到谷歌搞了个WebMCP，那我完全可以把这些函数做成WebMCP让AI来调用啊，相比于其他乱七八糟给AI的协议，至少WebMCP是有谷歌和微软撑腰的，所以加一个也不错。
最终我把WebMCP的文档以及之前实现的REPL代码直接发给AI，让它帮我实现对应的函数以及应用WebMCP API，没几分钟它就把所有功能实现好了，最终代码就是：blog-console.js。在开发的过程中，AI严格按照文档中的要求实现了，结果我在实际测试的时候，发现Chrome在实现registerTool方法的时候和文档的实现有差异，文档说这个方法会返回Promise，但Chrome实际并没有返回，所以AI调用.then()时直接报错了😓……没办法就只能手动给它套一层Promise，才算是能正常工作。毕竟这还是一个没有正式上线的API，出点问题倒也能理解吧。
虽然东西已经实现了，不过到底怎么才能让其他人知道我用了WebMCP的技术呢？我发现网上有一个WebMCP Directory专门收录使用了WebMCP技术的网站，于是我就随手提交了一下，现在在这里就能看到我博客的WebMCP信息了。
最近一段时间我在玩的一个使用Unity3D引擎制作的立直麻将游戏发了篇公告，说是要关服了，还给全服所有玩家送了一堆抽卡资源。我听到这个消息之后感觉还是挺遗憾的，那个游戏总体来说做得还算不错吧，而且画风也符合我的口味，角色模型有点米哈游的那种感觉。我曾经遇到过几次游戏关服的事件，但是每一次我都无能为力，只能看着游戏消失然后逐渐被遗忘。然而现在我有了AI，也许一切会变得不一样吧？这次我要使用AI，让即将消失的游戏不再消失。
最开始我其实想得很简单，游戏登录之后其实没有和服务器有太多的通信，像账号中有的角色以及各种材料等信息在登录的时候就已经发送了，直到进入对局之前客户端都没有和服务器有什么有效的沟通，而我主要需要的也就是能登进去看看角色，听听语音就好了。那么接下来我只需要让AI把整个游戏资源下载下来，然后抓个包，让它给客户端重放就好了，也不需要什么逆向操作。
AI分析了一下之后发现它用的是WebSocket+Protobuf协议，正常来说如果要使用Protobuf协议得要有.proto接口定义文件才可以，不过既然我的要求只是重放，它就没去考虑逆向了，只是简单分析了一下编码规则，就用Python把服务端给写出来了。
不过如果这个快照必须本地启动一个服务器，玩起来就有点麻烦了，这样搞得我估计就没心思玩它了，但是自从我做了终端版博客之后，我觉得浏览器可能没有做不到的事情，既然Python能运行服务器，那JavaScript一样可以，于是我就要求AI让游戏和服务端一起运行在浏览器中。
结果它还真折腾出来了，在JavaScript中似乎一切都可以被覆盖，于是AI就替换了浏览器的WebSocket对象，让客户端以为它连接到了真实的服务器，然后再按照之前录制的包发送就好了，折腾了一段时间之后游戏真的能正常启动了，真的是太离谱了😨。
其实这基本上已经满足我的要求了，但我还是有点好奇，这个客户端里真的没有.proto接口定义文件吗？正常来说既然客户端知道怎么和服务端通信，那么这个文件应该是必须要有的吧？反正目的已经达到了，接下来就让AI瞎折腾看看。
在我要求完之后它又折腾了好长时间，结果.proto也被它给搞出来了😨，也许对于会逆向Unity3D游戏的人来说这不算什么，但这已经让我很震惊了……既然有了这些，也许完成一个完整的服务端将不再是难事。
于是我让它录制了几把我实际对局的包，并理解整个对局中数据包的时序，开始考虑如何做出完整的麻将对局。首先作为离线运行的游戏来说，和人对局肯定是做不出来了，那就得想办法搓个人机出来。我最先想到的是著名的麻将AI Mortal，之前用它在雀魂上打了几把之后结果号没了😂，不过我确实觉得和人打麻将没什么意思，感觉麻将如果全按最优解打的话结果就全是靠运气……不过后来我想了一下还是不用Mortal了，首先这个模型不小，整个模型要好几百MiB，而且三麻和四麻的模型还不一样，还有三麻的模型还是收费的……另外真和Mortal打的话感觉难度也确实有点高，除了运气因素外，至少人机犯错的概率比人要低不少，那打起来估计会很难受。所以我就对AI说，做一个降低向听数为主的全攻型人机，其他细节让它自己考虑，另外我想到上次做的那个麻将Wordle它自己写的役种分析实在是有点烂，所以这次就告诉它必须要使用现成的Riichi库，免得出乱七八糟的问题。
在连续跑了好几个小时之后AI成功把完整的对局引擎给搓出来了，真的是相当了不起，要知道这个东西正常来说人开发怎么说也得好几周吧？它就几个小时开发出来了，还理解了数据包的时序，用之前分析的.proto文件成功接入到了游戏中，实在是太强了……当然因为我没有用大量真实牌谱去测试，有BUG也是正常的，比如目前的引擎还存在立直之后完全没法开杠这样的边缘问题，就只能慢慢修了，不过这都是小问题，它能做到目前的水平已经不是一般人能做到的了。
最终做出来的项目可以在这里看到，还可以通过GitHub Pages上的部署直接游玩。
在AI做完整个服务端之后，我闲得没事瞎改对战人机的角色和皮肤，因为游戏中的皮肤ID有一定的规律，所以可以根据角色ID来猜皮肤ID。在这个时候，我发现其中有一个角色在游戏中没有对应的泳装皮肤，但是我把人机的角色按照规律配置推测的泳装皮肤ID是能显示出来的……那也就是说这个游戏可能在配置中隐藏了这个皮肤，但是并没有删掉美术资源，所以接下来我就想着能不能让AI把这个隐藏的皮肤显示出来，这样就能在游戏中直接使用了。
所以我根据自己推测的信息，告诉AI：在这个游戏中有一个皮肤编号3708002，头像编号1018的资源被隐藏了，其中关于这些资源的配置应该在commonconfigs中，帮我解包并分析，考虑如何将这个皮肤和头像恢复显示，然后重新打包资源，使其可以正常加载。就这么个需求，它从Bundle的解包，到分析FlatBuffers，然后又处理XAsset的校验……最终花了3个小时就给我做出来了，而且我试了一下，非常完美，简直就像这个皮肤本来就存在一样，不过原本这就是被官方隐藏的嘛，解除隐藏自然就和真的存在一样啊😂。
之前我做完终端版博客之后就在想，让这个完整的Linux只用来展示博客似乎有点浪费，感觉它还能做更多事，但对我来说编译在这个系统中能运行的二进制程序还是有一点困难，虽然我有用AI，但是我的Agent是运行在macOS上的，所以编译Linux二进制程序有点困难……不过单是BusyBox其实就能做相当多的事情了，毕竟我当年可是在手机上用ash做出了论坛系统Mabbs，那它能做的事情应该还有很多。
最近我在和群友聊天的时候发现有人用C#做了个静态网站生成器，也是用AI做的，看起来挺有意思。其实SSG的原理还是挺简单的，说到底就是些文字转换罢了。于是我就想，要不然我就让AI搓个能在WASM Linux中运行的SSG出来吧？
当然直接搓我还是会有一点怀疑AI的能力，所以又找了点素材，之前在体验Tilde社区的时候发现了bashblog，其实这就已经是一个现成的SSG了，只不过它用的是Bash语法，得让AI转换成POSIX风格的语法。另外它还依赖使用Perl语言编写的Markdown库，我这个系统里可没有Perl啊，也得想办法找个可以替代的东西。
我想了一下，在BusyBox中最适合处理文本的大概就是awk了，所以我首先应该找找有没有用awk编写的Markdown转换程序。结果搜了一下还真被我找到了，有一个叫做d.awk的库，里面包含一个完全用awk写的Markdown转换程序，那么接下来我只需要把这两个项目丢给AI，让它以bashblog为基础，以d.awk为引擎制作一个兼容性极强的POSIX SSG吧。
这种问题果然还是太简单了，它只用了一两个小时的时间，就真的做出来了，最终它把这个项目起名叫做possg，并且我把这个程序粘贴到了我的终端版博客中运行了一下，效果也是非常完美……如果哪一天我抛弃了Jekyll，也许就会用这个possg吧😂。
最近一段时间我在我的终端版博客又有一些新的想法，于是就对着AI反复许愿，结果AI还真的能让愿望成真。以下就是从上次做完之后又新增的一些功能：
最初博客在终端中显示文章时是直接打印的Markdown原文，感觉不太好看，所以我就想着如果能加个高亮功能体验应该会更好吧？最开始我要求AI做这个功能的时候它选择使用Shell+sed来实现，那时候我还没有做possg，没考虑用awk来做，所以性能非常差。因为Shell在处理这种任务的时候要不停地调用sed，这个系统开进程的开销又很大，所以处理一篇文章得好几秒，后来才考虑用awk来做，结果性能确实好了不少，最终的代码可以在这里查看。
前段时间我看到作者合并了一个关于virtiofs的PR，当时我看到这个PR的时候就想，记得Chrome浏览器有一个showDirectoryPicker API，可以允许网页对指定的文件夹进行读写。现在有了这个驱动的话，让AI做个将本地文件夹挂载到Linux的功能应该是个很简单的事情，于是我直接就闭着眼睛让AI做了。
做完之后虽然能用，但是我又回想起上次让AI搓的通过Service Worker访问系统内网络服务的功能，结果人家原作者已经做了一样的功能，我还给作者发了Issue推荐😅，所以这次我得吸取教训看看作者是不是已经做了这个功能，毕竟作者实现virtiofs的时候估计就有这个打算吧？结果看了一眼还真做了😆，我闭着眼睛让这个垃圾AI去做看来确实有点着急了，虽然我不知道原来我的AI做得到底怎么样，但是我对它还是没什么信心，也许作者这个代码也是AI写的，但很明显他的AI比我的AI水平强得多，所以让它参考作者做的这个版本重新写了一版。总之最终都能用，写出来的代码就是：hostfs.js，再配合上mount_host脚本就可以实现挂载本地文件夹的功能了。
我在制作网络模块的时候一直在想，有没有办法能在不依靠服务器的情况下让两个这样的Linux系统互联呢？正常来说要想让两个浏览器内的服务互相访问就必须有一个中转服务器，因为浏览器肯定是不允许网页监听端口，而且就算能监听，我的这个网页还是放在GitHub Pages上，使用了HTTPS，也没办法访问HTTP资源，搞证书也是个麻烦事。
但其实浏览器是有P2P功能的，那就是WebRTC，只是我不太确定这个技术能不能做到无服务器，于是我问了问AI，这个功能能不能不要服务器，结果AI说可以，只要两个机器在同一个局域网，就可以通过手动交换SDP信令来实现P2P连接，而且考虑到我基本上完全没有锥形NAT网络可以用，所以就不搞STUN服务器了，反正也连不上。那既然理论上可以的话，就让AI实现吧。
最终花了一点时间AI就写出来了，使用webrtc-p2p.js配合connector就可以实现同一个局域网内两台电脑上的浏览器中的Linux互联了。
这个功能做出来之后直接就有了无限的想象空间，比如想用A电脑的浏览器控制B电脑浏览器中的Linux，就可以在B电脑中执行tcpsvd -v 0.0.0.0 4444 /usr/bin/script -qc 'stty -echo; /bin/sh -i' /dev/null，然后在A电脑用nc 10.0.2.x 4444去连接就可以了，效果就像SSH一样，而且配合之前做的挂载本地文件夹功能，甚至可以直接操作B电脑中的文件；另外之前做的Service Worker访问系统内网络服务也能用得上，假如B电脑有一个网页想要分享给A电脑，可以在B电脑上用mount_host把本地文件夹挂载到Linux的/mnt/host目录中，然后在那个目录执行httpd -p 8080，A电脑可以执行tcpsvd -v 0.0.0.0 8080 /usr/bin/nc 10.0.2.x 8080转发，接下来打开http/8080/就可以访问到B电脑上的网页了。
总之只要有了网络，这个系统的功能就能被无限扩大了，只是……说是这么说，其实还是依赖浏览器的功能，所以这些功能就算用JavaScript直接做也完全没问题，但毕竟是作为Linux终端，有了这些能力之后就显得更完整了。
最终我仔细确认了原作者确实没有做这个功能，所以就发了个Issue看看他有没有兴趣把这个功能加到他的库中，如果被接受了也算是做出贡献了吧。
虽然之前已经加了基于浏览器Fetch API的网关，但这个网关的局限性还是挺大的，基本上也就只能请求网站了。那其他能在浏览器上运行系统的项目是怎么做的呢？看了一下像v86用的是一个基于L2 Over WS的公共网关relay.widgetry.org，而WebVM用的则是Tailscale。这两种我都不喜欢，主要是这些东西基本上专门服务于这类在浏览器中运行的系统，普通人并不会特意去搭建，而我既不想用某个特定的服务，也不想自己搭……那么有什么样的网络服务既随处可见，又能在浏览器中使用呢？这时候我就想到了VLESS协议，它天生就适合运行在WebSocket中，而且搭建起来极其简单，比如有人用Cloudflare Workers做了一个叫做edgetunnel的脚本，可以轻松将Cloudflare作为代理使用，因此我决定接下来让AI实现一个基于VLESS+WSS的网关，这样除了请求网站之外还能做点别的事情，比如用telnet towel.blinkenlights.nl命令看ASCII版的《星球大战》😆。
只不过相比于其他项目用的那些L2的服务来说，VLESS承载的是L4的协议，所以我还得让AI实现TCP/IP协议栈……其实我不太想这样做的，因为AI做这种事情容易出错，但是它看了一下我之前用的tcpip.js库说是里面的API不满足要求……那没办法，AI既然自己想搓就让它搓吧。不过VLESS协议这个我觉得还是得给它个文档，免得它开始瞎编。
不管怎么说它还是搓出来了，虽然质量我不是特别满意……但毕竟手头用的AI不是最先进的AI，就先这样吧～最终的代码就是：vless-gateway.js，用法也很简单，首先要有一个证书合法的VLESS+WSS节点，然后在我的博客终端中按F12打开控制台，输入：switchToVless("wss://domain:port/path", "UUID")就可以将原来基于Fetch的网络切换到VLESS节点的网络了，考虑到避免被滥用的情况，我就不内置节点了，如果有人有兴趣可以用自己的节点进行连接。
做到这里，可以感受到最近的AI所做的一切真的是令人难以置信，它不仅知道几乎所有知识，还有不知疲倦的思考能力，以及足够的动手能力，而这也仅仅建立在一个很一般的免费模型上。再回想起前段时间用的Agent，做的什么订阅源预览，简直就太简单了😂。目前据我所知其他人使用更先进的AI已经可以解决人类没能解决的数学难题、自主设计处理器指令集架构在上面运行Doom、从零开发一个Classic Mac OS、制作直接在UEFI中运行的软件、以及完全逆向Unity3D游戏，把几款游戏混合起来做成一个新游戏……（至于网上总传的什么DeepSeek的所谓的灰测模型搞出来的各种Demo就算了，根本比不了）过去各种难以想象的东西都被轻松做出来了，而这一切就发生在最近几个月，我已经很难想象未来AI能做到什么样的地步了，也许AI真的会带领人类走向无法想象的高度吧～
2026-08-01 00:00:00
在浏览器上用真的Linux访问博客会是什么样的感觉？
在去年的时候我为了想在博客上加个终端，写过一篇关于在浏览器中运行Linux的文章。不过那时候只是调查了有哪些可能的方法，并没有真的做出来，主要是自己做感觉太麻烦了😂。
但从上次进行了Vibe Coding之后，我意识到这可能不再是什么麻烦事，也许只要和AI聊聊天就能轻松做出来了，于是接着之前的方案，我开始着手完成我的想法。
在之前的文章里，我其实更倾向于WASM Linux的方案，尤其是Thomas Stokes制作的WASM Linux，因为它相比于虚拟机更轻量，效率更高，而且还能完美地在Safari中运行（毕竟我用的是macOS😂），因此接下来我计划用它为基础制作我的终端版博客。
不过之前我也说过，目前这个WASM Linux还不能和外界通信，不能和外界通信就不能读取我的文章。于是我打算先让AI解决这个问题。但如果我直接给它说想要和外界通信也许它会给我整一堆奇怪的实现，所以我得自己先想一个实现起来最简单的方案。之前我记得Fabrice Bellard大神写的JSLinux中有一个剪贴板设备文件可以和外部通信，那么干脆让AI写一个能在浏览器中执行JS的设备文件吧，这样做出来只需要写简单的Shell脚本就可以和浏览器交互了，也不需要考虑如何编译WASM程序，于是就这样我开始让它实现这个功能。
最终经过几次调试和修改之后AI真的把这个设备文件给做出来了，不过刚做出来的时候有些问题。因为在这个Linux中的进程都是在Web Worker中运行，执行的JS也会在Web Worker中运行。但是Web Worker的局限性很大，并不能和当前网页进行交互，不过这种问题也都是小事，发现了之后只要给AI一反馈也是轻轻松松就改好了……最终实现的效果我感觉相当不错，既然如此，那我就把AI做的这个驱动给上游PR一下吧，也算做个贡献。不过最终原作者不太喜欢这种实现方式，他好像更喜欢标准化的方式，说是已经做了一个VsockDevice，可以直接使用Linux自带的驱动和外界通信，所以把我的PR拒了（看来AI Slop还是入不了大佬的法眼😂）。 不过我问了一下AI，它说用这个设备的话可能需要 虽然VsockDevice确实是不太好操作，但是大佬随后又新增了一个ethernetDevice可以用来上网，考虑到维护自己的分支实在是太麻烦了，所以我让AI写了一个虚拟的HTTP服务，使其能够执行JS，顺便还增加了连接互联网的功能。这样我就可以直接使用上游的代码，不需要再维护之前写的驱动了（其实以太网比Vsock复杂多了，只是因为可以用wget之类的客户端请求所以才简单😋）。 ioctl()系统调用，就没办法直接用Shell操作了，只能用别的语言写的程序才能用它通信，所以我只好自己用这种不标准的方式了。
在做好与外界通信的通道后我就开始让AI写能和我博客交互的脚本，不过在这个过程中也遇到了不少问题，最开始的时候发现这个Shell的管道有些问题，执行的时候会忽略管道前的命令，后面的命令就没法执行了……感觉有点奇怪，所以我要求它在实现交互脚本时不要使用管道符。除此之外我还发现反引号和$()这些能把执行结果插入字符串的语法也不能用……总之就是局限相当多，但它还是克服了这些困难最终把脚本给写出来了，最终实现了列出文章、查看文章、打开文章和查看评论等功能，效果还是相当不错。 唯一的缺陷就是它不能进行异步操作，只能同步请求，所以在进行请求的时候会把浏览器卡住，体验稍有打折。 后来用了网络版的JS执行服务后就可以异步了，避免出现把浏览器卡住的情况。
不过这个系统如果管道符不能用的话效果会差一大截，这可是Shell的灵魂啊，所以我提交了一个Issue，希望作者能修一下。结果作者真的强得离谱，提出问题之后第二天就修复了，实在是太强了😨。但修好这个问题之后作者又整了一堆别的花活，把内存限制给取消了，结果我在Safari上一运行就报OOM的错误……于是我把这个问题又反馈了一下，顺便让AI看看这个问题怎么解决，结果AI也不辜负我的期待，轻轻松松就解决了🤣。最终我除了会提出问题之外什么也不会做，只能看着大佬和AI完成我的想法了😝。
后来增加的以太网网关也挺折腾，原理其实并不复杂，但直接给AI说想用这个接口上网的时候它会自己想办法搓一套漏洞百出的TCP/IP协议栈，所以后来我在GitHub找了个tcpip.js库让它集成，BUG就比它自己手搓的少了不少。后来我又想加一个TLS终止功能，因为在浏览器上没办法直接让系统里的程序发送原始的TLS数据包去请求网站，但我又希望它能够请求带有https协议的网址，所以要提前对TLS进行解密再转换成fetch()，这时候我就没让AI自己手搓了，在网上找了个node-forge库让AI对接，就可以避免出现一堆奇怪的BUG了。
做好网关之后我又突发奇想，想做一个展示终端内状态的页面，但是目前JS并不能主动和里面的系统直接通信，所以又让AI搓了一个通过Service Worker访问系统内网络服务的功能，然后用busybox的httpd执行sh写的cgi程序，就可以被前端读取到了。随后我还给原作者发了个Issue展示这个Service Worker的效果，结果人家已经做了这个功能，我还是太Naive了😂。
最终做好的网站可以在这里查看，效果真的不是一般的好，相比于其他模仿终端形式的博客简直是降维打击，毕竟这是真的Linux啊～虽然目前依然有些BUG，但是原作者修BUG的速度还是挺快的，相信等他发布正式版之后，体验会更进一步。
虽然我觉得追求真实感很重要，但是我又回去看了一下为我提供想法的JSNix。虽然这个项目并不是真的Linux，但是我让AI看了一遍之后它对这个项目的评价也很高，说是作者对Linux的了解非常深入，很多细节模拟的也非常到位。而且为WASM Linux制作的那些脚本，在这个JSNix上其实完全可以实现，还更简单，于是我直接让AI参考给WASM Linux写的脚本把那些功能全部移植到这个JSNix上。也许是因为AI更擅长编写网页，很快它就完成了这些功能，而且效果非常不错，基本上和WASM Linux上面的效果一样，表现也相当不错，最终生成的结果可以在这里查看。
不过假的毕竟还是假的，和真的Linux差异还是有不少的，像正经的Shell可以在终端中直接用if、for和while之类的语法写一些东西，它这个就必须要编写成脚本才可以，而且解析也有相当多的问题。各种命令的运行结果很多都是硬编码的，只是编了一段对应命令的输出，参数做得也不全。另外著名的Linux文本三剑客grep、sed和awk在这个系统里实现得也非常差，只做了一些非常基础的功能，基本上就是完全不可用。所以说这个系统还是不够真实，只能做最基础的体验，不能够深究。
但并不是说它没有优点，它作为纯粹的JS不像WASM Linux那样容易出现兼容性问题，比如很多手机浏览器不支持跨源隔离，导致SharedArrayBuffer之类的API没法用，所以手机上只能体验这个JSNix的版本；另外WASM Linux的体积虽然说并不算很大，但也有好几MiB，不像这个JSNix连1MiB都不到；还有这个系统没有启动过程，相比于WASM Linux来说加载更快；而且在编写和修改功能的时候比WASM Linux简单不少，像给WASM Linux做的终端状态展示在JSNix上根本不需要折腾什么Service Worker，以及请求网络之类的也非常简单，直接fetch()就可以（也正是如此，JSNix的版本AI没一会儿就做完了）……总之就是只要不涉及具体功能方面JSNix都比WASM Linux要强不少，不过这样说有点“抛开事实不谈”的感觉了😂。
最终做出来的这两个版本的终端版博客基本上所有的代码都不是我写的，真的做到了仅仅是和AI聊聊天就把东西做出来了……只能说AI现在真的不是一般的强啊～当然方案还是得自己考虑，因为AI一旦走错了方向就很容易往一个方向不停地钻，结果就是花了很长时间可能效果会很差，尤其是现在的AI还特别喜欢自己造轮子，所以这时候还是自己找找看已经有的方案然后告诉它会比较好。
从效果来说这次写的东西也让我非常满意，我想我这次做的这个博客界面估计是前所未有的，毕竟这可是在浏览器上用真的Linux访问博客啊，其他人做的都是假的。至于虚拟机那些方案可能也是因为太重，所以基本上没人用，而WASM Linux只有Thomas Stokes做的效果最好，但是又没有受到应有的关注……总之从我个人的角度来看这次做的东西还挺Cool，就是不知道会不会吸引其他人来看呢……可能还是太小众了吧😂。
2026-07-01 00:00:00
All in AI!
前段时间，我写了一篇关于AI Agent的文章，虽然当时我说这并没有什么创新之处，但在那之后用着用着感觉还是挺好用的，至少相比以前在云端Vibe Coding来说又快又免费，所以我打算分享一下最近我用AI Agent的经历。
不过大多数情况下我写代码都可以直接让各路AI在对话中给我生成，那到底有什么样的场景才需要用AI Agent呢？一般应该就是需要边改边测的东西吧。
虽然说AI生成的东西可靠性并不是很高，但至少在网上已经有很多例子的代码应该不至于乱生成吧……其实我一直都想给我的博客增加全站无刷新的功能，但是每次看到PJAX都要自己重新调用当前页面的各种函数就懒得动手了😂。不过其实这也不是什么复杂的事情，主要是一个一个分析函数比较麻烦，所以既然有AI了，不如直接让AI帮我做吧。于是我让它克隆了我的仓库，分析并写出PJAX的代码。
用AI Agent最大的好处就是能让它了解整个项目的所有代码，如果是直接在对话里让AI修改的话就很难做到这一点。总之我告诉它之后等了一会它真的写出来了，我试了一下确实能用，虽然刚开始有不少问题，主要是因为我用了Live2D，不少问题也是这个功能引起的。不过只要我告诉它问题在哪里，它总是能给我改好，先不论它实现的代码怎么样，但至少从效果来说还是相当不错的。
最终生成的代码我看了一下，代码质量非常糟糕，把我在别处的代码到处复制粘贴到PJAX用的代码里了，但是体验效果不错……所以没办法，虽然代码很糟糕，但先用着吧。不过我觉得优化应该也能交给AI Agent，只是它有时候改来改去还会把正常代码改坏，所以优化看起来还是只能自己做……
最终的代码就是：pjax.js，虽然我对这个代码质量不是很满意，但是毕竟能用，效果也不错，就先这样用吧～
前段时间，我在Bilibili上看视频的时候发现了一个有趣的微信小游戏，叫做“雀谜”，它把立直麻将和Wordle两个玩法结合到一起，看起来挺有意思。不过它作为微信上的小游戏有些问题，一是看提示要看广告，二是换题也要看广告，而且看广告就算能换题也有次数限制，一天也就几次机会，让我有点不爽。
不过从功能上来说看起来并不复杂，感觉用AI直接Vibe Coding完全不是问题，不过AI Agent从零开始写比较考验模型的水平，所以我打算先让各路网页上的AI生成初版的Demo，之后再考虑怎么改。于是我对它们说：
帮我制作一款立直麻将Wordle网页小游戏，规则为：
- 系统随机生成一个有役能和的14张牌，必须按照万、筒、索、字的顺序排列，同花色内数字从小到大，但是第14张用来和的牌可以不遵守排列规则
- 让玩家猜6次来猜到这个牌型，每次猜的牌也必须是满足有役和牌的条件
- 位置和内容正确的牌用绿色标注、只有内容正确但是位置不正确的用黄色标注，不在生成范围内的牌用灰色标注
- 界面中所有的牌使用Unicode中的麻将字符，并且提供一组输入按钮用于输入这些麻将字符
- 输入按钮在游玩过程中也要标注颜色
- 随机生成的牌需要标注场风、自风和是否自摸，如果是荣和需要额外的役，需要考虑平和之类的役种
- 加一个提示键，可以查看答案牌型的役种
直接生成确实很考验水平，有很多AI直接生成出来的完全用不了，像DeepSeek生成出来的按钮按上去都没有反应😅，最终只有Gemini生成出来的感觉还不错，无论是界面还是输入的感觉都很好，于是接下来的修改我就打算以Gemini生成出来的代码为主。
刚生成出来的代码虽然能玩，但是每个AI直接生成都不想做完整的役种检测，说什么一个对话不够😥，明明各个都是几百k上下文的AI，有什么不够的……总之想一口气就完美做出来显然是不现实的，不过其实最开始我也不打算把所有役都做了，毕竟条件是“有役和牌”，而且没有副露，所以好多可以叠加的役其实都不需要做，尤其是役满那些，大多都可以用其他役的条件满足。
在这期间，“平和”算是实现起来最麻烦的，主要是这个役需要根据和牌判定，必须是顺子的两面听牌才可以……虽然AI理解这些知识，但是写的时候经常出问题，像边张之类的经常误判……但是不实现又不行，毕竟这是可以nomi的牌型。
最终写的差不多了，感觉还不错，于是在GitHub Pages上部署了一份，起名叫立直麻将 Wordle。
最开始我以为这个小游戏的玩法是他们原创的，于是把生成的代码又丢给AI，让它评价一下创新性，结果AI说已经有开源了😂，有个叫做Mahjong Handle的才是最早做这个玩法的……还好这代码不是我写的，要是我写了半天结果发现有现成的那就很无语了😆。
当然我让AI写的版本也有一些特色，Mahjong Handle那个和原版Wordle一样，每天只有一道题，而且这个题是从题库里抽的，我让AI Agent看了一下它的代码，似乎是用phoenix-logs这个脚本从天凤上下载的牌谱数据解析得到的题库，所以它的题目数量是有限的，而我的代码是完全随机生成的，所以结果是无限的。但AI似乎认为作为Wordle来说一天一题才是它的特色，无限就显得没意思了……既然如此，那干脆让它给我把两个游戏合并了，让它参考Mahjong Handle把一天一题的功能也给我加上，另外Mahjong Handle还有一个MPSZ快速输入法，也让AI给我合并进我的游戏里了。
这时候就体现了AI Agent的优势，如果直接用线上的AI把Mahjong Handle中的功能移植到我的代码，我还得完整描述要实现的功能，但是AI Agent能自己看人家仓库的代码，就知道要改什么东西了。
最终实现的效果也很不错，虽然我的代码没有题库，但是使用了当天日期作为随机种子，用mulberry32算法搞出了一天一题的功能，还挺有意思的。
另外在Mahjong Handle以及原版的Wordle中，还有一个困难模式，就是每次猜测必须使用之前的猜测条件，不过我觉得如果原模原样把这个功能抄过来似乎有点无聊，而且我发现玩这个东西一般直接一个国士+两个一气就基本上把要猜的牌找完了，有点简单，所以我想了一下，要不然限制一下提交的条件，提交的牌型必须包含目标的役种之类的……不过要是做这个功能就不能像之前那样只实现部分役种了，得把所有能用到的都实现一遍，还好这种事情肯定也不是我来做，肯定还得是让AI来😋。
实现的时候还不能让AI直接一口气全加上，还得一步一步的提示，而且每个牌型还得手动测一遍……最终虽然做的看起来似乎已经很完善了，不过逻辑的根本还是有点问题，AI设计役种计算的时候是按拆解后役种数量最多的情况来算，但事实上立直麻将算役种是按照高点法，要最终点数最大才行。虽然按照目前的方法绝大多数情况下都能正常工作，但偶尔也会遇到问题，比如像这副牌：
按照目前的方法计算的结果是：
然而正确答案是：
但是符数的计算又很麻烦……不过反正提示和验牌都是同一套程序，题倒是也能做出来……所以就懒得修了🤣。当然说到底还是因为AI没有用现有的计算库，像Mahjong Handle用的是现成的Riichi库，制作起来比我这个就简单多了。
在我制作“立直麻将 Wordle”这个小游戏的时候，我让AI使用的是Unicode中的麻将字符，毕竟有现成的字符总比去找牌的图片要方便，而且实现起来感觉更简单一些。不过用字符确实也会遇到一些问题，我不知道为什么在macOS和Android上其他牌都是黑白的，唯有🀄️是彩色的😅，Windows上使用的Segoe UI Emoji倒都是彩色的，但是牌很丑，像是小学生设计的……虽然不影响玩，但是看着很难受，至少牌的颜色统一一下也好吧？虽然也可以搞出黑白的🀄︎，但如果用这个，在Windows上又成了只有🀄️是黑白的😅。虽然Mahjong Handle是直接用了SVG格式的牌图片，但我觉得应该有现成的麻将字体，像这种问题应该有现成的解决方法才对吧？
于是我直接问AI有没有这样的字体，AI搜了一下看起来还真有，有个叫做Mahjong Colored就是基于那些SVG做的OpenType-SVG字体，我本来以为问题就会这样解决，然而当我应用了字体之后发现牌的位置变成了白色的方块，貌似只有Firefox才支持这种字体……然而现在谁还用Firefox啊？
那遇到这种问题怎么办？当然还是让AI解决，它说要在Safari和Chrome上使用彩色字体，用COLR或者SBIX格式比较好，COLR格式的兼容性最好，三个浏览器都能用，而SBIX的话Firefox没法用。不过Firefox既然基本上没人用了，我倒是不太在乎它的兼容性，但我肯定还是更希望能用兼容性更好的方案，于是我把OpenType-SVG字体转换为COLR的工作交给了AI Agent。
结果它整了半天也没整出来，貌似是因为SVG的特效太多，没办法转换为COLR格式，于是它选择了SBIX方案，不过COLR存储的是矢量图，而SBIX存储的是位图，大小要大一些，而且放大还会有锯齿。只是目前实在是没办法了，只能用这种格式。
最终我把它转换后的字体存了一份到仓库，但是看着它1MiB多的大小以及不太好的浏览器兼容性，再考虑到它还是位图，实在是不太想在我的游戏里应用它……
结果后来我又搜了一下，原来是有现成的麻将字体啊，有个叫做Mahjong Font的项目用的就是COLR格式，兼容性非常好，而且大小只有80KiB，算是最完美的解决方案了。垃圾AI给我推荐的Mahjong Colored真的是……浪费了不少时间，看来有的时候自己调查要比问AI好一些吧。
总的来看，目前Vibe Coding的体验确实是相当不错，只是问题也非常明显，完全让AI生成的代码要么实现的不优雅，要么就会出BUG，如果完全不懂要写的东西，纯粹由AI生成，那出问题的概率就非常高了。不过要是看不出来错误，完全被AI忽悠过去了，也许这时候的Vibe Coding在另一方面算是完美吧😂？至少对用的人来说也算是解决了问题呢。
