Python lzma 模块，LZMAFile() 实例源码

def extract_ar(archive, dst, *kwargs):
    if sys.version_info[0] < 3 and archive.endswith('.xz'):
        with contextlib.closing(lzma.LZMAFile(archive)) as xz:
            with tarfile.open(fileobj=xz, *kwargs) as f:
                f.extractall(dst)
    elif archive.endswith('.zip'):
        with zipfile.ZipFile(archive,'r') as f:
            f.extractall(dst)
    elif tarfile.is_tarfile(archive):
        if USE_CMAKE_TAR:
            cmd([which('cmake'), '-E', 'tar', 'xzf', os.path.abspath(archive)], cwd=dst)
        else:
            tarfile.open(archive, *kwargs).extractall(dst)
    else:
        # Treat as a single source file
        d = os.path.join(dst, 'header')
        mkdir(d)
        copy_to(archive, d)

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w"):
            raise ValueError("mode must be 'r' or 'w'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            raise ReadError("not an lzma file")
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def _write_fileobject(filename, compress=("zlib", 3)):
    """Return the right compressor file object in write mode."""
    compressmethod = compress[0]
    compresslevel = compress[1]
    if compressmethod == "gzip":
        return _buffered_write_file(BinaryGzipFile(filename, 'wb',
                                    compresslevel=compresslevel))
    elif compressmethod == "bz2" and bz2 is not None:
        return _buffered_write_file(bz2.BZ2File(filename, 'wb',
                                                compresslevel=compresslevel))
    elif lzma is not None and compressmethod == "xz":
        return _buffered_write_file(lzma.LZMAFile(filename, 'wb',
                                                  check=lzma.CHECK_NONE,
                                                  preset=compresslevel))
    elif lzma is not None and compressmethod == "lzma":
        return _buffered_write_file(lzma.LZMAFile(filename, 'wb',
                                                  preset=compresslevel,
                                                  format=lzma.FORMAT_ALONE))
    else:
        return _buffered_write_file(BinaryZlibFile(filename, 'wb',
                                    compresslevel=compresslevel))


###############################################################################
#  Joblib zlib compression file object definition

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w", "x"):
            raise ValueError("mode must be 'r', 'w' or 'x'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            if mode == 'r':
                raise ReadError("not an lzma file")
            raise
        except:
            fileobj.close()
            raise
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w", "x"):
            raise ValueError("mode must be 'r', 'w' or 'x'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            if mode == 'r':
                raise ReadError("not an lzma file")
            raise
        except:
            fileobj.close()
            raise
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def xzopen(cls, name, mode="r", fileobj=None, compresslevel=6):
        """
        Open xz compressed cpio archive name for reading or writing.
        Appending is not allowed.
        """
        if len(mode) > 1 or mode not in "rw":
            raise ValueError("mode must be 'r' or 'w'.")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        if fileobj is not None:
            fileobj = _XZProxy(fileobj, mode)
        else:
            fileobj = lzma.LZMAFile(name, mode) # options={'level': compresslevel, 'dict_size': 20 }

        try:
            t = cls.cpioopen(name, mode, fileobj)
        except IOError:
            raise ReadError("not a XZ file")
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def process_item(out, min_cache_zoom, cache, link,
                 part_zoom, tiled_countries, all_countries,
                 min_zoom, max_zoom, skip_empty_tiles):
    stat = Stat()
    date = get_date_from_link(link)

    for line in lzma.LZMAFile(get_tile_usage_dump(link)):
        path, count = line.decode().strip().split()
        z, x, y = path.split('/')

        x = int(x)
        y = int(y)
        z = int(z)

        if min_zoom is not None and z < min_zoom:
            continue
        if max_zoom is not None and z > max_zoom:
            continue

        twest, tsouth, teast, tnorth = b = mercantile.bounds(x, y, z)
        country = detect_country_with_cache(
            path, b, x, y, z, part_zoom, tiled_countries, all_countries,
            min_cache_zoom, cache, stat)

        if skip_empty_tiles and country == '??':
            continue

        lat = tnorth + (tnorth - tsouth) / 2
        lon = twest + (teast - twest) / 2

        out.write(('%s,%s,%s,%s,%s,%s,%s,%s\n' % (
            date, z, x, y, count, lat, lon, country)).encode())
    stat.log_stats(date, cache)

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w", "x"):
            raise ValueError("mode must be 'r', 'w' or 'x'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            if mode == 'r':
                raise ReadError("not an lzma file")
            raise
        except:
            fileobj.close()
            raise
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w", "x"):
            raise ValueError("mode must be 'r', 'w' or 'x'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            if mode == 'r':
                raise ReadError("not an lzma file")
            raise
        except:
            fileobj.close()
            raise
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def openCompressedFile(fileName, mode='r', *args, **kwargs):
    fileNameLower = fileName.lower()

    if fileNameLower.endswith('.xz'):
        fh = lzma.LZMAFile(fileName, mode, *args, **kwargs)
        dir(fh) # fix for wierd bug https://bugs.launchpad.net/pyliblzma/+bug/1219296 hopefully fixed in next release XXX - idfah
        return fh
    elif fileNameLower.endswith('.bz2'):
        return bz2.BZ2File(fileName, mode, *args, **kwargs)
    elif fileNameLower.endswith('.gz'):
        return gzip.GzipFile(fileName, mode, *args, **kwargs)
    else:
        return open(fileName, mode, *args, **kwargs)

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w", "x"):
            raise ValueError("mode must be 'r', 'w' or 'x'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            if mode == 'r':
                raise ReadError("not an lzma file")
            raise
        except:
            fileobj.close()
            raise
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def test_no_name_argument(self):
        self.skipTest("LZMAFile have no name attribute")

def extract_xz_tarball(path, folder):
  try:
    f = lzma.LZMAFile(path)

    tarball = tarfile.open(fileobj=f)
    tarball.extractall(path=folder)

    return True

  except Exception as e:
    print(" ! " + str(e))
    return False

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w", "x"):
            raise ValueError("mode must be 'r', 'w' or 'x'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            if mode == 'r':
                raise ReadError("not an lzma file")
            raise
        except:
            fileobj.close()
            raise
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w"):
            raise ValueError("mode must be 'r' or 'w'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            if mode == 'r':
                raise ReadError("not an lzma file")
            raise
        except:
            fileobj.close()
            raise
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def requires_name_attribute(self):
        self.skipTest("LZMAFile have no name attribute")

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w", "x"):
            raise ValueError("mode must be 'r', 'w' or 'x'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            if mode == 'r':
                raise ReadError("not an lzma file")
            raise
        except:
            fileobj.close()
            raise
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w", "x"):
            raise ValueError("mode must be 'r', 'w' or 'x'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            if mode == 'r':
                raise ReadError("not an lzma file")
            raise
        except:
            fileobj.close()
            raise
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w", "x"):
            raise ValueError("mode must be 'r', 'w' or 'x'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            if mode == 'r':
                raise ReadError("not an lzma file")
            raise
        except:
            fileobj.close()
            raise
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w", "x"):
            raise ValueError("mode must be 'r', 'w' or 'x'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            if mode == 'r':
                raise ReadError("not an lzma file")
            raise
        except:
            fileobj.close()
            raise
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w"):
            raise ValueError("mode must be 'r' or 'w'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            if mode == 'r':
                raise ReadError("not an lzma file")
            raise
        except:
            fileobj.close()
            raise
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def requires_name_attribute(self):
        self.skipTest("LZMAFile have no name attribute")

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w", "x"):
            raise ValueError("mode must be 'r', 'w' or 'x'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            if mode == 'r':
                raise ReadError("not an lzma file")
            raise
        except:
            fileobj.close()
            raise
        t._extfileobj = False
        return t

    # All *open() methods are registered here.

def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
        """Open lzma compressed tar archive name for reading or writing.
           Appending is not allowed.
        """
        if mode not in ("r", "w", "x"):
            raise ValueError("mode must be 'r', 'w' or 'x'")

        try:
            import lzma
        except ImportError:
            raise CompressionError("lzma module is not available")

        fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)

        try:
            t = cls.taropen(name, mode, fileobj, **kwargs)
        except (lzma.LZMAError, EOFError):
            fileobj.close()
            if mode == 'r':
                raise ReadError("not an lzma file")
            raise
        except:
            fileobj.close()
            raise
        t._extfileobj = False
        return t

    # All *open() methods are registered here.