def copy_resource_dir(src, dest):
"""
To copy package data directory to destination
"""
package_name = "mocha"
dest = (dest + "/" + os.path.basename(src)).rstrip("/")
if pkg_resources.resource_isdir(package_name, src):
if not os.path.isdir(dest):
os.makedirs(dest)
for res in pkg_resources.resource_listdir(__name__, src):
copy_resource_dir(src + "/" + res, dest)
else:
if not os.path.isfile(dest) and os.path.splitext(src)[1] not in [".pyc"]:
copy_resource_file(src, dest)
python类resource_listdir()的实例源码
def _init_plugins(self, plugindir, plugins_to_load=None):
if plugindir and not os.path.isdir(plugindir):
raise InvalidPluginDir(plugindir)
if not plugindir:
plugindir = DIR("plugins")
logger.debug("plugindir: {0}".format(plugindir))
if os.path.isdir(plugindir):
pluginfiles = glob(os.path.join(plugindir, "[!_]*.py"))
plugins = strip_extension(os.path.basename(p) for p in pluginfiles)
else:
# we might be in an egg; try to get the files that way
logger.debug("trying pkg_resources")
import pkg_resources
try:
plugins = strip_extension(
pkg_resources.resource_listdir(__name__, "plugins"))
except OSError:
raise InvalidPluginDir(plugindir)
hooks = {}
oldpath = copy.deepcopy(sys.path)
sys.path.insert(0, plugindir)
for plugin in plugins:
if plugins_to_load and plugin not in plugins_to_load:
logger.debug("skipping plugin {0}, not in plugins_to_load {1}".format(plugin, plugins_to_load))
continue
logger.debug("plugin: {0}".format(plugin))
try:
mod = importlib.import_module(plugin)
modname = mod.__name__
for hook in re.findall("on_(\w+)", " ".join(dir(mod))):
hookfun = getattr(mod, "on_" + hook)
logger.debug("plugin: attaching %s hook for %s", hook, modname)
hooks.setdefault(hook, []).append(hookfun)
if mod.__doc__:
# firstline = mod.__doc__.split('\n')[0]
part_attachment = json.loads(mod.__doc__)
hooks.setdefault('help', {})[modname] = part_attachment
hooks.setdefault('extendedhelp', {})[modname] = mod.__doc__
# bare except, because the modules could raise any number of errors
# on import, and we want them not to kill our server
except:
logger.warning("import failed on module {0}, module not loaded".format(plugin))
logger.warning("{0}".format(sys.exc_info()[0]))
logger.warning("{0}".format(traceback.format_exc()))
sys.path = oldpath
return hooks
def init_templates(config_file):
config = configparser.ConfigParser(interpolation=configparser.ExtendedInterpolation())
config.read(config_file)
template_dir = config.get('DEFAULT', 'TEMPLATE_DIR')
texmf_dir = os.path.expanduser('~/texmf/tex/latex/fensterbrief/')
# check if template directory exists
if not os.path.exists(template_dir):
answer = input("+ Shall directory %s be created? " % template_dir).lower()
if 'y' in answer:
os.makedirs(template_dir)
else:
return
# create user's 'texmf' directory
if not os.path.exists(texmf_dir):
answer = input("+ Shall directory %s be created? " % texmf_dir).lower()
if 'y' in answer:
os.makedirs(texmf_dir)
else:
return
# copy templates to tempalte directory
for res_name in resource_listdir('templates', ''):
if res_name.endswith(".tex") or res_name.endswith(".md") or res_name.endswith(".lco") or res_name.endswith(".sty"):
src_fd = resource_stream('templates', res_name)
if res_name.endswith(".tex") or res_name.endswith(".md"):
dst_file = os.path.join(template_dir, res_name)
else:
dst_file = os.path.join(texmf_dir, res_name)
print("+ Copy resource file to %s" % dst_file)
write_file = False
if os.path.exists(dst_file):
answer = input("+ Shall %s be overwritten? " % dst_file).lower()
if 'y' in answer:
write_file = True
else:
write_file = True
if write_file:
with open(dst_file, 'wb') as dst_fd:
shutil.copyfileobj(src_fd, dst_fd)
# update
fensterbrief.run_program('texhash')
def _validate_config_files(self):
"""
Validates the configuration files necessary for the application. An exception is thrown
if any of the required files are inaccessible.
"""
# Determine the module of the derived class
mod = self.__class__.__module__
# If the configuration directory exists in the library, create config files as necessary
# This check also provides backwards compatibility for projects that don't have the
# configuration files in the library.
if pkg_resources.resource_exists(mod, self.LIB_CONFIG_DIR):
# Create configuration directory if not found
if not os.access(self._config_dir, os.R_OK):
logger.info("Configuration directory '{0}' not found, creating...".format(self._config_dir))
os.makedirs(self._config_dir)
# Count of current configuration files
config_files_count = len([name for name in os.listdir(self._config_dir)
if os.path.isfile(os.path.join(self._config_dir, name))])
# Create configuration files if not found
files = pkg_resources.resource_listdir(mod, self.LIB_APP_CONFIG_DIR)
for f in files:
config_path = os.path.join(self._config_dir, f)
if not os.access(config_path, os.R_OK):
f_lower = f.lower()
# Copy configuration file. Only copy logging file if the directory was empty
if not(f_lower.endswith(".py") or f_lower.endswith(".pyc")) and \
(f_lower != Application.LOGGING_CONFIG_FILE or config_files_count == 0):
logger.info("Configuration file '{0}' not found, creating...".format(f))
shutil.copyfile(pkg_resources.resource_filename(
mod, self.LIB_APP_CONFIG_DIR + "/" + f), config_path)
if not os.access(self._dxlclient_config_path, os.R_OK):
raise Exception(
"Unable to access client configuration file: {0}".format(
self._dxlclient_config_path))
if not os.access(self._app_config_path, os.R_OK):
raise Exception(
"Unable to access application configuration file: {0}".format(
self._app_config_path))
def dbsource(dbname, var, resolution=None, tscale=None):
"""
Temporary solution, just to move on with CoTeDe.
"""
db_cfg = {}
cfg_dir = 'datasource'
for src_cfg in pkg_resources.resource_listdir('oceansdb', cfg_dir):
text = pkg_resources.resource_string(
'oceansdb', os.path.join(cfg_dir, src_cfg))
text = text.decode('UTF-8', 'replace')
cfg = json.loads(text)
for c in cfg:
assert c not in db_cfg, "Trying to overwrite %s"
db_cfg[c] = cfg[c]
dbpath = oceansdb_dir()
datafiles = []
cfg = db_cfg[dbname]
if (resolution is None):
resolution = cfg['vars'][var]['default_resolution']
if (tscale is None):
tscale = cfg['vars'][var][resolution]["default_tscale"]
for c in cfg['vars'][var][resolution][tscale]:
download_file(outputdir=dbpath, **c)
if 'filename' in c:
filename = os.path.join(dbpath, c['filename'])
else:
filename = os.path.join(dbpath,
os.path.basename(urlparse(c['url']).path))
if 'varnames' in cfg['vars'][var][resolution]:
datafiles.append(Dataset_flex(filename,
aliases=cfg['vars'][var][resolution]['varnames']))
else:
datafiles.append(Dataset_flex(filename))
return datafiles
def copy_dir(source, dest, variables, out_=sys.stdout, i=0):
"""
Copies the ``source`` directory to the ``dest`` directory, where
``source`` is some tuple representing an installed package and a
subdirectory in the package, e.g.,
('pecan', os.path.join('scaffolds', 'base'))
('pecan_extension', os.path.join('scaffolds', 'scaffold_name'))
``variables``: A dictionary of variables to use in any substitutions.
Substitution is performed via ``string.Template``.
``out_``: File object to write to (default is sys.stdout).
"""
def out(msg):
out_.write('%s%s' % (' ' * (i * 2), msg))
out_.write('\n')
out_.flush()
names = sorted(pkg_resources.resource_listdir(source[0], source[1]))
if not os.path.exists(dest):
out('Creating %s' % dest)
makedirs(dest)
else:
out('%s already exists' % dest)
return
for name in names:
full = '/'.join([source[1], name])
dest_full = os.path.join(dest, substitute_filename(name, variables))
sub_file = False
if dest_full.endswith('_tmpl'):
dest_full = dest_full[:-5]
sub_file = True
if pkg_resources.resource_isdir(source[0], full):
out('Recursing into %s' % os.path.basename(full))
copy_dir((source[0], full), dest_full, variables, out_, i + 1)
continue
else:
content = pkg_resources.resource_string(source[0], full)
if sub_file:
content = render_template(content, variables)
if content is None:
continue # pragma: no cover
out('Copying %s to %s' % (full, dest_full))
f = open(dest_full, 'wb')
f.write(content)
f.close()
def run(cls, args):
samples = """
conf.sample.yaml Hoaxy configuration file
domains_claim.sample.txt Claim domains
domains_factchecking.sample.txt Factchecking domains
site.sample.yaml Claim and/or factchecking sites
crontab.sample.txt Crontab sample
"""
if args['--home'] is None:
hoaxy_home = HOAXY_HOME
msg = """
Sample files are put into the default location:
'{}'.
Please edit and rename sample files to make Hoaxy work with them.
{}"""
msg = msg.format(hoaxy_home, samples)
else:
hoaxy_home = os.path.expanduser(args['--home'])
if not hoaxy_home.endswith('/'):
hoaxy_home += '/'
msg = """
Sample files are put into folder
'{}'.
You need to set environment
HOAXY_HOME={}
to activate this path.
Also please edit and rename samples to make Hoaxy work with them.
{}"""
msg = msg.format(hoaxy_home, hoaxy_home, samples)
if not os.path.exists(hoaxy_home):
try:
org_umask = os.umask(0)
os.makedirs(hoaxy_home, 0755)
finally:
os.umask(org_umask)
samples = resource_listdir('hoaxy.data', 'samples')
for sample in samples:
if not sample.startswith('__init__.'):
sample = resource_filename('hoaxy.data.samples', sample)
shutil.copy(sample, hoaxy_home)
os.chmod(
os.path.join(hoaxy_home, os.path.basename(sample)), 0644)
print(msg)
