"""

try:

if _EndRecData(fp):

return True # file has correct magic number

except OSError:

pass

"""Quickly see if a file is a ZIP file by checking the magic number.

result = False

try:

if hasattr(filename, "read"):

result = _check_zipfile(fp=filename)

else:

with open(filename, "rb") as fp:

result = _check_zipfile(fp)

except OSError:

pass

"""

try:

fpin.seek(offset - sizeEndCentDir64Locator, 2)

except OSError:

# If the seek fails, the file is not large enough to contain a ZIP64

# end-of-archive record, so just return the end record we were given.

return endrec

data = fpin.read(sizeEndCentDir64Locator)

if len(data) != sizeEndCentDir64Locator:

return endrec

sig, diskno, reloff, disks = struct.unpack(structEndArchive64Locator, data)

if sig != stringEndArchive64Locator:

return endrec

if diskno != 0 or disks != 1:

raise BadZipFile("zipfiles that span multiple disks are not supported")

# Assume no 'zip64 extensible data'

fpin.seek(offset - sizeEndCentDir64Locator - sizeEndCentDir64, 2)

data = fpin.read(sizeEndCentDir64)

if len(data) != sizeEndCentDir64:

return endrec

sig, sz, create_version, read_version, disk_num, disk_dir, \

dircount, dircount2, dirsize, diroffset = \

struct.unpack(structEndArchive64, data)

if sig != stringEndArchive64:

return endrec

# Update the original endrec using data from the ZIP64 record

endrec[_ECD_SIGNATURE] = sig

endrec[_ECD_DISK_NUMBER] = disk_num

endrec[_ECD_DISK_START] = disk_dir

endrec[_ECD_ENTRIES_THIS_DISK] = dircount

endrec[_ECD_ENTRIES_TOTAL] = dircount2

endrec[_ECD_SIZE] = dirsize

endrec[_ECD_OFFSET] = diroffset

"""Return data from the "End of Central Directory" record, or None.

# Determine file size

fpin.seek(0, 2)

filesize = fpin.tell()

# Check to see if this is ZIP file with no archive comment (the

# "end of central directory" structure should be the last item in the

# file if this is the case).

try:

fpin.seek(-sizeEndCentDir, 2)

except OSError:

return None

data = fpin.read()

if (len(data) == sizeEndCentDir and

data[0:4] == stringEndArchive and

data[-2:] == b"\000\000"):

# the signature is correct and there's no comment, unpack structure

endrec = struct.unpack(structEndArchive, data)

endrec=list(endrec)

# Append a blank comment and record start offset

endrec.append(b"")

endrec.append(filesize - sizeEndCentDir)

# Try to read the "Zip64 end of central directory" structure

return _EndRecData64(fpin, -sizeEndCentDir, endrec)

# Either this is not a ZIP file, or it is a ZIP file with an archive

# comment. Search the end of the file for the "end of central directory"

# record signature. The comment is the last item in the ZIP file and may be

# up to 64K long. It is assumed that the "end of central directory" magic

# number does not appear in the comment.

maxCommentStart = max(filesize - (1 << 16) - sizeEndCentDir, 0)

fpin.seek(maxCommentStart, 0)

data = fpin.read()

start = data.rfind(stringEndArchive)

if start >= 0:

# found the magic number; attempt to unpack and interpret

recData = data[start:start+sizeEndCentDir]

if len(recData) != sizeEndCentDir:

# Zip file is corrupted.

return None

endrec = list(struct.unpack(structEndArchive, recData))

commentSize = endrec[_ECD_COMMENT_SIZE] #as claimed by the zip file

comment = data[start+sizeEndCentDir:start+sizeEndCentDir+commentSize]

endrec.append(comment)

endrec.append(maxCommentStart + start)

# Try to read the "Zip64 end of central directory" structure

return _EndRecData64(fpin, maxCommentStart + start - filesize,

endrec)

# Unable to find a valid end of central directory structure

"""Class with attributes describing each file in the ZIP archive."""

__slots__ = (

'orig_filename',

'filename',

'date_time',

'compress_type',

'comment',

'extra',

'create_system',

'create_version',

'extract_version',

'reserved',

'flag_bits',

'volume',

'internal_attr',

'external_attr',

'header_offset',

'CRC',

'compress_size',

'file_size',

'_raw_time',

)

def __init__(self, filename="NoName", date_time=DEFAULT_DATE):

self.orig_filename = filename # Original file name in archive

# Terminate the file name at the first null byte. Null bytes in file

# names are used as tricks by viruses in archives.

null_byte = filename.find(chr(0))

if null_byte >= 0:

filename = filename[0:null_byte]

# This is used to ensure paths in generated ZIP files always use

# forward slashes as the directory separator, as required by the

# ZIP format specification.

self.filename = filename # Normalized file name

self.date_time = date_time # year, month, day, hour, min, sec

if date_time[0] < 1980:

raise ValueError('ZIP does not support timestamps before 1980')

# Standard values:

self.compress_type = ZIP_STORED # Type of compression for the file

self.comment = b"" # Comment for each file

self.extra = b"" # ZIP extra data

if sys.platform == 'win32':

self.create_system = 0 # System which created ZIP archive

else:

# Assume everything else is unix-y

self.create_system = 3 # System which created ZIP archive

self.create_version = DEFAULT_VERSION # Version which created ZIP archive

self.extract_version = DEFAULT_VERSION # Version needed to extract archive

self.reserved = 0 # Must be zero

self.flag_bits = 0 # ZIP flag bits

self.volume = 0 # Volume number of file header

self.internal_attr = 0 # Internal attributes

self.external_attr = 0 # External file attributes

# Other attributes are set by class ZipFile:

# header_offset Byte offset to the file header

# CRC CRC-32 of the uncompressed file

# compress_size Size of the compressed file

# file_size Size of the uncompressed file

def __repr__(self):

result = ['<%s filename=%r' % (self.__class__.__name__, self.filename)]

if self.compress_type != ZIP_STORED:

result.append(' compress_type=%s' %

compressor_names.get(self.compress_type,

self.compress_type))

hi = self.external_attr >> 16

lo = self.external_attr & 0xFFFF

if hi:

result.append(' filemode=%r' % stat.filemode(hi))

if lo:

result.append(' external_attr=%#x' % lo)

isdir = self.is_dir()

if not isdir or self.file_size:

result.append(' file_size=%r' % self.file_size)

if ((not isdir or self.compress_size) and

(self.compress_type != ZIP_STORED or

self.file_size != self.compress_size)):

result.append(' compress_size=%r' % self.compress_size)

result.append('>')

return ''.join(result)

def FileHeader(self, zip64=None):

"""Return the per-file header as a string."""

dt = self.date_time

dosdate = (dt[0] - 1980) << 9 | dt[1] << 5 | dt[2]

dostime = dt[3] << 11 | dt[4] << 5 | (dt[5] // 2)

if self.flag_bits & 0x08:

# Set these to zero because we write them after the file data

CRC = compress_size = file_size = 0

else:

CRC = self.CRC

compress_size = self.compress_size

file_size = self.file_size

extra = self.extra

min_version = 0

if zip64 is None:

zip64 = file_size > ZIP64_LIMIT or compress_size > ZIP64_LIMIT

if zip64:

fmt = '<HHQQ'

extra = extra + struct.pack(fmt,

1, struct.calcsize(fmt)-4, file_size, compress_size)

if file_size > ZIP64_LIMIT or compress_size > ZIP64_LIMIT:

if not zip64:

raise LargeZipFile("Filesize would require ZIP64 extensions")

# File is larger than what fits into a 4 byte integer,

# fall back to the ZIP64 extension

file_size = 0xffffffff

compress_size = 0xffffffff

min_version = ZIP64_VERSION

if self.compress_type == ZIP_BZIP2:

min_version = max(BZIP2_VERSION, min_version)

elif self.compress_type == ZIP_LZMA:

min_version = max(LZMA_VERSION, min_version)

self.extract_version = max(min_version, self.extract_version)

self.create_version = max(min_version, self.create_version)

filename, flag_bits = self._encodeFilenameFlags()

header = struct.pack(structFileHeader, stringFileHeader,

self.extract_version, self.reserved, flag_bits,

self.compress_type, dostime, dosdate, CRC,

compress_size, file_size,

len(filename), len(extra))

return header + filename + extra

def _encodeFilenameFlags(self):

try:

return self.filename.encode('ascii'), self.flag_bits

except UnicodeEncodeError:

return self.filename.encode('utf-8'), self.flag_bits | 0x800

def _decodeExtra(self):

# Try to decode the extra field.

extra = self.extra

unpack = struct.unpack

while len(extra) >= 4:

tp, ln = unpack('<HH', extra[:4])

if tp == 1:

if ln >= 24:

counts = unpack('<QQQ', extra[4:28])

elif ln == 16:

counts = unpack('<QQ', extra[4:20])

elif ln == 8:

counts = unpack('<Q', extra[4:12])

elif ln == 0:

counts = ()

else:

raise BadZipFile("Corrupt extra field %04x (size=%d)" % (tp, ln))

idx = 0

# ZIP64 extension (large files and/or large archives)

if self.file_size in (0xffffffffffffffff, 0xffffffff):

self.file_size = counts[idx]

idx += 1

if self.compress_size == 0xFFFFFFFF:

self.compress_size = counts[idx]

idx += 1

if self.header_offset == 0xffffffff:

old = self.header_offset

self.header_offset = counts[idx]

idx+=1

extra = extra[ln+4:]

@classmethod

def from_file(cls, filename, arcname=None, date_time=DEFAULT_DATE):

"""Construct an appropriate ZipInfo for a file on the filesystem.

if isinstance(filename, os.PathLike):

filename = os.fspath(filename)

st = os.stat(filename)

isdir = stat.S_ISDIR(st.st_mode)

# Create ZipInfo instance to store file information

if arcname is None:

arcname = filename

arcname = os.path.normpath(os.path.splitdrive(arcname)[1])

while arcname[0] in (os.sep, os.altsep):

arcname = arcname[1:]

if isdir:

arcname += '/'

zinfo = cls(arcname, date_time)

zinfo.external_attr = 25165824 # Unix attributes, hard-codeds

if isdir:

zinfo.file_size = 0

zinfo.external_attr |= 0x10 # MS-DOS directory flag

else:

zinfo.file_size = st.st_size

return zinfo

def is_dir(self):

"""Return True if this archive member is a directory."""

"""Class to handle decryption of files stored within a ZIP archive.

def _GenerateCRCTable():

"""Generate a CRC-32 table.

poly = 0xedb88320

table = [0] * 256

for i in range(256):

crc = i

for j in range(8):

if crc & 1:

crc = ((crc >> 1) & 0x7FFFFFFF) ^ poly

else:

crc = ((crc >> 1) & 0x7FFFFFFF)

table[i] = crc

return table

crctable = None

def _crc32(self, ch, crc):

"""Compute the CRC32 primitive on one byte."""

return ((crc >> 8) & 0xffffff) ^ self.crctable[(crc ^ ch) & 0xff]

def __init__(self, pwd):

if _ZipDecrypter.crctable is None:

_ZipDecrypter.crctable = _ZipDecrypter._GenerateCRCTable()

self.key0 = 305419896

self.key1 = 591751049

self.key2 = 878082192

for p in pwd:

self._UpdateKeys(p)

def _UpdateKeys(self, c):

self.key0 = self._crc32(c, self.key0)

self.key1 = (self.key1 + (self.key0 & 255)) & 4294967295

self.key1 = (self.key1 * 134775813 + 1) & 4294967295

self.key2 = self._crc32((self.key1 >> 24) & 255, self.key2)

def __call__(self, c):

"""Decrypt a single character."""

assert isinstance(c, int)

k = self.key2 | 2

c = c ^ (((k * (k^1)) >> 8) & 255)

self._UpdateKeys(c)

def __init__(self):

self._comp = None

def _init(self):

props = lzma._encode_filter_properties({'id': lzma.FILTER_LZMA1})

self._comp = lzma.LZMACompressor(lzma.FORMAT_RAW, filters=[

lzma._decode_filter_properties(lzma.FILTER_LZMA1, props)

])

return struct.pack('<BBH', 9, 4, len(props)) + props

def compress(self, data):

if self._comp is None:

return self._init() + self._comp.compress(data)

return self._comp.compress(data)

def flush(self):

if self._comp is None:

return self._init() + self._comp.flush()

def __init__(self):

self._decomp = None

self._unconsumed = b''

self.eof = False

def decompress(self, data):

if self._decomp is None:

self._unconsumed += data

if len(self._unconsumed) <= 4:

return b''

psize, = struct.unpack('<H', self._unconsumed[2:4])

if len(self._unconsumed) <= 4 + psize:

return b''

self._decomp = lzma.LZMADecompressor(lzma.FORMAT_RAW, filters=[

lzma._decode_filter_properties(lzma.FILTER_LZMA1,

self._unconsumed[4:4 + psize])

])

data = self._unconsumed[4 + psize:]

del self._unconsumed

result = self._decomp.decompress(data)

self.eof = self._decomp.eof

0: 'store',

1: 'shrink',

2: 'reduce',

3: 'reduce',

4: 'reduce',

5: 'reduce',

6: 'implode',

7: 'tokenize',

8: 'deflate',

9: 'deflate64',

10: 'implode',

12: 'bzip2',

14: 'lzma',

18: 'terse',

19: 'lz77',

97: 'wavpack',

98: 'ppmd',

if compression == ZIP_STORED:

pass

elif compression == ZIP_DEFLATED:

if not zlib:

raise RuntimeError(

"Compression requires the (missing) zlib module")

elif compression == ZIP_BZIP2:

if not bz2:

raise RuntimeError(

"Compression requires the (missing) bz2 module")

elif compression == ZIP_LZMA:

if not lzma:

raise RuntimeError(

"Compression requires the (missing) lzma module")

else:

if compress_type == ZIP_DEFLATED:

return zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION,

zlib.DEFLATED, -15)

elif compress_type == ZIP_BZIP2:

return bz2.BZ2Compressor()

elif compress_type == ZIP_LZMA:

return LZMACompressor()

else:

if compress_type == ZIP_STORED:

return None

elif compress_type == ZIP_DEFLATED:

return zlib.decompressobj(-15)

elif compress_type == ZIP_BZIP2:

return bz2.BZ2Decompressor()

elif compress_type == ZIP_LZMA:

return LZMADecompressor()

else:

descr = compressor_names.get(compress_type)

if descr:

raise NotImplementedError("compression type %d (%s)" % (compress_type, descr))

else:

def __init__(self, file, pos, close, lock, writing):

self._file = file

self._pos = pos

self._close = close

self._lock = lock

self._writing = writing

def read(self, n=-1):

with self._lock:

if self._writing():

raise ValueError("Can't read from the ZIP file while there "

"is an open writing handle on it. "

"Close the writing handle before trying to read.")

self._file.seek(self._pos)

data = self._file.read(n)

self._pos = self._file.tell()

return data

def close(self):

if self._file is not None:

fileobj = self._file

self._file = None

def __init__(self, fp):

self.fp = fp

self.offset = 0

def write(self, data):

n = self.fp.write(data)

self.offset += n

return n

def tell(self):

return self.offset

def flush(self):

self.fp.flush()

def close(self):

"""File-like object for reading an archive member.

# Max size supported by decompressor.

MAX_N = 1 << 31 - 1

# Read from compressed files in 4k blocks.

MIN_READ_SIZE = 4096

def __init__(self, fileobj, mode, zipinfo, decrypter=None,

close_fileobj=False):

self._fileobj = fileobj

self._decrypter = decrypter

self._close_fileobj = close_fileobj

self._compress_type = zipinfo.compress_type

self._compress_left = zipinfo.compress_size

self._left = zipinfo.file_size

self._decompressor = _get_decompressor(self._compress_type)

self._eof = False

self._readbuffer = b''

self._offset = 0

self.newlines = None

# Adjust read size for encrypted files since the first 12 bytes

# are for the encryption/password information.

if self._decrypter is not None:

self._compress_left -= 12

self.mode = mode

self.name = zipinfo.filename

if hasattr(zipinfo, 'CRC'):

self._expected_crc = zipinfo.CRC

self._running_crc = crc32(b'')

else:

self._expected_crc = None

def __repr__(self):

result = ['<%s.%s' % (self.__class__.__module__,

self.__class__.__qualname__)]

if not self.closed:

result.append(' name=%r mode=%r' % (self.name, self.mode))

if self._compress_type != ZIP_STORED:

result.append(' compress_type=%s' %

compressor_names.get(self._compress_type,

self._compress_type))

else:

result.append(' [closed]')

result.append('>')

return ''.join(result)

def readline(self, limit=-1):

"""Read and return a line from the stream.

if limit < 0:

# Shortcut common case - newline found in buffer.

i = self._readbuffer.find(b'\n', self._offset) + 1

if i > 0:

line = self._readbuffer[self._offset: i]

self._offset = i

return line

return io.BufferedIOBase.readline(self, limit)

def peek(self, n=1):

"""Returns buffered bytes without advancing the position."""

if n > len(self._readbuffer) - self._offset:

chunk = self.read(n)

if len(chunk) > self._offset:

self._readbuffer = chunk + self._readbuffer[self._offset:]

self._offset = 0

else:

self._offset -= len(chunk)

# Return up to 512 bytes to reduce allocation overhead for tight loops.

return self._readbuffer[self._offset: self._offset + 512]

def readable(self):

return True

def read(self, n=-1):

"""Read and return up to n bytes.

if n is None or n < 0:

buf = self._readbuffer[self._offset:]

self._readbuffer = b''

self._offset = 0

while not self._eof:

buf += self._read1(self.MAX_N)

return buf

end = n + self._offset

if end < len(self._readbuffer):

buf = self._readbuffer[self._offset:end]

self._offset = end

return buf

n = end - len(self._readbuffer)

buf = self._readbuffer[self._offset:]

self._readbuffer = b''

self._offset = 0

while n > 0 and not self._eof:

data = self._read1(n)

if n < len(data):

self._readbuffer = data

self._offset = n

buf += data[:n]

break

buf += data

n -= len(data)

return buf

def _update_crc(self, newdata):

# Update the CRC using the given data.

if self._expected_crc is None:

# No need to compute the CRC if we don't have a reference value

return

self._running_crc = crc32(newdata, self._running_crc)

# Check the CRC if we're at the end of the file

if self._eof and self._running_crc != self._expected_crc:

raise BadZipFile("Bad CRC-32 for file %r" % self.name)

def read1(self, n):

"""Read up to n bytes with at most one read() system call."""

if n is None or n < 0:

buf = self._readbuffer[self._offset:]

self._readbuffer = b''

self._offset = 0

while not self._eof:

data = self._read1(self.MAX_N)

if data:

buf += data

break

return buf

end = n + self._offset

if end < len(self._readbuffer):

buf = self._readbuffer[self._offset:end]

self._offset = end

return buf

n = end - len(self._readbuffer)

buf = self._readbuffer[self._offset:]

self._readbuffer = b''

self._offset = 0

if n > 0:

while not self._eof:

data = self._read1(n)

if n < len(data):

self._readbuffer = data

self._offset = n

buf += data[:n]

break

if data:

buf += data

break

return buf

def _read1(self, n):

# Read up to n compressed bytes with at most one read() system call,

# decrypt and decompress them.

if self._eof or n <= 0:

return b''

# Read from file.

if self._compress_type == ZIP_DEFLATED:

## Handle unconsumed data.

data = self._decompressor.unconsumed_tail

if n > len(data):

data += self._read2(n - len(data))

else:

data = self._read2(n)

if self._compress_type == ZIP_STORED:

self._eof = self._compress_left <= 0

elif self._compress_type == ZIP_DEFLATED:

n = max(n, self.MIN_READ_SIZE)

data = self._decompressor.decompress(data, n)

self._eof = (self._decompressor.eof or

self._compress_left <= 0 and

not self._decompressor.unconsumed_tail)

if self._eof:

data += self._decompressor.flush()

else:

data = self._decompressor.decompress(data)

self._eof = self._decompressor.eof or self._compress_left <= 0

data = data[:self._left]

self._left -= len(data)

if self._left <= 0:

self._eof = True

self._update_crc(data)

return data

def _read2(self, n):

if self._compress_left <= 0:

return b''

n = max(n, self.MIN_READ_SIZE)

n = min(n, self._compress_left)

data = self._fileobj.read(n)

self._compress_left -= len(data)

if not data:

raise EOFError

if self._decrypter is not None:

data = bytes(map(self._decrypter, data))

return data

def close(self):

try:

if self._close_fileobj:

self._fileobj.close()

finally:

def __init__(self, zf, zinfo, zip64):

self._zinfo = zinfo

self._zip64 = zip64

self._zipfile = zf

self._compressor = _get_compressor(zinfo.compress_type)

self._file_size = 0

self._compress_size = 0

self._crc = 0

@property

def _fileobj(self):

return self._zipfile.fp

def writable(self):

return True

def write(self, data):

if self.closed:

raise ValueError('I/O operation on closed file.')

nbytes = len(data)

self._file_size += nbytes

self._crc = crc32(data, self._crc)

if self._compressor:

data = self._compressor.compress(data)

self._compress_size += len(data)

self._fileobj.write(data)

return nbytes

def close(self):

if self.closed:

return

super().close()

# Flush any data from the compressor, and update header info