252 lines
7.1 KiB
Python
252 lines
7.1 KiB
Python
import os
|
|
import datetime
|
|
from pathlib import Path
|
|
import shutil
|
|
import sys
|
|
|
|
|
|
class CWD:
|
|
def __init__(self, push_path, pop_path=None):
|
|
self.push_path = push_path
|
|
self.pop_path = pop_path or os.getcwd()
|
|
|
|
def __enter__(self):
|
|
if self.push_path is not None:
|
|
os.chdir(self.push_path)
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
if self.pop_path is not None:
|
|
os.chdir(self.pop_path)
|
|
|
|
|
|
def _workspace_root():
|
|
return Path(os.path.realpath(__file__)).parent.parent.parent.parent.absolute()
|
|
|
|
|
|
def _default_build():
|
|
return _workspace_root().joinpath('build')
|
|
|
|
|
|
class SparkPaths:
|
|
@staticmethod
|
|
def root():
|
|
return _workspace_root()
|
|
|
|
@staticmethod
|
|
def client_csproj(name):
|
|
return _workspace_root().joinpath('src', 'client', 'name', f'{name}.csproj')
|
|
|
|
@staticmethod
|
|
def client_bins(build_root=_default_build(), configuration='Release'):
|
|
return build_root.joinpath(configuration, 'bin', 'net8.0-windows')
|
|
|
|
@staticmethod
|
|
def client_sln():
|
|
return _workspace_root().joinpath('spark_client.sln')
|
|
|
|
@staticmethod
|
|
def tools():
|
|
return _workspace_root().joinpath('tools')
|
|
|
|
@staticmethod
|
|
def default_build():
|
|
return _default_build()
|
|
|
|
@staticmethod
|
|
def externals():
|
|
return _workspace_root().joinpath('externals')
|
|
|
|
|
|
class PathOrNone:
|
|
def __init__(self, path):
|
|
if path is None:
|
|
return
|
|
if isinstance(path, str):
|
|
self.path = Path(path)
|
|
elif isinstance(path, Path):
|
|
self.path = path
|
|
else:
|
|
self.path = Path(str(path))
|
|
|
|
def value(self):
|
|
if self.path is None:
|
|
raise ValueError('path is None')
|
|
return self.path
|
|
|
|
def __str__(self):
|
|
return '' if self.path is None else str(self.path)
|
|
|
|
|
|
def as_path(path):
|
|
return PathOrNone(
|
|
path if isinstance(path, Path) else
|
|
Path(str(path)) if isinstance(path, str) else
|
|
None)
|
|
|
|
|
|
def as_absolute_paths(relative_paths: list, with_respect_to = None):
|
|
if relative_paths is None:
|
|
raise ValueError('relative_path is None')
|
|
if len(relative_paths) == 0:
|
|
return []
|
|
|
|
def impl():
|
|
return [as_path(p).value().absolute() for p in relative_paths]
|
|
|
|
if with_respect_to is None:
|
|
return impl()
|
|
with CWD(with_respect_to):
|
|
return impl()
|
|
|
|
|
|
def ensure_exists(path):
|
|
if not os.path.exists(path):
|
|
raise ValueError(f"Path '{path}' does not exist")
|
|
|
|
|
|
def fmt_datetime_suffix(dt: datetime.datetime = None):
|
|
return (dt or datetime.datetime.now()).strftime("%Y-%m%d-%H%M")
|
|
|
|
|
|
def append_datetime_suffix(path, dt: datetime.datetime = None):
|
|
return f'{path}_{fmt_datetime_suffix(dt)}'
|
|
|
|
|
|
def get_unique_path(path: str, max_tries: int = 100):
|
|
if path is None or len(path) == 0:
|
|
raise ValueError("'path' is none or empty")
|
|
|
|
if not os.path.exists(path):
|
|
return path
|
|
|
|
max_tries = 100
|
|
i = 1
|
|
orig_path = path
|
|
while i < max_tries:
|
|
path = f'{orig_path}({i})'
|
|
if not os.path.exists(path):
|
|
break
|
|
i += 1
|
|
if i == max_tries:
|
|
raise ValueError(
|
|
f"Unable to make path unique by appending appending an incrementing integer. "
|
|
f"Maximum attempts exceeded ({max_tries})")
|
|
return path
|
|
|
|
|
|
def unlink(path):
|
|
if not os.path.exists(path):
|
|
return
|
|
if not os.path.islink(path):
|
|
raise ValueError(
|
|
'The specified path already exists and is not a symbolic link')
|
|
os.unlink(path)
|
|
|
|
|
|
def rmdir_or_unlink(dir):
|
|
if os.path.islink(dir):
|
|
os.unlink(dir)
|
|
elif os.path.exists(dir):
|
|
shutil.rmtree(dir)
|
|
|
|
|
|
def link_dir(source_path, link_path):
|
|
if source_path is None:
|
|
raise TypeError('source_path is None')
|
|
if not os.path.isdir(source_path):
|
|
raise ValueError('source_path is not a directory')
|
|
if link_path is None:
|
|
raise TypeError('link_path is None')
|
|
if source_path == link_path:
|
|
return
|
|
|
|
if not os.path.exists(source_path):
|
|
os.makedirs(source_path)
|
|
if os.path.exists(link_path):
|
|
if os.path.islink(link_path):
|
|
if os.readlink(link_path) == source_path:
|
|
return
|
|
os.unlink(link_path)
|
|
elif os.path.isdir(link_path):
|
|
shutil.rmtree(link_path)
|
|
elif os.path.isfile(link_path):
|
|
os.remove(link_path)
|
|
else:
|
|
raise ValueError('link_path is an unexpected filetype')
|
|
os.symlink(source_path, link_path)
|
|
|
|
|
|
def link_file(source_path, link_path, mk_link_dirs = True, overwrite = False):
|
|
'''
|
|
Creates a symlink
|
|
:param source_path: The path to which the link points
|
|
:param link_path: The path to the link
|
|
:param mk_source_dirs: Whether or not to create missing directories in the source_path
|
|
:param mk_link_dirs: Whether or not to create missing directories in the link_path
|
|
'''
|
|
|
|
# Convert to Path
|
|
source_path = PathOrNone(source_path).value()
|
|
link_path = PathOrNone(link_path).value()
|
|
if source_path == link_path:
|
|
raise ValueError(f'source_path and link_path are the same: {source_path}')
|
|
|
|
if link_path.exists() and link_path.is_dir():
|
|
link_path = link_path.joinpath(source_path.name)
|
|
if link_path.is_symlink() and link_path.readlink() == source_path:
|
|
return
|
|
|
|
# Check 'if exists OR is_symlink' so it evalutates to True for broken
|
|
# symlinks (i.e., a symlink that exists, but points to a non-existant path)
|
|
if link_path.exists() or link_path.is_symlink():
|
|
if not link_path.is_dir():
|
|
if not overwrite:
|
|
raise ValueError(f"link_path already exists: '{link_path}'")
|
|
if link_path.is_symlink():
|
|
os.unlink(link_path)
|
|
elif link_path.is_file():
|
|
os.remove(link_path)
|
|
else:
|
|
raise ValueError(f"link_path is an unexpected filetype: '{link_path}'")
|
|
elif mk_link_dirs:
|
|
# When the link path doesn't exist, assume it represents a file, and
|
|
# create the missing dirs. Pass 'link_path.parent' to makedirs, so it
|
|
# doesn't create the end node as a directory.
|
|
if len(link_path.parents) > 0 and not link_path.parent.exists():
|
|
os.makedirs(link_path.parent)
|
|
os.symlink(source_path, link_path)
|
|
|
|
|
|
def create_dir(path, overwrite_if_exists=False):
|
|
if path is None:
|
|
raise TypeError('path is None')
|
|
|
|
if not os.path.exists(path):
|
|
os.makedirs(path)
|
|
elif overwrite_if_exists:
|
|
rmdir_or_unlink(path)
|
|
os.makedirs(path)
|
|
|
|
|
|
def copy(src, dest, verbose=True):
|
|
if verbose:
|
|
print(f"copy file '{src}' -> '{dest}'")
|
|
|
|
|
|
def find_files(src: Path, suffix: str):
|
|
dest = []
|
|
|
|
def add_files(src: Path):
|
|
if src.is_file():
|
|
return [src] if src.name.endswith(suffix) else []
|
|
if src.is_dir():
|
|
for c in src.iterdir():
|
|
if c.is_file():
|
|
if c.name.endswith(suffix):
|
|
dest.append(c)
|
|
elif c.is_dir():
|
|
add_files(c)
|
|
|
|
add_files(src)
|
|
return dest
|