diff --git a/Lib/functools.py b/Lib/functools.py index c8b79c2a7..b41dea790 100644 --- a/Lib/functools.py +++ b/Lib/functools.py @@ -11,12 +11,8 @@ __all__ = ['update_wrapper', 'wraps', 'WRAPPER_ASSIGNMENTS', 'WRAPPER_UPDATES', 'total_ordering', 'cmp_to_key', 'lru_cache', 'reduce', 'partial', - 'partialmethod', 'singledispatch'] + 'partialmethod', 'singledispatch', 'singledispatchmethod'] -try: - from _functools import reduce -except ImportError: - pass from abc import get_cache_token from collections import namedtuple # import types, weakref # Deferred to single_dispatch() @@ -226,6 +222,45 @@ except ImportError: pass +################################################################################ +### reduce() sequence to a single item +################################################################################ + +_initial_missing = object() + +def reduce(function, sequence, initial=_initial_missing): + """ + reduce(function, sequence[, initial]) -> value + + Apply a function of two arguments cumulatively to the items of a sequence, + from left to right, so as to reduce the sequence to a single value. + For example, reduce(lambda x, y: x+y, [1, 2, 3, 4, 5]) calculates + ((((1+2)+3)+4)+5). If initial is present, it is placed before the items + of the sequence in the calculation, and serves as a default when the + sequence is empty. + """ + + it = iter(sequence) + + if initial is _initial_missing: + try: + value = next(it) + except StopIteration: + raise TypeError("reduce() of empty sequence with no initial value") from None + else: + value = initial + + for element in it: + value = function(value, element) + + return value + +try: + from _functools import reduce +except ImportError: + pass + + ################################################################################ ### partial() argument application ################################################################################ @@ -238,22 +273,13 @@ class partial: __slots__ = "func", "args", "keywords", "__dict__", "__weakref__" - def __new__(*args, **keywords): - if not args: - raise TypeError("descriptor '__new__' of partial needs an argument") - if len(args) < 2: - raise TypeError("type 'partial' takes at least one argument") - cls, func, *args = args + def __new__(cls, func, /, *args, **keywords): if not callable(func): raise TypeError("the first argument must be callable") - args = tuple(args) if hasattr(func, "func"): args = func.args + args - tmpkw = func.keywords.copy() - tmpkw.update(keywords) - keywords = tmpkw - del tmpkw + keywords = {**func.keywords, **keywords} func = func.func self = super(partial, cls).__new__(cls) @@ -263,13 +289,9 @@ class partial: self.keywords = keywords return self - def __call__(*args, **keywords): - if not args: - raise TypeError("descriptor '__call__' of partial needs an argument") - self, *args = args - newkeywords = self.keywords.copy() - newkeywords.update(keywords) - return self.func(*self.args, *args, **newkeywords) + def __call__(self, /, *args, **keywords): + keywords = {**self.keywords, **keywords} + return self.func(*self.args, *args, **keywords) @recursive_repr() def __repr__(self): @@ -323,7 +345,23 @@ class partialmethod(object): callables as instance methods. """ - def __init__(self, func, *args, **keywords): + def __init__(*args, **keywords): + if len(args) >= 2: + self, func, *args = args + elif not args: + raise TypeError("descriptor '__init__' of partialmethod " + "needs an argument") + elif 'func' in keywords: + func = keywords.pop('func') + self, *args = args + import warnings + warnings.warn("Passing 'func' as keyword argument is deprecated", + DeprecationWarning, stacklevel=2) + else: + raise TypeError("type 'partialmethod' takes at least one argument, " + "got %d" % (len(args)-1)) + args = tuple(args) + if not callable(func) and not hasattr(func, "__get__"): raise TypeError("{!r} is not callable or a descriptor" .format(func)) @@ -336,12 +374,12 @@ class partialmethod(object): # it's also more efficient since only one function will be called self.func = func.func self.args = func.args + args - self.keywords = func.keywords.copy() - self.keywords.update(keywords) + self.keywords = {**func.keywords, **keywords} else: self.func = func self.args = args self.keywords = keywords + __init__.__text_signature__ = '($self, func, /, *args, **keywords)' def __repr__(self): args = ", ".join(map(repr, self.args)) @@ -355,17 +393,14 @@ class partialmethod(object): keywords=keywords) def _make_unbound_method(self): - def _method(*args, **keywords): - call_keywords = self.keywords.copy() - call_keywords.update(keywords) - cls_or_self, *rest = args - call_args = (cls_or_self,) + self.args + tuple(rest) - return self.func(*call_args, **call_keywords) + def _method(cls_or_self, /, *args, **keywords): + keywords = {**self.keywords, **keywords} + return self.func(cls_or_self, *self.args, *args, **keywords) _method.__isabstractmethod__ = self.__isabstractmethod__ _method._partialmethod = self return _method - def __get__(self, obj, cls): + def __get__(self, obj, cls=None): get = getattr(self.func, "__get__", None) result = None if get is not None: @@ -388,6 +423,12 @@ class partialmethod(object): def __isabstractmethod__(self): return getattr(self.func, "__isabstractmethod__", False) +# Helper functions + +def _unwrap_partial(func): + while isinstance(func, partial): + func = func.func + return func ################################################################################ ### LRU Cache function decorator @@ -413,7 +454,7 @@ class _HashedSeq(list): def _make_key(args, kwds, typed, kwd_mark = (object(),), - fasttypes = {int, str, frozenset, type(None)}, + fasttypes = {int, str}, tuple=tuple, type=type, len=len): """Make a cache key from optionally typed positional and keyword arguments @@ -458,7 +499,7 @@ def lru_cache(maxsize=128, typed=False): with f.cache_info(). Clear the cache and statistics with f.cache_clear(). Access the underlying function with f.__wrapped__. - See: http://en.wikipedia.org/wiki/Cache_algorithms#Least_Recently_Used + See: http://en.wikipedia.org/wiki/Cache_replacement_policies#Least_recently_used_(LRU) """ @@ -467,11 +508,18 @@ def lru_cache(maxsize=128, typed=False): # The internals of the lru_cache are encapsulated for thread safety and # to allow the implementation to change (including a possible C version). - # Early detection of an erroneous call to @lru_cache without any arguments - # resulting in the inner function being passed to maxsize instead of an - # integer or None. - if maxsize is not None and not isinstance(maxsize, int): - raise TypeError('Expected maxsize to be an integer or None') + if isinstance(maxsize, int): + # Negative maxsize is treated as 0 + if maxsize < 0: + maxsize = 0 + elif callable(maxsize) and isinstance(typed, bool): + # The user_function was passed in directly via the maxsize argument + user_function, maxsize = maxsize, 128 + wrapper = _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo) + return update_wrapper(wrapper, user_function) + elif maxsize is not None: + raise TypeError( + 'Expected first argument to be an integer, a callable, or None') def decorating_function(user_function): wrapper = _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo) @@ -497,10 +545,10 @@ def _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo): if maxsize == 0: def wrapper(*args, **kwds): - # No caching -- just a statistics update after a successful call + # No caching -- just a statistics update nonlocal misses - result = user_function(*args, **kwds) misses += 1 + result = user_function(*args, **kwds) return result elif maxsize is None: @@ -513,9 +561,9 @@ def _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo): if result is not sentinel: hits += 1 return result + misses += 1 result = user_function(*args, **kwds) cache[key] = result - misses += 1 return result else: @@ -537,6 +585,7 @@ def _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo): link[NEXT] = root hits += 1 return result + misses += 1 result = user_function(*args, **kwds) with lock: if key in cache: @@ -574,7 +623,6 @@ def _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo): # Use the cache_len bound method instead of the len() function # which could potentially be wrapped in an lru_cache itself. full = (cache_len() >= maxsize) - misses += 1 return result def cache_info(): @@ -807,9 +855,11 @@ def singledispatch(func): # only import typing if annotation parsing is necessary from typing import get_type_hints argname, cls = next(iter(get_type_hints(func).items())) - assert isinstance(cls, type), ( - f"Invalid annotation for {argname!r}. {cls!r} is not a class." - ) + if not isinstance(cls, type): + raise TypeError( + f"Invalid annotation for {argname!r}. " + f"{cls!r} is not a class." + ) registry[cls] = func if cache_token is None and hasattr(cls, '__abstractmethods__'): cache_token = get_cache_token() @@ -817,8 +867,13 @@ def singledispatch(func): return func def wrapper(*args, **kw): + if not args: + raise TypeError(f'{funcname} requires at least ' + '1 positional argument') + return dispatch(args[0].__class__)(*args, **kw) + funcname = getattr(func, '__name__', 'singledispatch function') registry[object] = func wrapper.register = register wrapper.dispatch = dispatch @@ -826,3 +881,95 @@ def singledispatch(func): wrapper._clear_cache = dispatch_cache.clear update_wrapper(wrapper, func) return wrapper + + +# Descriptor version +class singledispatchmethod: + """Single-dispatch generic method descriptor. + + Supports wrapping existing descriptors and handles non-descriptor + callables as instance methods. + """ + + def __init__(self, func): + if not callable(func) and not hasattr(func, "__get__"): + raise TypeError(f"{func!r} is not callable or a descriptor") + + self.dispatcher = singledispatch(func) + self.func = func + + def register(self, cls, method=None): + """generic_method.register(cls, func) -> func + + Registers a new implementation for the given *cls* on a *generic_method*. + """ + return self.dispatcher.register(cls, func=method) + + def __get__(self, obj, cls=None): + def _method(*args, **kwargs): + method = self.dispatcher.dispatch(args[0].__class__) + return method.__get__(obj, cls)(*args, **kwargs) + + _method.__isabstractmethod__ = self.__isabstractmethod__ + _method.register = self.register + update_wrapper(_method, self.func) + return _method + + @property + def __isabstractmethod__(self): + return getattr(self.func, '__isabstractmethod__', False) + + +################################################################################ +### cached_property() - computed once per instance, cached as attribute +################################################################################ + +_NOT_FOUND = object() + + +class cached_property: + def __init__(self, func): + self.func = func + self.attrname = None + self.__doc__ = func.__doc__ + self.lock = RLock() + + def __set_name__(self, owner, name): + if self.attrname is None: + self.attrname = name + elif name != self.attrname: + raise TypeError( + "Cannot assign the same cached_property to two different names " + f"({self.attrname!r} and {name!r})." + ) + + def __get__(self, instance, owner=None): + if instance is None: + return self + if self.attrname is None: + raise TypeError( + "Cannot use cached_property instance without calling __set_name__ on it.") + try: + cache = instance.__dict__ + except AttributeError: # not all objects have __dict__ (e.g. class defines slots) + msg = ( + f"No '__dict__' attribute on {type(instance).__name__!r} " + f"instance to cache {self.attrname!r} property." + ) + raise TypeError(msg) from None + val = cache.get(self.attrname, _NOT_FOUND) + if val is _NOT_FOUND: + with self.lock: + # check if another thread filled cache while we awaited lock + val = cache.get(self.attrname, _NOT_FOUND) + if val is _NOT_FOUND: + val = self.func(instance) + try: + cache[self.attrname] = val + except TypeError: + msg = ( + f"The '__dict__' attribute on {type(instance).__name__!r} instance " + f"does not support item assignment for caching {self.attrname!r} property." + ) + raise TypeError(msg) from None + return val diff --git a/Lib/inspect.py b/Lib/inspect.py index e08e9f578..3ff395ca3 100644 --- a/Lib/inspect.py +++ b/Lib/inspect.py @@ -18,7 +18,7 @@ Here are some of the useful functions provided by this module: getargvalues(), getcallargs() - get info about function arguments getfullargspec() - same, with support for Python 3 features - formatargspec(), formatargvalues() - format an argument spec + formatargvalues() - format an argument spec getouterframes(), getinnerframes() - get info about frames currentframe() - get the current stack frame stack(), trace() - get info about frames on the stack or in a traceback @@ -31,7 +31,7 @@ Here are some of the useful functions provided by this module: __author__ = ('Ka-Ping Yee ', 'Yury Selivanov ') -import ast +import abc import dis import collections.abc import enum @@ -110,7 +110,7 @@ def ismethoddescriptor(object): def isdatadescriptor(object): """Return true if the object is a data descriptor. - Data descriptors have both a __get__ and a __set__ attribute. Examples are + Data descriptors have a __set__ or a __delete__ attribute. Examples are properties (defined in Python) and getsets and members (defined in C). Typically, data descriptors will also have __name__ and __doc__ attributes (properties, getsets, and members have both of these attributes), but this @@ -119,7 +119,7 @@ def isdatadescriptor(object): # mutual exclusion return False tp = type(object) - return hasattr(tp, "__set__") and hasattr(tp, "__get__") + return hasattr(tp, "__set__") or hasattr(tp, "__delete__") if hasattr(types, 'MemberDescriptorType'): # CPython and equivalent @@ -168,30 +168,38 @@ def isfunction(object): __kwdefaults__ dict of keyword only parameters with defaults""" return isinstance(object, types.FunctionType) -def isgeneratorfunction(object): +def _has_code_flag(f, flag): + """Return true if ``f`` is a function (or a method or functools.partial + wrapper wrapping a function) whose code object has the given ``flag`` + set in its flags.""" + while ismethod(f): + f = f.__func__ + f = functools._unwrap_partial(f) + if not isfunction(f): + return False + return bool(f.__code__.co_flags & flag) + +def isgeneratorfunction(obj): """Return true if the object is a user-defined generator function. Generator function objects provide the same attributes as functions. See help(isfunction) for a list of attributes.""" - return bool((isfunction(object) or ismethod(object)) and - object.__code__.co_flags & CO_GENERATOR) + return _has_code_flag(obj, CO_GENERATOR) -def iscoroutinefunction(object): +def iscoroutinefunction(obj): """Return true if the object is a coroutine function. Coroutine functions are defined with "async def" syntax. """ - return bool((isfunction(object) or ismethod(object)) and - object.__code__.co_flags & CO_COROUTINE) + return _has_code_flag(obj, CO_COROUTINE) -def isasyncgenfunction(object): +def isasyncgenfunction(obj): """Return true if the object is an asynchronous generator function. Asynchronous generator functions are defined with "async def" syntax and have "yield" expressions in their body. """ - return bool((isfunction(object) or ismethod(object)) and - object.__code__.co_flags & CO_ASYNC_GENERATOR) + return _has_code_flag(obj, CO_ASYNC_GENERATOR) def isasyncgen(object): """Return true if the object is an asynchronous generator.""" @@ -253,18 +261,25 @@ def iscode(object): """Return true if the object is a code object. Code objects provide these attributes: - co_argcount number of arguments (not including * or ** args) - co_code string of raw compiled bytecode - co_consts tuple of constants used in the bytecode - co_filename name of file in which this code object was created - co_firstlineno number of first line in Python source code - co_flags bitmap: 1=optimized | 2=newlocals | 4=*arg | 8=**arg - co_lnotab encoded mapping of line numbers to bytecode indices - co_name name with which this code object was defined - co_names tuple of names of local variables - co_nlocals number of local variables - co_stacksize virtual machine stack space required - co_varnames tuple of names of arguments and local variables""" + co_argcount number of arguments (not including *, ** args + or keyword only arguments) + co_code string of raw compiled bytecode + co_cellvars tuple of names of cell variables + co_consts tuple of constants used in the bytecode + co_filename name of file in which this code object was created + co_firstlineno number of first line in Python source code + co_flags bitmap: 1=optimized | 2=newlocals | 4=*arg | 8=**arg + | 16=nested | 32=generator | 64=nofree | 128=coroutine + | 256=iterable_coroutine | 512=async_generator + co_freevars tuple of names of free variables + co_posonlyargcount number of positional only arguments + co_kwonlyargcount number of keyword only arguments (not including ** arg) + co_lnotab encoded mapping of line numbers to bytecode indices + co_name name with which this code object was defined + co_names tuple of names of local variables + co_nlocals number of local variables + co_stacksize virtual machine stack space required + co_varnames tuple of names of arguments and local variables""" return isinstance(object, types.CodeType) def isbuiltin(object): @@ -285,7 +300,27 @@ def isroutine(object): def isabstract(object): """Return true if the object is an abstract base class (ABC).""" - return bool(isinstance(object, type) and object.__flags__ & TPFLAGS_IS_ABSTRACT) + if not isinstance(object, type): + return False + if object.__flags__ & TPFLAGS_IS_ABSTRACT: + return True + if not issubclass(type(object), abc.ABCMeta): + return False + if hasattr(object, '__abstractmethods__'): + # It looks like ABCMeta.__new__ has finished running; + # TPFLAGS_IS_ABSTRACT should have been accurate. + return False + # It looks like ABCMeta.__new__ has not finished running yet; we're + # probably in __init_subclass__. We'll look for abstractmethods manually. + for name, value in object.__dict__.items(): + if getattr(value, "__isabstractmethod__", False): + return True + for base in object.__bases__: + for name in getattr(base, "__abstractmethods__", ()): + value = getattr(object, name, None) + if getattr(value, "__isabstractmethod__", False): + return True + return False def getmembers(object, predicate=None): """Return all members of an object as (name, value) pairs sorted by name. @@ -362,7 +397,7 @@ def classify_class_attrs(cls): mro = getmro(cls) metamro = getmro(type(cls)) # for attributes stored in the metaclass - metamro = tuple([cls for cls in metamro if cls not in (type, object)]) + metamro = tuple(cls for cls in metamro if cls not in (type, object)) class_bases = (cls,) + mro all_bases = class_bases + metamro names = dir(cls) @@ -430,10 +465,10 @@ def classify_class_attrs(cls): continue obj = get_obj if get_obj is not None else dict_obj # Classify the object or its descriptor. - if isinstance(dict_obj, staticmethod): + if isinstance(dict_obj, (staticmethod, types.BuiltinMethodType)): kind = "static method" obj = dict_obj - elif isinstance(dict_obj, classmethod): + elif isinstance(dict_obj, (classmethod, types.ClassMethodDescriptorType)): kind = "class method" obj = dict_obj elif isinstance(dict_obj, property): @@ -478,13 +513,16 @@ def unwrap(func, *, stop=None): def _is_wrapper(f): return hasattr(f, '__wrapped__') and not stop(f) f = func # remember the original func for error reporting - memo = {id(f)} # Memoise by id to tolerate non-hashable objects + # Memoise by id to tolerate non-hashable objects, but store objects to + # ensure they aren't destroyed, which would allow their IDs to be reused. + memo = {id(f): f} + recursion_limit = sys.getrecursionlimit() while _is_wrapper(func): func = func.__wrapped__ id_func = id(func) - if id_func in memo: + if (id_func in memo) or (len(memo) >= recursion_limit): raise ValueError('wrapper loop when unwrapping {!r}'.format(f)) - memo.add(id_func) + memo[id_func] = func return func # -------------------------------------------------- source code extraction @@ -550,9 +588,12 @@ def _finddoc(obj): cls = obj.__objclass__ if getattr(cls, name) is not obj: return None + if ismemberdescriptor(obj): + slots = getattr(cls, '__slots__', None) + if isinstance(slots, dict) and name in slots: + return slots[name] else: return None - for base in cls.__mro__: try: doc = getattr(base, name).__doc__ @@ -613,14 +654,14 @@ def cleandoc(doc): def getfile(object): """Work out which source or compiled file an object was defined in.""" if ismodule(object): - if hasattr(object, '__file__'): + if getattr(object, '__file__', None): return object.__file__ raise TypeError('{!r} is a built-in module'.format(object)) if isclass(object): if hasattr(object, '__module__'): - object = sys.modules.get(object.__module__) - if hasattr(object, '__file__'): - return object.__file__ + module = sys.modules.get(object.__module__) + if getattr(module, '__file__', None): + return module.__file__ raise TypeError('{!r} is a built-in class'.format(object)) if ismethod(object): object = object.__func__ @@ -632,8 +673,9 @@ def getfile(object): object = object.f_code if iscode(object): return object.co_filename - raise TypeError('{!r} is not a module, class, method, ' - 'function, traceback, frame, or code object'.format(object)) + raise TypeError('module, class, method, function, traceback, frame, or ' + 'code object was expected, got {}'.format( + type(object).__name__)) def getmodulename(path): """Return the module name for a given file, or None.""" @@ -924,7 +966,12 @@ def getsourcelines(object): object = unwrap(object) lines, lnum = findsource(object) - if ismodule(object): + if istraceback(object): + object = object.tb_frame + + # for module or frame that corresponds to module, return all source lines + if (ismodule(object) or + (isframe(object) and object.f_code.co_name == "")): return lines, 0 else: return getblock(lines[lnum:]), lnum + 1 @@ -963,7 +1010,7 @@ def getclasstree(classes, unique=False): for c in classes: if c.__bases__: for parent in c.__bases__: - if not parent in children: + if parent not in children: children[parent] = [] if c not in children[parent]: children[parent].append(c) @@ -985,21 +1032,11 @@ def getargs(co): 'args' is the list of argument names. Keyword-only arguments are appended. 'varargs' and 'varkw' are the names of the * and ** arguments or None.""" - args, varargs, kwonlyargs, varkw = _getfullargs(co) - return Arguments(args + kwonlyargs, varargs, varkw) - -def _getfullargs(co): - """Get information about the arguments accepted by a code object. - - Four things are returned: (args, varargs, kwonlyargs, varkw), where - 'args' and 'kwonlyargs' are lists of argument names, and 'varargs' - and 'varkw' are the names of the * and ** arguments or None.""" - if not iscode(co): raise TypeError('{!r} is not a code object'.format(co)) - nargs = co.co_argcount names = co.co_varnames + nargs = co.co_argcount nkwargs = co.co_kwonlyargcount args = list(names[:nargs]) kwonlyargs = list(names[nargs:nargs+nkwargs]) @@ -1013,8 +1050,7 @@ def _getfullargs(co): varkw = None if co.co_flags & CO_VARKEYWORDS: varkw = co.co_varnames[nargs] - return args, varargs, kwonlyargs, varkw - + return Arguments(args + kwonlyargs, varargs, varkw) ArgSpec = namedtuple('ArgSpec', 'args varargs keywords defaults') @@ -1035,15 +1071,17 @@ def getargspec(func): Alternatively, use getfullargspec() for an API with a similar namedtuple based interface, but full support for annotations and keyword-only parameters. + + Deprecated since Python 3.5, use `inspect.getfullargspec()`. """ - warnings.warn("inspect.getargspec() is deprecated, " + warnings.warn("inspect.getargspec() is deprecated since Python 3.0, " "use inspect.signature() or inspect.getfullargspec()", DeprecationWarning, stacklevel=2) args, varargs, varkw, defaults, kwonlyargs, kwonlydefaults, ann = \ getfullargspec(func) if kwonlyargs or ann: raise ValueError("Function has keyword-only parameters or annotations" - ", use getfullargspec() API which can support them") + ", use inspect.signature() API which can support them") return ArgSpec(args, varargs, varkw, defaults) FullArgSpec = namedtuple('FullArgSpec', @@ -1065,7 +1103,6 @@ def getfullargspec(func): - the "self" parameter is always reported, even for bound methods - wrapper chains defined by __wrapped__ *not* unwrapped automatically """ - try: # Re: `skip_bound_arg=False` # @@ -1097,6 +1134,7 @@ def getfullargspec(func): args = [] varargs = None varkw = None + posonlyargs = [] kwonlyargs = [] defaults = () annotations = {} @@ -1111,7 +1149,9 @@ def getfullargspec(func): name = param.name if kind is _POSITIONAL_ONLY: - args.append(name) + posonlyargs.append(name) + if param.default is not param.empty: + defaults += (param.default,) elif kind is _POSITIONAL_OR_KEYWORD: args.append(name) if param.default is not param.empty: @@ -1136,7 +1176,7 @@ def getfullargspec(func): # compatibility with 'func.__defaults__' defaults = None - return FullArgSpec(args, varargs, varkw, defaults, + return FullArgSpec(posonlyargs + args, varargs, varkw, defaults, kwonlyargs, kwdefaults, annotations) @@ -1181,7 +1221,19 @@ def formatargspec(args, varargs=None, varkw=None, defaults=None, kwonlyargs, kwonlydefaults, annotations). The other five arguments are the corresponding optional formatting functions that are called to turn names and values into strings. The last argument is an optional - function to format the sequence of arguments.""" + function to format the sequence of arguments. + + Deprecated since Python 3.5: use the `signature` function and `Signature` + objects. + """ + + from warnings import warn + + warn("`formatargspec` is deprecated since Python 3.5. Use `signature` and " + "the `Signature` object directly", + DeprecationWarning, + stacklevel=2) + def formatargandannotation(arg): result = formatarg(arg) if arg in annotations: @@ -1273,14 +1325,12 @@ def _too_many(f_name, args, kwonly, varargs, defcount, given, values): (f_name, sig, "s" if plural else "", given, kwonly_sig, "was" if given == 1 and not kwonly_given else "were")) -def getcallargs(*func_and_positional, **named): +def getcallargs(func, /, *positional, **named): """Get the mapping of arguments to values. A dict is returned, with keys the function argument names (including the names of the * and ** arguments, if any), and values the respective bound values from 'positional' and 'named'.""" - func = func_and_positional[0] - positional = func_and_positional[1:] spec = getfullargspec(func) args, varargs, varkw, defaults, kwonlyargs, kwonlydefaults, ann = spec f_name = func.__name__ @@ -1350,7 +1400,7 @@ def getclosurevars(func): func = func.__func__ if not isfunction(func): - raise TypeError("'{!r}' is not a Python function".format(func)) + raise TypeError("{!r} is not a Python function".format(func)) code = func.__code__ # Nonlocal references are named in co_freevars and resolved @@ -1416,7 +1466,6 @@ def getframeinfo(frame, context=1): except OSError: lines = index = None else: - start = max(start, 1) start = max(0, min(start, len(lines) - context)) lines = lines[start:start+context] index = lineno - 1 - start @@ -1594,7 +1643,7 @@ def getgeneratorlocals(generator): bound values.""" if not isgenerator(generator): - raise TypeError("'{!r}' is not a Python generator".format(generator)) + raise TypeError("{!r} is not a Python generator".format(generator)) frame = getattr(generator, "gi_frame", None) if frame is not None: @@ -1910,6 +1959,9 @@ def _signature_fromstr(cls, obj, s, skip_bound_arg=True): """Private helper to parse content of '__text_signature__' and return a Signature based on it. """ + # Lazy import ast because it's relatively heavy and + # it's not used for other than this function. + import ast Parameter = cls._parameter_cls @@ -1939,11 +1991,11 @@ def _signature_fromstr(cls, obj, s, skip_bound_arg=True): module = sys.modules.get(module_name, None) if module: module_dict = module.__dict__ - sys_module_dict = sys.modules + sys_module_dict = sys.modules.copy() def parse_name(node): assert isinstance(node, ast.arg) - if node.annotation != None: + if node.annotation is not None: raise ValueError("Annotations are not currently supported") return node.arg @@ -1956,14 +2008,8 @@ def _signature_fromstr(cls, obj, s, skip_bound_arg=True): except NameError: raise RuntimeError() - if isinstance(value, str): - return ast.Str(value) - if isinstance(value, (int, float)): - return ast.Num(value) - if isinstance(value, bytes): - return ast.Bytes(value) - if value in (True, False, None): - return ast.NameConstant(value) + if isinstance(value, (str, int, float, bytes, bool, type(None))): + return ast.Constant(value) raise RuntimeError() class RewriteSymbolics(ast.NodeTransformer): @@ -2063,7 +2109,7 @@ def _signature_from_builtin(cls, func, skip_bound_arg=True): return _signature_fromstr(cls, func, s, skip_bound_arg) -def _signature_from_function(cls, func): +def _signature_from_function(cls, func, skip_bound_arg=True): """Private helper: constructs Signature for the given python function.""" is_duck_function = False @@ -2075,15 +2121,20 @@ def _signature_from_function(cls, func): # of pure function: raise TypeError('{!r} is not a Python function'.format(func)) + s = getattr(func, "__text_signature__", None) + if s: + return _signature_fromstr(cls, func, s, skip_bound_arg) + Parameter = cls._parameter_cls # Parameter information. func_code = func.__code__ pos_count = func_code.co_argcount arg_names = func_code.co_varnames - positional = tuple(arg_names[:pos_count]) + posonly_count = func_code.co_posonlyargcount + positional = arg_names[:pos_count] keyword_only_count = func_code.co_kwonlyargcount - keyword_only = arg_names[pos_count:(pos_count + keyword_only_count)] + keyword_only = arg_names[pos_count:pos_count + keyword_only_count] annotations = func.__annotations__ defaults = func.__defaults__ kwdefaults = func.__kwdefaults__ @@ -2095,19 +2146,27 @@ def _signature_from_function(cls, func): parameters = [] - # Non-keyword-only parameters w/o defaults. non_default_count = pos_count - pos_default_count + posonly_left = posonly_count + + # Non-keyword-only parameters w/o defaults. for name in positional[:non_default_count]: + kind = _POSITIONAL_ONLY if posonly_left else _POSITIONAL_OR_KEYWORD annotation = annotations.get(name, _empty) parameters.append(Parameter(name, annotation=annotation, - kind=_POSITIONAL_OR_KEYWORD)) + kind=kind)) + if posonly_left: + posonly_left -= 1 # ... w/ defaults. for offset, name in enumerate(positional[non_default_count:]): + kind = _POSITIONAL_ONLY if posonly_left else _POSITIONAL_OR_KEYWORD annotation = annotations.get(name, _empty) parameters.append(Parameter(name, annotation=annotation, - kind=_POSITIONAL_OR_KEYWORD, + kind=kind, default=defaults[offset])) + if posonly_left: + posonly_left -= 1 # *args if func_code.co_flags & CO_VARARGS: @@ -2215,16 +2274,23 @@ def _signature_from_callable(obj, *, sigcls=sigcls) sig = _signature_get_partial(wrapped_sig, partialmethod, (None,)) - first_wrapped_param = tuple(wrapped_sig.parameters.values())[0] - new_params = (first_wrapped_param,) + tuple(sig.parameters.values()) - - return sig.replace(parameters=new_params) + if first_wrapped_param.kind is Parameter.VAR_POSITIONAL: + # First argument of the wrapped callable is `*args`, as in + # `partialmethod(lambda *args)`. + return sig + else: + sig_params = tuple(sig.parameters.values()) + assert (not sig_params or + first_wrapped_param is not sig_params[0]) + new_params = (first_wrapped_param,) + sig_params + return sig.replace(parameters=new_params) if isfunction(obj) or _signature_is_functionlike(obj): # If it's a pure Python function, or an object that is duck type # of a Python function (Cython functions, for instance), then: - return _signature_from_function(sigcls, obj) + return _signature_from_function(sigcls, obj, + skip_bound_arg=skip_bound_arg) if _signature_is_builtin(obj): return _signature_from_builtin(sigcls, obj, @@ -2301,7 +2367,7 @@ def _signature_from_callable(obj, *, if (obj.__init__ is object.__init__ and obj.__new__ is object.__new__): # Return a signature of 'object' builtin. - return signature(object) + return sigcls.from_callable(object) else: raise ValueError( 'no signature found for builtin type {!r}'.format(obj)) @@ -2357,6 +2423,9 @@ class _ParameterKind(enum.IntEnum): def __str__(self): return self._name_ + @property + def description(self): + return _PARAM_NAME_MAPPING[self] _POSITIONAL_ONLY = _ParameterKind.POSITIONAL_ONLY _POSITIONAL_OR_KEYWORD = _ParameterKind.POSITIONAL_OR_KEYWORD @@ -2364,6 +2433,14 @@ _VAR_POSITIONAL = _ParameterKind.VAR_POSITIONAL _KEYWORD_ONLY = _ParameterKind.KEYWORD_ONLY _VAR_KEYWORD = _ParameterKind.VAR_KEYWORD +_PARAM_NAME_MAPPING = { + _POSITIONAL_ONLY: 'positional-only', + _POSITIONAL_OR_KEYWORD: 'positional or keyword', + _VAR_POSITIONAL: 'variadic positional', + _KEYWORD_ONLY: 'keyword-only', + _VAR_KEYWORD: 'variadic keyword' +} + class Parameter: """Represents a parameter in a function signature. @@ -2398,15 +2475,14 @@ class Parameter: empty = _empty def __init__(self, name, kind, *, default=_empty, annotation=_empty): - - if kind not in (_POSITIONAL_ONLY, _POSITIONAL_OR_KEYWORD, - _VAR_POSITIONAL, _KEYWORD_ONLY, _VAR_KEYWORD): - raise ValueError("invalid value for 'Parameter.kind' attribute") - self._kind = kind - + try: + self._kind = _ParameterKind(kind) + except ValueError: + raise ValueError(f'value {kind!r} is not a valid Parameter.kind') if default is not _empty: - if kind in (_VAR_POSITIONAL, _VAR_KEYWORD): - msg = '{} parameters cannot have default values'.format(kind) + if self._kind in (_VAR_POSITIONAL, _VAR_KEYWORD): + msg = '{} parameters cannot have default values' + msg = msg.format(self._kind.description) raise ValueError(msg) self._default = default self._annotation = annotation @@ -2415,19 +2491,21 @@ class Parameter: raise ValueError('name is a required attribute for Parameter') if not isinstance(name, str): - raise TypeError("name must be a str, not a {!r}".format(name)) + msg = 'name must be a str, not a {}'.format(type(name).__name__) + raise TypeError(msg) if name[0] == '.' and name[1:].isdigit(): # These are implicit arguments generated by comprehensions. In # order to provide a friendlier interface to users, we recast # their name as "implicitN" and treat them as positional-only. # See issue 19611. - if kind != _POSITIONAL_OR_KEYWORD: - raise ValueError( - 'implicit arguments must be passed in as {}'.format( - _POSITIONAL_OR_KEYWORD - ) + if self._kind != _POSITIONAL_OR_KEYWORD: + msg = ( + 'implicit arguments must be passed as ' + 'positional or keyword arguments, not {}' ) + msg = msg.format(self._kind.description) + raise ValueError(msg) self._kind = _POSITIONAL_ONLY name = 'implicit{}'.format(name[1:]) @@ -2486,11 +2564,14 @@ class Parameter: # Add annotation and default value if self._annotation is not _empty: - formatted = '{}:{}'.format(formatted, + formatted = '{}: {}'.format(formatted, formatannotation(self._annotation)) if self._default is not _empty: - formatted = '{}={}'.format(formatted, repr(self._default)) + if self._annotation is not _empty: + formatted = '{} = {}'.format(formatted, repr(self._default)) + else: + formatted = '{}={}'.format(formatted, repr(self._default)) if kind == _VAR_POSITIONAL: formatted = '*' + formatted @@ -2695,8 +2776,12 @@ class Signature: name = param.name if kind < top_kind: - msg = 'wrong parameter order: {!r} before {!r}' - msg = msg.format(top_kind, kind) + msg = ( + 'wrong parameter order: {} parameter before {} ' + 'parameter' + ) + msg = msg.format(top_kind.description, + kind.description) raise ValueError(msg) elif kind > top_kind: kind_defaults = False @@ -2729,19 +2814,25 @@ class Signature: @classmethod def from_function(cls, func): - """Constructs Signature for the given python function.""" + """Constructs Signature for the given python function. - warnings.warn("inspect.Signature.from_function() is deprecated, " - "use Signature.from_callable()", + Deprecated since Python 3.5, use `Signature.from_callable()`. + """ + + warnings.warn("inspect.Signature.from_function() is deprecated since " + "Python 3.5, use Signature.from_callable()", DeprecationWarning, stacklevel=2) return _signature_from_function(cls, func) @classmethod def from_builtin(cls, func): - """Constructs Signature for the given builtin function.""" + """Constructs Signature for the given builtin function. - warnings.warn("inspect.Signature.from_builtin() is deprecated, " - "use Signature.from_callable()", + Deprecated since Python 3.5, use `Signature.from_callable()`. + """ + + warnings.warn("inspect.Signature.from_builtin() is deprecated since " + "Python 3.5, use Signature.from_callable()", DeprecationWarning, stacklevel=2) return _signature_from_builtin(cls, func) @@ -2869,7 +2960,7 @@ class Signature: arguments[param.name] = tuple(values) break - if param.name in kwargs: + if param.name in kwargs and param.kind != _POSITIONAL_ONLY: raise TypeError( 'multiple values for argument {arg!r}'.format( arg=param.name)) from None @@ -2926,19 +3017,19 @@ class Signature: return self._bound_arguments_cls(self, arguments) - def bind(*args, **kwargs): + def bind(self, /, *args, **kwargs): """Get a BoundArguments object, that maps the passed `args` and `kwargs` to the function's signature. Raises `TypeError` if the passed arguments can not be bound. """ - return args[0]._bind(args[1:], kwargs) + return self._bind(args, kwargs) - def bind_partial(*args, **kwargs): + def bind_partial(self, /, *args, **kwargs): """Get a BoundArguments object, that partially maps the passed `args` and `kwargs` to the function's signature. Raises `TypeError` if the passed arguments can not be bound. """ - return args[0]._bind(args[1:], kwargs, partial=True) + return self._bind(args, kwargs, partial=True) def __reduce__(self): return (type(self), @@ -3027,7 +3118,7 @@ def _main(): type(exc).__name__, exc) print(msg, file=sys.stderr) - exit(2) + sys.exit(2) if has_attrs: parts = attrs.split(".") @@ -3037,7 +3128,7 @@ def _main(): if module.__name__ in sys.builtin_module_names: print("Can't get info for builtin modules.", file=sys.stderr) - exit(1) + sys.exit(1) if args.details: print('Target: {}'.format(target)) diff --git a/Lib/test/support/script_helper.py b/Lib/test/support/script_helper.py index 9b43c3330..83519988e 100644 --- a/Lib/test/support/script_helper.py +++ b/Lib/test/support/script_helper.py @@ -8,7 +8,7 @@ import os import os.path import subprocess import py_compile -# import zipfile XXX RustPython +import zipfile from importlib.util import source_from_cache from test.support import make_legacy_pyc, strip_python_stderr @@ -137,7 +137,7 @@ def run_python_until_end(*args, **env_vars): err = strip_python_stderr(err) return _PythonRunResult(rc, out, err), cmd_line -def _assert_python(expected_success, *args, **env_vars): +def _assert_python(expected_success, /, *args, **env_vars): res, cmd_line = run_python_until_end(*args, **env_vars) if (res.rc and expected_success) or (not res.rc and not expected_success): res.fail(cmd_line) diff --git a/Lib/test/test_typechecks.py b/Lib/test/test_typechecks.py new file mode 100644 index 000000000..9e6628346 --- /dev/null +++ b/Lib/test/test_typechecks.py @@ -0,0 +1,73 @@ +"""Unit tests for __instancecheck__ and __subclasscheck__.""" + +import unittest + + +class ABC(type): + + def __instancecheck__(cls, inst): + """Implement isinstance(inst, cls).""" + return any(cls.__subclasscheck__(c) + for c in {type(inst), inst.__class__}) + + def __subclasscheck__(cls, sub): + """Implement issubclass(sub, cls).""" + candidates = cls.__dict__.get("__subclass__", set()) | {cls} + return any(c in candidates for c in sub.mro()) + + +class Integer(metaclass=ABC): + __subclass__ = {int} + + +class SubInt(Integer): + pass + + +class TypeChecksTest(unittest.TestCase): + + def testIsSubclassInternal(self): + self.assertEqual(Integer.__subclasscheck__(int), True) + self.assertEqual(Integer.__subclasscheck__(float), False) + + def testIsSubclassBuiltin(self): + self.assertEqual(issubclass(int, Integer), True) + self.assertEqual(issubclass(int, (Integer,)), True) + self.assertEqual(issubclass(float, Integer), False) + self.assertEqual(issubclass(float, (Integer,)), False) + + def testIsInstanceBuiltin(self): + self.assertEqual(isinstance(42, Integer), True) + self.assertEqual(isinstance(42, (Integer,)), True) + self.assertEqual(isinstance(3.14, Integer), False) + self.assertEqual(isinstance(3.14, (Integer,)), False) + + def testIsInstanceActual(self): + self.assertEqual(isinstance(Integer(), Integer), True) + self.assertEqual(isinstance(Integer(), (Integer,)), True) + + def testIsSubclassActual(self): + self.assertEqual(issubclass(Integer, Integer), True) + self.assertEqual(issubclass(Integer, (Integer,)), True) + + # TODO: RUSTPYTHON + @unittest.expectedFailure + def testSubclassBehavior(self): + self.assertEqual(issubclass(SubInt, Integer), True) + self.assertEqual(issubclass(SubInt, (Integer,)), True) + self.assertEqual(issubclass(SubInt, SubInt), True) + self.assertEqual(issubclass(SubInt, (SubInt,)), True) + self.assertEqual(issubclass(Integer, SubInt), False) + self.assertEqual(issubclass(Integer, (SubInt,)), False) + self.assertEqual(issubclass(int, SubInt), False) + self.assertEqual(issubclass(int, (SubInt,)), False) + self.assertEqual(isinstance(SubInt(), Integer), True) + self.assertEqual(isinstance(SubInt(), (Integer,)), True) + self.assertEqual(isinstance(SubInt(), SubInt), True) + self.assertEqual(isinstance(SubInt(), (SubInt,)), True) + self.assertEqual(isinstance(42, SubInt), False) + self.assertEqual(isinstance(42, (SubInt,)), False) + + +if __name__ == "__main__": + unittest.main() diff --git a/Lib/test/test_unary.py b/Lib/test/test_unary.py new file mode 100644 index 000000000..c3c17cc9f --- /dev/null +++ b/Lib/test/test_unary.py @@ -0,0 +1,53 @@ +"""Test compiler changes for unary ops (+, -, ~) introduced in Python 2.2""" + +import unittest + +class UnaryOpTestCase(unittest.TestCase): + + def test_negative(self): + self.assertTrue(-2 == 0 - 2) + self.assertEqual(-0, 0) + self.assertEqual(--2, 2) + self.assertTrue(-2 == 0 - 2) + self.assertTrue(-2.0 == 0 - 2.0) + self.assertTrue(-2j == 0 - 2j) + + def test_positive(self): + self.assertEqual(+2, 2) + self.assertEqual(+0, 0) + self.assertEqual(++2, 2) + self.assertEqual(+2, 2) + self.assertEqual(+2.0, 2.0) + self.assertEqual(+2j, 2j) + + def test_invert(self): + self.assertTrue(-2 == 0 - 2) + self.assertEqual(-0, 0) + self.assertEqual(--2, 2) + self.assertTrue(-2 == 0 - 2) + + def test_no_overflow(self): + nines = "9" * 32 + self.assertTrue(eval("+" + nines) == 10**32-1) + self.assertTrue(eval("-" + nines) == -(10**32-1)) + self.assertTrue(eval("~" + nines) == ~(10**32-1)) + + def test_negation_of_exponentiation(self): + # Make sure '**' does the right thing; these form a + # regression test for SourceForge bug #456756. + self.assertEqual(-2 ** 3, -8) + self.assertEqual((-2) ** 3, -8) + self.assertEqual(-2 ** 4, -16) + self.assertEqual((-2) ** 4, 16) + + def test_bad_types(self): + for op in '+', '-', '~': + self.assertRaises(TypeError, eval, op + "b'a'") + self.assertRaises(TypeError, eval, op + "'a'") + + self.assertRaises(TypeError, eval, "~2j") + self.assertRaises(TypeError, eval, "~2.0") + + +if __name__ == "__main__": + unittest.main() diff --git a/Lib/zipfile.py b/Lib/zipfile.py new file mode 100644 index 000000000..5dc6516cc --- /dev/null +++ b/Lib/zipfile.py @@ -0,0 +1,2450 @@ +""" +Read and write ZIP files. + +XXX references to utf-8 need further investigation. +""" +import binascii +import functools +import importlib.util +import io +import itertools +import os +import posixpath +import shutil +import stat +import struct +import sys +import threading +import time +import contextlib +from collections import OrderedDict + +try: + import zlib # We may need its compression method + crc32 = zlib.crc32 +except ImportError: + zlib = None + crc32 = binascii.crc32 + +try: + import bz2 # We may need its compression method +except ImportError: + bz2 = None + +try: + import lzma # We may need its compression method +except ImportError: + lzma = None + +__all__ = ["BadZipFile", "BadZipfile", "error", + "ZIP_STORED", "ZIP_DEFLATED", "ZIP_BZIP2", "ZIP_LZMA", + "is_zipfile", "ZipInfo", "ZipFile", "PyZipFile", "LargeZipFile"] + +class BadZipFile(Exception): + pass + + +class LargeZipFile(Exception): + """ + Raised when writing a zipfile, the zipfile requires ZIP64 extensions + and those extensions are disabled. + """ + +error = BadZipfile = BadZipFile # Pre-3.2 compatibility names + + +ZIP64_LIMIT = (1 << 31) - 1 +ZIP_FILECOUNT_LIMIT = (1 << 16) - 1 +ZIP_MAX_COMMENT = (1 << 16) - 1 + +# constants for Zip file compression methods +ZIP_STORED = 0 +ZIP_DEFLATED = 8 +ZIP_BZIP2 = 12 +ZIP_LZMA = 14 +# Other ZIP compression methods not supported + +DEFAULT_VERSION = 20 +ZIP64_VERSION = 45 +BZIP2_VERSION = 46 +LZMA_VERSION = 63 +# we recognize (but not necessarily support) all features up to that version +MAX_EXTRACT_VERSION = 63 + +# Below are some formats and associated data for reading/writing headers using +# the struct module. The names and structures of headers/records are those used +# in the PKWARE description of the ZIP file format: +# http://www.pkware.com/documents/casestudies/APPNOTE.TXT +# (URL valid as of January 2008) + +# The "end of central directory" structure, magic number, size, and indices +# (section V.I in the format document) +structEndArchive = b"<4s4H2LH" +stringEndArchive = b"PK\005\006" +sizeEndCentDir = struct.calcsize(structEndArchive) + +_ECD_SIGNATURE = 0 +_ECD_DISK_NUMBER = 1 +_ECD_DISK_START = 2 +_ECD_ENTRIES_THIS_DISK = 3 +_ECD_ENTRIES_TOTAL = 4 +_ECD_SIZE = 5 +_ECD_OFFSET = 6 +_ECD_COMMENT_SIZE = 7 +# These last two indices are not part of the structure as defined in the +# spec, but they are used internally by this module as a convenience +_ECD_COMMENT = 8 +_ECD_LOCATION = 9 + +# The "central directory" structure, magic number, size, and indices +# of entries in the structure (section V.F in the format document) +structCentralDir = "<4s4B4HL2L5H2L" +stringCentralDir = b"PK\001\002" +sizeCentralDir = struct.calcsize(structCentralDir) + +# indexes of entries in the central directory structure +_CD_SIGNATURE = 0 +_CD_CREATE_VERSION = 1 +_CD_CREATE_SYSTEM = 2 +_CD_EXTRACT_VERSION = 3 +_CD_EXTRACT_SYSTEM = 4 +_CD_FLAG_BITS = 5 +_CD_COMPRESS_TYPE = 6 +_CD_TIME = 7 +_CD_DATE = 8 +_CD_CRC = 9 +_CD_COMPRESSED_SIZE = 10 +_CD_UNCOMPRESSED_SIZE = 11 +_CD_FILENAME_LENGTH = 12 +_CD_EXTRA_FIELD_LENGTH = 13 +_CD_COMMENT_LENGTH = 14 +_CD_DISK_NUMBER_START = 15 +_CD_INTERNAL_FILE_ATTRIBUTES = 16 +_CD_EXTERNAL_FILE_ATTRIBUTES = 17 +_CD_LOCAL_HEADER_OFFSET = 18 + +# The "local file header" structure, magic number, size, and indices +# (section V.A in the format document) +structFileHeader = "<4s2B4HL2L2H" +stringFileHeader = b"PK\003\004" +sizeFileHeader = struct.calcsize(structFileHeader) + +_FH_SIGNATURE = 0 +_FH_EXTRACT_VERSION = 1 +_FH_EXTRACT_SYSTEM = 2 +_FH_GENERAL_PURPOSE_FLAG_BITS = 3 +_FH_COMPRESSION_METHOD = 4 +_FH_LAST_MOD_TIME = 5 +_FH_LAST_MOD_DATE = 6 +_FH_CRC = 7 +_FH_COMPRESSED_SIZE = 8 +_FH_UNCOMPRESSED_SIZE = 9 +_FH_FILENAME_LENGTH = 10 +_FH_EXTRA_FIELD_LENGTH = 11 + +# The "Zip64 end of central directory locator" structure, magic number, and size +structEndArchive64Locator = "<4sLQL" +stringEndArchive64Locator = b"PK\x06\x07" +sizeEndCentDir64Locator = struct.calcsize(structEndArchive64Locator) + +# The "Zip64 end of central directory" record, magic number, size, and indices +# (section V.G in the format document) +structEndArchive64 = "<4sQ2H2L4Q" +stringEndArchive64 = b"PK\x06\x06" +sizeEndCentDir64 = struct.calcsize(structEndArchive64) + +_CD64_SIGNATURE = 0 +_CD64_DIRECTORY_RECSIZE = 1 +_CD64_CREATE_VERSION = 2 +_CD64_EXTRACT_VERSION = 3 +_CD64_DISK_NUMBER = 4 +_CD64_DISK_NUMBER_START = 5 +_CD64_NUMBER_ENTRIES_THIS_DISK = 6 +_CD64_NUMBER_ENTRIES_TOTAL = 7 +_CD64_DIRECTORY_SIZE = 8 +_CD64_OFFSET_START_CENTDIR = 9 + +_DD_SIGNATURE = 0x08074b50 + +_EXTRA_FIELD_STRUCT = struct.Struct(' 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 endrec + + +def _EndRecData(fpin): + """Return data from the "End of Central Directory" record, or None. + + The data is a list of the nine items in the ZIP "End of central dir" + record followed by a tenth item, the file seek offset of this record.""" + + # 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 + return None + + +class ZipInfo (object): + """Class with attributes describing each file in the ZIP archive.""" + + __slots__ = ( + 'orig_filename', + 'filename', + 'date_time', + 'compress_type', + '_compresslevel', + '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=(1980,1,1,0,0,0)): + 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. + if os.sep != "/" and os.sep in filename: + filename = filename.replace(os.sep, "/") + + 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._compresslevel = None # Level for the compressor + 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 bytes object.""" + 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 = ' 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(' len(extra): + raise BadZipFile("Corrupt extra field %04x (size=%d)" % (tp, ln)) + if tp == 0x0001: + if ln >= 24: + counts = unpack(' 2107: + date_time = (2107, 12, 31, 23, 59, 59) + # 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 = (st.st_mode & 0xFFFF) << 16 # Unix attributes + 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.""" + return self.filename[-1] == '/' + + +# ZIP encryption uses the CRC32 one-byte primitive for scrambling some +# internal keys. We noticed that a direct implementation is faster than +# relying on binascii.crc32(). + +_crctable = None +def _gen_crc(crc): + for j in range(8): + if crc & 1: + crc = (crc >> 1) ^ 0xEDB88320 + else: + crc >>= 1 + return crc + +# ZIP supports a password-based form of encryption. Even though known +# plaintext attacks have been found against it, it is still useful +# to be able to get data out of such a file. +# +# Usage: +# zd = _ZipDecrypter(mypwd) +# plain_bytes = zd(cypher_bytes) + +def _ZipDecrypter(pwd): + key0 = 305419896 + key1 = 591751049 + key2 = 878082192 + + global _crctable + if _crctable is None: + _crctable = list(map(_gen_crc, range(256))) + crctable = _crctable + + def crc32(ch, crc): + """Compute the CRC32 primitive on one byte.""" + return (crc >> 8) ^ crctable[(crc ^ ch) & 0xFF] + + def update_keys(c): + nonlocal key0, key1, key2 + key0 = crc32(c, key0) + key1 = (key1 + (key0 & 0xFF)) & 0xFFFFFFFF + key1 = (key1 * 134775813 + 1) & 0xFFFFFFFF + key2 = crc32(key1 >> 24, key2) + + for p in pwd: + update_keys(p) + + def decrypter(data): + """Decrypt a bytes object.""" + result = bytearray() + append = result.append + for c in data: + k = key2 | 2 + c ^= ((k * (k^1)) >> 8) & 0xFF + update_keys(c) + append(c) + return bytes(result) + + return decrypter + + +class LZMACompressor: + + 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('> 8) & 0xff + else: + # compare against the CRC otherwise + check_byte = (zipinfo.CRC >> 24) & 0xff + h = self._init_decrypter() + if h != check_byte: + raise RuntimeError("Bad password for file %r" % zipinfo.orig_filename) + + + def _init_decrypter(self): + self._decrypter = _ZipDecrypter(self._pwd) + # The first 12 bytes in the cypher stream is an encryption header + # used to strengthen the algorithm. The first 11 bytes are + # completely random, while the 12th contains the MSB of the CRC, + # or the MSB of the file time depending on the header type + # and is used to check the correctness of the password. + header = self._fileobj.read(12) + self._compress_left -= 12 + return self._decrypter(header)[11] + + 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 is specified, at most limit bytes will be read. + """ + + 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 the argument is omitted, None, or negative, data is read and returned until EOF is reached. + """ + 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 = self._decrypter(data) + return data + + def close(self): + try: + if self._close_fileobj: + self._fileobj.close() + finally: + super().close() + + def seekable(self): + return self._seekable + + def seek(self, offset, whence=0): + if not self._seekable: + raise io.UnsupportedOperation("underlying stream is not seekable") + curr_pos = self.tell() + if whence == 0: # Seek from start of file + new_pos = offset + elif whence == 1: # Seek from current position + new_pos = curr_pos + offset + elif whence == 2: # Seek from EOF + new_pos = self._orig_file_size + offset + else: + raise ValueError("whence must be os.SEEK_SET (0), " + "os.SEEK_CUR (1), or os.SEEK_END (2)") + + if new_pos > self._orig_file_size: + new_pos = self._orig_file_size + + if new_pos < 0: + new_pos = 0 + + read_offset = new_pos - curr_pos + buff_offset = read_offset + self._offset + + if buff_offset >= 0 and buff_offset < len(self._readbuffer): + # Just move the _offset index if the new position is in the _readbuffer + self._offset = buff_offset + read_offset = 0 + elif read_offset < 0: + # Position is before the current position. Reset the ZipExtFile + self._fileobj.seek(self._orig_compress_start) + self._running_crc = self._orig_start_crc + self._compress_left = self._orig_compress_size + self._left = self._orig_file_size + self._readbuffer = b'' + self._offset = 0 + self._decompressor = _get_decompressor(self._compress_type) + self._eof = False + read_offset = new_pos + if self._decrypter is not None: + self._init_decrypter() + + while read_offset > 0: + read_len = min(self.MAX_SEEK_READ, read_offset) + self.read(read_len) + read_offset -= read_len + + return self.tell() + + def tell(self): + if not self._seekable: + raise io.UnsupportedOperation("underlying stream is not seekable") + filepos = self._orig_file_size - self._left - len(self._readbuffer) + self._offset + return filepos + + +class _ZipWriteFile(io.BufferedIOBase): + def __init__(self, zf, zinfo, zip64): + self._zinfo = zinfo + self._zip64 = zip64 + self._zipfile = zf + self._compressor = _get_compressor(zinfo.compress_type, + zinfo._compresslevel) + 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 + try: + super().close() + # Flush any data from the compressor, and update header info + if self._compressor: + buf = self._compressor.flush() + self._compress_size += len(buf) + self._fileobj.write(buf) + self._zinfo.compress_size = self._compress_size + else: + self._zinfo.compress_size = self._file_size + self._zinfo.CRC = self._crc + self._zinfo.file_size = self._file_size + + # Write updated header info + if self._zinfo.flag_bits & 0x08: + # Write CRC and file sizes after the file data + fmt = ' ZIP64_LIMIT: + raise RuntimeError( + 'File size unexpectedly exceeded ZIP64 limit') + if self._compress_size > ZIP64_LIMIT: + raise RuntimeError( + 'Compressed size unexpectedly exceeded ZIP64 limit') + # Seek backwards and write file header (which will now include + # correct CRC and file sizes) + + # Preserve current position in file + self._zipfile.start_dir = self._fileobj.tell() + self._fileobj.seek(self._zinfo.header_offset) + self._fileobj.write(self._zinfo.FileHeader(self._zip64)) + self._fileobj.seek(self._zipfile.start_dir) + + # Successfully written: Add file to our caches + self._zipfile.filelist.append(self._zinfo) + self._zipfile.NameToInfo[self._zinfo.filename] = self._zinfo + finally: + self._zipfile._writing = False + + + +class ZipFile: + """ Class with methods to open, read, write, close, list zip files. + + z = ZipFile(file, mode="r", compression=ZIP_STORED, allowZip64=True, + compresslevel=None) + + file: Either the path to the file, or a file-like object. + If it is a path, the file will be opened and closed by ZipFile. + mode: The mode can be either read 'r', write 'w', exclusive create 'x', + or append 'a'. + compression: ZIP_STORED (no compression), ZIP_DEFLATED (requires zlib), + ZIP_BZIP2 (requires bz2) or ZIP_LZMA (requires lzma). + allowZip64: if True ZipFile will create files with ZIP64 extensions when + needed, otherwise it will raise an exception when this would + be necessary. + compresslevel: None (default for the given compression type) or an integer + specifying the level to pass to the compressor. + When using ZIP_STORED or ZIP_LZMA this keyword has no effect. + When using ZIP_DEFLATED integers 0 through 9 are accepted. + When using ZIP_BZIP2 integers 1 through 9 are accepted. + + """ + + fp = None # Set here since __del__ checks it + _windows_illegal_name_trans_table = None + + def __init__(self, file, mode="r", compression=ZIP_STORED, allowZip64=True, + compresslevel=None, *, strict_timestamps=True): + """Open the ZIP file with mode read 'r', write 'w', exclusive create 'x', + or append 'a'.""" + if mode not in ('r', 'w', 'x', 'a'): + raise ValueError("ZipFile requires mode 'r', 'w', 'x', or 'a'") + + _check_compression(compression) + + self._allowZip64 = allowZip64 + self._didModify = False + self.debug = 0 # Level of printing: 0 through 3 + self.NameToInfo = {} # Find file info given name + self.filelist = [] # List of ZipInfo instances for archive + self.compression = compression # Method of compression + self.compresslevel = compresslevel + self.mode = mode + self.pwd = None + self._comment = b'' + self._strict_timestamps = strict_timestamps + + # Check if we were passed a file-like object + if isinstance(file, os.PathLike): + file = os.fspath(file) + if isinstance(file, str): + # No, it's a filename + self._filePassed = 0 + self.filename = file + modeDict = {'r' : 'rb', 'w': 'w+b', 'x': 'x+b', 'a' : 'r+b', + 'r+b': 'w+b', 'w+b': 'wb', 'x+b': 'xb'} + filemode = modeDict[mode] + while True: + try: + self.fp = io.open(file, filemode) + except OSError: + if filemode in modeDict: + filemode = modeDict[filemode] + continue + raise + break + else: + self._filePassed = 1 + self.fp = file + self.filename = getattr(file, 'name', None) + self._fileRefCnt = 1 + self._lock = threading.RLock() + self._seekable = True + self._writing = False + + try: + if mode == 'r': + self._RealGetContents() + elif mode in ('w', 'x'): + # set the modified flag so central directory gets written + # even if no files are added to the archive + self._didModify = True + try: + self.start_dir = self.fp.tell() + except (AttributeError, OSError): + self.fp = _Tellable(self.fp) + self.start_dir = 0 + self._seekable = False + else: + # Some file-like objects can provide tell() but not seek() + try: + self.fp.seek(self.start_dir) + except (AttributeError, OSError): + self._seekable = False + elif mode == 'a': + try: + # See if file is a zip file + self._RealGetContents() + # seek to start of directory and overwrite + self.fp.seek(self.start_dir) + except BadZipFile: + # file is not a zip file, just append + self.fp.seek(0, 2) + + # set the modified flag so central directory gets written + # even if no files are added to the archive + self._didModify = True + self.start_dir = self.fp.tell() + else: + raise ValueError("Mode must be 'r', 'w', 'x', or 'a'") + except: + fp = self.fp + self.fp = None + self._fpclose(fp) + raise + + def __enter__(self): + return self + + def __exit__(self, type, value, traceback): + self.close() + + def __repr__(self): + result = ['<%s.%s' % (self.__class__.__module__, + self.__class__.__qualname__)] + if self.fp is not None: + if self._filePassed: + result.append(' file=%r' % self.fp) + elif self.filename is not None: + result.append(' filename=%r' % self.filename) + result.append(' mode=%r' % self.mode) + else: + result.append(' [closed]') + result.append('>') + return ''.join(result) + + def _RealGetContents(self): + """Read in the table of contents for the ZIP file.""" + fp = self.fp + try: + endrec = _EndRecData(fp) + except OSError: + raise BadZipFile("File is not a zip file") + if not endrec: + raise BadZipFile("File is not a zip file") + if self.debug > 1: + print(endrec) + size_cd = endrec[_ECD_SIZE] # bytes in central directory + offset_cd = endrec[_ECD_OFFSET] # offset of central directory + self._comment = endrec[_ECD_COMMENT] # archive comment + + # "concat" is zero, unless zip was concatenated to another file + concat = endrec[_ECD_LOCATION] - size_cd - offset_cd + if endrec[_ECD_SIGNATURE] == stringEndArchive64: + # If Zip64 extension structures are present, account for them + concat -= (sizeEndCentDir64 + sizeEndCentDir64Locator) + + if self.debug > 2: + inferred = concat + offset_cd + print("given, inferred, offset", offset_cd, inferred, concat) + # self.start_dir: Position of start of central directory + self.start_dir = offset_cd + concat + fp.seek(self.start_dir, 0) + data = fp.read(size_cd) + fp = io.BytesIO(data) + total = 0 + while total < size_cd: + centdir = fp.read(sizeCentralDir) + if len(centdir) != sizeCentralDir: + raise BadZipFile("Truncated central directory") + centdir = struct.unpack(structCentralDir, centdir) + if centdir[_CD_SIGNATURE] != stringCentralDir: + raise BadZipFile("Bad magic number for central directory") + if self.debug > 2: + print(centdir) + filename = fp.read(centdir[_CD_FILENAME_LENGTH]) + flags = centdir[5] + if flags & 0x800: + # UTF-8 file names extension + filename = filename.decode('utf-8') + else: + # Historical ZIP filename encoding + filename = filename.decode('cp437') + # Create ZipInfo instance to store file information + x = ZipInfo(filename) + x.extra = fp.read(centdir[_CD_EXTRA_FIELD_LENGTH]) + x.comment = fp.read(centdir[_CD_COMMENT_LENGTH]) + x.header_offset = centdir[_CD_LOCAL_HEADER_OFFSET] + (x.create_version, x.create_system, x.extract_version, x.reserved, + x.flag_bits, x.compress_type, t, d, + x.CRC, x.compress_size, x.file_size) = centdir[1:12] + if x.extract_version > MAX_EXTRACT_VERSION: + raise NotImplementedError("zip file version %.1f" % + (x.extract_version / 10)) + x.volume, x.internal_attr, x.external_attr = centdir[15:18] + # Convert date/time code to (year, month, day, hour, min, sec) + x._raw_time = t + x.date_time = ( (d>>9)+1980, (d>>5)&0xF, d&0x1F, + t>>11, (t>>5)&0x3F, (t&0x1F) * 2 ) + + x._decodeExtra() + x.header_offset = x.header_offset + concat + self.filelist.append(x) + self.NameToInfo[x.filename] = x + + # update total bytes read from central directory + total = (total + sizeCentralDir + centdir[_CD_FILENAME_LENGTH] + + centdir[_CD_EXTRA_FIELD_LENGTH] + + centdir[_CD_COMMENT_LENGTH]) + + if self.debug > 2: + print("total", total) + + + def namelist(self): + """Return a list of file names in the archive.""" + return [data.filename for data in self.filelist] + + def infolist(self): + """Return a list of class ZipInfo instances for files in the + archive.""" + return self.filelist + + def printdir(self, file=None): + """Print a table of contents for the zip file.""" + print("%-46s %19s %12s" % ("File Name", "Modified ", "Size"), + file=file) + for zinfo in self.filelist: + date = "%d-%02d-%02d %02d:%02d:%02d" % zinfo.date_time[:6] + print("%-46s %s %12d" % (zinfo.filename, date, zinfo.file_size), + file=file) + + def testzip(self): + """Read all the files and check the CRC.""" + chunk_size = 2 ** 20 + for zinfo in self.filelist: + try: + # Read by chunks, to avoid an OverflowError or a + # MemoryError with very large embedded files. + with self.open(zinfo.filename, "r") as f: + while f.read(chunk_size): # Check CRC-32 + pass + except BadZipFile: + return zinfo.filename + + def getinfo(self, name): + """Return the instance of ZipInfo given 'name'.""" + info = self.NameToInfo.get(name) + if info is None: + raise KeyError( + 'There is no item named %r in the archive' % name) + + return info + + def setpassword(self, pwd): + """Set default password for encrypted files.""" + if pwd and not isinstance(pwd, bytes): + raise TypeError("pwd: expected bytes, got %s" % type(pwd).__name__) + if pwd: + self.pwd = pwd + else: + self.pwd = None + + @property + def comment(self): + """The comment text associated with the ZIP file.""" + return self._comment + + @comment.setter + def comment(self, comment): + if not isinstance(comment, bytes): + raise TypeError("comment: expected bytes, got %s" % type(comment).__name__) + # check for valid comment length + if len(comment) > ZIP_MAX_COMMENT: + import warnings + warnings.warn('Archive comment is too long; truncating to %d bytes' + % ZIP_MAX_COMMENT, stacklevel=2) + comment = comment[:ZIP_MAX_COMMENT] + self._comment = comment + self._didModify = True + + def read(self, name, pwd=None): + """Return file bytes for name.""" + with self.open(name, "r", pwd) as fp: + return fp.read() + + def open(self, name, mode="r", pwd=None, *, force_zip64=False): + """Return file-like object for 'name'. + + name is a string for the file name within the ZIP file, or a ZipInfo + object. + + mode should be 'r' to read a file already in the ZIP file, or 'w' to + write to a file newly added to the archive. + + pwd is the password to decrypt files (only used for reading). + + When writing, if the file size is not known in advance but may exceed + 2 GiB, pass force_zip64 to use the ZIP64 format, which can handle large + files. If the size is known in advance, it is best to pass a ZipInfo + instance for name, with zinfo.file_size set. + """ + if mode not in {"r", "w"}: + raise ValueError('open() requires mode "r" or "w"') + if pwd and not isinstance(pwd, bytes): + raise TypeError("pwd: expected bytes, got %s" % type(pwd).__name__) + if pwd and (mode == "w"): + raise ValueError("pwd is only supported for reading files") + if not self.fp: + raise ValueError( + "Attempt to use ZIP archive that was already closed") + + # Make sure we have an info object + if isinstance(name, ZipInfo): + # 'name' is already an info object + zinfo = name + elif mode == 'w': + zinfo = ZipInfo(name) + zinfo.compress_type = self.compression + zinfo._compresslevel = self.compresslevel + else: + # Get info object for name + zinfo = self.getinfo(name) + + if mode == 'w': + return self._open_to_write(zinfo, force_zip64=force_zip64) + + 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.") + + # Open for reading: + self._fileRefCnt += 1 + zef_file = _SharedFile(self.fp, zinfo.header_offset, + self._fpclose, self._lock, lambda: self._writing) + try: + # Skip the file header: + fheader = zef_file.read(sizeFileHeader) + if len(fheader) != sizeFileHeader: + raise BadZipFile("Truncated file header") + fheader = struct.unpack(structFileHeader, fheader) + if fheader[_FH_SIGNATURE] != stringFileHeader: + raise BadZipFile("Bad magic number for file header") + + fname = zef_file.read(fheader[_FH_FILENAME_LENGTH]) + if fheader[_FH_EXTRA_FIELD_LENGTH]: + zef_file.read(fheader[_FH_EXTRA_FIELD_LENGTH]) + + if zinfo.flag_bits & 0x20: + # Zip 2.7: compressed patched data + raise NotImplementedError("compressed patched data (flag bit 5)") + + if zinfo.flag_bits & 0x40: + # strong encryption + raise NotImplementedError("strong encryption (flag bit 6)") + + if zinfo.flag_bits & 0x800: + # UTF-8 filename + fname_str = fname.decode("utf-8") + else: + fname_str = fname.decode("cp437") + + if fname_str != zinfo.orig_filename: + raise BadZipFile( + 'File name in directory %r and header %r differ.' + % (zinfo.orig_filename, fname)) + + # check for encrypted flag & handle password + is_encrypted = zinfo.flag_bits & 0x1 + if is_encrypted: + if not pwd: + pwd = self.pwd + if not pwd: + raise RuntimeError("File %r is encrypted, password " + "required for extraction" % name) + else: + pwd = None + + return ZipExtFile(zef_file, mode, zinfo, pwd, True) + except: + zef_file.close() + raise + + def _open_to_write(self, zinfo, force_zip64=False): + if force_zip64 and not self._allowZip64: + raise ValueError( + "force_zip64 is True, but allowZip64 was False when opening " + "the ZIP file." + ) + if self._writing: + raise ValueError("Can't write to the ZIP file while there is " + "another write handle open on it. " + "Close the first handle before opening another.") + + # Sizes and CRC are overwritten with correct data after processing the file + if not hasattr(zinfo, 'file_size'): + zinfo.file_size = 0 + zinfo.compress_size = 0 + zinfo.CRC = 0 + + zinfo.flag_bits = 0x00 + if zinfo.compress_type == ZIP_LZMA: + # Compressed data includes an end-of-stream (EOS) marker + zinfo.flag_bits |= 0x02 + if not self._seekable: + zinfo.flag_bits |= 0x08 + + if not zinfo.external_attr: + zinfo.external_attr = 0o600 << 16 # permissions: ?rw------- + + # Compressed size can be larger than uncompressed size + zip64 = self._allowZip64 and \ + (force_zip64 or zinfo.file_size * 1.05 > ZIP64_LIMIT) + + if self._seekable: + self.fp.seek(self.start_dir) + zinfo.header_offset = self.fp.tell() + + self._writecheck(zinfo) + self._didModify = True + + self.fp.write(zinfo.FileHeader(zip64)) + + self._writing = True + return _ZipWriteFile(self, zinfo, zip64) + + def extract(self, member, path=None, pwd=None): + """Extract a member from the archive to the current working directory, + using its full name. Its file information is extracted as accurately + as possible. `member' may be a filename or a ZipInfo object. You can + specify a different directory using `path'. + """ + if path is None: + path = os.getcwd() + else: + path = os.fspath(path) + + return self._extract_member(member, path, pwd) + + def extractall(self, path=None, members=None, pwd=None): + """Extract all members from the archive to the current working + directory. `path' specifies a different directory to extract to. + `members' is optional and must be a subset of the list returned + by namelist(). + """ + if members is None: + members = self.namelist() + + if path is None: + path = os.getcwd() + else: + path = os.fspath(path) + + for zipinfo in members: + self._extract_member(zipinfo, path, pwd) + + @classmethod + def _sanitize_windows_name(cls, arcname, pathsep): + """Replace bad characters and remove trailing dots from parts.""" + table = cls._windows_illegal_name_trans_table + if not table: + illegal = ':<>|"?*' + table = str.maketrans(illegal, '_' * len(illegal)) + cls._windows_illegal_name_trans_table = table + arcname = arcname.translate(table) + # remove trailing dots + arcname = (x.rstrip('.') for x in arcname.split(pathsep)) + # rejoin, removing empty parts. + arcname = pathsep.join(x for x in arcname if x) + return arcname + + def _extract_member(self, member, targetpath, pwd): + """Extract the ZipInfo object 'member' to a physical + file on the path targetpath. + """ + if not isinstance(member, ZipInfo): + member = self.getinfo(member) + + # build the destination pathname, replacing + # forward slashes to platform specific separators. + arcname = member.filename.replace('/', os.path.sep) + + if os.path.altsep: + arcname = arcname.replace(os.path.altsep, os.path.sep) + # interpret absolute pathname as relative, remove drive letter or + # UNC path, redundant separators, "." and ".." components. + arcname = os.path.splitdrive(arcname)[1] + invalid_path_parts = ('', os.path.curdir, os.path.pardir) + arcname = os.path.sep.join(x for x in arcname.split(os.path.sep) + if x not in invalid_path_parts) + if os.path.sep == '\\': + # filter illegal characters on Windows + arcname = self._sanitize_windows_name(arcname, os.path.sep) + + targetpath = os.path.join(targetpath, arcname) + targetpath = os.path.normpath(targetpath) + + # Create all upper directories if necessary. + upperdirs = os.path.dirname(targetpath) + if upperdirs and not os.path.exists(upperdirs): + os.makedirs(upperdirs) + + if member.is_dir(): + if not os.path.isdir(targetpath): + os.mkdir(targetpath) + return targetpath + + with self.open(member, pwd=pwd) as source, \ + open(targetpath, "wb") as target: + shutil.copyfileobj(source, target) + + return targetpath + + def _writecheck(self, zinfo): + """Check for errors before writing a file to the archive.""" + if zinfo.filename in self.NameToInfo: + import warnings + warnings.warn('Duplicate name: %r' % zinfo.filename, stacklevel=3) + if self.mode not in ('w', 'x', 'a'): + raise ValueError("write() requires mode 'w', 'x', or 'a'") + if not self.fp: + raise ValueError( + "Attempt to write ZIP archive that was already closed") + _check_compression(zinfo.compress_type) + if not self._allowZip64: + requires_zip64 = None + if len(self.filelist) >= ZIP_FILECOUNT_LIMIT: + requires_zip64 = "Files count" + elif zinfo.file_size > ZIP64_LIMIT: + requires_zip64 = "Filesize" + elif zinfo.header_offset > ZIP64_LIMIT: + requires_zip64 = "Zipfile size" + if requires_zip64: + raise LargeZipFile(requires_zip64 + + " would require ZIP64 extensions") + + def write(self, filename, arcname=None, + compress_type=None, compresslevel=None): + """Put the bytes from filename into the archive under the name + arcname.""" + if not self.fp: + raise ValueError( + "Attempt to write to ZIP archive that was already closed") + if self._writing: + raise ValueError( + "Can't write to ZIP archive while an open writing handle exists" + ) + + zinfo = ZipInfo.from_file(filename, arcname, + strict_timestamps=self._strict_timestamps) + + if zinfo.is_dir(): + zinfo.compress_size = 0 + zinfo.CRC = 0 + else: + if compress_type is not None: + zinfo.compress_type = compress_type + else: + zinfo.compress_type = self.compression + + if compresslevel is not None: + zinfo._compresslevel = compresslevel + else: + zinfo._compresslevel = self.compresslevel + + if zinfo.is_dir(): + with self._lock: + if self._seekable: + self.fp.seek(self.start_dir) + zinfo.header_offset = self.fp.tell() # Start of header bytes + if zinfo.compress_type == ZIP_LZMA: + # Compressed data includes an end-of-stream (EOS) marker + zinfo.flag_bits |= 0x02 + + self._writecheck(zinfo) + self._didModify = True + + self.filelist.append(zinfo) + self.NameToInfo[zinfo.filename] = zinfo + self.fp.write(zinfo.FileHeader(False)) + self.start_dir = self.fp.tell() + else: + with open(filename, "rb") as src, self.open(zinfo, 'w') as dest: + shutil.copyfileobj(src, dest, 1024*8) + + def writestr(self, zinfo_or_arcname, data, + compress_type=None, compresslevel=None): + """Write a file into the archive. The contents is 'data', which + may be either a 'str' or a 'bytes' instance; if it is a 'str', + it is encoded as UTF-8 first. + 'zinfo_or_arcname' is either a ZipInfo instance or + the name of the file in the archive.""" + if isinstance(data, str): + data = data.encode("utf-8") + if not isinstance(zinfo_or_arcname, ZipInfo): + zinfo = ZipInfo(filename=zinfo_or_arcname, + date_time=time.localtime(time.time())[:6]) + zinfo.compress_type = self.compression + zinfo._compresslevel = self.compresslevel + if zinfo.filename[-1] == '/': + zinfo.external_attr = 0o40775 << 16 # drwxrwxr-x + zinfo.external_attr |= 0x10 # MS-DOS directory flag + else: + zinfo.external_attr = 0o600 << 16 # ?rw------- + else: + zinfo = zinfo_or_arcname + + if not self.fp: + raise ValueError( + "Attempt to write to ZIP archive that was already closed") + if self._writing: + raise ValueError( + "Can't write to ZIP archive while an open writing handle exists." + ) + + if compress_type is not None: + zinfo.compress_type = compress_type + + if compresslevel is not None: + zinfo._compresslevel = compresslevel + + zinfo.file_size = len(data) # Uncompressed size + with self._lock: + with self.open(zinfo, mode='w') as dest: + dest.write(data) + + def __del__(self): + """Call the "close()" method in case the user forgot.""" + self.close() + + def close(self): + """Close the file, and for mode 'w', 'x' and 'a' write the ending + records.""" + if self.fp is None: + return + + if self._writing: + raise ValueError("Can't close the ZIP file while there is " + "an open writing handle on it. " + "Close the writing handle before closing the zip.") + + try: + if self.mode in ('w', 'x', 'a') and self._didModify: # write ending records + with self._lock: + if self._seekable: + self.fp.seek(self.start_dir) + self._write_end_record() + finally: + fp = self.fp + self.fp = None + self._fpclose(fp) + + def _write_end_record(self): + for zinfo in self.filelist: # write central directory + dt = zinfo.date_time + dosdate = (dt[0] - 1980) << 9 | dt[1] << 5 | dt[2] + dostime = dt[3] << 11 | dt[4] << 5 | (dt[5] // 2) + extra = [] + if zinfo.file_size > ZIP64_LIMIT \ + or zinfo.compress_size > ZIP64_LIMIT: + extra.append(zinfo.file_size) + extra.append(zinfo.compress_size) + file_size = 0xffffffff + compress_size = 0xffffffff + else: + file_size = zinfo.file_size + compress_size = zinfo.compress_size + + if zinfo.header_offset > ZIP64_LIMIT: + extra.append(zinfo.header_offset) + header_offset = 0xffffffff + else: + header_offset = zinfo.header_offset + + extra_data = zinfo.extra + min_version = 0 + if extra: + # Append a ZIP64 field to the extra's + extra_data = _strip_extra(extra_data, (1,)) + extra_data = struct.pack( + ' ZIP_FILECOUNT_LIMIT: + requires_zip64 = "Files count" + elif centDirOffset > ZIP64_LIMIT: + requires_zip64 = "Central directory offset" + elif centDirSize > ZIP64_LIMIT: + requires_zip64 = "Central directory size" + if requires_zip64: + # Need to write the ZIP64 end-of-archive records + if not self._allowZip64: + raise LargeZipFile(requires_zip64 + + " would require ZIP64 extensions") + zip64endrec = struct.pack( + structEndArchive64, stringEndArchive64, + 44, 45, 45, 0, 0, centDirCount, centDirCount, + centDirSize, centDirOffset) + self.fp.write(zip64endrec) + + zip64locrec = struct.pack( + structEndArchive64Locator, + stringEndArchive64Locator, 0, pos2, 1) + self.fp.write(zip64locrec) + centDirCount = min(centDirCount, 0xFFFF) + centDirSize = min(centDirSize, 0xFFFFFFFF) + centDirOffset = min(centDirOffset, 0xFFFFFFFF) + + endrec = struct.pack(structEndArchive, stringEndArchive, + 0, 0, centDirCount, centDirCount, + centDirSize, centDirOffset, len(self._comment)) + self.fp.write(endrec) + self.fp.write(self._comment) + self.fp.flush() + + def _fpclose(self, fp): + assert self._fileRefCnt > 0 + self._fileRefCnt -= 1 + if not self._fileRefCnt and not self._filePassed: + fp.close() + + +class PyZipFile(ZipFile): + """Class to create ZIP archives with Python library files and packages.""" + + def __init__(self, file, mode="r", compression=ZIP_STORED, + allowZip64=True, optimize=-1): + ZipFile.__init__(self, file, mode=mode, compression=compression, + allowZip64=allowZip64) + self._optimize = optimize + + def writepy(self, pathname, basename="", filterfunc=None): + """Add all files from "pathname" to the ZIP archive. + + If pathname is a package directory, search the directory and + all package subdirectories recursively for all *.py and enter + the modules into the archive. If pathname is a plain + directory, listdir *.py and enter all modules. Else, pathname + must be a Python *.py file and the module will be put into the + archive. Added modules are always module.pyc. + This method will compile the module.py into module.pyc if + necessary. + If filterfunc(pathname) is given, it is called with every argument. + When it is False, the file or directory is skipped. + """ + pathname = os.fspath(pathname) + if filterfunc and not filterfunc(pathname): + if self.debug: + label = 'path' if os.path.isdir(pathname) else 'file' + print('%s %r skipped by filterfunc' % (label, pathname)) + return + dir, name = os.path.split(pathname) + if os.path.isdir(pathname): + initname = os.path.join(pathname, "__init__.py") + if os.path.isfile(initname): + # This is a package directory, add it + if basename: + basename = "%s/%s" % (basename, name) + else: + basename = name + if self.debug: + print("Adding package in", pathname, "as", basename) + fname, arcname = self._get_codename(initname[0:-3], basename) + if self.debug: + print("Adding", arcname) + self.write(fname, arcname) + dirlist = sorted(os.listdir(pathname)) + dirlist.remove("__init__.py") + # Add all *.py files and package subdirectories + for filename in dirlist: + path = os.path.join(pathname, filename) + root, ext = os.path.splitext(filename) + if os.path.isdir(path): + if os.path.isfile(os.path.join(path, "__init__.py")): + # This is a package directory, add it + self.writepy(path, basename, + filterfunc=filterfunc) # Recursive call + elif ext == ".py": + if filterfunc and not filterfunc(path): + if self.debug: + print('file %r skipped by filterfunc' % path) + continue + fname, arcname = self._get_codename(path[0:-3], + basename) + if self.debug: + print("Adding", arcname) + self.write(fname, arcname) + else: + # This is NOT a package directory, add its files at top level + if self.debug: + print("Adding files from directory", pathname) + for filename in sorted(os.listdir(pathname)): + path = os.path.join(pathname, filename) + root, ext = os.path.splitext(filename) + if ext == ".py": + if filterfunc and not filterfunc(path): + if self.debug: + print('file %r skipped by filterfunc' % path) + continue + fname, arcname = self._get_codename(path[0:-3], + basename) + if self.debug: + print("Adding", arcname) + self.write(fname, arcname) + else: + if pathname[-3:] != ".py": + raise RuntimeError( + 'Files added with writepy() must end with ".py"') + fname, arcname = self._get_codename(pathname[0:-3], basename) + if self.debug: + print("Adding file", arcname) + self.write(fname, arcname) + + def _get_codename(self, pathname, basename): + """Return (filename, archivename) for the path. + + Given a module name path, return the correct file path and + archive name, compiling if necessary. For example, given + /python/lib/string, return (/python/lib/string.pyc, string). + """ + def _compile(file, optimize=-1): + import py_compile + if self.debug: + print("Compiling", file) + try: + py_compile.compile(file, doraise=True, optimize=optimize) + except py_compile.PyCompileError as err: + print(err.msg) + return False + return True + + file_py = pathname + ".py" + file_pyc = pathname + ".pyc" + pycache_opt0 = importlib.util.cache_from_source(file_py, optimization='') + pycache_opt1 = importlib.util.cache_from_source(file_py, optimization=1) + pycache_opt2 = importlib.util.cache_from_source(file_py, optimization=2) + if self._optimize == -1: + # legacy mode: use whatever file is present + if (os.path.isfile(file_pyc) and + os.stat(file_pyc).st_mtime >= os.stat(file_py).st_mtime): + # Use .pyc file. + arcname = fname = file_pyc + elif (os.path.isfile(pycache_opt0) and + os.stat(pycache_opt0).st_mtime >= os.stat(file_py).st_mtime): + # Use the __pycache__/*.pyc file, but write it to the legacy pyc + # file name in the archive. + fname = pycache_opt0 + arcname = file_pyc + elif (os.path.isfile(pycache_opt1) and + os.stat(pycache_opt1).st_mtime >= os.stat(file_py).st_mtime): + # Use the __pycache__/*.pyc file, but write it to the legacy pyc + # file name in the archive. + fname = pycache_opt1 + arcname = file_pyc + elif (os.path.isfile(pycache_opt2) and + os.stat(pycache_opt2).st_mtime >= os.stat(file_py).st_mtime): + # Use the __pycache__/*.pyc file, but write it to the legacy pyc + # file name in the archive. + fname = pycache_opt2 + arcname = file_pyc + else: + # Compile py into PEP 3147 pyc file. + if _compile(file_py): + if sys.flags.optimize == 0: + fname = pycache_opt0 + elif sys.flags.optimize == 1: + fname = pycache_opt1 + else: + fname = pycache_opt2 + arcname = file_pyc + else: + fname = arcname = file_py + else: + # new mode: use given optimization level + if self._optimize == 0: + fname = pycache_opt0 + arcname = file_pyc + else: + arcname = file_pyc + if self._optimize == 1: + fname = pycache_opt1 + elif self._optimize == 2: + fname = pycache_opt2 + else: + msg = "invalid value for 'optimize': {!r}".format(self._optimize) + raise ValueError(msg) + if not (os.path.isfile(fname) and + os.stat(fname).st_mtime >= os.stat(file_py).st_mtime): + if not _compile(file_py, optimize=self._optimize): + fname = arcname = file_py + archivename = os.path.split(arcname)[1] + if basename: + archivename = "%s/%s" % (basename, archivename) + return (fname, archivename) + + +def _unique_everseen(iterable, key=None): + "List unique elements, preserving order. Remember all elements ever seen." + # unique_everseen('AAAABBBCCDAABBB') --> A B C D + # unique_everseen('ABBCcAD', str.lower) --> A B C D + seen = set() + seen_add = seen.add + if key is None: + for element in itertools.filterfalse(seen.__contains__, iterable): + seen_add(element) + yield element + else: + for element in iterable: + k = key(element) + if k not in seen: + seen_add(k) + yield element + + +def _parents(path): + """ + Given a path with elements separated by + posixpath.sep, generate all parents of that path. + + >>> list(_parents('b/d')) + ['b'] + >>> list(_parents('/b/d/')) + ['/b'] + >>> list(_parents('b/d/f/')) + ['b/d', 'b'] + >>> list(_parents('b')) + [] + >>> list(_parents('')) + [] + """ + return itertools.islice(_ancestry(path), 1, None) + + +def _ancestry(path): + """ + Given a path with elements separated by + posixpath.sep, generate all elements of that path + + >>> list(_ancestry('b/d')) + ['b/d', 'b'] + >>> list(_ancestry('/b/d/')) + ['/b/d', '/b'] + >>> list(_ancestry('b/d/f/')) + ['b/d/f', 'b/d', 'b'] + >>> list(_ancestry('b')) + ['b'] + >>> list(_ancestry('')) + [] + """ + path = path.rstrip(posixpath.sep) + while path and path != posixpath.sep: + yield path + path, tail = posixpath.split(path) + + +class CompleteDirs(ZipFile): + """ + A ZipFile subclass that ensures that implied directories + are always included in the namelist. + """ + + @staticmethod + def _implied_dirs(names): + parents = itertools.chain.from_iterable(map(_parents, names)) + # Deduplicate entries in original order + implied_dirs = OrderedDict.fromkeys( + p + posixpath.sep for p in parents + # Cast names to a set for O(1) lookups + if p + posixpath.sep not in set(names) + ) + return implied_dirs + + def namelist(self): + names = super(CompleteDirs, self).namelist() + return names + list(self._implied_dirs(names)) + + def _name_set(self): + return set(self.namelist()) + + def resolve_dir(self, name): + """ + If the name represents a directory, return that name + as a directory (with the trailing slash). + """ + names = self._name_set() + dirname = name + '/' + dir_match = name not in names and dirname in names + return dirname if dir_match else name + + @classmethod + def make(cls, source): + """ + Given a source (filename or zipfile), return an + appropriate CompleteDirs subclass. + """ + if isinstance(source, CompleteDirs): + return source + + if not isinstance(source, ZipFile): + return cls(source) + + # Only allow for FastPath when supplied zipfile is read-only + if 'r' not in source.mode: + cls = CompleteDirs + + res = cls.__new__(cls) + vars(res).update(vars(source)) + return res + + +class FastLookup(CompleteDirs): + """ + ZipFile subclass to ensure implicit + dirs exist and are resolved rapidly. + """ + def namelist(self): + with contextlib.suppress(AttributeError): + return self.__names + self.__names = super(FastLookup, self).namelist() + return self.__names + + def _name_set(self): + with contextlib.suppress(AttributeError): + return self.__lookup + self.__lookup = super(FastLookup, self)._name_set() + return self.__lookup + + +class Path: + """ + A pathlib-compatible interface for zip files. + + Consider a zip file with this structure:: + + . + ├── a.txt + └── b + ├── c.txt + └── d + └── e.txt + + >>> data = io.BytesIO() + >>> zf = ZipFile(data, 'w') + >>> zf.writestr('a.txt', 'content of a') + >>> zf.writestr('b/c.txt', 'content of c') + >>> zf.writestr('b/d/e.txt', 'content of e') + >>> zf.filename = 'abcde.zip' + + Path accepts the zipfile object itself or a filename + + >>> root = Path(zf) + + From there, several path operations are available. + + Directory iteration (including the zip file itself): + + >>> a, b = root.iterdir() + >>> a + Path('abcde.zip', 'a.txt') + >>> b + Path('abcde.zip', 'b/') + + name property: + + >>> b.name + 'b' + + join with divide operator: + + >>> c = b / 'c.txt' + >>> c + Path('abcde.zip', 'b/c.txt') + >>> c.name + 'c.txt' + + Read text: + + >>> c.read_text() + 'content of c' + + existence: + + >>> c.exists() + True + >>> (b / 'missing.txt').exists() + False + + Coercion to string: + + >>> str(c) + 'abcde.zip/b/c.txt' + """ + + __repr = "{self.__class__.__name__}({self.root.filename!r}, {self.at!r})" + + def __init__(self, root, at=""): + self.root = FastLookup.make(root) + self.at = at + + @property + def open(self): + return functools.partial(self.root.open, self.at) + + @property + def name(self): + return posixpath.basename(self.at.rstrip("/")) + + def read_text(self, *args, **kwargs): + with self.open() as strm: + return io.TextIOWrapper(strm, *args, **kwargs).read() + + def read_bytes(self): + with self.open() as strm: + return strm.read() + + def _is_child(self, path): + return posixpath.dirname(path.at.rstrip("/")) == self.at.rstrip("/") + + def _next(self, at): + return Path(self.root, at) + + def is_dir(self): + return not self.at or self.at.endswith("/") + + def is_file(self): + return not self.is_dir() + + def exists(self): + return self.at in self.root._name_set() + + def iterdir(self): + if not self.is_dir(): + raise ValueError("Can't listdir a file") + subs = map(self._next, self.root.namelist()) + return filter(self._is_child, subs) + + def __str__(self): + return posixpath.join(self.root.filename, self.at) + + def __repr__(self): + return self.__repr.format(self=self) + + def joinpath(self, add): + next = posixpath.join(self.at, add) + return self._next(self.root.resolve_dir(next)) + + __truediv__ = joinpath + + @property + def parent(self): + parent_at = posixpath.dirname(self.at.rstrip('/')) + if parent_at: + parent_at += '/' + return self._next(parent_at) + + +def main(args=None): + import argparse + + description = 'A simple command-line interface for zipfile module.' + parser = argparse.ArgumentParser(description=description) + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument('-l', '--list', metavar='', + help='Show listing of a zipfile') + group.add_argument('-e', '--extract', nargs=2, + metavar=('', ''), + help='Extract zipfile into target dir') + group.add_argument('-c', '--create', nargs='+', + metavar=('', ''), + help='Create zipfile from sources') + group.add_argument('-t', '--test', metavar='', + help='Test if a zipfile is valid') + args = parser.parse_args(args) + + if args.test is not None: + src = args.test + with ZipFile(src, 'r') as zf: + badfile = zf.testzip() + if badfile: + print("The following enclosed file is corrupted: {!r}".format(badfile)) + print("Done testing") + + elif args.list is not None: + src = args.list + with ZipFile(src, 'r') as zf: + zf.printdir() + + elif args.extract is not None: + src, curdir = args.extract + with ZipFile(src, 'r') as zf: + zf.extractall(curdir) + + elif args.create is not None: + zip_name = args.create.pop(0) + files = args.create + + def addToZip(zf, path, zippath): + if os.path.isfile(path): + zf.write(path, zippath, ZIP_DEFLATED) + elif os.path.isdir(path): + if zippath: + zf.write(path, zippath) + for nm in sorted(os.listdir(path)): + addToZip(zf, + os.path.join(path, nm), os.path.join(zippath, nm)) + # else: ignore + + with ZipFile(zip_name, 'w') as zf: + for path in files: + zippath = os.path.basename(path) + if not zippath: + zippath = os.path.basename(os.path.dirname(path)) + if zippath in ('', os.curdir, os.pardir): + zippath = '' + addToZip(zf, path, zippath) + + +if __name__ == "__main__": + main()