Yanz Mini Shell
[_]
[-]
[X]
[
HomeShell 1
] [
HomeShell 2
] [
Upload
] [
Command Shell
] [
Scripting
] [
About
]
[ Directory ] =>
/
home
veronikagstoette
public_html
wp-content
Action
[*]
New File
[*]
New Folder
Sensitive File
[*]
/etc/passwd
[*]
/etc/shadow
[*]
/etc/resolv.conf
[
Delete
] [
Edit
] [
Rename
] [
Back
]
usr/lib64/python3.12/collections/__init__.py 0000644 00000146232 15246343252 0014567 0 ustar 00 '''This module implements specialized container datatypes providing alternatives to Python's general purpose built-in containers, dict, list, set, and tuple. * namedtuple factory function for creating tuple subclasses with named fields * deque list-like container with fast appends and pops on either end * ChainMap dict-like class for creating a single view of multiple mappings * Counter dict subclass for counting hashable objects * OrderedDict dict subclass that remembers the order entries were added * defaultdict dict subclass that calls a factory function to supply missing values * UserDict wrapper around dictionary objects for easier dict subclassing * UserList wrapper around list objects for easier list subclassing * UserString wrapper around string objects for easier string subclassing ''' __all__ = [ 'ChainMap', 'Counter', 'OrderedDict', 'UserDict', 'UserList', 'UserString', 'defaultdict', 'deque', 'namedtuple', ] import _collections_abc import sys as _sys from itertools import chain as _chain from itertools import repeat as _repeat from itertools import starmap as _starmap from keyword import iskeyword as _iskeyword from operator import eq as _eq from operator import itemgetter as _itemgetter from reprlib import recursive_repr as _recursive_repr from _weakref import proxy as _proxy try: from _collections import deque except ImportError: pass else: _collections_abc.MutableSequence.register(deque) try: from _collections import _deque_iterator except ImportError: pass try: from _collections import defaultdict except ImportError: pass ################################################################################ ### OrderedDict ################################################################################ class _OrderedDictKeysView(_collections_abc.KeysView): def __reversed__(self): yield from reversed(self._mapping) class _OrderedDictItemsView(_collections_abc.ItemsView): def __reversed__(self): for key in reversed(self._mapping): yield (key, self._mapping[key]) class _OrderedDictValuesView(_collections_abc.ValuesView): def __reversed__(self): for key in reversed(self._mapping): yield self._mapping[key] class _Link(object): __slots__ = 'prev', 'next', 'key', '__weakref__' class OrderedDict(dict): 'Dictionary that remembers insertion order' # An inherited dict maps keys to values. # The inherited dict provides __getitem__, __len__, __contains__, and get. # The remaining methods are order-aware. # Big-O running times for all methods are the same as regular dictionaries. # The internal self.__map dict maps keys to links in a doubly linked list. # The circular doubly linked list starts and ends with a sentinel element. # The sentinel element never gets deleted (this simplifies the algorithm). # The sentinel is in self.__hardroot with a weakref proxy in self.__root. # The prev links are weakref proxies (to prevent circular references). # Individual links are kept alive by the hard reference in self.__map. # Those hard references disappear when a key is deleted from an OrderedDict. def __new__(cls, /, *args, **kwds): "Create the ordered dict object and set up the underlying structures." self = dict.__new__(cls) self.__hardroot = _Link() self.__root = root = _proxy(self.__hardroot) root.prev = root.next = root self.__map = {} return self def __init__(self, other=(), /, **kwds): '''Initialize an ordered dictionary. The signature is the same as regular dictionaries. Keyword argument order is preserved. ''' self.__update(other, **kwds) def __setitem__(self, key, value, dict_setitem=dict.__setitem__, proxy=_proxy, Link=_Link): 'od.__setitem__(i, y) <==> od[i]=y' # Setting a new item creates a new link at the end of the linked list, # and the inherited dictionary is updated with the new key/value pair. if key not in self: self.__map[key] = link = Link() root = self.__root last = root.prev link.prev, link.next, link.key = last, root, key last.next = link root.prev = proxy(link) dict_setitem(self, key, value) def __delitem__(self, key, dict_delitem=dict.__delitem__): 'od.__delitem__(y) <==> del od[y]' # Deleting an existing item uses self.__map to find the link which gets # removed by updating the links in the predecessor and successor nodes. dict_delitem(self, key) link = self.__map.pop(key) link_prev = link.prev link_next = link.next link_prev.next = link_next link_next.prev = link_prev link.prev = None link.next = None def __iter__(self): 'od.__iter__() <==> iter(od)' # Traverse the linked list in order. root = self.__root curr = root.next while curr is not root: yield curr.key curr = curr.next def __reversed__(self): 'od.__reversed__() <==> reversed(od)' # Traverse the linked list in reverse order. root = self.__root curr = root.prev while curr is not root: yield curr.key curr = curr.prev def clear(self): 'od.clear() -> None. Remove all items from od.' root = self.__root root.prev = root.next = root self.__map.clear() dict.clear(self) def popitem(self, last=True): '''Remove and return a (key, value) pair from the dictionary. Pairs are returned in LIFO order if last is true or FIFO order if false. ''' if not self: raise KeyError('dictionary is empty') root = self.__root if last: link = root.prev link_prev = link.prev link_prev.next = root root.prev = link_prev else: link = root.next link_next = link.next root.next = link_next link_next.prev = root key = link.key del self.__map[key] value = dict.pop(self, key) return key, value def move_to_end(self, key, last=True): '''Move an existing element to the end (or beginning if last is false). Raise KeyError if the element does not exist. ''' link = self.__map[key] link_prev = link.prev link_next = link.next soft_link = link_next.prev link_prev.next = link_next link_next.prev = link_prev root = self.__root if last: last = root.prev link.prev = last link.next = root root.prev = soft_link last.next = link else: first = root.next link.prev = root link.next = first first.prev = soft_link root.next = link def __sizeof__(self): sizeof = _sys.getsizeof n = len(self) + 1 # number of links including root size = sizeof(self.__dict__) # instance dictionary size += sizeof(self.__map) * 2 # internal dict and inherited dict size += sizeof(self.__hardroot) * n # link objects size += sizeof(self.__root) * n # proxy objects return size update = __update = _collections_abc.MutableMapping.update def keys(self): "D.keys() -> a set-like object providing a view on D's keys" return _OrderedDictKeysView(self) def items(self): "D.items() -> a set-like object providing a view on D's items" return _OrderedDictItemsView(self) def values(self): "D.values() -> an object providing a view on D's values" return _OrderedDictValuesView(self) __ne__ = _collections_abc.MutableMapping.__ne__ __marker = object() def pop(self, key, default=__marker): '''od.pop(k[,d]) -> v, remove specified key and return the corresponding value. If key is not found, d is returned if given, otherwise KeyError is raised. ''' marker = self.__marker result = dict.pop(self, key, marker) if result is not marker: # The same as in __delitem__(). link = self.__map.pop(key) link_prev = link.prev link_next = link.next link_prev.next = link_next link_next.prev = link_prev link.prev = None link.next = None return result if default is marker: raise KeyError(key) return default def setdefault(self, key, default=None): '''Insert key with a value of default if key is not in the dictionary. Return the value for key if key is in the dictionary, else default. ''' if key in self: return self[key] self[key] = default return default @_recursive_repr() def __repr__(self): 'od.__repr__() <==> repr(od)' if not self: return '%s()' % (self.__class__.__name__,) return '%s(%r)' % (self.__class__.__name__, dict(self.items())) def __reduce__(self): 'Return state information for pickling' state = self.__getstate__() if state: if isinstance(state, tuple): state, slots = state else: slots = {} state = state.copy() slots = slots.copy() for k in vars(OrderedDict()): state.pop(k, None) slots.pop(k, None) if slots: state = state, slots else: state = state or None return self.__class__, (), state, None, iter(self.items()) def copy(self): 'od.copy() -> a shallow copy of od' return self.__class__(self) @classmethod def fromkeys(cls, iterable, value=None): '''Create a new ordered dictionary with keys from iterable and values set to value. ''' self = cls() for key in iterable: self[key] = value return self def __eq__(self, other): '''od.__eq__(y) <==> od==y. Comparison to another OD is order-sensitive while comparison to a regular mapping is order-insensitive. ''' if isinstance(other, OrderedDict): return dict.__eq__(self, other) and all(map(_eq, self, other)) return dict.__eq__(self, other) def __ior__(self, other): self.update(other) return self def __or__(self, other): if not isinstance(other, dict): return NotImplemented new = self.__class__(self) new.update(other) return new def __ror__(self, other): if not isinstance(other, dict): return NotImplemented new = self.__class__(other) new.update(self) return new try: from _collections import OrderedDict except ImportError: # Leave the pure Python version in place. pass ################################################################################ ### namedtuple ################################################################################ try: from _collections import _tuplegetter except ImportError: _tuplegetter = lambda index, doc: property(_itemgetter(index), doc=doc) def namedtuple(typename, field_names, *, rename=False, defaults=None, module=None): """Returns a new subclass of tuple with named fields. >>> Point = namedtuple('Point', ['x', 'y']) >>> Point.__doc__ # docstring for the new class 'Point(x, y)' >>> p = Point(11, y=22) # instantiate with positional args or keywords >>> p[0] + p[1] # indexable like a plain tuple 33 >>> x, y = p # unpack like a regular tuple >>> x, y (11, 22) >>> p.x + p.y # fields also accessible by name 33 >>> d = p._asdict() # convert to a dictionary >>> d['x'] 11 >>> Point(**d) # convert from a dictionary Point(x=11, y=22) >>> p._replace(x=100) # _replace() is like str.replace() but targets named fields Point(x=100, y=22) """ # Validate the field names. At the user's option, either generate an error # message or automatically replace the field name with a valid name. if isinstance(field_names, str): field_names = field_names.replace(',', ' ').split() field_names = list(map(str, field_names)) typename = _sys.intern(str(typename)) if rename: seen = set() for index, name in enumerate(field_names): if (not name.isidentifier() or _iskeyword(name) or name.startswith('_') or name in seen): field_names[index] = f'_{index}' seen.add(name) for name in [typename] + field_names: if type(name) is not str: raise TypeError('Type names and field names must be strings') if not name.isidentifier(): raise ValueError('Type names and field names must be valid ' f'identifiers: {name!r}') if _iskeyword(name): raise ValueError('Type names and field names cannot be a ' f'keyword: {name!r}') seen = set() for name in field_names: if name.startswith('_') and not rename: raise ValueError('Field names cannot start with an underscore: ' f'{name!r}') if name in seen: raise ValueError(f'Encountered duplicate field name: {name!r}') seen.add(name) field_defaults = {} if defaults is not None: defaults = tuple(defaults) if len(defaults) > len(field_names): raise TypeError('Got more default values than field names') field_defaults = dict(reversed(list(zip(reversed(field_names), reversed(defaults))))) # Variables used in the methods and docstrings field_names = tuple(map(_sys.intern, field_names)) num_fields = len(field_names) arg_list = ', '.join(field_names) if num_fields == 1: arg_list += ',' repr_fmt = '(' + ', '.join(f'{name}=%r' for name in field_names) + ')' tuple_new = tuple.__new__ _dict, _tuple, _len, _map, _zip = dict, tuple, len, map, zip # Create all the named tuple methods to be added to the class namespace namespace = { '_tuple_new': tuple_new, '__builtins__': {}, '__name__': f'namedtuple_{typename}', } code = f'lambda _cls, {arg_list}: _tuple_new(_cls, ({arg_list}))' __new__ = eval(code, namespace) __new__.__name__ = '__new__' __new__.__doc__ = f'Create new instance of {typename}({arg_list})' if defaults is not None: __new__.__defaults__ = defaults @classmethod def _make(cls, iterable): result = tuple_new(cls, iterable) if _len(result) != num_fields: raise TypeError(f'Expected {num_fields} arguments, got {len(result)}') return result _make.__func__.__doc__ = (f'Make a new {typename} object from a sequence ' 'or iterable') def _replace(self, /, **kwds): result = self._make(_map(kwds.pop, field_names, self)) if kwds: raise ValueError(f'Got unexpected field names: {list(kwds)!r}') return result _replace.__doc__ = (f'Return a new {typename} object replacing specified ' 'fields with new values') def __repr__(self): 'Return a nicely formatted representation string' return self.__class__.__name__ + repr_fmt % self def _asdict(self): 'Return a new dict which maps field names to their values.' return _dict(_zip(self._fields, self)) def __getnewargs__(self): 'Return self as a plain tuple. Used by copy and pickle.' return _tuple(self) # Modify function metadata to help with introspection and debugging for method in ( __new__, _make.__func__, _replace, __repr__, _asdict, __getnewargs__, ): method.__qualname__ = f'{typename}.{method.__name__}' # Build-up the class namespace dictionary # and use type() to build the result class class_namespace = { '__doc__': f'{typename}({arg_list})', '__slots__': (), '_fields': field_names, '_field_defaults': field_defaults, '__new__': __new__, '_make': _make, '_replace': _replace, '__repr__': __repr__, '_asdict': _asdict, '__getnewargs__': __getnewargs__, '__match_args__': field_names, } for index, name in enumerate(field_names): doc = _sys.intern(f'Alias for field number {index}') class_namespace[name] = _tuplegetter(index, doc) result = type(typename, (tuple,), class_namespace) # For pickling to work, the __module__ variable needs to be set to the frame # where the named tuple is created. Bypass this step in environments where # sys._getframe is not defined (Jython for example) or sys._getframe is not # defined for arguments greater than 0 (IronPython), or where the user has # specified a particular module. if module is None: try: module = _sys._getframemodulename(1) or '__main__' except AttributeError: try: module = _sys._getframe(1).f_globals.get('__name__', '__main__') except (AttributeError, ValueError): pass if module is not None: result.__module__ = module return result ######################################################################## ### Counter ######################################################################## def _count_elements(mapping, iterable): 'Tally elements from the iterable.' mapping_get = mapping.get for elem in iterable: mapping[elem] = mapping_get(elem, 0) + 1 try: # Load C helper function if available from _collections import _count_elements except ImportError: pass class Counter(dict): '''Dict subclass for counting hashable items. Sometimes called a bag or multiset. Elements are stored as dictionary keys and their counts are stored as dictionary values. >>> c = Counter('abcdeabcdabcaba') # count elements from a string >>> c.most_common(3) # three most common elements [('a', 5), ('b', 4), ('c', 3)] >>> sorted(c) # list all unique elements ['a', 'b', 'c', 'd', 'e'] >>> ''.join(sorted(c.elements())) # list elements with repetitions 'aaaaabbbbcccdde' >>> sum(c.values()) # total of all counts 15 >>> c['a'] # count of letter 'a' 5 >>> for elem in 'shazam': # update counts from an iterable ... c[elem] += 1 # by adding 1 to each element's count >>> c['a'] # now there are seven 'a' 7 >>> del c['b'] # remove all 'b' >>> c['b'] # now there are zero 'b' 0 >>> d = Counter('simsalabim') # make another counter >>> c.update(d) # add in the second counter >>> c['a'] # now there are nine 'a' 9 >>> c.clear() # empty the counter >>> c Counter() Note: If a count is set to zero or reduced to zero, it will remain in the counter until the entry is deleted or the counter is cleared: >>> c = Counter('aaabbc') >>> c['b'] -= 2 # reduce the count of 'b' by two >>> c.most_common() # 'b' is still in, but its count is zero [('a', 3), ('c', 1), ('b', 0)] ''' # References: # http://en.wikipedia.org/wiki/Multiset # http://www.gnu.org/software/smalltalk/manual-base/html_node/Bag.html # http://www.java2s.com/Tutorial/Cpp/0380__set-multiset/Catalog0380__set-multiset.htm # http://code.activestate.com/recipes/259174/ # Knuth, TAOCP Vol. II section 4.6.3 def __init__(self, iterable=None, /, **kwds): '''Create a new, empty Counter object. And if given, count elements from an input iterable. Or, initialize the count from another mapping of elements to their counts. >>> c = Counter() # a new, empty counter >>> c = Counter('gallahad') # a new counter from an iterable >>> c = Counter({'a': 4, 'b': 2}) # a new counter from a mapping >>> c = Counter(a=4, b=2) # a new counter from keyword args ''' super().__init__() self.update(iterable, **kwds) def __missing__(self, key): 'The count of elements not in the Counter is zero.' # Needed so that self[missing_item] does not raise KeyError return 0 def total(self): 'Sum of the counts' return sum(self.values()) def most_common(self, n=None): '''List the n most common elements and their counts from the most common to the least. If n is None, then list all element counts. >>> Counter('abracadabra').most_common(3) [('a', 5), ('b', 2), ('r', 2)] ''' # Emulate Bag.sortedByCount from Smalltalk if n is None: return sorted(self.items(), key=_itemgetter(1), reverse=True) # Lazy import to speedup Python startup time import heapq return heapq.nlargest(n, self.items(), key=_itemgetter(1)) def elements(self): '''Iterator over elements repeating each as many times as its count. >>> c = Counter('ABCABC') >>> sorted(c.elements()) ['A', 'A', 'B', 'B', 'C', 'C'] Knuth's example for prime factors of 1836: 2**2 * 3**3 * 17**1 >>> import math >>> prime_factors = Counter({2: 2, 3: 3, 17: 1}) >>> math.prod(prime_factors.elements()) 1836 Note, if an element's count has been set to zero or is a negative number, elements() will ignore it. ''' # Emulate Bag.do from Smalltalk and Multiset.begin from C++. return _chain.from_iterable(_starmap(_repeat, self.items())) # Override dict methods where necessary @classmethod def fromkeys(cls, iterable, v=None): # There is no equivalent method for counters because the semantics # would be ambiguous in cases such as Counter.fromkeys('aaabbc', v=2). # Initializing counters to zero values isn't necessary because zero # is already the default value for counter lookups. Initializing # to one is easily accomplished with Counter(set(iterable)). For # more exotic cases, create a dictionary first using a dictionary # comprehension or dict.fromkeys(). raise NotImplementedError( 'Counter.fromkeys() is undefined. Use Counter(iterable) instead.') def update(self, iterable=None, /, **kwds): '''Like dict.update() but add counts instead of replacing them. Source can be an iterable, a dictionary, or another Counter instance. >>> c = Counter('which') >>> c.update('witch') # add elements from another iterable >>> d = Counter('watch') >>> c.update(d) # add elements from another counter >>> c['h'] # four 'h' in which, witch, and watch 4 ''' # The regular dict.update() operation makes no sense here because the # replace behavior results in some of the original untouched counts # being mixed-in with all of the other counts for a mismash that # doesn't have a straight-forward interpretation in most counting # contexts. Instead, we implement straight-addition. Both the inputs # and outputs are allowed to contain zero and negative counts. if iterable is not None: if isinstance(iterable, _collections_abc.Mapping): if self: self_get = self.get for elem, count in iterable.items(): self[elem] = count + self_get(elem, 0) else: # fast path when counter is empty super().update(iterable) else: _count_elements(self, iterable) if kwds: self.update(kwds) def subtract(self, iterable=None, /, **kwds): '''Like dict.update() but subtracts counts instead of replacing them. Counts can be reduced below zero. Both the inputs and outputs are allowed to contain zero and negative counts. Source can be an iterable, a dictionary, or another Counter instance. >>> c = Counter('which') >>> c.subtract('witch') # subtract elements from another iterable >>> c.subtract(Counter('watch')) # subtract elements from another counter >>> c['h'] # 2 in which, minus 1 in witch, minus 1 in watch 0 >>> c['w'] # 1 in which, minus 1 in witch, minus 1 in watch -1 ''' if iterable is not None: self_get = self.get if isinstance(iterable, _collections_abc.Mapping): for elem, count in iterable.items(): self[elem] = self_get(elem, 0) - count else: for elem in iterable: self[elem] = self_get(elem, 0) - 1 if kwds: self.subtract(kwds) def copy(self): 'Return a shallow copy.' return self.__class__(self) def __reduce__(self): return self.__class__, (dict(self),) def __delitem__(self, elem): 'Like dict.__delitem__() but does not raise KeyError for missing values.' if elem in self: super().__delitem__(elem) def __repr__(self): if not self: return f'{self.__class__.__name__}()' try: # dict() preserves the ordering returned by most_common() d = dict(self.most_common()) except TypeError: # handle case where values are not orderable d = dict(self) return f'{self.__class__.__name__}({d!r})' # Multiset-style mathematical operations discussed in: # Knuth TAOCP Volume II section 4.6.3 exercise 19 # and at http://en.wikipedia.org/wiki/Multiset # # Outputs guaranteed to only include positive counts. # # To strip negative and zero counts, add-in an empty counter: # c += Counter() # # Results are ordered according to when an element is first # encountered in the left operand and then by the order # encountered in the right operand. # # When the multiplicities are all zero or one, multiset operations # are guaranteed to be equivalent to the corresponding operations # for regular sets. # Given counter multisets such as: # cp = Counter(a=1, b=0, c=1) # cq = Counter(c=1, d=0, e=1) # The corresponding regular sets would be: # sp = {'a', 'c'} # sq = {'c', 'e'} # All of the following relations would hold: # set(cp + cq) == sp | sq # set(cp - cq) == sp - sq # set(cp | cq) == sp | sq # set(cp & cq) == sp & sq # (cp == cq) == (sp == sq) # (cp != cq) == (sp != sq) # (cp <= cq) == (sp <= sq) # (cp < cq) == (sp < sq) # (cp >= cq) == (sp >= sq) # (cp > cq) == (sp > sq) def __eq__(self, other): 'True if all counts agree. Missing counts are treated as zero.' if not isinstance(other, Counter): return NotImplemented return all(self[e] == other[e] for c in (self, other) for e in c) def __ne__(self, other): 'True if any counts disagree. Missing counts are treated as zero.' if not isinstance(other, Counter): return NotImplemented return not self == other def __le__(self, other): 'True if all counts in self are a subset of those in other.' if not isinstance(other, Counter): return NotImplemented return all(self[e] <= other[e] for c in (self, other) for e in c) def __lt__(self, other): 'True if all counts in self are a proper subset of those in other.' if not isinstance(other, Counter): return NotImplemented return self <= other and self != other def __ge__(self, other): 'True if all counts in self are a superset of those in other.' if not isinstance(other, Counter): return NotImplemented return all(self[e] >= other[e] for c in (self, other) for e in c) def __gt__(self, other): 'True if all counts in self are a proper superset of those in other.' if not isinstance(other, Counter): return NotImplemented return self >= other and self != other def __add__(self, other): '''Add counts from two counters. >>> Counter('abbb') + Counter('bcc') Counter({'b': 4, 'c': 2, 'a': 1}) ''' if not isinstance(other, Counter): return NotImplemented result = Counter() for elem, count in self.items(): newcount = count + other[elem] if newcount > 0: result[elem] = newcount for elem, count in other.items(): if elem not in self and count > 0: result[elem] = count return result def __sub__(self, other): ''' Subtract count, but keep only results with positive counts. >>> Counter('abbbc') - Counter('bccd') Counter({'b': 2, 'a': 1}) ''' if not isinstance(other, Counter): return NotImplemented result = Counter() for elem, count in self.items(): newcount = count - other[elem] if newcount > 0: result[elem] = newcount for elem, count in other.items(): if elem not in self and count < 0: result[elem] = 0 - count return result def __or__(self, other): '''Union is the maximum of value in either of the input counters. >>> Counter('abbb') | Counter('bcc') Counter({'b': 3, 'c': 2, 'a': 1}) ''' if not isinstance(other, Counter): return NotImplemented result = Counter() for elem, count in self.items(): other_count = other[elem] newcount = other_count if count < other_count else count if newcount > 0: result[elem] = newcount for elem, count in other.items(): if elem not in self and count > 0: result[elem] = count return result def __and__(self, other): ''' Intersection is the minimum of corresponding counts. >>> Counter('abbb') & Counter('bcc') Counter({'b': 1}) ''' if not isinstance(other, Counter): return NotImplemented result = Counter() for elem, count in self.items(): other_count = other[elem] newcount = count if count < other_count else other_count if newcount > 0: result[elem] = newcount return result def __pos__(self): 'Adds an empty counter, effectively stripping negative and zero counts' result = Counter() for elem, count in self.items(): if count > 0: result[elem] = count return result def __neg__(self): '''Subtracts from an empty counter. Strips positive and zero counts, and flips the sign on negative counts. ''' result = Counter() for elem, count in self.items(): if count < 0: result[elem] = 0 - count return result def _keep_positive(self): '''Internal method to strip elements with a negative or zero count''' nonpositive = [elem for elem, count in self.items() if not count > 0] for elem in nonpositive: del self[elem] return self def __iadd__(self, other): '''Inplace add from another counter, keeping only positive counts. >>> c = Counter('abbb') >>> c += Counter('bcc') >>> c Counter({'b': 4, 'c': 2, 'a': 1}) ''' for elem, count in other.items(): self[elem] += count return self._keep_positive() def __isub__(self, other): '''Inplace subtract counter, but keep only results with positive counts. >>> c = Counter('abbbc') >>> c -= Counter('bccd') >>> c Counter({'b': 2, 'a': 1}) ''' for elem, count in other.items(): self[elem] -= count return self._keep_positive() def __ior__(self, other): '''Inplace union is the maximum of value from either counter. >>> c = Counter('abbb') >>> c |= Counter('bcc') >>> c Counter({'b': 3, 'c': 2, 'a': 1}) ''' for elem, other_count in other.items(): count = self[elem] if other_count > count: self[elem] = other_count return self._keep_positive() def __iand__(self, other): '''Inplace intersection is the minimum of corresponding counts. >>> c = Counter('abbb') >>> c &= Counter('bcc') >>> c Counter({'b': 1}) ''' for elem, count in self.items(): other_count = other[elem] if other_count < count: self[elem] = other_count return self._keep_positive() ######################################################################## ### ChainMap ######################################################################## class ChainMap(_collections_abc.MutableMapping): ''' A ChainMap groups multiple dicts (or other mappings) together to create a single, updateable view. The underlying mappings are stored in a list. That list is public and can be accessed or updated using the *maps* attribute. There is no other state. Lookups search the underlying mappings successively until a key is found. In contrast, writes, updates, and deletions only operate on the first mapping. ''' def __init__(self, *maps): '''Initialize a ChainMap by setting *maps* to the given mappings. If no mappings are provided, a single empty dictionary is used. ''' self.maps = list(maps) or [{}] # always at least one map def __missing__(self, key): raise KeyError(key) def __getitem__(self, key): for mapping in self.maps: try: return mapping[key] # can't use 'key in mapping' with defaultdict except KeyError: pass return self.__missing__(key) # support subclasses that define __missing__ def get(self, key, default=None): return self[key] if key in self else default def __len__(self): return len(set().union(*self.maps)) # reuses stored hash values if possible def __iter__(self): d = {} for mapping in map(dict.fromkeys, reversed(self.maps)): d |= mapping # reuses stored hash values if possible return iter(d) def __contains__(self, key): return any(key in m for m in self.maps) def __bool__(self): return any(self.maps) @_recursive_repr() def __repr__(self): return f'{self.__class__.__name__}({", ".join(map(repr, self.maps))})' @classmethod def fromkeys(cls, iterable, *args): 'Create a ChainMap with a single dict created from the iterable.' return cls(dict.fromkeys(iterable, *args)) def copy(self): 'New ChainMap or subclass with a new copy of maps[0] and refs to maps[1:]' return self.__class__(self.maps[0].copy(), *self.maps[1:]) __copy__ = copy def new_child(self, m=None, **kwargs): # like Django's Context.push() '''New ChainMap with a new map followed by all previous maps. If no map is provided, an empty dict is used. Keyword arguments update the map or new empty dict. ''' if m is None: m = kwargs elif kwargs: m.update(kwargs) return self.__class__(m, *self.maps) @property def parents(self): # like Django's Context.pop() 'New ChainMap from maps[1:].' return self.__class__(*self.maps[1:]) def __setitem__(self, key, value): self.maps[0][key] = value def __delitem__(self, key): try: del self.maps[0][key] except KeyError: raise KeyError(f'Key not found in the first mapping: {key!r}') def popitem(self): 'Remove and return an item pair from maps[0]. Raise KeyError is maps[0] is empty.' try: return self.maps[0].popitem() except KeyError: raise KeyError('No keys found in the first mapping.') def pop(self, key, *args): 'Remove *key* from maps[0] and return its value. Raise KeyError if *key* not in maps[0].' try: return self.maps[0].pop(key, *args) except KeyError: raise KeyError(f'Key not found in the first mapping: {key!r}') def clear(self): 'Clear maps[0], leaving maps[1:] intact.' self.maps[0].clear() def __ior__(self, other): self.maps[0].update(other) return self def __or__(self, other): if not isinstance(other, _collections_abc.Mapping): return NotImplemented m = self.copy() m.maps[0].update(other) return m def __ror__(self, other): if not isinstance(other, _collections_abc.Mapping): return NotImplemented m = dict(other) for child in reversed(self.maps): m.update(child) return self.__class__(m) ################################################################################ ### UserDict ################################################################################ class UserDict(_collections_abc.MutableMapping): # Start by filling-out the abstract methods def __init__(self, dict=None, /, **kwargs): self.data = {} if dict is not None: self.update(dict) if kwargs: self.update(kwargs) def __len__(self): return len(self.data) def __getitem__(self, key): if key in self.data: return self.data[key] if hasattr(self.__class__, "__missing__"): return self.__class__.__missing__(self, key) raise KeyError(key) def __setitem__(self, key, item): self.data[key] = item def __delitem__(self, key): del self.data[key] def __iter__(self): return iter(self.data) # Modify __contains__ and get() to work like dict # does when __missing__ is present. def __contains__(self, key): return key in self.data def get(self, key, default=None): if key in self: return self[key] return default # Now, add the methods in dicts but not in MutableMapping def __repr__(self): return repr(self.data) def __or__(self, other): if isinstance(other, UserDict): return self.__class__(self.data | other.data) if isinstance(other, dict): return self.__class__(self.data | other) return NotImplemented def __ror__(self, other): if isinstance(other, UserDict): return self.__class__(other.data | self.data) if isinstance(other, dict): return self.__class__(other | self.data) return NotImplemented def __ior__(self, other): if isinstance(other, UserDict): self.data |= other.data else: self.data |= other return self def __copy__(self): inst = self.__class__.__new__(self.__class__) inst.__dict__.update(self.__dict__) # Create a copy and avoid triggering descriptors inst.__dict__["data"] = self.__dict__["data"].copy() return inst def copy(self): if self.__class__ is UserDict: return UserDict(self.data.copy()) import copy data = self.data try: self.data = {} c = copy.copy(self) finally: self.data = data c.update(self) return c @classmethod def fromkeys(cls, iterable, value=None): d = cls() for key in iterable: d[key] = value return d ################################################################################ ### UserList ################################################################################ class UserList(_collections_abc.MutableSequence): """A more or less complete user-defined wrapper around list objects.""" def __init__(self, initlist=None): self.data = [] if initlist is not None: # XXX should this accept an arbitrary sequence? if type(initlist) == type(self.data): self.data[:] = initlist elif isinstance(initlist, UserList): self.data[:] = initlist.data[:] else: self.data = list(initlist) def __repr__(self): return repr(self.data) def __lt__(self, other): return self.data < self.__cast(other) def __le__(self, other): return self.data <= self.__cast(other) def __eq__(self, other): return self.data == self.__cast(other) def __gt__(self, other): return self.data > self.__cast(other) def __ge__(self, other): return self.data >= self.__cast(other) def __cast(self, other): return other.data if isinstance(other, UserList) else other def __contains__(self, item): return item in self.data def __len__(self): return len(self.data) def __getitem__(self, i): if isinstance(i, slice): return self.__class__(self.data[i]) else: return self.data[i] def __setitem__(self, i, item): self.data[i] = item def __delitem__(self, i): del self.data[i] def __add__(self, other): if isinstance(other, UserList): return self.__class__(self.data + other.data) elif isinstance(other, type(self.data)): return self.__class__(self.data + other) return self.__class__(self.data + list(other)) def __radd__(self, other): if isinstance(other, UserList): return self.__class__(other.data + self.data) elif isinstance(other, type(self.data)): return self.__class__(other + self.data) return self.__class__(list(other) + self.data) def __iadd__(self, other): if isinstance(other, UserList): self.data += other.data elif isinstance(other, type(self.data)): self.data += other else: self.data += list(other) return self def __mul__(self, n): return self.__class__(self.data * n) __rmul__ = __mul__ def __imul__(self, n): self.data *= n return self def __copy__(self): inst = self.__class__.__new__(self.__class__) inst.__dict__.update(self.__dict__) # Create a copy and avoid triggering descriptors inst.__dict__["data"] = self.__dict__["data"][:] return inst def append(self, item): self.data.append(item) def insert(self, i, item): self.data.insert(i, item) def pop(self, i=-1): return self.data.pop(i) def remove(self, item): self.data.remove(item) def clear(self): self.data.clear() def copy(self): return self.__class__(self) def count(self, item): return self.data.count(item) def index(self, item, *args): return self.data.index(item, *args) def reverse(self): self.data.reverse() def sort(self, /, *args, **kwds): self.data.sort(*args, **kwds) def extend(self, other): if isinstance(other, UserList): self.data.extend(other.data) else: self.data.extend(other) ################################################################################ ### UserString ################################################################################ class UserString(_collections_abc.Sequence): def __init__(self, seq): if isinstance(seq, str): self.data = seq elif isinstance(seq, UserString): self.data = seq.data[:] else: self.data = str(seq) def __str__(self): return str(self.data) def __repr__(self): return repr(self.data) def __int__(self): return int(self.data) def __float__(self): return float(self.data) def __complex__(self): return complex(self.data) def __hash__(self): return hash(self.data) def __getnewargs__(self): return (self.data[:],) def __eq__(self, string): if isinstance(string, UserString): return self.data == string.data return self.data == string def __lt__(self, string): if isinstance(string, UserString): return self.data < string.data return self.data < string def __le__(self, string): if isinstance(string, UserString): return self.data <= string.data return self.data <= string def __gt__(self, string): if isinstance(string, UserString): return self.data > string.data return self.data > string def __ge__(self, string): if isinstance(string, UserString): return self.data >= string.data return self.data >= string def __contains__(self, char): if isinstance(char, UserString): char = char.data return char in self.data def __len__(self): return len(self.data) def __getitem__(self, index): return self.__class__(self.data[index]) def __add__(self, other): if isinstance(other, UserString): return self.__class__(self.data + other.data) elif isinstance(other, str): return self.__class__(self.data + other) return self.__class__(self.data + str(other)) def __radd__(self, other): if isinstance(other, str): return self.__class__(other + self.data) return self.__class__(str(other) + self.data) def __mul__(self, n): return self.__class__(self.data * n) __rmul__ = __mul__ def __mod__(self, args): return self.__class__(self.data % args) def __rmod__(self, template): return self.__class__(str(template) % self) # the following methods are defined in alphabetical order: def capitalize(self): return self.__class__(self.data.capitalize()) def casefold(self): return self.__class__(self.data.casefold()) def center(self, width, *args): return self.__class__(self.data.center(width, *args)) def count(self, sub, start=0, end=_sys.maxsize): if isinstance(sub, UserString): sub = sub.data return self.data.count(sub, start, end) def removeprefix(self, prefix, /): if isinstance(prefix, UserString): prefix = prefix.data return self.__class__(self.data.removeprefix(prefix)) def removesuffix(self, suffix, /): if isinstance(suffix, UserString): suffix = suffix.data return self.__class__(self.data.removesuffix(suffix)) def encode(self, encoding='utf-8', errors='strict'): encoding = 'utf-8' if encoding is None else encoding errors = 'strict' if errors is None else errors return self.data.encode(encoding, errors) def endswith(self, suffix, start=0, end=_sys.maxsize): return self.data.endswith(suffix, start, end) def expandtabs(self, tabsize=8): return self.__class__(self.data.expandtabs(tabsize)) def find(self, sub, start=0, end=_sys.maxsize): if isinstance(sub, UserString): sub = sub.data return self.data.find(sub, start, end) def format(self, /, *args, **kwds): return self.data.format(*args, **kwds) def format_map(self, mapping): return self.data.format_map(mapping) def index(self, sub, start=0, end=_sys.maxsize): return self.data.index(sub, start, end) def isalpha(self): return self.data.isalpha() def isalnum(self): return self.data.isalnum() def isascii(self): return self.data.isascii() def isdecimal(self): return self.data.isdecimal() def isdigit(self): return self.data.isdigit() def isidentifier(self): return self.data.isidentifier() def islower(self): return self.data.islower() def isnumeric(self): return self.data.isnumeric() def isprintable(self): return self.data.isprintable() def isspace(self): return self.data.isspace() def istitle(self): return self.data.istitle() def isupper(self): return self.data.isupper() def join(self, seq): return self.data.join(seq) def ljust(self, width, *args): return self.__class__(self.data.ljust(width, *args)) def lower(self): return self.__class__(self.data.lower()) def lstrip(self, chars=None): return self.__class__(self.data.lstrip(chars)) maketrans = str.maketrans def partition(self, sep): return self.data.partition(sep) def replace(self, old, new, maxsplit=-1): if isinstance(old, UserString): old = old.data if isinstance(new, UserString): new = new.data return self.__class__(self.data.replace(old, new, maxsplit)) def rfind(self, sub, start=0, end=_sys.maxsize): if isinstance(sub, UserString): sub = sub.data return self.data.rfind(sub, start, end) def rindex(self, sub, start=0, end=_sys.maxsize): return self.data.rindex(sub, start, end) def rjust(self, width, *args): return self.__class__(self.data.rjust(width, *args)) def rpartition(self, sep): return self.data.rpartition(sep) def rstrip(self, chars=None): return self.__class__(self.data.rstrip(chars)) def split(self, sep=None, maxsplit=-1): return self.data.split(sep, maxsplit) def rsplit(self, sep=None, maxsplit=-1): return self.data.rsplit(sep, maxsplit) def splitlines(self, keepends=False): return self.data.splitlines(keepends) def startswith(self, prefix, start=0, end=_sys.maxsize): return self.data.startswith(prefix, start, end) def strip(self, chars=None): return self.__class__(self.data.strip(chars)) def swapcase(self): return self.__class__(self.data.swapcase()) def title(self): return self.__class__(self.data.title()) def translate(self, *args): return self.__class__(self.data.translate(*args)) def upper(self): return self.__class__(self.data.upper()) def zfill(self, width): return self.__class__(self.data.zfill(width)) usr/lib64/python3.12/lib2to3/__init__.py 0000644 00000000234 15246346167 0013526 0 ustar 00 import warnings warnings.warn( "lib2to3 package is deprecated and may not be able to parse Python 3.10+", DeprecationWarning, stacklevel=2, ) usr/lib64/python3.12/xml/__init__.py 0000644 00000001055 15246346755 0013055 0 ustar 00 """Core XML support for Python. This package contains four sub-packages: dom -- The W3C Document Object Model. This supports DOM Level 1 + Namespaces. parsers -- Python wrappers for XML parsers (currently only supports Expat). sax -- The Simple API for XML, developed by XML-Dev, led by David Megginson and ported to Python by Lars Marius Garshol. This supports the SAX 2 API. etree -- The ElementTree XML library. This is a subset of the full ElementTree XML release. """ __all__ = ["dom", "parsers", "sax", "etree"] usr/lib64/python3.12/importlib/__init__.py 0000644 00000011260 15246347013 0014241 0 ustar 00 """A pure Python implementation of import.""" __all__ = ['__import__', 'import_module', 'invalidate_caches', 'reload'] # Bootstrap help ##################################################### # Until bootstrapping is complete, DO NOT import any modules that attempt # to import importlib._bootstrap (directly or indirectly). Since this # partially initialised package would be present in sys.modules, those # modules would get an uninitialised copy of the source version, instead # of a fully initialised version (either the frozen one or the one # initialised below if the frozen one is not available). import _imp # Just the builtin component, NOT the full Python module import sys try: import _frozen_importlib as _bootstrap except ImportError: from . import _bootstrap _bootstrap._setup(sys, _imp) else: # importlib._bootstrap is the built-in import, ensure we don't create # a second copy of the module. _bootstrap.__name__ = 'importlib._bootstrap' _bootstrap.__package__ = 'importlib' try: _bootstrap.__file__ = __file__.replace('__init__.py', '_bootstrap.py') except NameError: # __file__ is not guaranteed to be defined, e.g. if this code gets # frozen by a tool like cx_Freeze. pass sys.modules['importlib._bootstrap'] = _bootstrap try: import _frozen_importlib_external as _bootstrap_external except ImportError: from . import _bootstrap_external _bootstrap_external._set_bootstrap_module(_bootstrap) _bootstrap._bootstrap_external = _bootstrap_external else: _bootstrap_external.__name__ = 'importlib._bootstrap_external' _bootstrap_external.__package__ = 'importlib' try: _bootstrap_external.__file__ = __file__.replace('__init__.py', '_bootstrap_external.py') except NameError: # __file__ is not guaranteed to be defined, e.g. if this code gets # frozen by a tool like cx_Freeze. pass sys.modules['importlib._bootstrap_external'] = _bootstrap_external # To simplify imports in test code _pack_uint32 = _bootstrap_external._pack_uint32 _unpack_uint32 = _bootstrap_external._unpack_uint32 # Fully bootstrapped at this point, import whatever you like, circular # dependencies and startup overhead minimisation permitting :) import warnings # Public API ######################################################### from ._bootstrap import __import__ def invalidate_caches(): """Call the invalidate_caches() method on all meta path finders stored in sys.meta_path (where implemented).""" for finder in sys.meta_path: if hasattr(finder, 'invalidate_caches'): finder.invalidate_caches() def import_module(name, package=None): """Import a module. The 'package' argument is required when performing a relative import. It specifies the package to use as the anchor point from which to resolve the relative import to an absolute import. """ level = 0 if name.startswith('.'): if not package: raise TypeError("the 'package' argument is required to perform a " f"relative import for {name!r}") for character in name: if character != '.': break level += 1 return _bootstrap._gcd_import(name[level:], package, level) _RELOADING = {} def reload(module): """Reload the module and return it. The module must have been successfully imported before. """ try: name = module.__spec__.name except AttributeError: try: name = module.__name__ except AttributeError: raise TypeError("reload() argument must be a module") from None if sys.modules.get(name) is not module: raise ImportError(f"module {name} not in sys.modules", name=name) if name in _RELOADING: return _RELOADING[name] _RELOADING[name] = module try: parent_name = name.rpartition('.')[0] if parent_name: try: parent = sys.modules[parent_name] except KeyError: raise ImportError(f"parent {parent_name!r} not in sys.modules", name=parent_name) from None else: pkgpath = parent.__path__ else: pkgpath = None target = module spec = module.__spec__ = _bootstrap._find_spec(name, pkgpath, target) if spec is None: raise ModuleNotFoundError(f"spec not found for the module {name!r}", name=name) _bootstrap._exec(spec, module) # The module may have replaced itself in sys.modules! return sys.modules[name] finally: try: del _RELOADING[name] except KeyError: pass usr/lib64/python3.12/turtledemo/__init__.py 0000644 00000000472 15246361744 0014436 0 ustar 00 """ -------------------------------------- About this viewer -------------------------------------- Tiny demo viewer to view turtle graphics example scripts. Quickly and dirtyly assembled by Gregor Lingl. June, 2006 For more information see: turtledemo - Help Have fun! """ usr/lib64/python3.12/curses/__init__.py 0000644 00000006451 15246362003 0013546 0 ustar 00 """curses The main package for curses support for Python. Normally used by importing the package, and perhaps a particular module inside it. import curses from curses import textpad curses.initscr() ... """ from _curses import * import os as _os import sys as _sys # Some constants, most notably the ACS_* ones, are only added to the C # _curses module's dictionary after initscr() is called. (Some # versions of SGI's curses don't define values for those constants # until initscr() has been called.) This wrapper function calls the # underlying C initscr(), and then copies the constants from the # _curses module to the curses package's dictionary. Don't do 'from # curses import *' if you'll be needing the ACS_* constants. def initscr(): import _curses, curses # we call setupterm() here because it raises an error # instead of calling exit() in error cases. setupterm(term=_os.environ.get("TERM", "unknown"), fd=_sys.__stdout__.fileno()) stdscr = _curses.initscr() for key, value in _curses.__dict__.items(): if key[0:4] == 'ACS_' or key in ('LINES', 'COLS'): setattr(curses, key, value) return stdscr # This is a similar wrapper for start_color(), which adds the COLORS and # COLOR_PAIRS variables which are only available after start_color() is # called. def start_color(): import _curses, curses retval = _curses.start_color() if hasattr(_curses, 'COLORS'): curses.COLORS = _curses.COLORS if hasattr(_curses, 'COLOR_PAIRS'): curses.COLOR_PAIRS = _curses.COLOR_PAIRS return retval # Import Python has_key() implementation if _curses doesn't contain has_key() try: has_key except NameError: from .has_key import has_key # Wrapper for the entire curses-based application. Runs a function which # should be the rest of your curses-based application. If the application # raises an exception, wrapper() will restore the terminal to a sane state so # you can read the resulting traceback. def wrapper(func, /, *args, **kwds): """Wrapper function that initializes curses and calls another function, restoring normal keyboard/screen behavior on error. The callable object 'func' is then passed the main window 'stdscr' as its first argument, followed by any other arguments passed to wrapper(). """ try: # Initialize curses stdscr = initscr() # Turn off echoing of keys, and enter cbreak mode, # where no buffering is performed on keyboard input noecho() cbreak() # In keypad mode, escape sequences for special keys # (like the cursor keys) will be interpreted and # a special value like curses.KEY_LEFT will be returned stdscr.keypad(1) # Start color, too. Harmless if the terminal doesn't have # color; user can test with has_color() later on. The try/catch # works around a minor bit of over-conscientiousness in the curses # module -- the error return from C start_color() is ignorable. try: start_color() except: pass return func(stdscr, *args, **kwds) finally: # Set everything back to normal if 'stdscr' in locals(): stdscr.keypad(0) echo() nocbreak() endwin() usr/lib64/python3.12/zipfile/__init__.py 0000644 00000255646 15246367773 0013743 0 ustar 00 """ Read and write ZIP files. XXX references to utf-8 need further investigation. """ import binascii import importlib.util import io import os import shutil import stat import struct import sys import threading import time 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", "Path"] 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 # General purpose bit flags # Zip Appnote: 4.4.4 general purpose bit flag: (2 bytes) _MASK_ENCRYPTED = 1 << 0 # Bits 1 and 2 have different meanings depending on the compression used. _MASK_COMPRESS_OPTION_1 = 1 << 1 # _MASK_COMPRESS_OPTION_2 = 1 << 2 # _MASK_USE_DATA_DESCRIPTOR: If set, crc-32, compressed size and uncompressed # size are zero in the local header and the real values are written in the data # descriptor immediately following the compressed data. _MASK_USE_DATA_DESCRIPTOR = 1 << 3 # Bit 4: Reserved for use with compression method 8, for enhanced deflating. # _MASK_RESERVED_BIT_4 = 1 << 4 _MASK_COMPRESSED_PATCH = 1 << 5 _MASK_STRONG_ENCRYPTION = 1 << 6 # _MASK_UNUSED_BIT_7 = 1 << 7 # _MASK_UNUSED_BIT_8 = 1 << 8 # _MASK_UNUSED_BIT_9 = 1 << 9 # _MASK_UNUSED_BIT_10 = 1 << 10 _MASK_UTF_FILENAME = 1 << 11 # Bit 12: Reserved by PKWARE for enhanced compression. # _MASK_RESERVED_BIT_12 = 1 << 12 # _MASK_ENCRYPTED_CENTRAL_DIR = 1 << 13 # Bit 14, 15: Reserved by PKWARE # _MASK_RESERVED_BIT_14 = 1 << 14 # _MASK_RESERVED_BIT_15 = 1 << 15 # 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('<HH') def _strip_extra(extra, xids): # Remove Extra Fields with specified IDs. unpack = _EXTRA_FIELD_STRUCT.unpack modified = False buffer = [] start = i = 0 while i + 4 <= len(extra): xid, xlen = unpack(extra[i : i + 4]) j = i + 4 + xlen if xid in xids: if i != start: buffer.append(extra[start : i]) start = j modified = True i = j if not modified: return extra if start != len(extra): buffer.append(extra[start:]) return b''.join(buffer) def _check_zipfile(fp): try: if _EndRecData(fp): return True # file has correct magic number except OSError: pass return False def is_zipfile(filename): """Quickly see if a file is a ZIP file by checking the magic number. The filename argument may be a file or file-like object too. """ result = False try: if hasattr(filename, "read"): result = _check_zipfile(fp=filename) else: with open(filename, "rb") as fp: result = _check_zipfile(fp) except (OSError, BadZipFile): pass return result def _EndRecData64(fpin, offset, endrec): """ Read the ZIP64 end-of-archive records and use that to update endrec """ offset -= sizeEndCentDir64Locator if offset < 0: # The file is not large enough to contain a ZIP64 # end-of-archive record, so just return the end record we were given. return endrec fpin.seek(offset) data = fpin.read(sizeEndCentDir64Locator) if len(data) != sizeEndCentDir64Locator: raise OSError("Unknown I/O error") sig, diskno, reloff, disks = struct.unpack(structEndArchive64Locator, data) if sig != stringEndArchive64Locator: return endrec if diskno != 0 or disks > 1: raise BadZipFile("zipfiles that span multiple disks are not supported") offset -= sizeEndCentDir64 if reloff > offset: raise BadZipFile("Corrupt zip64 end of central directory locator") # First, check the assumption that there is no prepended data. fpin.seek(reloff) extrasz = offset - reloff data = fpin.read(sizeEndCentDir64) if len(data) != sizeEndCentDir64: raise OSError("Unknown I/O error") if not data.startswith(stringEndArchive64) and reloff != offset: # Since we already have seen the Zip64 EOCD Locator, it's # possible we got here because there is prepended data. # Assume no 'zip64 extensible data' fpin.seek(offset) extrasz = 0 data = fpin.read(sizeEndCentDir64) if len(data) != sizeEndCentDir64: raise OSError("Unknown I/O error") if not data.startswith(stringEndArchive64): raise BadZipFile("Zip64 end of central directory record not found") sig, sz, create_version, read_version, disk_num, disk_dir, \ dircount, dircount2, dirsize, diroffset = \ struct.unpack(structEndArchive64, data) if (diroffset + dirsize != reloff or sz + 12 != sizeEndCentDir64 + extrasz): raise BadZipFile("Corrupt zip64 end of central directory record") # 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 endrec[_ECD_LOCATION] = offset - extrasz 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(sizeEndCentDir) 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, filesize - 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 - ZIP_MAX_COMMENT - sizeEndCentDir, 0) fpin.seek(maxCommentStart, 0) data = fpin.read(ZIP_MAX_COMMENT + sizeEndCentDir) 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, endrec) # Unable to find a valid end of central directory structure return None def _sanitize_filename(filename): """Terminate the file name at the first null byte and ensure paths always use forward slashes as the directory separator.""" # 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, "/") if os.altsep and os.altsep != "/" and os.altsep in filename: filename = filename.replace(os.altsep, "/") return filename 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', '_end_offset', ) 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 and # ensure paths always use forward slashes as the directory separator. filename = _sanitize_filename(filename) 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 self.compress_size = 0 # Size of the compressed file self.file_size = 0 # Size of the uncompressed file self._end_offset = None # Start of the next local header or central directory # Other attributes are set by class ZipFile: # header_offset Byte offset to the file header # CRC CRC-32 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. When the optional zip64 arg is None rather than a bool, we will decide based upon the file_size and compress_size, if known, False otherwise. """ 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 & _MASK_USE_DATA_DESCRIPTOR: # 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: # We always explicitly pass zip64 within this module.... This # remains for anyone using ZipInfo.FileHeader as a public API. zip64 = file_size > ZIP64_LIMIT or compress_size > ZIP64_LIMIT if zip64: fmt = '<HHQQ' extra = extra + struct.pack(fmt, 1, struct.calcsize(fmt)-4, file_size, compress_size) 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 | _MASK_UTF_FILENAME def _decodeExtra(self, filename_crc): # Try to decode the extra field. extra = self.extra unpack = struct.unpack while len(extra) >= 4: tp, ln = unpack('<HH', extra[:4]) if ln+4 > len(extra): raise BadZipFile("Corrupt extra field %04x (size=%d)" % (tp, ln)) if tp == 0x0001: data = extra[4:ln+4] # ZIP64 extension (large files and/or large archives) try: if self.file_size in (0xFFFF_FFFF_FFFF_FFFF, 0xFFFF_FFFF): field = "File size" self.file_size, = unpack('<Q', data[:8]) data = data[8:] if self.compress_size == 0xFFFF_FFFF: field = "Compress size" self.compress_size, = unpack('<Q', data[:8]) data = data[8:] if self.header_offset == 0xFFFF_FFFF: field = "Header offset" self.header_offset, = unpack('<Q', data[:8]) except struct.error: raise BadZipFile(f"Corrupt zip64 extra field. " f"{field} not found.") from None elif tp == 0x7075: data = extra[4:ln+4] # Unicode Path Extra Field try: up_version, up_name_crc = unpack('<BL', data[:5]) if up_version == 1 and up_name_crc == filename_crc: up_unicode_name = data[5:].decode('utf-8') if up_unicode_name: self.filename = _sanitize_filename(up_unicode_name) else: import warnings warnings.warn("Empty unicode path extra field (0x7075)", stacklevel=2) except struct.error as e: raise BadZipFile("Corrupt unicode path extra field (0x7075)") from e except UnicodeDecodeError as e: raise BadZipFile('Corrupt unicode path extra field (0x7075): invalid utf-8 bytes') from e extra = extra[ln+4:] @classmethod def from_file(cls, filename, arcname=None, *, strict_timestamps=True): """Construct an appropriate ZipInfo for a file on the filesystem. filename should be the path to a file or directory on the filesystem. arcname is the name which it will have within the archive (by default, this will be the same as filename, but without a drive letter and with leading path separators removed). """ if isinstance(filename, os.PathLike): filename = os.fspath(filename) st = os.stat(filename) isdir = stat.S_ISDIR(st.st_mode) mtime = time.localtime(st.st_mtime) date_time = mtime[0:6] if not strict_timestamps and date_time[0] < 1980: date_time = (1980, 1, 1, 0, 0, 0) elif not strict_timestamps and date_time[0] > 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.""" if self.filename.endswith('/'): return True # The ZIP format specification requires to use forward slashes # as the directory separator, but in practice some ZIP files # created on Windows can use backward slashes. For compatibility # with the extraction code which already handles this: if os.path.altsep: return self.filename.endswith((os.path.sep, os.path.altsep)) return False # 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('<BBH', 9, 4, len(props)) + props def compress(self, data): if self._comp is None: return self._init() + self._comp.compress(data) return self._comp.compress(data) def flush(self): if self._comp is None: return self._init() + self._comp.flush() return self._comp.flush() class LZMADecompressor: def __init__(self): self._decomp = None self._unconsumed = b'' self.eof = False def decompress(self, data): if self._decomp is None: self._unconsumed += data if len(self._unconsumed) <= 4: return b'' psize, = struct.unpack('<H', self._unconsumed[2:4]) if len(self._unconsumed) <= 4 + psize: return b'' self._decomp = lzma.LZMADecompressor(lzma.FORMAT_RAW, filters=[ lzma._decode_filter_properties(lzma.FILTER_LZMA1, self._unconsumed[4:4 + psize]) ]) data = self._unconsumed[4 + psize:] del self._unconsumed result = self._decomp.decompress(data) self.eof = self._decomp.eof return result compressor_names = { 0: 'store', 1: 'shrink', 2: 'reduce', 3: 'reduce', 4: 'reduce', 5: 'reduce', 6: 'implode', 7: 'tokenize', 8: 'deflate', 9: 'deflate64', 10: 'implode', 12: 'bzip2', 14: 'lzma', 18: 'terse', 19: 'lz77', 97: 'wavpack', 98: 'ppmd', } def _check_compression(compression): if compression == ZIP_STORED: pass elif compression == ZIP_DEFLATED: if not zlib: raise RuntimeError( "Compression requires the (missing) zlib module") elif compression == ZIP_BZIP2: if not bz2: raise RuntimeError( "Compression requires the (missing) bz2 module") elif compression == ZIP_LZMA: if not lzma: raise RuntimeError( "Compression requires the (missing) lzma module") else: raise NotImplementedError("That compression method is not supported") def _get_compressor(compress_type, compresslevel=None): if compress_type == ZIP_DEFLATED: if compresslevel is not None: return zlib.compressobj(compresslevel, zlib.DEFLATED, -15) return zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -15) elif compress_type == ZIP_BZIP2: if compresslevel is not None: return bz2.BZ2Compressor(compresslevel) return bz2.BZ2Compressor() # compresslevel is ignored for ZIP_LZMA elif compress_type == ZIP_LZMA: return LZMACompressor() else: return None def _get_decompressor(compress_type): _check_compression(compress_type) if compress_type == ZIP_STORED: return None elif compress_type == ZIP_DEFLATED: return zlib.decompressobj(-15) elif compress_type == ZIP_BZIP2: return bz2.BZ2Decompressor() elif compress_type == ZIP_LZMA: return LZMADecompressor() else: descr = compressor_names.get(compress_type) if descr: raise NotImplementedError("compression type %d (%s)" % (compress_type, descr)) else: raise NotImplementedError("compression type %d" % (compress_type,)) class _SharedFile: def __init__(self, file, pos, close, lock, writing): self._file = file self._pos = pos self._close = close self._lock = lock self._writing = writing self.seekable = file.seekable def tell(self): return self._pos def seek(self, offset, whence=0): with self._lock: if self._writing(): raise ValueError("Can't reposition in the ZIP file while " "there is an open writing handle on it. " "Close the writing handle before trying to read.") if whence == os.SEEK_CUR: self._file.seek(self._pos + offset) else: self._file.seek(offset, whence) self._pos = self._file.tell() return self._pos def read(self, n=-1): with self._lock: if self._writing(): raise ValueError("Can't read from the ZIP file while there " "is an open writing handle on it. " "Close the writing handle before trying to read.") self._file.seek(self._pos) data = self._file.read(n) self._pos = self._file.tell() return data def close(self): if self._file is not None: fileobj = self._file self._file = None self._close(fileobj) # Provide the tell method for unseekable stream class _Tellable: def __init__(self, fp): self.fp = fp self.offset = 0 def write(self, data): n = self.fp.write(data) self.offset += n return n def tell(self): return self.offset def flush(self): self.fp.flush() def close(self): self.fp.close() class ZipExtFile(io.BufferedIOBase): """File-like object for reading an archive member. Is returned by ZipFile.open(). """ # Max size supported by decompressor. MAX_N = 1 << 31 - 1 # Read from compressed files in 4k blocks. MIN_READ_SIZE = 4096 # Chunk size to read during seek MAX_SEEK_READ = 1 << 24 def __init__(self, fileobj, mode, zipinfo, pwd=None, close_fileobj=False): self._fileobj = fileobj self._pwd = pwd self._close_fileobj = close_fileobj self._compress_type = zipinfo.compress_type self._compress_left = zipinfo.compress_size self._left = zipinfo.file_size self._decompressor = _get_decompressor(self._compress_type) self._eof = False self._readbuffer = b'' self._offset = 0 self.newlines = None self.mode = mode self.name = zipinfo.filename if hasattr(zipinfo, 'CRC'): self._expected_crc = zipinfo.CRC self._running_crc = crc32(b'') else: self._expected_crc = None self._seekable = False try: if fileobj.seekable(): self._orig_compress_start = fileobj.tell() self._orig_compress_size = zipinfo.compress_size self._orig_file_size = zipinfo.file_size self._orig_start_crc = self._running_crc self._orig_crc = self._expected_crc self._seekable = True except AttributeError: pass self._decrypter = None if pwd: if zipinfo.flag_bits & _MASK_USE_DATA_DESCRIPTOR: # compare against the file type from extended local headers check_byte = (zipinfo._raw_time >> 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): if self.closed: raise ValueError("I/O operation on closed file.") 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 self.closed: raise ValueError("read from closed file.") 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): if self.closed: raise ValueError("I/O operation on closed file.") return self._seekable def seek(self, offset, whence=os.SEEK_SET): if self.closed: raise ValueError("seek on closed file.") if not self._seekable: raise io.UnsupportedOperation("underlying stream is not seekable") curr_pos = self.tell() if whence == os.SEEK_SET: new_pos = offset elif whence == os.SEEK_CUR: new_pos = curr_pos + offset elif whence == os.SEEK_END: 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 # Fast seek uncompressed unencrypted file elif self._compress_type == ZIP_STORED and self._decrypter is None and read_offset != 0: # disable CRC checking after first seeking - it would be invalid self._expected_crc = None # seek actual file taking already buffered data into account read_offset -= len(self._readbuffer) - self._offset self._fileobj.seek(read_offset, os.SEEK_CUR) self._left -= read_offset self._compress_left -= read_offset self._eof = self._left <= 0 read_offset = 0 # flush read buffer self._readbuffer = b'' self._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._expected_crc = self._orig_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 self.closed: raise ValueError("tell on closed file.") 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.') # Accept any data that supports the buffer protocol if isinstance(data, (bytes, bytearray)): nbytes = len(data) else: data = memoryview(data) nbytes = data.nbytes 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 if not self._zip64: if self._file_size > ZIP64_LIMIT: raise RuntimeError("File size too large, try using force_zip64") if self._compress_size > ZIP64_LIMIT: raise RuntimeError("Compressed size too large, try using force_zip64") # Write updated header info if self._zinfo.flag_bits & _MASK_USE_DATA_DESCRIPTOR: # Write CRC and file sizes after the file data fmt = '<LLQQ' if self._zip64 else '<LLLL' self._fileobj.write(struct.pack(fmt, _DD_SIGNATURE, self._zinfo.CRC, self._zinfo.compress_size, self._zinfo.file_size)) self._zipfile.start_dir = self._fileobj.tell() else: # 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, metadata_encoding=None): """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 self.metadata_encoding = metadata_encoding # Check that we don't try to write with nonconforming codecs if self.metadata_encoding and mode != 'r': raise ValueError( "metadata_encoding is only supported for reading files") # 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 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 if self.start_dir < 0: raise BadZipFile("Bad offset for central directory") 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]) orig_filename_crc = crc32(filename) flags = centdir[_CD_FLAG_BITS] if flags & _MASK_UTF_FILENAME: # UTF-8 file names extension filename = filename.decode('utf-8') else: # Historical ZIP filename encoding filename = filename.decode(self.metadata_encoding or '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(orig_filename_crc) 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) end_offset = self.start_dir for zinfo in reversed(sorted(self.filelist, key=lambda zinfo: zinfo.header_offset)): zinfo._end_offset = end_offset end_offset = zinfo.header_offset 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. Return None if all files could be read successfully, or the name of the offending file otherwise.""" 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. 'pwd' is the password to decrypt encrypted files.""" 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 (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.seek(fheader[_FH_EXTRA_FIELD_LENGTH], whence=1) if zinfo.flag_bits & _MASK_COMPRESSED_PATCH: # Zip 2.7: compressed patched data raise NotImplementedError("compressed patched data (flag bit 5)") if zinfo.flag_bits & _MASK_STRONG_ENCRYPTION: # strong encryption raise NotImplementedError("strong encryption (flag bit 6)") if fheader[_FH_GENERAL_PURPOSE_FLAG_BITS] & _MASK_UTF_FILENAME: # UTF-8 filename fname_str = fname.decode("utf-8") else: fname_str = fname.decode(self.metadata_encoding or "cp437") if fname_str != zinfo.orig_filename: raise BadZipFile( 'File name in directory %r and header %r differ.' % (zinfo.orig_filename, fname)) if (zinfo._end_offset is not None and zef_file.tell() + zinfo.compress_size > zinfo._end_offset): if zinfo._end_offset == zinfo.header_offset: import warnings warnings.warn( f"Overlapped entries: {zinfo.orig_filename!r} " f"(possible zip bomb)", skip_file_prefixes=(os.path.dirname(__file__),)) else: raise BadZipFile( f"Overlapped entries: {zinfo.orig_filename!r} " f"(possible zip bomb)") # check for encrypted flag & handle password is_encrypted = zinfo.flag_bits & _MASK_ENCRYPTED if is_encrypted: if not pwd: pwd = self.pwd if pwd and not isinstance(pwd, bytes): raise TypeError("pwd: expected bytes, got %s" % type(pwd).__name__) 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.") # Size and CRC are overwritten with correct data after processing the file 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 |= _MASK_COMPRESS_OPTION_1 if not self._seekable: zinfo.flag_bits |= _MASK_USE_DATA_DESCRIPTOR if not zinfo.external_attr: zinfo.external_attr = 0o600 << 16 # permissions: ?rw------- # Compressed size can be larger than uncompressed size zip64 = force_zip64 or (zinfo.file_size * 1.05 > ZIP64_LIMIT) if not self._allowZip64 and zip64: raise LargeZipFile("Filesize would require ZIP64 extensions") 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'. You can specify the password to decrypt the file using 'pwd'. """ 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(). You can specify the password to decrypt all files using 'pwd'. """ 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 and spaces 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) if not arcname and not member.is_dir(): raise ValueError("Empty filename.") 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 self.mkdir(zinfo) 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 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.endswith('/'): 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 mkdir(self, zinfo_or_directory_name, mode=511): """Creates a directory inside the zip archive.""" if isinstance(zinfo_or_directory_name, ZipInfo): zinfo = zinfo_or_directory_name if not zinfo.is_dir(): raise ValueError("The given ZipInfo does not describe a directory") elif isinstance(zinfo_or_directory_name, str): directory_name = zinfo_or_directory_name if not directory_name.endswith("/"): directory_name += "/" zinfo = ZipInfo(directory_name) zinfo.compress_size = 0 zinfo.CRC = 0 zinfo.external_attr = ((0o40000 | mode) & 0xFFFF) << 16 zinfo.file_size = 0 zinfo.external_attr |= 0x10 else: raise TypeError("Expected type str or ZipInfo") 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 |= _MASK_COMPRESS_OPTION_1 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() 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( '<HH' + 'Q'*len(extra), 1, 8*len(extra), *extra) + extra_data min_version = ZIP64_VERSION if zinfo.compress_type == ZIP_BZIP2: min_version = max(BZIP2_VERSION, min_version) elif zinfo.compress_type == ZIP_LZMA: min_version = max(LZMA_VERSION, min_version) extract_version = max(min_version, zinfo.extract_version) create_version = max(min_version, zinfo.create_version) filename, flag_bits = zinfo._encodeFilenameFlags() centdir = struct.pack(structCentralDir, stringCentralDir, create_version, zinfo.create_system, extract_version, zinfo.reserved, flag_bits, zinfo.compress_type, dostime, dosdate, zinfo.CRC, compress_size, file_size, len(filename), len(extra_data), len(zinfo.comment), 0, zinfo.internal_attr, zinfo.external_attr, header_offset) self.fp.write(centdir) self.fp.write(filename) self.fp.write(extra_data) self.fp.write(zinfo.comment) pos2 = self.fp.tell() # Write end-of-zip-archive record centDirCount = len(self.filelist) centDirSize = pos2 - self.start_dir centDirOffset = self.start_dir requires_zip64 = None if centDirCount > 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, sizeEndCentDir64 - 12, 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) if self.mode == "a": self.fp.truncate() 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 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='<zipfile>', help='Show listing of a zipfile') group.add_argument('-e', '--extract', nargs=2, metavar=('<zipfile>', '<output_dir>'), help='Extract zipfile into target dir') group.add_argument('-c', '--create', nargs='+', metavar=('<name>', '<file>'), help='Create zipfile from sources') group.add_argument('-t', '--test', metavar='<zipfile>', help='Test if a zipfile is valid') parser.add_argument('--metadata-encoding', metavar='<encoding>', help='Specify encoding of member names for -l, -e and -t') args = parser.parse_args(args) encoding = args.metadata_encoding if args.test is not None: src = args.test with ZipFile(src, 'r', metadata_encoding=encoding) 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', metadata_encoding=encoding) as zf: zf.printdir() elif args.extract is not None: src, curdir = args.extract with ZipFile(src, 'r', metadata_encoding=encoding) as zf: zf.extractall(curdir) elif args.create is not None: if encoding: print("Non-conforming encodings not supported with -c.", file=sys.stderr) sys.exit(1) 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) from ._path import ( # noqa: E402 Path, # used privately for tests CompleteDirs, # noqa: F401 ) usr/lib64/python3.12/sqlite3/__init__.py 0000644 00000004705 15246406453 0013636 0 ustar 00 # pysqlite2/__init__.py: the pysqlite2 package. # # Copyright (C) 2005 Gerhard Häring <gh@ghaering.de> # # This file is part of pysqlite. # # This software is provided 'as-is', without any express or implied # warranty. In no event will the authors be held liable for any damages # arising from the use of this software. # # Permission is granted to anyone to use this software for any purpose, # including commercial applications, and to alter it and redistribute it # freely, subject to the following restrictions: # # 1. The origin of this software must not be misrepresented; you must not # claim that you wrote the original software. If you use this software # in a product, an acknowledgment in the product documentation would be # appreciated but is not required. # 2. Altered source versions must be plainly marked as such, and must not be # misrepresented as being the original software. # 3. This notice may not be removed or altered from any source distribution. """ The sqlite3 extension module provides a DB-API 2.0 (PEP 249) compliant interface to the SQLite library, and requires SQLite 3.7.15 or newer. To use the module, start by creating a database Connection object: import sqlite3 cx = sqlite3.connect("test.db") # test.db will be created or opened The special path name ":memory:" can be provided to connect to a transient in-memory database: cx = sqlite3.connect(":memory:") # connect to a database in RAM Once a connection has been established, create a Cursor object and call its execute() method to perform SQL queries: cu = cx.cursor() # create a table cu.execute("create table lang(name, first_appeared)") # insert values into a table cu.execute("insert into lang values (?, ?)", ("C", 1972)) # execute a query and iterate over the result for row in cu.execute("select * from lang"): print(row) cx.close() The sqlite3 module is written by Gerhard Häring <gh@ghaering.de>. """ from sqlite3.dbapi2 import * from sqlite3.dbapi2 import (_deprecated_names, _deprecated_version_info, _deprecated_version) def __getattr__(name): if name in _deprecated_names: from warnings import warn warn(f"{name} is deprecated and will be removed in Python 3.14", DeprecationWarning, stacklevel=2) return globals()[f"_deprecated_{name}"] raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
Free Space : 748127674368 Byte