TermiNexus
2992674783
把此前未提交的 rowptr 迁移整体入库（42 文件，含 ast/llvmlite/hashlib/json/zlib 等签名 rowptr 化）。临时诊断（用后移除）：linkedlist.vp GSList[T].append 链完整性四检、__module.vp PrintModule Function.Params/param 指针野值护栏（[DBGCHAIN] 前缀 + fflush + exit）。
650 lines
24 KiB
Plaintext
650 lines
24 KiB
Plaintext
# -*- coding: utf-8 -*-
|
||
# zipfile.vp — ZIP 归档读写模块（Python zipfile 子集）
|
||
#
|
||
# 支持:
|
||
# - 写入成员: ZIP_STORED / ZIP_DEFLATED（raw deflate，wbits=-15，CRC32 校验）
|
||
# - 读取成员: 自动识别 ZIP_STORED / ZIP_DEFLATED，解压后做 CRC32 校验
|
||
# - 读取时从文件尾部回扫 EOCD（容忍注释/前导数据）
|
||
#
|
||
# 不支持: zip64（>4GB/成员数>65535）、加密、分卷、数据描述符(flag bit 3)等。
|
||
#
|
||
# 成员名为原始 UTF-8 字节（写入时置 flag bit 11，读取时不自动解码）。
|
||
#
|
||
# 用法（对齐 Python 语义，只读模式自动把整个 zip 读入内存）:
|
||
# w: zipfile.ZipFile | t.CPtr = zipfile.ZipFile("out.zip", "w")
|
||
# w.write("a.txt", buf, buf_len) # 默认 ZIP_DEFLATED
|
||
# w.close()
|
||
#
|
||
# r: zipfile.ZipFile | t.CPtr = zipfile.ZipFile("out.zip", "r")
|
||
# olen: t.CSizeT = 0
|
||
# data: bytes = r.read("a.txt", c.Addr(olen)) # malloc 缓冲，用后 free
|
||
# err = zipfile.error_code() / zipfile.error()
|
||
|
||
import t, c
|
||
from stdint import *
|
||
import stdlib
|
||
import string
|
||
import viperlib
|
||
import w32.win32base as wb
|
||
import w32.win32file as wf
|
||
import zlib.pyzlib as zpy
|
||
|
||
|
||
# ============================================================
|
||
# 常量（对齐 Python zipfile）
|
||
# ============================================================
|
||
|
||
ZIP_STORED: t.CDefine = 0
|
||
ZIP_DEFLATED: t.CDefine = 8
|
||
ZIP_LEVEL_DEFAULT: t.CDefine = 6 # 压缩级别（zlib 默认 6）
|
||
_ZIP_FLAG_UTF8: t.CDefine = 0x0800 # 文件名按 UTF-8（flag bit 11）
|
||
_ZIP_VERSION: t.CDefine = 20 # 所需版本 2.0
|
||
_ZIP_DOS_DATE_EPOCH: t.CDefine = 0x21 # 1980-01-01
|
||
_ZIP_MAX_COMMENT: t.CDefine = 65535
|
||
|
||
# 错误码（方法返回值；全局最后错误见 error_code()/error()）
|
||
ZIP_ERR_OK: t.CDefine = 0
|
||
ZIP_ERR_OPEN: t.CDefine = -1
|
||
ZIP_ERR_NOT_ZIP: t.CDefine = -2
|
||
ZIP_ERR_MEMORY: t.CDefine = -3
|
||
ZIP_ERR_NOT_FOUND: t.CDefine = -4
|
||
ZIP_ERR_INVALID: t.CDefine = -5
|
||
ZIP_ERR_UNSUPPORTED: t.CDefine = -6
|
||
ZIP_ERR_CRC: t.CDefine = -7
|
||
ZIP_ERR_IO: t.CDefine = -8
|
||
ZIP_ERR_MODE: t.CDefine = -9
|
||
ZIP_ERR_CLOSED: t.CDefine = -10
|
||
|
||
# ============================================================
|
||
# 模块级错误状态（仿 zlib.pyzlib）
|
||
# ============================================================
|
||
|
||
zipfile_error_code_val: t.CInt = 0
|
||
zipfile_error_msg: t.CArray[t.CChar, 256] = ""
|
||
|
||
|
||
def zipfile_set_error(code: t.CInt, msg: str):
|
||
global zipfile_error_code_val, zipfile_error_msg
|
||
zipfile_error_code_val = code
|
||
if msg:
|
||
string.strncpy(zipfile_error_msg, msg, 255)
|
||
zipfile_error_msg[255] = '\0'
|
||
else:
|
||
zipfile_error_msg[0] = '\0'
|
||
|
||
|
||
def zipfile_set_error_win(code: t.CInt, msg: str, err: ULONG):
|
||
global zipfile_error_code_val, zipfile_error_msg
|
||
zipfile_error_code_val = code
|
||
if err != 0:
|
||
viperlib.snprintf(zipfile_error_msg, 256, "%s (win32 err=%u)", msg, err)
|
||
else:
|
||
string.strncpy(zipfile_error_msg, msg, 255)
|
||
zipfile_error_msg[255] = '\0'
|
||
|
||
|
||
def error_code() -> t.CInt:
|
||
return zipfile_error_code_val
|
||
|
||
|
||
def error() -> str:
|
||
return zipfile_error_msg
|
||
|
||
|
||
# ============================================================
|
||
# 小工具
|
||
# ============================================================
|
||
|
||
def zf_strlen(s: bytes) -> t.CSizeT:
|
||
n: t.CSizeT = 0
|
||
while s is not None and c.Deref(s) != 0:
|
||
n += 1
|
||
s += 1
|
||
return n
|
||
|
||
|
||
def zf_dup_name(src: bytes, length: t.CSizeT) -> bytes:
|
||
buf: BYTEPTR = BYTEPTR(stdlib.malloc(length + 1))
|
||
if not buf:
|
||
return None
|
||
if length > 0:
|
||
string.memcpy(buf, src, length)
|
||
buf[length] = 0
|
||
return buf
|
||
|
||
|
||
def zf_u16le(p: UINT8PTR) -> t.CUInt32T:
|
||
return t.CUInt32T(p[0]) | (t.CUInt32T(p[1]) << 8)
|
||
|
||
|
||
def zf_u32le(p: UINT8PTR) -> t.CUInt32T:
|
||
return (t.CUInt32T(p[0]) |
|
||
(t.CUInt32T(p[1]) << 8) |
|
||
(t.CUInt32T(p[2]) << 16) |
|
||
(t.CUInt32T(p[3]) << 24))
|
||
|
||
|
||
def zf_w16le(h: BYTEPTR, v: t.CUInt32T):
|
||
h[0] = v & 0xFF
|
||
h[1] = (v >> 8) & 0xFF
|
||
|
||
|
||
def zf_w32le(h: BYTEPTR, v: t.CUInt32T):
|
||
h[0] = v & 0xFF
|
||
h[1] = (v >> 8) & 0xFF
|
||
h[2] = (v >> 16) & 0xFF
|
||
h[3] = (v >> 24) & 0xFF
|
||
|
||
|
||
# 所有文件打开都允许读/写/删共享，避免句柄互斥
|
||
_ZF_SHARE: t.CDefine = 7 # FILE_SHARE_READ|WRITE|DELETE
|
||
|
||
|
||
def zf_share_flags() -> ULONG:
|
||
return ULONG(_ZF_SHARE)
|
||
|
||
|
||
def zf_write_all(h: wb.HANDLE, p: BYTEPTR, n: t.CSizeT) -> t.CInt:
|
||
done: t.CSizeT = 0
|
||
while done < n:
|
||
chunk: t.CSizeT = n - done
|
||
if chunk > 0xFFFFFF00:
|
||
chunk = 0xFFFFFF00
|
||
written: ULONG = 0
|
||
ok: BOOL = wf.WriteFile(h, p + done, ULONG(chunk), c.Addr(written), None)
|
||
if ok == 0 or written == 0:
|
||
global zipfile_error_code_val, zipfile_error_msg
|
||
zipfile_error_code_val = ZIP_ERR_IO
|
||
viperlib.snprintf(zipfile_error_msg, 256,
|
||
"write failed (win32_err=%u)", wb.GetLastError())
|
||
return ZIP_ERR_IO
|
||
done += t.CSizeT(written)
|
||
return ZIP_ERR_OK
|
||
|
||
|
||
def zf_read_all(h: wb.HANDLE, p: BYTEPTR, n: t.CSizeT) -> t.CInt:
|
||
done: t.CSizeT = 0
|
||
while done < n:
|
||
chunk: t.CSizeT = n - done
|
||
if chunk > 0xFFFFFF00:
|
||
chunk = 0xFFFFFF00
|
||
bytes_read: ULONG = 0
|
||
ok: BOOL = wf.ReadFile(h, p + done, ULONG(chunk), c.Addr(bytes_read), None)
|
||
if ok == 0 or bytes_read == 0:
|
||
global zipfile_error_code_val, zipfile_error_msg
|
||
zipfile_error_code_val = ZIP_ERR_IO
|
||
viperlib.snprintf(zipfile_error_msg, 256,
|
||
"read failed (win32_err=%u)", wb.GetLastError())
|
||
return ZIP_ERR_IO
|
||
done += t.CSizeT(bytes_read)
|
||
return ZIP_ERR_OK
|
||
|
||
|
||
def zf_read_whole(path: str, out_len: t.CSizeT | t.CPtr) -> BYTEPTR:
|
||
if not out_len:
|
||
zipfile_set_error(ZIP_ERR_OPEN, "out_len is None")
|
||
return None
|
||
h: wb.HANDLE = wf.CreateFileA(
|
||
path, wf.GENERIC_READ, zf_share_flags(),
|
||
None, wf.OPEN_EXISTING, wf.FILE_ATTRIBUTE_NORMAL, None)
|
||
if h == wb.INVALID_HANDLE_VALUE:
|
||
zipfile_set_error_win(ZIP_ERR_OPEN, "cannot open file for reading", wb.GetLastError())
|
||
return None
|
||
li: wb.LARGE_INTEGER = wb.LARGE_INTEGER()
|
||
ok_size: BOOL = wf.GetFileSizeEx(h, t.CPtr(c.Addr(li)))
|
||
if ok_size == 0:
|
||
wb.CloseHandle(h)
|
||
zipfile_set_error(ZIP_ERR_IO, "cannot get file size")
|
||
return None
|
||
sz: t.CSizeT = t.CSizeT(li.QuadPart)
|
||
if sz == 0:
|
||
wb.CloseHandle(h)
|
||
zipfile_set_error(ZIP_ERR_NOT_ZIP, "empty file is not a zip archive")
|
||
return None
|
||
buf: BYTEPTR = BYTEPTR(stdlib.malloc(sz))
|
||
if not buf:
|
||
wb.CloseHandle(h)
|
||
zipfile_set_error(ZIP_ERR_MEMORY, "out of memory")
|
||
return None
|
||
rc: t.CInt = zf_read_all(h, buf, sz)
|
||
wb.CloseHandle(h)
|
||
if rc != ZIP_ERR_OK:
|
||
stdlib.free(buf)
|
||
return None
|
||
c.Set(c.Deref(out_len), sz)
|
||
return buf
|
||
|
||
|
||
# ============================================================
|
||
# 条目信息（目录项）
|
||
# 注意: 节点为显式 calloc 分配（@t.Object 裸结构，无 GC 追踪），
|
||
# 由 ZipFile 负责释放——避免 GC 实例经 | t.CPtr 联合字段引用被回收。
|
||
# ============================================================
|
||
|
||
@t.Object
|
||
class ZipInfo:
|
||
filename: bytes # 原始字节名（含 \0，zipfile 持有并负责释放）
|
||
name_len: t.CSizeT
|
||
compress_type: t.CInt # ZIP_STORED / ZIP_DEFLATED
|
||
crc32: t.CUInt32T
|
||
compress_size: t.CSizeT
|
||
file_size: t.CSizeT
|
||
header_offset: t.CSizeT # 本地文件头偏移
|
||
next: "ZipInfo" | t.CPtr # 链表（保持写入/目录顺序）
|
||
|
||
|
||
# ============================================================
|
||
# ZipFile
|
||
# ============================================================
|
||
|
||
class ZipFile:
|
||
path: bytes
|
||
mode: t.CInt # 0 = 读，1 = 写
|
||
compression: t.CInt # write() 未显式指定时的压缩类型
|
||
fh: wb.HANDLE # 写模式句柄（裸 win32 CreateFileA）
|
||
data: BYTEPTR # 读模式：整个归档的内存副本
|
||
data_len: t.CSizeT
|
||
info_head: ZipInfo | t.CPtr
|
||
info_tail: ZipInfo | t.CPtr
|
||
info_count: t.CSizeT
|
||
pos: t.CSizeT # 写模式当前偏移
|
||
ok: t.CInt # 1 = 可正常使用
|
||
closed: t.CInt
|
||
|
||
def __init__(self, path: str, mode: str = "r", compression: t.CInt = ZIP_DEFLATED):
|
||
self.path = path
|
||
self.mode = -1
|
||
self.compression = compression
|
||
self.fh = None
|
||
self.data = None
|
||
self.data_len = 0
|
||
self.info_head = None
|
||
self.info_tail = None
|
||
self.info_count = 0
|
||
self.pos = 0
|
||
self.ok = 0
|
||
self.closed = 0
|
||
# 注意: 字符串必须用 strcmp 比较（== 只会比较指针）
|
||
if string.strcmp(mode, "r") == 0:
|
||
self.mode = 0
|
||
elif string.strcmp(mode, "w") == 0:
|
||
self.mode = 1
|
||
else:
|
||
zipfile_set_error(ZIP_ERR_OPEN, "mode must be 'r' or 'w'")
|
||
self.closed = 1
|
||
return
|
||
if self.mode == 0:
|
||
self.data = zf_read_whole(path, c.Addr(self.data_len))
|
||
if not self.data:
|
||
self.closed = 1
|
||
return
|
||
if self._parse() != ZIP_ERR_OK:
|
||
stdlib.free(self.data)
|
||
self.data = None
|
||
self.data_len = 0
|
||
self._clear_infos()
|
||
self.closed = 1
|
||
return
|
||
else:
|
||
self.fh = wf.CreateFileA(
|
||
path, wf.GENERIC_WRITE, zf_share_flags(),
|
||
None, wf.CREATE_ALWAYS, wf.FILE_ATTRIBUTE_NORMAL, None)
|
||
if self.fh == wb.INVALID_HANDLE_VALUE:
|
||
zipfile_set_error_win(ZIP_ERR_OPEN, "cannot create file for writing", wb.GetLastError())
|
||
self.fh = None
|
||
self.closed = 1
|
||
return
|
||
self.pos = 0
|
||
self.ok = 1
|
||
|
||
def _clear_infos(self):
|
||
n: ZipInfo | t.CPtr = self.info_head
|
||
while n is not None:
|
||
if n.filename is not None:
|
||
stdlib.free(n.filename)
|
||
n.filename = None
|
||
nxt: ZipInfo | t.CPtr = n.next
|
||
n.next = None
|
||
stdlib.free(n)
|
||
n = nxt
|
||
self.info_head = None
|
||
self.info_tail = None
|
||
self.info_count = 0
|
||
|
||
def _append_info(self) -> ZipInfo | t.CPtr:
|
||
node: ZipInfo | t.CPtr = stdlib.calloc(1, ZipInfo.__sizeof__())
|
||
if not node:
|
||
zipfile_set_error(ZIP_ERR_MEMORY, "out of memory")
|
||
return None
|
||
node.next = None
|
||
if self.info_tail is None:
|
||
self.info_head = node
|
||
else:
|
||
self.info_tail.next = node
|
||
self.info_tail = node
|
||
self.info_count += 1
|
||
return node
|
||
|
||
# ---- 读模式：解析中央目录 ----
|
||
def _parse(self) -> t.CInt:
|
||
if self.data_len < 22:
|
||
zipfile_set_error(ZIP_ERR_NOT_ZIP, "file too small to be a zip")
|
||
return ZIP_ERR_NOT_ZIP
|
||
d: UINT8PTR = UINT8PTR(self.data)
|
||
|
||
# 从尾部回扫 EOCD（容忍尾部注释）
|
||
max_back: t.CSizeT = self.data_len - 22
|
||
if max_back > _ZIP_MAX_COMMENT + 20:
|
||
max_back = _ZIP_MAX_COMMENT + 20
|
||
eocd: t.CSizeT = 0
|
||
found: t.CInt = 0
|
||
i: t.CSizeT = 0
|
||
while i <= max_back and found == 0:
|
||
p: t.CSizeT = self.data_len - 22 - i
|
||
if d[p] == 0x50 and d[p + 1] == 0x4B and d[p + 2] == 0x05 and d[p + 3] == 0x06:
|
||
eocd = p
|
||
found = 1
|
||
i += 1
|
||
if found == 0:
|
||
zipfile_set_error(ZIP_ERR_NOT_ZIP, "EOCD not found")
|
||
return ZIP_ERR_NOT_ZIP
|
||
|
||
total: t.CUInt32T = zf_u16le(d + eocd + 10)
|
||
cd_off: t.CUInt32T = zf_u32le(d + eocd + 16)
|
||
cd_size: t.CUInt32T = zf_u32le(d + eocd + 12)
|
||
if t.CSizeT(cd_off) + t.CSizeT(cd_size) > self.data_len:
|
||
zipfile_set_error(ZIP_ERR_NOT_ZIP, "central directory out of range")
|
||
return ZIP_ERR_NOT_ZIP
|
||
|
||
cp: t.CSizeT = t.CSizeT(cd_off)
|
||
end: t.CSizeT = t.CSizeT(cd_off) + t.CSizeT(cd_size)
|
||
idx: t.CSizeT = 0
|
||
while idx < total:
|
||
if cp + 46 > end:
|
||
zipfile_set_error(ZIP_ERR_NOT_ZIP, "central directory entry truncated")
|
||
return ZIP_ERR_NOT_ZIP
|
||
if zf_u32le(d + cp) != 0x02014B50:
|
||
zipfile_set_error(ZIP_ERR_NOT_ZIP, "bad central directory signature")
|
||
return ZIP_ERR_NOT_ZIP
|
||
method: t.CUInt32T = zf_u16le(d + cp + 10)
|
||
crc: t.CUInt32T = zf_u32le(d + cp + 16)
|
||
csize: t.CUInt32T = zf_u32le(d + cp + 20)
|
||
usize: t.CUInt32T = zf_u32le(d + cp + 24)
|
||
nlen: t.CUInt32T = zf_u16le(d + cp + 28)
|
||
elen: t.CUInt32T = zf_u16le(d + cp + 30)
|
||
clen: t.CUInt32T = zf_u16le(d + cp + 32)
|
||
hoff: t.CUInt32T = zf_u32le(d + cp + 42)
|
||
rec_len: t.CSizeT = 46 + t.CSizeT(nlen) + t.CSizeT(elen) + t.CSizeT(clen)
|
||
if cp + rec_len > end:
|
||
zipfile_set_error(ZIP_ERR_NOT_ZIP, "central directory entry out of range")
|
||
return ZIP_ERR_NOT_ZIP
|
||
if nlen == 0:
|
||
zipfile_set_error(ZIP_ERR_NOT_ZIP, "central directory entry has empty name")
|
||
return ZIP_ERR_NOT_ZIP
|
||
dup: BYTEPTR = zf_dup_name(BYTEPTR(d + cp + 46), t.CSizeT(nlen))
|
||
if not dup:
|
||
zipfile_set_error(ZIP_ERR_MEMORY, "out of memory")
|
||
return ZIP_ERR_MEMORY
|
||
node: ZipInfo | t.CPtr = self._append_info()
|
||
if node is None:
|
||
stdlib.free(dup)
|
||
return ZIP_ERR_MEMORY
|
||
node.filename = dup
|
||
node.name_len = t.CSizeT(nlen)
|
||
node.compress_type = t.CInt(method)
|
||
node.crc32 = crc
|
||
node.compress_size = t.CSizeT(csize)
|
||
node.file_size = t.CSizeT(usize)
|
||
node.header_offset = t.CSizeT(hoff)
|
||
cp = cp + rec_len
|
||
idx += 1
|
||
return ZIP_ERR_OK
|
||
|
||
# ---- 公共查询 ----
|
||
def namelist_at(self, i: t.CSizeT) -> bytes:
|
||
if self.ok == 0 or self.info_count == 0 or i >= self.info_count:
|
||
return None
|
||
n: ZipInfo | t.CPtr = self.info_head
|
||
k: t.CSizeT = 0
|
||
while n is not None:
|
||
if k == i:
|
||
return n.filename
|
||
n = n.next
|
||
k += 1
|
||
return None
|
||
|
||
def getinfo(self, name: str) -> ZipInfo | t.CPtr:
|
||
if self.ok == 0:
|
||
return None
|
||
n: ZipInfo | t.CPtr = self.info_head
|
||
while n is not None:
|
||
if n.filename is not None and string.strcmp(n.filename, name) == 0:
|
||
return n
|
||
n = n.next
|
||
return None
|
||
|
||
# ---- 读成员：解压 + CRC32 校验 ----
|
||
def read(self, name: str, out_len: t.CSizeT | t.CPtr) -> BYTEPTR:
|
||
if not out_len:
|
||
zipfile_set_error(ZIP_ERR_INVALID, "out_len is None")
|
||
return None
|
||
c.Set(c.Deref(out_len), 0)
|
||
if self.closed or self.ok == 0:
|
||
zipfile_set_error(ZIP_ERR_CLOSED, "zip file is closed")
|
||
return None
|
||
if self.mode != 0:
|
||
zipfile_set_error(ZIP_ERR_MODE, "zip file is not open for reading")
|
||
return None
|
||
info: ZipInfo | t.CPtr = self.getinfo(name)
|
||
if info is None:
|
||
zipfile_set_error(ZIP_ERR_NOT_FOUND, "member not found")
|
||
return None
|
||
|
||
d: UINT8PTR = UINT8PTR(self.data)
|
||
off: t.CSizeT = t.CSizeT(info.header_offset)
|
||
if off + 30 > self.data_len:
|
||
zipfile_set_error(ZIP_ERR_NOT_ZIP, "local file header out of range")
|
||
return None
|
||
if zf_u32le(d + off) != 0x04034B50:
|
||
zipfile_set_error(ZIP_ERR_NOT_ZIP, "bad local file header signature")
|
||
return None
|
||
nlen: t.CUInt32T = zf_u16le(d + off + 26)
|
||
elen: t.CUInt32T = zf_u16le(d + off + 28)
|
||
start: t.CSizeT = off + 30 + t.CSizeT(nlen) + t.CSizeT(elen)
|
||
csz: t.CSizeT = t.CSizeT(info.compress_size)
|
||
if start + csz > self.data_len:
|
||
zipfile_set_error(ZIP_ERR_NOT_ZIP, "member data out of range")
|
||
return None
|
||
|
||
payload: BYTEPTR = BYTEPTR(d + start)
|
||
out: BYTEPTR = None
|
||
ctype: t.CInt = info.compress_type
|
||
if ctype == ZIP_STORED:
|
||
out = BYTEPTR(stdlib.malloc(t.CSizeT(info.file_size) + 1))
|
||
if not out:
|
||
zipfile_set_error(ZIP_ERR_MEMORY, "out of memory")
|
||
return None
|
||
if csz > 0:
|
||
string.memcpy(out, payload, csz)
|
||
c.Set(c.Deref(out_len), t.CSizeT(info.file_size))
|
||
elif ctype == ZIP_DEFLATED:
|
||
olen: t.CSizeT = 0
|
||
out = zpy.decompress(None, payload, csz, -15, 0, c.Addr(olen))
|
||
if not out:
|
||
zipfile_set_error(ZIP_ERR_IO, "deflate decompression failed")
|
||
return None
|
||
c.Set(c.Deref(out_len), olen)
|
||
else:
|
||
zipfile_set_error(ZIP_ERR_UNSUPPORTED, "unsupported compression method")
|
||
return None
|
||
|
||
got_len: t.CSizeT = c.Deref(out_len)
|
||
if got_len != t.CSizeT(info.file_size):
|
||
stdlib.free(out)
|
||
c.Set(c.Deref(out_len), 0)
|
||
zipfile_set_error(ZIP_ERR_NOT_ZIP, "member size mismatch")
|
||
return None
|
||
crc_calc: t.CUInt32T = t.CUInt32T(zpy.zlib_crc32(out, got_len, 0))
|
||
if crc_calc != info.crc32:
|
||
stdlib.free(out)
|
||
c.Set(c.Deref(out_len), 0)
|
||
zipfile_set_error(ZIP_ERR_CRC, "CRC32 mismatch")
|
||
return None
|
||
return out
|
||
|
||
# ---- 写成员 ----
|
||
def write(self, arcname: str, data: BYTEPTR, data_len: t.CSizeT,
|
||
compress_type: t.CInt = -1) -> t.CInt:
|
||
if self.closed or self.ok == 0:
|
||
zipfile_set_error(ZIP_ERR_CLOSED, "zip file is closed")
|
||
return ZIP_ERR_CLOSED
|
||
if self.mode != 1:
|
||
zipfile_set_error(ZIP_ERR_MODE, "zip file is not open for writing")
|
||
return ZIP_ERR_MODE
|
||
if not arcname:
|
||
zipfile_set_error(ZIP_ERR_INVALID, "arcname is required")
|
||
return ZIP_ERR_INVALID
|
||
ctype: t.CInt = compress_type
|
||
if ctype == -1:
|
||
ctype = self.compression
|
||
if ctype != ZIP_STORED and ctype != ZIP_DEFLATED:
|
||
zipfile_set_error(ZIP_ERR_UNSUPPORTED, "compress_type must be ZIP_STORED or ZIP_DEFLATED")
|
||
return ZIP_ERR_UNSUPPORTED
|
||
if data_len > 0 and not data:
|
||
zipfile_set_error(ZIP_ERR_INVALID, "data is None but data_len > 0")
|
||
return ZIP_ERR_INVALID
|
||
name_len: t.CSizeT = zf_strlen(arcname)
|
||
if name_len == 0:
|
||
zipfile_set_error(ZIP_ERR_INVALID, "arcname is empty")
|
||
return ZIP_ERR_INVALID
|
||
if name_len > 0xFFFF or data_len > 0xFFFFFFFF:
|
||
zipfile_set_error(ZIP_ERR_UNSUPPORTED, "name or data too large for zip32")
|
||
return ZIP_ERR_UNSUPPORTED
|
||
|
||
crc: t.CUInt32T = t.CUInt32T(zpy.zlib_crc32(data, data_len, 0))
|
||
payload: BYTEPTR = data
|
||
plen: t.CSizeT = data_len
|
||
if ctype == ZIP_DEFLATED:
|
||
tmp: t.CSizeT = 0
|
||
payload = zpy.compress(None, data, data_len, ZIP_LEVEL_DEFAULT, -15, c.Addr(tmp))
|
||
if not payload:
|
||
zipfile_set_error(ZIP_ERR_IO, "deflate compression failed")
|
||
return ZIP_ERR_IO
|
||
plen = tmp
|
||
|
||
hoff: t.CSizeT = self.pos
|
||
hdr: BYTEPTR = BYTEPTR(stdlib.malloc(30))
|
||
if not hdr:
|
||
if ctype == ZIP_DEFLATED:
|
||
stdlib.free(payload)
|
||
zipfile_set_error(ZIP_ERR_MEMORY, "out of memory")
|
||
return ZIP_ERR_MEMORY
|
||
string.memset(hdr, 0, 30)
|
||
zf_w32le(hdr, 0x04034B50)
|
||
zf_w16le(hdr + 4, _ZIP_VERSION)
|
||
zf_w16le(hdr + 6, _ZIP_FLAG_UTF8)
|
||
zf_w16le(hdr + 8, t.CUInt32T(ctype))
|
||
zf_w16le(hdr + 12, _ZIP_DOS_DATE_EPOCH)
|
||
zf_w32le(hdr + 14, crc)
|
||
zf_w32le(hdr + 18, t.CUInt32T(plen))
|
||
zf_w32le(hdr + 22, t.CUInt32T(data_len))
|
||
zf_w16le(hdr + 26, t.CUInt32T(name_len))
|
||
rc: t.CInt = zf_write_all(self.fh, hdr, 30)
|
||
stdlib.free(hdr)
|
||
if rc == ZIP_ERR_OK:
|
||
rc = zf_write_all(self.fh, arcname, name_len)
|
||
if rc == ZIP_ERR_OK:
|
||
rc = zf_write_all(self.fh, payload, plen)
|
||
if ctype == ZIP_DEFLATED:
|
||
stdlib.free(payload)
|
||
if rc != ZIP_ERR_OK:
|
||
return rc
|
||
|
||
self.pos = self.pos + 30 + name_len + plen
|
||
dup: BYTEPTR = zf_dup_name(arcname, name_len)
|
||
if not dup:
|
||
zipfile_set_error(ZIP_ERR_MEMORY, "out of memory")
|
||
return ZIP_ERR_MEMORY
|
||
node: ZipInfo | t.CPtr = self._append_info()
|
||
if node is None:
|
||
stdlib.free(dup)
|
||
return ZIP_ERR_MEMORY
|
||
node.filename = dup
|
||
node.name_len = name_len
|
||
node.compress_type = ctype
|
||
node.crc32 = crc
|
||
node.compress_size = plen
|
||
node.file_size = data_len
|
||
node.header_offset = hoff
|
||
return ZIP_ERR_OK
|
||
|
||
# ---- 写模式收尾：中央目录 + EOCD ----
|
||
def _finish_write(self) -> t.CInt:
|
||
if self.info_count > _ZIP_MAX_COMMENT:
|
||
zipfile_set_error(ZIP_ERR_UNSUPPORTED, "too many entries (>65535)")
|
||
return ZIP_ERR_UNSUPPORTED
|
||
cd_start: t.CSizeT = self.pos
|
||
hdr: BYTEPTR = BYTEPTR(stdlib.malloc(46))
|
||
if not hdr:
|
||
zipfile_set_error(ZIP_ERR_MEMORY, "out of memory")
|
||
return ZIP_ERR_MEMORY
|
||
rc: t.CInt = ZIP_ERR_OK
|
||
n: ZipInfo | t.CPtr = self.info_head
|
||
while n is not None and rc == ZIP_ERR_OK:
|
||
string.memset(hdr, 0, 46)
|
||
zf_w32le(hdr, 0x02014B50)
|
||
zf_w16le(hdr + 4, _ZIP_VERSION) # version made by
|
||
zf_w16le(hdr + 6, _ZIP_VERSION) # version needed
|
||
zf_w16le(hdr + 8, _ZIP_FLAG_UTF8)
|
||
zf_w16le(hdr + 10, t.CUInt32T(n.compress_type))
|
||
zf_w16le(hdr + 14, _ZIP_DOS_DATE_EPOCH)
|
||
zf_w32le(hdr + 16, n.crc32)
|
||
zf_w32le(hdr + 20, t.CUInt32T(n.compress_size))
|
||
zf_w32le(hdr + 24, t.CUInt32T(n.file_size))
|
||
zf_w16le(hdr + 28, t.CUInt32T(n.name_len))
|
||
zf_w32le(hdr + 42, t.CUInt32T(n.header_offset))
|
||
rc = zf_write_all(self.fh, hdr, 46)
|
||
if rc == ZIP_ERR_OK:
|
||
rc = zf_write_all(self.fh, n.filename, n.name_len)
|
||
if rc == ZIP_ERR_OK:
|
||
self.pos = self.pos + 46 + n.name_len
|
||
n = n.next
|
||
stdlib.free(hdr)
|
||
if rc != ZIP_ERR_OK:
|
||
return rc
|
||
|
||
cd_size: t.CSizeT = self.pos - cd_start
|
||
e: BYTEPTR = BYTEPTR(stdlib.malloc(22))
|
||
if not e:
|
||
zipfile_set_error(ZIP_ERR_MEMORY, "out of memory")
|
||
return ZIP_ERR_MEMORY
|
||
string.memset(e, 0, 22)
|
||
zf_w32le(e, 0x06054B50)
|
||
zf_w16le(e + 8, t.CUInt32T(self.info_count))
|
||
zf_w16le(e + 10, t.CUInt32T(self.info_count))
|
||
zf_w32le(e + 12, t.CUInt32T(cd_size))
|
||
zf_w32le(e + 16, t.CUInt32T(cd_start))
|
||
rc = zf_write_all(self.fh, e, 22)
|
||
stdlib.free(e)
|
||
if rc != ZIP_ERR_OK:
|
||
zipfile_set_error(ZIP_ERR_IO, "write EOCD failed")
|
||
return rc
|
||
|
||
def close(self) -> t.CInt:
|
||
if self.closed:
|
||
return ZIP_ERR_OK
|
||
rc: t.CInt = ZIP_ERR_OK
|
||
if self.mode == 1 and self.fh is not None:
|
||
rc = self._finish_write()
|
||
wb.CloseHandle(self.fh)
|
||
self.fh = None
|
||
elif self.mode == 0:
|
||
if self.data is not None:
|
||
stdlib.free(self.data)
|
||
self.data = None
|
||
self.data_len = 0
|
||
self._clear_infos()
|
||
self.closed = 1
|
||
self.ok = 0
|
||
return rc
|