# -*- 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