def uploadfile(srcfile, dstfolder):
'''
Copy a file to the upload folder
srcfile: Full source filename including path
'''
uploadfile = os.path.join(dstfolder, os.path.basename(srcfile))
folderinit(dstfolder, 'Upload file destination folder')
try:
shutil.copy2(srcfile, uploadfile)
except Exception as e:
logger.error('File "{}" could not be copied to upload folder "{}".\n\tException Message: {}'.format(srcfile, dstfolder))
return
python类copy2()的实例源码
def copy_cwl_files(from_dir=CWL_PATH):
"""Copy cwl files to a directory where the cwl-runner can find them.
cwl files are copied to $XDG_DATA_HOME/commonwl/ This is one of the default
locations where the cwl-runner looks for cwl files.
Args:
from_dir (str): Path to directory where to copy files from (default:
the cwl directory of nlppln).
"""
cwl_data_dir = os.environ.get('XDG_DATA_HOME')
if not cwl_data_dir:
cwl_data_dir = DEFAULT_DATA_DIR
cwl_data_dir = os.path.join(cwl_data_dir, CWL_DATA_DIR_PREFIX)
create_dirs(cwl_data_dir)
cwl_files = glob.glob('{}{}*.cwl'.format(from_dir, os.sep))
for fi in cwl_files:
fo = os.path.join(cwl_data_dir, os.path.basename(fi))
shutil.copy2(fi, fo)
def rename_file(entry, pdf, bib, dry_run):
if entry['author']:
authors = entry['author'].split(',')
if len(authors) <= 3:
author = ', '.join(authors[:-1])
else:
author = authors[0] + ' et al.'
if author and 'year' in entry and 'title' in entry:
newname = author + ' - ' + '{}'.format(entry['year']) + ' - ' + algo.tex_to_unicode(algo.title_case(entry['title'])).replace("/", " ") + '.pdf'
if os.path.exists(pdf):
shutil.copy2(pdf, os.path.expanduser("~") + papers_path + newname)
entry.set_tag('file', ':' + pdf + ':PDF' )
if not dry_run:
shutil.move(pdf, os.path.expanduser("~") + '/.local/share/Trash/files/')
return True
return False
def copy_nrpe_checks():
"""
Copy the nrpe checks into place
"""
NAGIOS_PLUGINS = '/usr/local/lib/nagios/plugins'
nrpe_files_dir = os.path.join(os.getenv('CHARM_DIR'), 'hooks',
'charmhelpers', 'contrib', 'openstack',
'files')
if not os.path.exists(NAGIOS_PLUGINS):
os.makedirs(NAGIOS_PLUGINS)
for fname in glob.glob(os.path.join(nrpe_files_dir, "check_*")):
if os.path.isfile(fname):
shutil.copy2(fname,
os.path.join(NAGIOS_PLUGINS, os.path.basename(fname)))
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 gen_setup(self, filename, fragment, tmpdir):
match = EGG_FRAGMENT.match(fragment)
dists = match and [
d for d in
interpret_distro_name(filename, match.group(1), None) if d.version
] or []
if len(dists) == 1: # unambiguous ``#egg`` fragment
basename = os.path.basename(filename)
# Make sure the file has been downloaded to the temp dir.
if os.path.dirname(filename) != tmpdir:
dst = os.path.join(tmpdir, basename)
from setuptools.command.easy_install import samefile
if not samefile(filename, dst):
shutil.copy2(filename, dst)
filename = dst
with open(os.path.join(tmpdir, 'setup.py'), 'w') as file:
file.write(
"from setuptools import setup\n"
"setup(name=%r, version=%r, py_modules=[%r])\n"
% (
dists[0].project_name, dists[0].version,
os.path.splitext(basename)[0]
)
)
return filename
elif match:
raise DistutilsError(
"Can't unambiguously interpret project/version identifier %r; "
"any dashes in the name or version should be escaped using "
"underscores. %r" % (fragment, dists)
)
else:
raise DistutilsError(
"Can't process plain .py files without an '#egg=name-version'"
" suffix to enable automatic setup script generation."
)
def gen_setup(self, filename, fragment, tmpdir):
match = EGG_FRAGMENT.match(fragment)
dists = match and [
d for d in
interpret_distro_name(filename, match.group(1), None) if d.version
] or []
if len(dists) == 1: # unambiguous ``#egg`` fragment
basename = os.path.basename(filename)
# Make sure the file has been downloaded to the temp dir.
if os.path.dirname(filename) != tmpdir:
dst = os.path.join(tmpdir, basename)
from setuptools.command.easy_install import samefile
if not samefile(filename, dst):
shutil.copy2(filename, dst)
filename = dst
with open(os.path.join(tmpdir, 'setup.py'), 'w') as file:
file.write(
"from setuptools import setup\n"
"setup(name=%r, version=%r, py_modules=[%r])\n"
% (
dists[0].project_name, dists[0].version,
os.path.splitext(basename)[0]
)
)
return filename
elif match:
raise DistutilsError(
"Can't unambiguously interpret project/version identifier %r; "
"any dashes in the name or version should be escaped using "
"underscores. %r" % (fragment, dists)
)
else:
raise DistutilsError(
"Can't process plain .py files without an '#egg=name-version'"
" suffix to enable automatic setup script generation."
)
def local_update(repo_path, deb_paths=[]):
"""
Updates a local deb repository by copying debs and running scanpackages.
Args:
repo_path: the path to the local repository.
dep_paths: list of problem deb paths to copy.
"""
if not exists(repo_path):
logger.info("Creating repository at '%s'.", repo_path)
makedirs(repo_path)
elif not isdir(repo_path):
logger.error("Repository '%s' is not a directory!", repo_path)
raise FatalException
[copy2(deb_path, repo_path) for deb_path in deb_paths]
shell = spur.LocalShell()
result = shell.run(["dpkg-scanpackages", ".", "/dev/null"], cwd=repo_path)
packages_path = join(repo_path, "Packages.gz")
with gzip.open(packages_path, "wb") as packages:
packages.write(result.output)
logger.info("Repository '%s' updated successfully. Copied %d packages.", repo_path, len(deb_paths))
def deploy_files(staging_directory, instance_directory, file_list, username, problem_class):
"""
Copies the list of files from the staging directory to the instance directory.
Will properly set permissions and setgid files based on their type.
"""
# get uid and gid for default and problem user
user = getpwnam(username)
default = getpwnam(deploy_config.default_user)
for f in file_list:
# copy the file over, making the directories as needed
output_path = join(instance_directory, f.path)
if not os.path.isdir(os.path.dirname(output_path)):
os.makedirs(os.path.dirname(output_path))
if not isinstance(f, Directory):
if isinstance(f, PreTemplatedFile):
file_source = join(staging_directory, "__pre_templated", f.path)
else:
file_source = join(staging_directory, f.path)
shutil.copy2(file_source, output_path)
# set the ownership based on the type of file
if isinstance(f, ProtectedFile) or isinstance(f, ExecutableFile):
os.chown(output_path, default.pw_uid, user.pw_gid)
else:
uid = default.pw_uid if f.user is None else getpwnam(f.user).pw_uid
gid = default.pw_gid if f.group is None else getgrnam(f.group).gr_gid
os.chown(output_path, uid, gid)
# set the permissions appropriately
os.chmod(output_path, f.permissions)
if issubclass(problem_class, Service):
os.chown(instance_directory, default.pw_uid, user.pw_gid)
os.chmod(instance_directory, 0o750)
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_nrpe_checks():
"""
Copy the nrpe checks into place
"""
NAGIOS_PLUGINS = '/usr/local/lib/nagios/plugins'
nrpe_files_dir = os.path.join(os.getenv('CHARM_DIR'), 'hooks',
'charmhelpers', 'contrib', 'openstack',
'files')
if not os.path.exists(NAGIOS_PLUGINS):
os.makedirs(NAGIOS_PLUGINS)
for fname in glob.glob(os.path.join(nrpe_files_dir, "check_*")):
if os.path.isfile(fname):
shutil.copy2(fname,
os.path.join(NAGIOS_PLUGINS, os.path.basename(fname)))
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 main():
beam_data = np.load(ARGS.data)
# Optionally load vocabulary data
vocab = None
if ARGS.vocab:
with open(ARGS.vocab) as file:
vocab = file.readlines()
vocab = [_.strip() for _ in vocab]
vocab += ["UNK", "SEQUENCE_START", "SEQUENCE_END"]
if not os.path.exists(ARGS.output_dir):
os.makedirs(ARGS.output_dir)
# Copy required files
shutil.copy2("./bin/tools/beam_search_viz/tree.css", ARGS.output_dir)
shutil.copy2("./bin/tools/beam_search_viz/tree.js", ARGS.output_dir)
for idx in range(len(beam_data["predicted_ids"])):
predicted_ids = beam_data["predicted_ids"][idx]
parent_ids = beam_data["beam_parent_ids"][idx]
scores = beam_data["scores"][idx]
graph = create_graph(
predicted_ids=predicted_ids,
parent_ids=parent_ids,
scores=scores,
vocab=vocab)
json_str = json.dumps(
json_graph.tree_data(graph, (0, 0)),
ensure_ascii=False)
html_str = HTML_TEMPLATE.substitute(DATA=json_str)
output_path = os.path.join(ARGS.output_dir, "{:06d}.html".format(idx))
with open(output_path, "w") as file:
file.write(html_str)
print(output_path)
def refilemessages(self, list, tofolder, keepsequences=0):
"""Refile one or more messages -- may raise os.error.
'tofolder' is an open folder object."""
errors = []
refiled = {}
for n in list:
ton = tofolder.getlast() + 1
path = self.getmessagefilename(n)
topath = tofolder.getmessagefilename(ton)
try:
os.rename(path, topath)
except os.error:
# Try copying
try:
shutil.copy2(path, topath)
os.unlink(path)
except (IOError, os.error), msg:
errors.append(msg)
try:
os.unlink(topath)
except os.error:
pass
continue
tofolder.setlast(ton)
refiled[n] = ton
if refiled:
if keepsequences:
tofolder._copysequences(self, refiled.items())
self.removefromallsequences(refiled.keys())
if errors:
if len(errors) == 1:
raise os.error, errors[0]
else:
raise os.error, ('multiple errors:', errors)
def movemessage(self, n, tofolder, ton):
"""Move one message over a specific destination message,
which may or may not already exist."""
path = self.getmessagefilename(n)
# Open it to check that it exists
f = open(path)
f.close()
del f
topath = tofolder.getmessagefilename(ton)
backuptopath = tofolder.getmessagefilename(',%d' % ton)
try:
os.rename(topath, backuptopath)
except os.error:
pass
try:
os.rename(path, topath)
except os.error:
# Try copying
ok = 0
try:
tofolder.setlast(None)
shutil.copy2(path, topath)
ok = 1
finally:
if not ok:
try:
os.unlink(topath)
except os.error:
pass
os.unlink(path)
self.removefromallsequences([n])
def extract_targz(tarname, filename=None, target_dir='.'):
"""filename must be a valid path in the tar"""
import tarfile
tmp_dir = '._tmp_'
if filename is None:
tarfile.TarFile.gzopen(tarname).extractall(target_dir)
else:
import shutil
tarfile.TarFile.gzopen(tarname).extractall(tmp_dir)
shutil.copy2(os.path.join(tmp_dir, filename),
os.path.join(target_dir, filename.split(os.path.sep)[-1]))
shutil.rmtree(tmp_dir)
def prepare_model_dir(self):
if self.config.load_path:
self.model_dir=self.config.load_path
else:
pth=datetime.now().strftime("%m%d_%H%M%S")+'_'+self.data_type
self.model_dir=os.path.join(self.config.model_dir,pth)
if not os.path.exists(self.model_dir):
os.mkdir(self.model_dir)
print('Model directory is ',self.model_dir)
self.save_model_dir=os.path.join(self.model_dir,'checkpoints')
if not os.path.exists(self.save_model_dir):
os.mkdir(self.save_model_dir)
self.save_model_name=os.path.join(self.save_model_dir,'Model')
param_path = os.path.join(self.model_dir, "params.json")
print("[*] MODEL dir: %s" % self.model_dir)
print("[*] PARAM path: %s" % param_path)
with open(param_path, 'w') as fp:
json.dump(self.config.__dict__, fp, indent=4, sort_keys=True)
config=self.config
if config.is_train and not config.load_path:
config.log_code_dir=os.path.join(self.model_dir,'code')
for path in [self.model_dir, config.log_code_dir]:
if not os.path.exists(path):
os.makedirs(path)
#Copy python code in directory into model_dir/code for future reference:
code_dir=os.path.dirname(os.path.realpath(sys.argv[0]))
model_files = [f for f in listdir(code_dir) if isfile(join(code_dir, f))]
for f in model_files:
if f.endswith('.py'):
shutil.copy2(f,config.log_code_dir)
def prepare_dirs_and_logger(config):
formatter = logging.Formatter("%(asctime)s:%(levelname)s::%(message)s")
logger = logging.getLogger()
for hdlr in logger.handlers:
logger.removeHandler(hdlr)
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)
if config.load_path:
if config.load_path.startswith(config.log_dir):
config.model_dir = config.load_path
else:
if config.load_path.startswith(config.dataset):
config.model_name = config.load_path
else:
config.model_name = "{}_{}".format(config.dataset, config.load_path)
else:
config.model_name = "{}_{}".format(config.dataset, get_time())
if not hasattr(config, 'model_dir'):
config.model_dir = os.path.join(config.log_dir, config.model_name)
config.data_path = os.path.join(config.data_dir, config.dataset)
if config.is_train:
config.log_code_dir=os.path.join(config.model_dir,'code')
for path in [config.log_dir, config.data_dir,
config.model_dir, config.log_code_dir]:
if not os.path.exists(path):
os.makedirs(path)
#Copy python code in directory into model_dir/code for future reference:
code_dir=os.path.dirname(os.path.realpath(sys.argv[0]))
model_files = [f for f in listdir(code_dir) if isfile(join(code_dir, f))]
for f in model_files:
if f.endswith('.py'):
shutil.copy2(f,config.log_code_dir)
def prepare_dirs_and_logger(config):
formatter = logging.Formatter("%(asctime)s:%(levelname)s::%(message)s")
logger = logging.getLogger()
for hdlr in logger.handlers:
logger.removeHandler(hdlr)
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)
if config.load_path:
if config.load_path.startswith(config.log_dir):
config.model_dir = config.load_path
else:
if config.load_path.startswith(config.dataset):
config.model_name = config.load_path
else:
config.model_name = "{}_{}".format(config.dataset, config.load_path)
else:
config.model_name = "{}_{}".format(config.dataset, get_time())
if not hasattr(config, 'model_dir'):
config.model_dir = os.path.join(config.log_dir, config.model_name)
config.data_path = os.path.join(config.data_dir, config.dataset)
if not config.load_path:
config.log_code_dir=os.path.join(config.model_dir,'code')
for path in [config.log_dir, config.data_dir,
config.model_dir, config.log_code_dir]:
if not os.path.exists(path):
os.makedirs(path)
#Copy python code in directory into model_dir/code for future reference:
code_dir=os.path.dirname(os.path.realpath(sys.argv[0]))
model_files = [f for f in listdir(code_dir) if isfile(join(code_dir, f))]
for f in model_files:
if f.endswith('.py'):
shutil.copy2(f,config.log_code_dir)
def set_image(self, file_):
"""Select currently active image"""
shutil.copy2(file_, self.temporary.name) # create temporary copy
self.current = self.load_image_data(file_, self.temporary.name) # parse SVG data
