Unzip filename to a temporary directory, set to the cwd.

    The unzipped target is cleaned up after.
778644
def archive_context(filename):
    
    tmpdir = tempfile.mkdtemp()
    log.warn('Extracting in %s', tmpdir)
    old_wd = os.getcwd()
    try:
        os.chdir(tmpdir)
        try:
            with ContextualZipFile(filename) as archive:
                archive.extractall()
        except zipfile.BadZipfile as err:
            if not err.args:
                err.args = ('', )
            err.args = err.args + (
                MEANINGFUL_INVALID_ZIP_ERR_MSG.format(filename),
            )
            raise

        # going in the directory
        subdir = os.path.join(tmpdir, os.listdir(tmpdir)[0])
        os.chdir(subdir)
        log.warn('Now working in %s', subdir)
        yield

    finally:
        os.chdir(old_wd)
        shutil.rmtree(tmpdir)
755661
def _clear_dir(dirName):
    
    # If we got here, clear dir  
    for fname in os.listdir(dirName):
        try:
            os.remove( os.path.join(dirName, fname) )
        except Exception:
            pass
    try:
        os.rmdir(dirName)
    except Exception:
        pass
618474
def clear_temp(wdir, rmdir=True):
    """"""
    wdir = pathlib.Path(wdir)
    extensions = ["*.log", "*.dat", "*.txt"]

    for ext in extensions:
        for ff in wdir.glob(ext):
            ff.unlink()

    if rmdir:
        wdir.rmdir()
7673
def uncompress_archives(file_names, temp_dir):
    
    result = []
    for filename in file_names:
        if filename.endswith(".tar") or filename.endswith(".tar.gz") or filename.endswith(".tar.bz2"):
            obj = tarfile.open(filename, "r")
            extracted_file_names = []
            for j, member in enumerate(obj.getmembers()):
                obj.extract(member, temp_dir)
                extracted_filename = os.path.join(temp_dir, obj.getnames()[j])
                extracted_file_names.append(extracted_filename)
                archives[extracted_filename] = filename
            result.extend(extracted_file_names[:])
        elif filename.endswith(".zip"):
            obj = zipfile.ZipFile(filename)
            extracted_file_names = []
            for j, info in enumerate(obj.infolist()):
                obj.extract(info, path=temp_dir)
                extracted_filename = os.path.join(temp_dir, obj.namelist()[j])
                extracted_file_names.append(extracted_filename)
                archives[extracted_filename] = filename
            result.extend(extracted_file_names[:])
        else:
            result.append(filename)

    return result
1336553
def truncate_cwd(full_filename):
    """"""
    if full_filename.startswith(os.getcwd()):
        truncated_filename = full_filename.split(os.getcwd(), 1)[1]
        truncated_filename = truncated_filename.split(os.sep, 1)[1]
    else:
        truncated_filename = full_filename
    return truncated_filename
1353157
def create_files(filedef, cleanup=True):
    
    cwd = os.getcwd()
    tmpdir = tempfile.mkdtemp()
    try:
        Filemaker(tmpdir, filedef)
        if not cleanup:  # pragma: nocover
            pass
            # print("TMPDIR =", tmpdir)
        os.chdir(tmpdir)
        yield tmpdir
    finally:
        os.chdir(cwd)
        if cleanup:  # pragma: nocover
            shutil.rmtree(tmpdir, ignore_errors=True)
1364451
def cleanup(self):
        

        self.log.debug("Cleaning up %s temporary directory" % self.tmp_dir)
        shutil.rmtree(self.tmp_dir, ignore_errors=True)
1165537
def untar_file(filename, location):
    """"""
    if not os.path.exists(location):
        os.makedirs(location)
    if filename.lower().endswith('.gz') or filename.lower().endswith('.tgz'):
        mode = 'r:gz'
    elif filename.lower().endswith('.bz2') or filename.lower().endswith('.tbz'):
        mode = 'r:bz2'
    elif filename.lower().endswith('.tar'):
        mode = 'r'
    else:
        logger.warn('Cannot determine compression type for file %s' % filename)
        mode = 'r:*'
    tar = tarfile.open(filename, mode)
    try:
        # note: python<=2.5 doesnt seem to know about pax headers, filter them
        leading = has_leading_dir([
            member.name for member in tar.getmembers()
            if member.name != 'pax_global_header'
        ])
        for member in tar.getmembers():
            fn = member.name
            if fn == 'pax_global_header':
                continue
            if leading:
                fn = split_leading_dir(fn)[1]
            path = os.path.join(location, fn)
            if member.isdir():
                if not os.path.exists(path):
                    os.makedirs(path)
            else:
                try:
                    fp = tar.extractfile(member)
                except (KeyError, AttributeError):
                    e = sys.exc_info()[1]
                    # Some corrupt tar files seem to produce this
                    # (specifically bad symlinks)
                    logger.warn(
                        'In the tar file %s the member %s is invalid: %s'
                        % (filename, member.name, e))
                    continue
                if not os.path.exists(os.path.dirname(path)):
                    os.makedirs(os.path.dirname(path))
                destfp = open(path, 'wb')
                try:
                    shutil.copyfileobj(fp, destfp)
                finally:
                    destfp.close()
                fp.close()
    finally:
        tar.close()
1255762
def _move_file_to_dest_dir(self, filename):
        ''''''
        assert self._params.move_to

        if os.path.isdir(self._params.move_to):
            _logger.debug('Moved {} to {}.', self._warc_filename,
                          self._params.move_to)
            shutil.move(filename, self._params.move_to)
        else:
            _logger.error('{} is not a directory; not moving {}.',
                          self._params.move_to, filename)
540371
def unpack_zipfile(filename):
    """"""
    with open(filename, "rb") as fzip:
        z = zipfile.ZipFile(fzip)
        for name in z.namelist():
            print(("      extracting {}".format(name)))
            ensure_dirs(name)
            z.extract(name)
630541
def download_and_unpack(self, url, *paths, **kw):
        
        with self.temp_download(url, 'ds.zip', log=kw.pop('log', None)) as zipp:
            with TemporaryDirectory() as tmpdir:
                with zipfile.ZipFile(zipp.as_posix()) as zipf:
                    for path in paths:
                        zipf.extract(as_posix(path), path=tmpdir.as_posix())
                        copy(tmpdir.joinpath(path), self)
132855
def mkzip(source_dir, output_filename):
        '''  
        '''
        zipf = zipfile.ZipFile(output_filename, 'w', zipfile.zlib.DEFLATED)
        pre_len = len(os.path.dirname(source_dir))
        for parent, dirnames, filenames in os.walk(source_dir):
            for filename in filenames:
                pathfile = os.path.join(parent, filename)
                arcname = pathfile[pre_len:].strip(os.path.sep);#相对路径
                zipf.write(pathfile, arcname)
        zipf.close()
615658
def remove_gzip_timestamp(filename: str,
                          gunzip_executable: str = "gunzip",
                          gzip_executable: str = "gzip",
                          gzip_args: List[str] = None) -> None:
    
    gzip_args = gzip_args or [
        "-9",  # maximum compression (or Lintian moans)
        "-n",
    ]
    # gzip/gunzip operate on SINGLE files
    with tempfile.TemporaryDirectory() as dir_:
        basezipfilename = os.path.basename(filename)
        newzip = os.path.join(dir_, basezipfilename)
        with open(newzip, 'wb') as z:
            log.info(
                "Removing gzip timestamp: "
                "{} -> gunzip -c -> gzip -n -> {}",
                basezipfilename, newzip)
            p1 = subprocess.Popen([gunzip_executable, "-c", filename],
                                  stdout=subprocess.PIPE)
            p2 = subprocess.Popen([gzip_executable] + gzip_args,
                                  stdin=p1.stdout, stdout=z)
            p2.communicate()
        shutil.copyfile(newzip, filename)
351801
private static void unzip(final ZipFile zip, final File patchDir) throws IOException {
        final Enumeration<? extends ZipEntry> entries = zip.entries();
        while(entries.hasMoreElements()) {
            final ZipEntry entry = entries.nextElement();
            final String name = entry.getName();
            final File current = new File(patchDir, name);
            if(entry.isDirectory()) {
                continue;
            } else {
                if(! current.getParentFile().exists()) {
                    current.getParentFile().mkdirs();
                }
                try (final InputStream eis = zip.getInputStream(entry)){
                    Files.copy(eis, current.toPath());
                    //copy(eis, current);
                }
            }
        }
    }
172562
def unzip_file(filename):
    """'"""

    if filename.endswith('bz2'):
        bz2file = bz2.BZ2File(filename)
        fdn, tmpfilepath = tempfile.mkstemp()
        with closing(os.fdopen(fdn, 'wb')) as ofpt:
            try:
                ofpt.write(bz2file.read())
            except IOError:
                import traceback
                traceback.print_exc()
                LOGGER.info("Failed to read bzipped file %s", str(filename))
                os.remove(tmpfilepath)
                return None

        return tmpfilepath

    return None
670866
def unarchive(self, target_path=None, zip_path=None):
        
        if target_path:
            self.target_path = target_path

        if zip_path:
            self.zip_path = zip_path

        if self.has_path is False:
            raise RuntimeError("")

        if os.path.isdir(self.target_path) is False:
            os.mkdir(self.target_path)

        with zipfile.ZipFile(self.zip_path, 'r') as zip:
            zip.extractall(self.target_path)
1345828
def unpack(filename)
      @destination = Dir.mktmpdir
      Zip::File.open(filename){ |zip_file|
        zip_file.each{ |f|
          f_path=File.join(@destination, f.name)
          FileUtils.mkdir_p(File.dirname(f_path))
          zip_file.extract(f, f_path) unless File.exists?(f_path)
        }
      }
    end