python类copytree()的实例源码

def build_wheel(wheel_directory, config_settings=None,
                metadata_directory=None):
    config_settings = _fix_config(config_settings)
    wheel_directory = os.path.abspath(wheel_directory)
    sys.argv = sys.argv[:1] + ['bdist_wheel'] + \
        config_settings["--global-option"]
    _run_setup()
    if wheel_directory != 'dist':
        shutil.rmtree(wheel_directory)
        shutil.copytree('dist', wheel_directory)

    wheels = [f for f in os.listdir(wheel_directory)
              if f.endswith('.whl')]

    assert len(wheels) == 1
    return wheels[0]

def src_proc_dispatcher(pkg_name, src_tbl_name, src_loc):
    tobj = tempfile.mkdtemp(dir='/var/cache/acbs/build/', prefix='acbs.')
    src_tbl_loc = os.path.join(src_loc, src_tbl_name)
    shadow_ark_loc = os.path.join(tobj, src_tbl_name)
    if os.path.isdir(src_tbl_loc):
        print('[I] Making a copy of the source directory...', end='')
        try:
            shutil.copytree(src=src_tbl_loc, dst=shadow_ark_loc)
        except:
            print('Failed!')
            return False
        print('Done!')
        return True, tobj
    else:
        os.symlink(src_tbl_loc, shadow_ark_loc)
        # print('[D] Source location: {}, Shadow link: {}'.format(src_tbl_loc, shadow_ark_loc))
        return decomp_file(shadow_ark_loc, tobj), tobj

def createTestDomain():
    domainName = getTestDomainName()
    print domainName

    domainPath = os.path.join(getSdrPath(), "dom", "domain")
    templatePath = os.path.join(getSdrPath(), "templates", "domain")
    # Create a test domain
    if os.path.isdir(domainPath):
        shutil.rmtree(domainPath)
    # Copy the template over
    shutil.copytree(templatePath, domainPath)
    # Open the DMD file and replace the name, using a very naive method
    dmd = open(os.path.join(domainPath, "DomainManager.dmd.xml"), "r")
    lines = dmd.read()
    dmd.close()
    lines = lines.replace("${DOMAINNAME}", domainName)
    dmd = open(os.path.join(domainPath, "DomainManager.dmd.xml"), "w+")
    dmd.write(lines)
    dmd.close()

    setupDeviceAndDomainMgrPackage()

def copy_abd(tmp_dir_loc, repo_dir, pkg_info):
    if pkg_info['SUBDIR'] != '':
        try:
            os.chdir(pkg_info['SUBDIR'])
        except:
            err_msg('Failed to enter sub-directory!')
            return False
    else:
        try:
            os.chdir(pkg_info['NAME'] + '-' + pkg_info['VER'])
        except:
            try:
                os.chdir(pkg_info['NAME'])
            except:
                err_msg(
                    'Failed to determine sub-directory, please specify manually.')
                return False
    try:
        shutil.copytree(repo_dir,
                        os.path.abspath(os.path.curdir) + '/autobuild/', symlinks=True)
    except:
        err_msg('Error occurred when copying files from tree!')
        return False
    return True

def build(**args):
    freezer = Freezer(app)
    if args.get('base_url'):
        app.config['FREEZER_BASE_URL'] = args.get('base_url')
    app.config['mathjax_node'] = args.get('node_mathjax', False)
    app.config['MATHJAX_WHOLEBOOK'] = args.get('node_mathjax', False)
    app.config['FREEZER_DESTINATION'] = os.path.join(curdir, "build")
    app.config['freeze'] = True
    freezer.freeze()

    if args.get('copy_mathjax'):

        mathjax_postfix = os.path.join('assets', "js", "mathjax")
        mathjax_from = os.path.join(scriptdir, mathjax_postfix)
        mathjax_to = os.path.join(curdir, "build", mathjax_postfix)

        try:
            shutil.rmtree(mathjax_to)
        except FileNotFoundError:
            pass

        shutil.copytree(mathjax_from, mathjax_to)

def copyFromReader(self, reader, sourcePath, destPath):
        """
        Copy the sourcePath in the provided UFOReader to destPath
        in this writer. The paths must be relative. They may represent
        directories or paths. This uses the most memory efficient
        method possible for copying the data possible.
        """
        if not isinstance(reader, UFOReader):
            raise UFOLibError("The reader must be an instance of UFOReader.")
        fullSourcePath = os.path.join(reader._path, sourcePath)
        if not reader._checkForFile(fullSourcePath):
            raise UFOLibError("No file named \"%s\" to copy from." % sourcePath)
        fullDestPath = os.path.join(self._path, destPath)
        if os.path.exists(fullDestPath):
            raise UFOLibError("A file named \"%s\" already exists." % sourcePath)
        self._buildDirectoryTree(destPath)
        if os.path.isdir(fullSourcePath):
            shutil.copytree(fullSourcePath, fullDestPath)
        else:
            shutil.copy(fullSourcePath, fullDestPath)

    # UFO mod time

def sourcecpy(src, des):
    src = os.path.normpath(src)
    des = os.path.normpath(des)
    if not os.path.exists(src) or not os.path.exists(src):
        print("folder is not exist")
        sys.exit(1)
    # ?????????????????????????
    os.chdir(src)
    src_file = [os.path.join(src, file) for file in os.listdir()]
    for source in src_file:
        # ????
        if os.path.isfile(source):
            shutil.copy(source, des)  # ????????????????
        # ????
        if os.path.isdir(source):
            p, src_name = os.path.split(source)
            des = os.path.join(des, src_name)
            shutil.copytree(source, des)  # ??????????????????

# ??CLI?????

def setUp(self):
        """Create a :class:`CaptchaFetchResource`."""
        secretKey, publicKey = crypto.getRSAKey('captcha.key', bits=1024)

        # Set up our resources to fake a minimal HTTP(S) server:
        self.pagename = b'fetch'
        self.root = Resource()

        shutil.copytree('../captchas', os.path.sep.join([os.getcwd(), 'captchas']))

        self.captchaDir = os.path.sep.join([os.getcwd(), 'captchas'])
        self.captchaResource = server.CaptchaFetchResource(
            secretKey=secretKey,
            publicKey=publicKey,
            hmacKey='abcdefghijklmnopqrstuvwxyz012345',
            captchaDir=self.captchaDir,
            useForwardedHeader=True)

        self.root.putChild(self.pagename, self.captchaResource)

        # Set up the basic parts of our faked request:
        self.request = DummyRequest([self.pagename])

def setUp(self):
        """Create a :class:`CaptchaFetchResource`.
        """
        secretKey, publicKey = crypto.getRSAKey('captcha.key', bits=1024)

        # Set up our resources to fake a minimal HTTP(S) server:
        self.pagename = b'check'
        self.root = Resource()

        shutil.copytree('../captchas', os.path.sep.join([os.getcwd(), 'captchas']))

        self.captchaDir = os.path.sep.join([os.getcwd(), 'captchas'])
        self.captchaResource = server.CaptchaCheckResource(
            secretKey=secretKey,
            publicKey=publicKey,
            hmacKey='abcdefghijklmnopqrstuvwxyz012345',
            useForwardedHeader=True)

        self.root.putChild(self.pagename, self.captchaResource)

        # Set up the basic parts of our faked request:
        self.request = DummyRequest([self.pagename])

def create_backup(self):
        if self._verbose:print("Backing up current addon folder")
        local = os.path.join(self._updater_path,"backup")
        tempdest = os.path.join(self._addon_root,
                        os.pardir,
                        self._addon+"_updater_backup_temp")

        if os.path.isdir(local) == True:
            shutil.rmtree(local)
        if self._verbose:print("Backup destination path: ",local)

        # make the copy
        shutil.copytree(self._addon_root,tempdest)
        shutil.move(tempdest,local)

        # save the date for future ref
        now = datetime.now()
        self._json["backup_date"] = "{m}-{d}-{yr}".format(
                m=now.strftime("%B"),d=now.day,yr=now.year)
        self.save_updater_json()

def copy(self, target, nameIsLeaf=False):
        """
        same as rename - except for copying.  returns the new target name
        """
        if self.isfile():
            target = Path(target)
            if nameIsLeaf:
                target = self.up() / target

            if self == target:
                return target

            targetDirpath = target.up()
            if not targetDirpath.exists():
                targetDirpath.create()

            shutil.copy2(str(self), str(target))

            return target
        elif self.isdir():
            shutil.copytree(str(self), str(target))

def copytree(src, dst, symlinks=False, ignore=None):
    """
    Correctly copy an entire directory

    :param src: source dir
    :param dst: destination dir
    :param symlinks: copy content of symlinks
    :param ignore: ignore
    """
    for item in os.listdir(src):
        s = os.path.join(src, item)
        d = os.path.join(dst, item)
        if os.path.isdir(s):
            shutil.copytree(s, d, symlinks, ignore)
        else:
            shutil.copy2(s, d)

def copy(src, dst):
    """
    Handle the copying of a file or directory.

    The destination basedir _must_ exist.

    :param src: A string containing the path of the source to copy.  If the
     source ends with a '/', will become a recursive directory copy of source.
    :param dst: A string containing the path to the destination.  If the
     destination ends with a '/', will copy into the target directory.
    :return: None
    """
    try:
        shutil.copytree(src, dst)
    except OSError as exc:
        if exc.errno == errno.ENOTDIR:
            shutil.copy(src, dst)
        else:
            raise

def main(argv):
    parser = argparse.ArgumentParser(
        description="Add DATA of multiple MeasurementSets")
    parser.add_argument("-i", "--input", dest="input", nargs="+",
                        help="two or more MeasurementSets to be added")
    parser.add_argument("-o", "--output", dest="output", required=True,
                        help="output MeasurementSet name")
    args = parser.parse_args(argv)

    nms = len(args.input)
    if nms < 2:
        raise RuntimeError("Two or more input MeasurementSets required")

    print("Copying the first input MS to be the output MS ...")
    ms1 = args.input[0]
    msout = args.output
    shutil.copytree(ms1, msout)

    data2 = []
    for msname in args.input[1:]:
        data2.append(get_data(msname))
    add_ms(msout, data2)

def execute(self, context):
        if os.path.exists(self.filepath):
            self.report({'ERROR_INVALID_INPUT'}, "Error creating new project," + self.filepath + " file or directory already exists.")
            return {'CANCELLED'}

        name = os.path.basename(self.filepath) 
        if name == "":
             self.report({'ERROR_INVALID_INPUT'}, "Error creating new project, project name can not be empty.")
             return {'CANCELLED'}

        tfolder = bpy.utils.user_resource('DATAFILES') + LIBNAME + os.sep + 'template' + os.sep
        shutil.copytree(tfolder, self.filepath)

        #Open the new blend
        bpy.ops.wm.open_mainfile(filepath=self.filepath + os.sep + 'project'  + os.sep + 'main.blend')

        utils.loadProjectFile(self.filepath + os.sep + 'project.json')

        return {'FINISHED'}

def copytree(src, dst, symlinks=False, ignore=None):
    names = os.listdir(src)
    if ignore is not None:
        ignored_names = ignore(src, names)
    else:
        ignored_names = set()

    if not os.path.isdir(dst):
        os.makedirs(dst)

    errors = []
    for name in names:
        if name in ignored_names:
            continue

        srcname = os.path.join(src, name)
        dstname = os.path.join(dst, name)
        if os.path.isdir(srcname):
            copytree(srcname, dstname, symlinks, ignore)
        else:
            shutil.copy2(srcname, dstname)

    return

# ????????????????????

def _create(self, fixture_file_path, less_verbose=0):
        try:
            self.write_info('Creating fixture %s' % fixture_file_path, 1+less_verbose)
            fixture_file_path = re.sub(r'\.zip$', '', fixture_file_path)  # we strip away .zip if given
            tmp_dir = tempfile.mkdtemp()
            # copy media root
            shutil.copytree(self._media_root, join(tmp_dir, 'MEDIA_ROOT'))
            # database dump
            with open(join(tmp_dir, 'db.sql'), 'w') as fp:
                return_code = subprocess.call(['pg_dump', '--clean', '--no-owner', self._database_name], stdout=fp)
                if return_code != 0:
                    raise CommandError('pg_dump failed with exit code {}'.format(return_code))
            # creating the fixture archive
            archive_name = shutil.make_archive(fixture_file_path, 'zip', root_dir=tmp_dir)
            self.write_debug(subprocess.check_output(['unzip', '-l', archive_name]))
        except:
            self.write_debug('Temporary directory %s kept due to exception.' % tmp_dir)
            raise
        else:
            self.write_info('... fixture created', 1+less_verbose)
            shutil.rmtree(tmp_dir)

def _build_platform(self, context):
            context.try_skip(self.install_path)
            tools_dir = pj(self.install_path, 'tools')
            shutil.copytree(self.source_path, tools_dir)
            script = pj(tools_dir, 'android')
            command = '{script} --verbose update sdk -a --no-ui --filter {packages}'
            command = command.format(
                script=script,
                packages = ','.join(str(i) for i in [1,2,8,34,162])
            )
            # packages correspond to :
            # - 1 : Android SDK Tools, revision 25.2.5
            # - 2 : Android SDK Platform-tools, revision 25.0.3
            # - 8 : Android SDK Build-tools, revision 24.0.1
            # - 34 : SDK Platform Android 7.0, API 24, revision 2
            # - 162 : Android Support Repository, revision 44
            self.buildEnv.run_command(command, self.install_path, context, input="y\n")

def _configure(self, context):
            if not os.path.exists(self.build_path):
                shutil.copytree(self.source_path, self.build_path, symlinks=True)
            try:
                shutil.rmtree(pj(self.build_path, 'kiwixlib', 'src', 'main'))
            except FileNotFoundError:
                pass
            shutil.copytree(pj(self.buildEnv.install_dir, 'kiwix-lib'),
                            pj(self.build_path, 'kiwixlib', 'src', 'main'))
            os.makedirs(
                pj(self.build_path, 'app', 'src', 'main', 'assets', 'icu'),
                exist_ok=True)
            shutil.copy2(pj(self.buildEnv.install_dir, 'share', 'icu', '58.2',
                            'icudt58l.dat'),
                         pj(self.build_path, 'app', 'src', 'main', 'assets',
                            'icu', 'icudt58l.dat'))

def ExecRecursiveMirror(self, source, dest):
    """Emulation of rm -rf out && cp -af in out."""
    if os.path.exists(dest):
      if os.path.isdir(dest):
        def _on_error(fn, path, dummy_excinfo):
          # The operation failed, possibly because the file is set to
          # read-only. If that's why, make it writable and try the op again.
          if not os.access(path, os.W_OK):
            os.chmod(path, stat.S_IWRITE)
          fn(path)
        shutil.rmtree(dest, onerror=_on_error)
      else:
        if not os.access(dest, os.W_OK):
          # Attempt to make the file writable before deleting it.
          os.chmod(dest, stat.S_IWRITE)
        os.unlink(dest)

    if os.path.isdir(source):
      shutil.copytree(source, dest)
    else:
      shutil.copy2(source, dest)
      # Try to diagnose crbug.com/741603
      if not os.path.exists(dest):
        raise Exception("Copying of %s to %s failed" % (source, dest))

def _Execute(args):
    if not args.identity:
      args.identity = '-'

    install_path = args.path
    if args.output:

      if os.path.isfile(args.output):
        os.unlink(args.output)
      elif os.path.isdir(args.output):
        shutil.rmtree(args.output)

      if os.path.isfile(args.path):
        shutil.copy(args.path, args.output)
      elif os.path.isdir(args.path):
        shutil.copytree(args.path, args.output)

      install_path = args.output

    CodeSignBundle(install_path, args.identity,
      ['--deep', '--preserve-metadata=identifier,entitlements'])

def load_config(self):
        """Loads the next configuration for the plugin to test"""
        config = super(Proxy, self).load_config()
        self._all_names, self._test_names = _get_names(config)

        server_root = _get_server_root(config)

        # XXX: Deleting all of this is kind of scary unless the test
        #      instances really each have a complete configuration!
        shutil.rmtree("/etc/nginx")
        shutil.copytree(server_root, "/etc/nginx", symlinks=True)

        self._prepare_configurator()

        try:
            subprocess.check_call("service nginx reload".split())
        except errors.Error:
            raise errors.Error(
                "Nginx failed to load {0} before tests started".format(
                    config))

        return config

def load_config(self):
        """Loads the next configuration for the plugin to test"""
        config = super(Proxy, self).load_config()
        self._all_names, self._test_names = _get_names(config)

        server_root = _get_server_root(config)
        shutil.rmtree("/etc/apache2")
        shutil.copytree(server_root, "/etc/apache2", symlinks=True)

        self._prepare_configurator()

        try:
            subprocess.check_call("apachectl -k restart".split())
        except errors.Error:
            raise errors.Error(
                "Apache failed to load {0} before tests started".format(
                    config))

        return config

def create_backup(self):
        if self._verbose:print("Backing up current addon folder")
        local = os.path.join(self._updater_path,"backup")
        tempdest = os.path.join(self._addon_root,
                        os.pardir,
                        self._addon+"_updater_backup_temp")

        if os.path.isdir(local) == True:
            shutil.rmtree(local)
        if self._verbose:print("Backup destination path: ",local)

        # make the copy
        shutil.copytree(self._addon_root,tempdest)
        shutil.move(tempdest,local)

        # save the date for future ref
        now = datetime.now()
        self._json["backup_date"] = "{m}-{d}-{yr}".format(
                m=now.strftime("%B"),d=now.day,yr=now.year)
        self.save_updater_json()

def CopyPythonLibs(dst, overwrite_lib, report=print):
    import platform

    # use python module to find pytohn's libpath
    src = os.path.dirname(platform.__file__)

    # dst points to lib/, but src points to current python's library path, eg:
    #  '/usr/lib/python3.2' vs '/usr/lib'
    # append python's library dir name to destination, so only python's
    # libraries would be copied
    if os.name == 'posix':
        dst = os.path.join(dst, os.path.basename(src))

    if os.path.exists(src):
        write = False
        if os.path.exists(dst):
            if overwrite_lib:
                shutil.rmtree(dst)
                write = True
        else:
            write = True
        if write:
            shutil.copytree(src, dst, ignore=lambda dir, contents: [i for i in contents if i == '__pycache__'])
    else:
        report({'WARNING'}, "Python not found in %r, skipping pythn copy" % src)

def upgrade(theme):
    """
    upgrade the theme
    """
    run_in_root(os.getcwd())
    theme_path = os.path.join(os.getcwd(), 'theme/%s' % theme)
    templates_path = os.path.join(theme_path, 'templates')
    templates_target_path = os.path.join(os.getcwd(), 'templates')
    static_path = os.path.join(theme_path, 'static')
    static_target_path = os.path.join(os.getcwd(), 'static')

    shutil.rmtree(templates_target_path)
    shutil.rmtree(static_target_path)
    shutil.copytree(templates_path, templates_target_path)
    shutil.copytree(static_path, static_target_path)

    logger.info('''
        \033[33m{info}\033[0m
        ==> upgrade the theme!
        '''
    )

def create_backup(self):
        if self._verbose:print("Backing up current addon folder")
        local = os.path.join(self._updater_path,"backup")
        tempdest = os.path.join(self._addon_root,
                        os.pardir,
                        self._addon+"_updater_backup_temp")

        if os.path.isdir(local) == True:
            shutil.rmtree(local)
        if self._verbose:print("Backup destination path: ",local)

        # make the copy
        shutil.copytree(self._addon_root,tempdest)
        shutil.move(tempdest,local)

        # save the date for future ref
        now = datetime.now()
        self._json["backup_date"] = "{m}-{d}-{yr}".format(
                m=now.strftime("%B"),d=now.day,yr=now.year)
        self.save_updater_json()

def sync_directory(src, dest, opts=None):
    if os.path.exists(dest):
        logging.debug('Removing existing directory: %s' % dest)
        shutil.rmtree(dest)
    logging.info('Syncing directory: %s -> %s.' % (src, dest))

    shutil.copytree(src, dest, ignore=get_filter(opts))
    ensure_init(dest)

def copy_files(src, dst, symlinks=False, ignore=None):
    """Copy files from src to dst."""
    for item in os.listdir(src):
        s = os.path.join(src, item)
        d = os.path.join(dst, item)
        if os.path.isdir(s):
            shutil.copytree(s, d, symlinks, ignore)
        else:
            shutil.copy2(s, d)

def _copy_dist_from_dir(link_path, location):
    """Copy distribution files in `link_path` to `location`.

    Invoked when user requests to install a local directory. E.g.:

        pip install .
        pip install ~/dev/git-repos/python-prompt-toolkit

    """

    # Note: This is currently VERY SLOW if you have a lot of data in the
    # directory, because it copies everything with `shutil.copytree`.
    # What it should really do is build an sdist and install that.
    # See https://github.com/pypa/pip/issues/2195

    if os.path.isdir(location):
        rmtree(location)

    # build an sdist
    setup_py = 'setup.py'
    sdist_args = [sys.executable]
    sdist_args.append('-c')
    sdist_args.append(SETUPTOOLS_SHIM % setup_py)
    sdist_args.append('sdist')
    sdist_args += ['--dist-dir', location]
    logger.info('Running setup.py sdist for %s', link_path)

    with indent_log():
        call_subprocess(sdist_args, cwd=link_path, show_stdout=False)

    # unpack sdist into `location`
    sdist = os.path.join(location, os.listdir(location)[0])
    logger.info('Unpacking sdist %s into %s', sdist, location)
    unpack_file(sdist, location, content_type=None, link=None)