Base¶

class Ignore:
    """A class that does nothing. No-operation faster Mock"""
    def __nop__(self, *args, **kwargs) -> Self:
        return self
    def __call__(self, *args, **kwargs) -> Self:
        return self
    def __getattr__(self, _):
        return self.__nop__

@define
class SameTracker:
    """Doumo same desu"""
    value: Any = None

    def __call__(self, value: Any) -> bool:
        """
        If a value is the same, returns True, else updates it and returns False
        • Useful on ignoring expensive calls where parameters doesn't changes
        """
        if self.value != value:
            self.value = value
            return False
        return True

class BrokenAttrs:
    """
    Walk over an @attrs.defined class and call __post__ on all classes in the MRO.
    # Warn: Must NOT define __attrs_post_init__ in an inheriting class
    # Fixme: Can improve by starting on BrokenAttrs itself
    """
    def __attrs_post_init__(self):
        for cls in reversed(type(self).mro()):
            if method := cls.__dict__.get("__post__"):
                method(self)

    def __post__(self) -> None:
        pass

class BrokenPlatform:
    """
    Host Platform information, Cross Compilation targets and some utilities
    """

    # Name of the current platform - (linux, windows, macos, bsd)
    Name:         str  = platform.system().lower().replace("darwin", "macos")

    # Booleans if the current platform is the following
    OnLinux:      bool = (Name == "linux")
    OnWindows:    bool = (Name == "windows")
    OnMacOS:      bool = (Name == "macos")
    OnBSD:        bool = (Name == "bsd")

    # Platform release binaries extension and CPU architecture
    Extension:    str  = (".exe" if OnWindows else ".bin")
    Architecture: str  = (platform.machine().lower().replace("x86_64", "amd64"))

    # Family of platforms
    OnUnix:       bool = (OnLinux or OnMacOS or OnBSD)

    # Distro IDs: https://distro.readthedocs.io/en/latest/
    LinuxDistro:  str  = distro.id()

    # # Booleans if the current platform is the following

    # Ubuntu-like
    OnUbuntu:     bool = (LinuxDistro == "ubuntu")
    OnDebian:     bool = (LinuxDistro == "debian")
    OnMint:       bool = (LinuxDistro == "linuxmint")
    OnRaspberry:  bool = (LinuxDistro == "raspbian")
    OnUbuntuLike: bool = (OnUbuntu or OnDebian or OnMint or OnRaspberry)

    # Arch-like
    OnArch:       bool = (LinuxDistro == "arch")
    OnManjaro:    bool = (LinuxDistro == "manjaro")
    OnArchLike:   bool = (OnArch or OnManjaro)

    # RedHat-like
    OnFedora:     bool = (LinuxDistro == "fedora")
    OnCentOS:     bool = (LinuxDistro == "centos")
    OnRedHat:     bool = (LinuxDistro == "rhel")
    OnRedHatLike: bool = (OnFedora or OnCentOS or OnRedHat)

    # Others
    OnGentoo:     bool = (LinuxDistro == "gentoo")

    # BSD - I have never used it
    OnOpenBSD:    bool = (LinuxDistro == "openbsd")
    OnNetBSD:     bool = (LinuxDistro == "netbsd")
    OnBSDLike:    bool = (OnOpenBSD or OnNetBSD)

    class Targets(BrokenEnum):
        """List of common platforms targets for releases"""
        LinuxAMD64:   str = "linux-amd64"
        LinuxARM:     str = "linux-arm64"
        WindowsAMD64: str = "windows-amd64"
        WindowsARM:   str = "windows-arm64"
        MacosAMD64:   str = "macos-amd64"
        MacosARM:     str = "macos-arm64"

        @property
        def rust(self) -> str:
            return {
                self.LinuxAMD64:   "x86_64-unknown-linux-gnu",
                self.LinuxARM:     "aarch64-unknown-linux-gnu",
                self.WindowsAMD64: "x86_64-pc-windows-gnu",
                self.WindowsARM:   "aarch64-pc-windows-gnu",
                self.MacosAMD64:   "x86_64-apple-darwin",
                self.MacosARM:     "aarch64-apple-darwin",
            }[self]

        @property
        def extension(self) -> str:
            """Same as BrokenPlatform.Extension"""
            return ".exe" if ("windows" in self.value) else ".bin"

        @property
        def name(self) -> str:
            """Same as BrokenPlatform.Name"""
            return self.value.split("-")[0]

        @property
        def architecture(self) -> str:
            """Same as BrokenPlatform.Architecture"""
            return self.value.split("-")[1]

    CurrentTarget: str = f"{Name}-{Architecture}"

    @staticmethod
    def clear_terminal(**kwargs):
        if not kwargs.get("do", True):
            return

        if BrokenPlatform.OnWindows:
            os.system("cls")
        else:
            shell("clear", **kwargs)

    @staticmethod
    def log_system_info():
        log.info(f"• System Info: {platform.system()} {platform.release()}, Python {platform.python_version()} {platform.machine()}")

    # Simply, why Windows/Python have different directory names for scripts? ...
    # https://github.com/pypa/virtualenv/commit/993ba1316a83b760370f5a3872b3f5ef4dd904c1
    PyScripts         = ("Scripts" if OnWindows else "bin")
    PyScriptExtension = (".cmd" if OnWindows else "")

class ShutilFormat(BrokenEnum):
    Zip   = "zip"
    Tar   = "tar"
    GzTar = "tar.gz"
    BzTar = "tar.bz2"
    XzTar = "tar.xz"

class BrokenPath(Path):
    """
    BrokenPath is not supposed to be instantiated as a class but used as a namespace (static class)
    • The __new__ method returns the value of .get method, which is pathlib.Path
    • Many Path utilities as staticmethod and contextmanager
    """
    def __new__(cls, *paths: Path, valid: bool=False) -> Optional[Path]:
        return cls.get(*paths, valid=valid)

    @staticmethod
    def get(*paths: Path, valid: bool=False) -> Optional[Path] | List[Optional[Path]]:
        """Resolve paths to absolute None-safe, optionally verify their existence, None if invalid"""
        paths = map(lambda path: Path(path).expanduser().resolve() if path else None, paths)
        paths = map(lambda path: (path if (path and path.exists()) else None) if valid else path, paths)
        paths = list(paths)
        return paths[0] if (len(paths) == 1) else paths

    @staticmethod
    def copy(src: Path, dst: Path, *, echo=True) -> Path:
        src, dst = BrokenPath(src, dst)
        BrokenPath.mkdir(dst.parent, echo=False)
        if src.is_dir():
            log.info(f"Copying Directory ({src})\n → ({dst})", echo=echo)
            shutil.copytree(src, dst)
        else:
            log.info(f"Copying File ({src})\n → ({dst})", echo=echo)
            shutil.copy2(src, dst)
        return dst

    @staticmethod
    def move(src: Path, dst: Path, *, echo=True) -> Path:
        src, dst = BrokenPath(src, dst)
        log.info(f"Moving ({src})\n → ({dst})", echo=echo)
        shutil.move(src, dst)
        return dst

    @staticmethod
    def remove(path: Path, *, confirm=False, echo=True) -> Path:

        # Already removed or doesn't exist
        if not (path := BrokenPath(path, valid=True)):
            return path

        log.info(f"Removing Path ({path})", echo=echo)

        # Safety: Must not be common
        if path in (Path.cwd(), Path.home()):
            log.error(f"Avoided catastrophic failure by not removing ({path})")
            exit(1)

        # Symlinks are safe to remove
        if path.is_symlink():
            path.unlink()
            return path

        # Confirm removal: directory contains data
        if confirm and (not click.confirm(f"• Confirm removing path ({path})")):
            return path

        # Remove the path
        if path.is_dir():
            shutil.rmtree(path, ignore_errors=True)
        else:
            path.unlink()

        return path

    @staticmethod
    def touch(path: Path, *, echo=True) -> Path:
        """Creates a file, fail safe™"""
        if (path := BrokenPath(path)).exists():
            log.success(f"File ({path}) already touched", echo=echo)
            return
        log.info(f"Touching file {path}", echo=echo)
        path.touch()
        return path

    @staticmethod
    def mkdir(path: Path, parent: bool=False, *, echo=True) -> Path:
        """Creates a directory and its parents, fail safe™"""
        path = BrokenPath(path)
        path = path.parent if parent else path
        if path.exists():
            log.success(f"Directory ({path}) already exists", echo=echo)
            return path
        log.info(f"Creating directory {path}", echo=echo)
        path.mkdir(parents=True, exist_ok=True)
        return path

    @staticmethod
    def resetdir(path: Path, *, echo=True) -> Path:
        """Creates a directory and its parents, fail safe™"""
        BrokenPath.remove(path, echo=echo)
        return BrokenPath.mkdir(path, echo=echo)

    @contextlib.contextmanager
    def pushd(path: Path, *, echo: bool=True) -> Generator[Path, None, None]:
        """Change directory, then change back when done"""
        path = BrokenPath(path)
        cwd = os.getcwd()
        log.info(f"Pushd ({path})", echo=echo)
        os.chdir(path)
        yield path
        log.info(f"Popd  ({path})", echo=echo)
        os.chdir(cwd)

    @staticmethod
    def symlink(virtual: Path, real: Path, *, echo: bool=True) -> Path:
        """
        Symlink [virtual] -> [real], `virtual` being the symlink file and `real` the target

        Args:
            virtual (Path): Symlink path (file)
            real (Path): Target path (real path)

        Returns:
            None if it fails, else `virtual` Path
        """
        log.info(f"Symlinking ({virtual})\n → ({real})", echo=echo)

        # Return if already symlinked
        if (BrokenPath(virtual) == BrokenPath(real)):
            return virtual

        # Make Virtual's parent directory
        BrokenPath.mkdir(virtual.parent, echo=False)

        # Remove old symlink if it points to a non existing directory
        if virtual.is_symlink() and (not virtual.resolve().exists()):
            virtual.unlink()

        # Virtual doesn't exist, ok to create
        elif not virtual.exists():
            pass

        # File exists and is a symlink - safe to remove
        elif virtual.is_symlink():
            virtual.unlink()

        # Virtual is a directory and not empty
        elif virtual.is_dir() and (not os.listdir(virtual)):
            BrokenPath.remove(virtual, echo=False)

        else:
            if click.confirm(f"• Path ({virtual}) exists, but Broken wants to create a symlink to ({real})\nConfirm removing the 'virtual' path and continuing? (It might contain data or be a important symlink)"):
                BrokenPath.remove(virtual, echo=False)
            else:
                return

        # Actually symlink
        virtual.symlink_to(real)
        return virtual

    @staticmethod
    def make_executable(path: Path, *, echo=False) -> Path:
        """Make a file executable"""
        path = BrokenPath(path)
        if BrokenPlatform.OnUnix:
            shell("chmod", "+x", path, echo=echo)
        elif BrokenPlatform.OnWindows:
            shell("attrib", "+x", path, echo=echo)
        return path

    @staticmethod
    def zip(path: Path, output: Path=None, *, format: ShutilFormat=ShutilFormat.Zip, echo: bool=True) -> Path:
        format = ShutilFormat(format)
        output = BrokenPath(output or path).with_suffix(f".{format}")
        path   = BrokenPath(path)
        BrokenPath.remove(output, echo=echo)
        log.info(f"Zipping ({path})\n → ({output})", echo=echo)
        shutil.make_archive(output.with_suffix(""), format, path)
        return output

    @staticmethod
    def stem(path: Path) -> str:
        """
        Get the "true stem" of a path, as pathlib's only gets the last dot one
        • "/path/with/many.ext.ens.ions" -> "many" instead of "many.ext.ens"
        """
        stem = Path(Path(path).stem)
        while (stem := Path(stem).with_suffix("")).suffix:
            continue
        return str(stem)

    @staticmethod
    def sha256sum(data: Union[Path, str, bytes]) -> Optional[str]:
        """Get the sha256sum of a file, directory or bytes"""

        # Nibble the bytes !
        if isinstance(data, bytes):
            return hashlib.sha256(data).hexdigest()

        # String or Path is a valid path
        elif (path := BrokenPath(data, valid=True)):
            with BrokenSpinner(log.info(f"Calculating sha256sum of ({path})")):
                if path.is_file():
                    return hashlib.sha256(path.read_bytes()).hexdigest()

                # Iterate on all files for low memory footprint
                feed = hashlib.sha256()
                for file in path.rglob("*"):
                    if not file.is_file():
                        continue
                    with open(file, "rb") as file:
                        while (chunk := file.read(8192)):
                            feed.update(chunk)
                return feed.hexdigest()

        elif isinstance(data, str):
            return hashlib.sha256(data.encode("utf-8")).hexdigest()

        return

    @staticmethod
    def extract(
        path: Path,
        output: Path=None,
        *,
        overwrite: bool=False,
        PATH: bool=False,
        echo: bool=True
    ) -> Path:
        path, output = BrokenPath(path, output)

        # Output is input without suffix if not given
        if (output is None):
            output = path.parent/BrokenPath.stem(path)

        # Add stem to the output as some archives might be flat
        output /= BrokenPath.stem(path)

        # Re-extract on order
        Maybe(BrokenPath.remove, overwrite, echo=echo)

        # A file to skip if it exists, created after successful extraction
        if (extract_flag := (output/"BrokenPath.extract.ok")).exists():
            log.minor(f"Already extracted ({output})", echo=echo)
            Maybe(BrokenPath.add_to_path, PATH, path=output, recursively=True, echo=echo)
            return output

        # Show progress as this might take a while on slower IOs
        log.info(f"Extracting ({path})\n → ({output})", echo=echo)
        with BrokenSpinner("Extracting archive.."):
            shutil.unpack_archive(path, output)

        extract_flag.touch()
        Maybe(BrokenPath.add_to_path, PATH, path=output, recursively=True, echo=echo)
        return output/BrokenPath.stem(path)

    @staticmethod
    def download(
        url: str,
        output: Path=None,
        *,
        size_check: bool=True,
        chunk: int=1024,
        echo: bool=True
    ) -> Optional[Path]:

        # Link must be valid
        if not validators.url(url):
            log.error(f"The string ({url}) doesn't look like a valid URL", echo=echo)
            return

        import requests

        # Default to Broken's Download directory
        if (output is None):
            output = Broken.BROKEN.DIRECTORIES.DOWNLOADS

        # Append url's file name to the output path
        if (output := BrokenPath(output)).is_dir():
            output /= Path(url.split("#")[0].split("?")[0].split("/")[-1])

        # Without size check, the existence of the file is enough
        if output.exists() and (not size_check):
            log.info(f"Already Downloaded ({output})", echo=echo)
            log.minor("• Size check was skipped, the file might be incomplete", echo=echo)
            return

        # Send the GET request, we might be offline!
        try:
            response = requests.get(url, stream=True)
        except requests.exceptions.RequestException as error:
            log.error(f"Failed to download file ({url}) → ({output}): {error}", echo=echo)
            return output

        size = int(response.headers.get('content-length', 0))

        # The file might already be (partially) downloaded
        if output.exists():
            A, B = (output.stat().st_size, size)
            if (A == B):
                log.info(f"Already Downloaded ({output})", echo=echo)
                return output
            else:
                log.warning(f"Wrong Download at ({output})", echo=echo)

        log.info(f"Downloading file at ({url}):", echo=echo)
        log.info(f"• Output: ({output})", echo=echo)

        # It is binary prefix, right? kibi, mebi, gibi, etc. as we're dealing with raw bytes
        with open(output, "wb") as file, tqdm.tqdm(
            desc=f"Downloading ({output.name})",
            total=size, unit="iB", unit_scale=True, unit_divisor=1024,
            mininterval=1/30, maxinterval=0.5, leave=False
        ) as progress:
            for data in response.iter_content(chunk_size=chunk):
                progress.update(file.write(data))

        # Url was invalid
        if (response.status_code != 200):
            log.error(f"Failed to Download File at ({url}):", echo=echo)
            log.error(f"• HTTP Error: {response.status_code}", echo=echo)
            return

        # Wrong downloaded and expected size
        elif (output.stat().st_size != size):
            log.error(f"File ({output}) was not downloaded correctly", echo=echo)
            return

        log.success(f"Downloaded file ({output}) from ({url})", echo=echo)
        return output

    @staticmethod
    def get_external(url: str, *, subdir: str="", echo: bool=True) -> Path:
        file = BrokenPath.download(denum(url), echo=echo)

        # File is a Archive, extract
        if any((str(file).endswith(ext) for ext in ShutilFormat.values)):
            directory = Broken.BROKEN.DIRECTORIES.EXTERNAL_ARCHIVES
            return BrokenPath.extract(file, directory, PATH=True, echo=echo)

        # File is some known type, move to their own external directory
        if bool(subdir):
            directory = Broken.BROKEN.DIRECTORIES.EXTERNALS/subdir
        elif file.suffix in AUDIO_EXTENSIONS:
            directory = Broken.BROKEN.DIRECTORIES.EXTERNAL_AUDIO
        elif file.suffix in IMAGE_EXTENSIONS:
            directory = Broken.BROKEN.DIRECTORIES.EXTERNAL_IMAGES
        elif file.suffix in FONTS_EXTENSIONS:
            directory = Broken.BROKEN.DIRECTORIES.EXTERNAL_FONTS
        elif file.suffix in SOUNDFONTS_EXTENSIONS:
            directory = Broken.BROKEN.DIRECTORIES.EXTERNAL_SOUNDFONTS
        elif file.suffix in MIDI_EXTENSIONS:
            directory = Broken.BROKEN.DIRECTORIES.EXTERNAL_MIDIS
        else:
            directory = Broken.BROKEN.DIRECTORIES.EXTERNALS
        return BrokenPath.move(file, directory/subdir, echo=echo)

    @staticmethod
    @easy_multiple
    def which(*name: str, echo=True) -> Optional[Path]:
        BrokenPath.update_externals_path()
        return apply(shutil.which, name)

    @staticmethod
    def update_externals_path(path: Path=None, *, echo: bool=True) -> Optional[Path]:
        path = path or Broken.BROKEN.DIRECTORIES.EXTERNALS
        return BrokenPath.add_to_path(path, recursively=True, echo=echo)

    @staticmethod
    def on_path(path: Path) -> bool:
        """Check if a path is on PATH, works with symlinks"""
        return BrokenPath(path) in map(BrokenPath, os.environ.get("PATH", "").split(os.pathsep))

    @staticmethod
    def add_to_path(
        path: Path,
        *,
        recursively: bool=False,
        persistent: bool=False,
        preferential: bool=True,
        echo: bool=True
    ) -> Path:
        """
        Add a path, recursively or not, to System's Path or this Python process's Path

        Args:
            recursively: Also add all subdirectories of the given path
            persistent: Use 'userpath' package to add to the Shell's or Registry PATH
            preferential: Prepends the path for less priority on system binaries

        Returns:
            The Path argument itself
        """
        path = BrokenPath(path)

        # Can't recurse on file or non existing directories
        if (not path.exists()) and path.is_file() and recursively:
            log.warning(f"Can't add non existing path or file recursively to Path ({path})", echo=echo)
            return path

        log.debug(f"Adding to Path (Recursively: {recursively}, Persistent: {persistent}): ({path})", echo=echo)

        for other in (path.rglob("*") if recursively else [path]):
            if other.is_file():
                continue
            if BrokenPath.on_path(other):
                continue
            if persistent:
                import userpath
                userpath.append(str(other))
            else:
                if preferential:
                    os.environ["PATH"] = (str(other)+os.pathsep+os.environ["PATH"])
                    sys.path.insert(0, str(other))
                else:
                    os.environ["PATH"] = (os.environ["PATH"]+os.pathsep+str(other))
                    sys.path.append(str(other))
        return path

    # # Specific / "Utils"

    @staticmethod
    def open_in_file_explorer(path: Path):
        """Opens a path in the file explorer"""
        path = BrokenPath(path)
        if BrokenPlatform.OnWindows:
            os.startfile(str(path))
        elif BrokenPlatform.OnLinux:
            shell("xdg-open", path)
        elif BrokenPlatform.OnMacOS:
            shell("open", path)

    # Fixme: Untested functions, needs better name; are these useful?

    @staticmethod
    def non_empty_file(path: Path) -> bool:
        return path.exists() and path.is_file() and path.stat().st_size > 0

    @staticmethod
    def empty_file(path: Path, create: bool=True) -> bool:
        if create and not path.exists():
            path.parent.mkdir(parents=True, exist_ok=True)
            path.touch()
        return path.exists() and path.is_file() and len(path.read_text()) == 0

    @contextlib.contextmanager
    def PATH(*,
        directories: List[Path],
        recursive: bool=True,
        prepend: bool=True,
        clean: bool=False,
        restore: bool=True,
    ):
        """
        Temporarily limits the PATH to given directories
        - directories: List of directories to add to PATH
        - recursive: Whether to add subdirectories of given directories to PATH
        - prepend: Prioritize binaries found in input directories
        - restricted: Do not include current PATH in the new PATH
        """

        # Make Path objects
        directories = apply(Path, flatten(directories))

        # Get current PATH
        old = os.environ["PATH"]

        # List of all directories in PATH
        PATH = [] if clean else os.environ["PATH"].split(os.pathsep)

        # Add directories to PATH
        for directory in directories:
            PATH.append(directory)

            # Do not recurse if so
            if not recursive:
                continue

            # WARN: This could be slow on too many directories (wrong input?)
            # Find all subdirectories of a path
            for path in directory.rglob("*"):
                if path.is_dir():
                    if prepend:
                        PATH.insert(0, path)
                    else:
                        PATH.append(path)

        # Set new PATH
        os.environ["PATH"] = os.pathsep.join(map(str, PATH))

        yield os.environ["PATH"]

        # Restore PATH
        os.environ["PATH"] = old