You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
392 lines
12 KiB
Python
392 lines
12 KiB
Python
2 years ago
|
# encoding: utf-8
|
||
|
|
"""
|
||
|
|
Utilities for path handling.
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Copyright (c) IPython Development Team.
|
||
|
|
# Distributed under the terms of the Modified BSD License.
|
||
|
|
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import errno
|
||
|
|
import shutil
|
||
|
|
import random
|
||
|
|
import glob
|
||
|
|
|
||
|
|
from IPython.utils.process import system
|
||
|
|
|
||
|
|
#-----------------------------------------------------------------------------
|
||
|
|
# Code
|
||
|
|
#-----------------------------------------------------------------------------
|
||
|
|
fs_encoding = sys.getfilesystemencoding()
|
||
|
|
|
||
|
|
def _writable_dir(path):
|
||
|
|
"""Whether `path` is a directory, to which the user has write access."""
|
||
|
|
return os.path.isdir(path) and os.access(path, os.W_OK)
|
||
|
|
|
||
|
|
if sys.platform == 'win32':
|
||
|
|
def _get_long_path_name(path):
|
||
|
|
"""Get a long path name (expand ~) on Windows using ctypes.
|
||
|
|
|
||
|
|
Examples
|
||
|
|
--------
|
||
|
|
|
||
|
|
>>> get_long_path_name('c:\\\\docume~1')
|
||
|
|
'c:\\\\Documents and Settings'
|
||
|
|
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
import ctypes
|
||
|
|
except ImportError as e:
|
||
|
|
raise ImportError('you need to have ctypes installed for this to work') from e
|
||
|
|
_GetLongPathName = ctypes.windll.kernel32.GetLongPathNameW
|
||
|
|
_GetLongPathName.argtypes = [ctypes.c_wchar_p, ctypes.c_wchar_p,
|
||
|
|
ctypes.c_uint ]
|
||
|
|
|
||
|
|
buf = ctypes.create_unicode_buffer(260)
|
||
|
|
rv = _GetLongPathName(path, buf, 260)
|
||
|
|
if rv == 0 or rv > 260:
|
||
|
|
return path
|
||
|
|
else:
|
||
|
|
return buf.value
|
||
|
|
else:
|
||
|
|
def _get_long_path_name(path):
|
||
|
|
"""Dummy no-op."""
|
||
|
|
return path
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def get_long_path_name(path):
|
||
|
|
"""Expand a path into its long form.
|
||
|
|
|
||
|
|
On Windows this expands any ~ in the paths. On other platforms, it is
|
||
|
|
a null operation.
|
||
|
|
"""
|
||
|
|
return _get_long_path_name(path)
|
||
|
|
|
||
|
|
|
||
|
|
def compress_user(path):
|
||
|
|
"""Reverse of :func:`os.path.expanduser`
|
||
|
|
"""
|
||
|
|
home = os.path.expanduser('~')
|
||
|
|
if path.startswith(home):
|
||
|
|
path = "~" + path[len(home):]
|
||
|
|
return path
|
||
|
|
|
||
|
|
def get_py_filename(name):
|
||
|
|
"""Return a valid python filename in the current directory.
|
||
|
|
|
||
|
|
If the given name is not a file, it adds '.py' and searches again.
|
||
|
|
Raises IOError with an informative message if the file isn't found.
|
||
|
|
"""
|
||
|
|
|
||
|
|
name = os.path.expanduser(name)
|
||
|
|
if os.path.isfile(name):
|
||
|
|
return name
|
||
|
|
if not name.endswith(".py"):
|
||
|
|
py_name = name + ".py"
|
||
|
|
if os.path.isfile(py_name):
|
||
|
|
return py_name
|
||
|
|
raise IOError("File `%r` not found." % name)
|
||
|
|
|
||
|
|
|
||
|
|
def filefind(filename: str, path_dirs=None) -> str:
|
||
|
|
"""Find a file by looking through a sequence of paths.
|
||
|
|
|
||
|
|
This iterates through a sequence of paths looking for a file and returns
|
||
|
|
the full, absolute path of the first occurrence of the file. If no set of
|
||
|
|
path dirs is given, the filename is tested as is, after running through
|
||
|
|
:func:`expandvars` and :func:`expanduser`. Thus a simple call::
|
||
|
|
|
||
|
|
filefind('myfile.txt')
|
||
|
|
|
||
|
|
will find the file in the current working dir, but::
|
||
|
|
|
||
|
|
filefind('~/myfile.txt')
|
||
|
|
|
||
|
|
Will find the file in the users home directory. This function does not
|
||
|
|
automatically try any paths, such as the cwd or the user's home directory.
|
||
|
|
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
filename : str
|
||
|
|
The filename to look for.
|
||
|
|
path_dirs : str, None or sequence of str
|
||
|
|
The sequence of paths to look for the file in. If None, the filename
|
||
|
|
need to be absolute or be in the cwd. If a string, the string is
|
||
|
|
put into a sequence and the searched. If a sequence, walk through
|
||
|
|
each element and join with ``filename``, calling :func:`expandvars`
|
||
|
|
and :func:`expanduser` before testing for existence.
|
||
|
|
|
||
|
|
Returns
|
||
|
|
-------
|
||
|
|
path : str
|
||
|
|
returns absolute path to file.
|
||
|
|
|
||
|
|
Raises
|
||
|
|
------
|
||
|
|
IOError
|
||
|
|
"""
|
||
|
|
|
||
|
|
# If paths are quoted, abspath gets confused, strip them...
|
||
|
|
filename = filename.strip('"').strip("'")
|
||
|
|
# If the input is an absolute path, just check it exists
|
||
|
|
if os.path.isabs(filename) and os.path.isfile(filename):
|
||
|
|
return filename
|
||
|
|
|
||
|
|
if path_dirs is None:
|
||
|
|
path_dirs = ("",)
|
||
|
|
elif isinstance(path_dirs, str):
|
||
|
|
path_dirs = (path_dirs,)
|
||
|
|
|
||
|
|
for path in path_dirs:
|
||
|
|
if path == '.': path = os.getcwd()
|
||
|
|
testname = expand_path(os.path.join(path, filename))
|
||
|
|
if os.path.isfile(testname):
|
||
|
|
return os.path.abspath(testname)
|
||
|
|
|
||
|
|
raise IOError("File %r does not exist in any of the search paths: %r" %
|
||
|
|
(filename, path_dirs) )
|
||
|
|
|
||
|
|
|
||
|
|
class HomeDirError(Exception):
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
def get_home_dir(require_writable=False) -> str:
|
||
|
|
"""Return the 'home' directory, as a unicode string.
|
||
|
|
|
||
|
|
Uses os.path.expanduser('~'), and checks for writability.
|
||
|
|
|
||
|
|
See stdlib docs for how this is determined.
|
||
|
|
For Python <3.8, $HOME is first priority on *ALL* platforms.
|
||
|
|
For Python >=3.8 on Windows, %HOME% is no longer considered.
|
||
|
|
|
||
|
|
Parameters
|
||
|
|
----------
|
||
|
|
require_writable : bool [default: False]
|
||
|
|
if True:
|
||
|
|
guarantees the return value is a writable directory, otherwise
|
||
|
|
raises HomeDirError
|
||
|
|
if False:
|
||
|
|
The path is resolved, but it is not guaranteed to exist or be writable.
|
||
|
|
"""
|
||
|
|
|
||
|
|
homedir = os.path.expanduser('~')
|
||
|
|
# Next line will make things work even when /home/ is a symlink to
|
||
|
|
# /usr/home as it is on FreeBSD, for example
|
||
|
|
homedir = os.path.realpath(homedir)
|
||
|
|
|
||
|
|
if not _writable_dir(homedir) and os.name == 'nt':
|
||
|
|
# expanduser failed, use the registry to get the 'My Documents' folder.
|
||
|
|
try:
|
||
|
|
import winreg as wreg
|
||
|
|
with wreg.OpenKey(
|
||
|
|
wreg.HKEY_CURRENT_USER,
|
||
|
|
r"Software\Microsoft\Windows\CurrentVersion\Explorer\Shell Folders"
|
||
|
|
) as key:
|
||
|
|
homedir = wreg.QueryValueEx(key,'Personal')[0]
|
||
|
|
except:
|
||
|
|
pass
|
||
|
|
|
||
|
|
if (not require_writable) or _writable_dir(homedir):
|
||
|
|
assert isinstance(homedir, str), "Homedir should be unicode not bytes"
|
||
|
|
return homedir
|
||
|
|
else:
|
||
|
|
raise HomeDirError('%s is not a writable dir, '
|
||
|
|
'set $HOME environment variable to override' % homedir)
|
||
|
|
|
||
|
|
def get_xdg_dir():
|
||
|
|
"""Return the XDG_CONFIG_HOME, if it is defined and exists, else None.
|
||
|
|
|
||
|
|
This is only for non-OS X posix (Linux,Unix,etc.) systems.
|
||
|
|
"""
|
||
|
|
|
||
|
|
env = os.environ
|
||
|
|
|
||
|
|
if os.name == "posix":
|
||
|
|
# Linux, Unix, AIX, etc.
|
||
|
|
# use ~/.config if empty OR not set
|
||
|
|
xdg = env.get("XDG_CONFIG_HOME", None) or os.path.join(get_home_dir(), '.config')
|
||
|
|
if xdg and _writable_dir(xdg):
|
||
|
|
assert isinstance(xdg, str)
|
||
|
|
return xdg
|
||
|
|
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def get_xdg_cache_dir():
|
||
|
|
"""Return the XDG_CACHE_HOME, if it is defined and exists, else None.
|
||
|
|
|
||
|
|
This is only for non-OS X posix (Linux,Unix,etc.) systems.
|
||
|
|
"""
|
||
|
|
|
||
|
|
env = os.environ
|
||
|
|
|
||
|
|
if os.name == "posix":
|
||
|
|
# Linux, Unix, AIX, etc.
|
||
|
|
# use ~/.cache if empty OR not set
|
||
|
|
xdg = env.get("XDG_CACHE_HOME", None) or os.path.join(get_home_dir(), '.cache')
|
||
|
|
if xdg and _writable_dir(xdg):
|
||
|
|
assert isinstance(xdg, str)
|
||
|
|
return xdg
|
||
|
|
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def expand_path(s):
|
||
|
|
"""Expand $VARS and ~names in a string, like a shell
|
||
|
|
|
||
|
|
:Examples:
|
||
|
|
|
||
|
|
In [2]: os.environ['FOO']='test'
|
||
|
|
|
||
|
|
In [3]: expand_path('variable FOO is $FOO')
|
||
|
|
Out[3]: 'variable FOO is test'
|
||
|
|
"""
|
||
|
|
# This is a pretty subtle hack. When expand user is given a UNC path
|
||
|
|
# on Windows (\\server\share$\%username%), os.path.expandvars, removes
|
||
|
|
# the $ to get (\\server\share\%username%). I think it considered $
|
||
|
|
# alone an empty var. But, we need the $ to remains there (it indicates
|
||
|
|
# a hidden share).
|
||
|
|
if os.name=='nt':
|
||
|
|
s = s.replace('$\\', 'IPYTHON_TEMP')
|
||
|
|
s = os.path.expandvars(os.path.expanduser(s))
|
||
|
|
if os.name=='nt':
|
||
|
|
s = s.replace('IPYTHON_TEMP', '$\\')
|
||
|
|
return s
|
||
|
|
|
||
|
|
|
||
|
|
def unescape_glob(string):
|
||
|
|
"""Unescape glob pattern in `string`."""
|
||
|
|
def unescape(s):
|
||
|
|
for pattern in '*[]!?':
|
||
|
|
s = s.replace(r'\{0}'.format(pattern), pattern)
|
||
|
|
return s
|
||
|
|
return '\\'.join(map(unescape, string.split('\\\\')))
|
||
|
|
|
||
|
|
|
||
|
|
def shellglob(args):
|
||
|
|
"""
|
||
|
|
Do glob expansion for each element in `args` and return a flattened list.
|
||
|
|
|
||
|
|
Unmatched glob pattern will remain as-is in the returned list.
|
||
|
|
|
||
|
|
"""
|
||
|
|
expanded = []
|
||
|
|
# Do not unescape backslash in Windows as it is interpreted as
|
||
|
|
# path separator:
|
||
|
|
unescape = unescape_glob if sys.platform != 'win32' else lambda x: x
|
||
|
|
for a in args:
|
||
|
|
expanded.extend(glob.glob(a) or [unescape(a)])
|
||
|
|
return expanded
|
||
|
|
|
||
|
|
|
||
|
|
def target_outdated(target,deps):
|
||
|
|
"""Determine whether a target is out of date.
|
||
|
|
|
||
|
|
target_outdated(target,deps) -> 1/0
|
||
|
|
|
||
|
|
deps: list of filenames which MUST exist.
|
||
|
|
target: single filename which may or may not exist.
|
||
|
|
|
||
|
|
If target doesn't exist or is older than any file listed in deps, return
|
||
|
|
true, otherwise return false.
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
target_time = os.path.getmtime(target)
|
||
|
|
except os.error:
|
||
|
|
return 1
|
||
|
|
for dep in deps:
|
||
|
|
dep_time = os.path.getmtime(dep)
|
||
|
|
if dep_time > target_time:
|
||
|
|
#print "For target",target,"Dep failed:",dep # dbg
|
||
|
|
#print "times (dep,tar):",dep_time,target_time # dbg
|
||
|
|
return 1
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
def target_update(target,deps,cmd):
|
||
|
|
"""Update a target with a given command given a list of dependencies.
|
||
|
|
|
||
|
|
target_update(target,deps,cmd) -> runs cmd if target is outdated.
|
||
|
|
|
||
|
|
This is just a wrapper around target_outdated() which calls the given
|
||
|
|
command if target is outdated."""
|
||
|
|
|
||
|
|
if target_outdated(target,deps):
|
||
|
|
system(cmd)
|
||
|
|
|
||
|
|
|
||
|
|
ENOLINK = 1998
|
||
|
|
|
||
|
|
def link(src, dst):
|
||
|
|
"""Hard links ``src`` to ``dst``, returning 0 or errno.
|
||
|
|
|
||
|
|
Note that the special errno ``ENOLINK`` will be returned if ``os.link`` isn't
|
||
|
|
supported by the operating system.
|
||
|
|
"""
|
||
|
|
|
||
|
|
if not hasattr(os, "link"):
|
||
|
|
return ENOLINK
|
||
|
|
link_errno = 0
|
||
|
|
try:
|
||
|
|
os.link(src, dst)
|
||
|
|
except OSError as e:
|
||
|
|
link_errno = e.errno
|
||
|
|
return link_errno
|
||
|
|
|
||
|
|
|
||
|
|
def link_or_copy(src, dst):
|
||
|
|
"""Attempts to hardlink ``src`` to ``dst``, copying if the link fails.
|
||
|
|
|
||
|
|
Attempts to maintain the semantics of ``shutil.copy``.
|
||
|
|
|
||
|
|
Because ``os.link`` does not overwrite files, a unique temporary file
|
||
|
|
will be used if the target already exists, then that file will be moved
|
||
|
|
into place.
|
||
|
|
"""
|
||
|
|
|
||
|
|
if os.path.isdir(dst):
|
||
|
|
dst = os.path.join(dst, os.path.basename(src))
|
||
|
|
|
||
|
|
link_errno = link(src, dst)
|
||
|
|
if link_errno == errno.EEXIST:
|
||
|
|
if os.stat(src).st_ino == os.stat(dst).st_ino:
|
||
|
|
# dst is already a hard link to the correct file, so we don't need
|
||
|
|
# to do anything else. If we try to link and rename the file
|
||
|
|
# anyway, we get duplicate files - see http://bugs.python.org/issue21876
|
||
|
|
return
|
||
|
|
|
||
|
|
new_dst = dst + "-temp-%04X" %(random.randint(1, 16**4), )
|
||
|
|
try:
|
||
|
|
link_or_copy(src, new_dst)
|
||
|
|
except:
|
||
|
|
try:
|
||
|
|
os.remove(new_dst)
|
||
|
|
except OSError:
|
||
|
|
pass
|
||
|
|
raise
|
||
|
|
os.rename(new_dst, dst)
|
||
|
|
elif link_errno != 0:
|
||
|
|
# Either link isn't supported, or the filesystem doesn't support
|
||
|
|
# linking, or 'src' and 'dst' are on different filesystems.
|
||
|
|
shutil.copy(src, dst)
|
||
|
|
|
||
|
|
def ensure_dir_exists(path, mode=0o755):
|
||
|
|
"""ensure that a directory exists
|
||
|
|
|
||
|
|
If it doesn't exist, try to create it and protect against a race condition
|
||
|
|
if another process is doing the same.
|
||
|
|
|
||
|
|
The default permissions are 755, which differ from os.makedirs default of 777.
|
||
|
|
"""
|
||
|
|
if not os.path.exists(path):
|
||
|
|
try:
|
||
|
|
os.makedirs(path, mode=mode)
|
||
|
|
except OSError as e:
|
||
|
|
if e.errno != errno.EEXIST:
|
||
|
|
raise
|
||
|
|
elif not os.path.isdir(path):
|
||
|
|
raise IOError("%r exists but is not a directory" % path)
|