File: Broken/Core/BrokenProject.py¶

@define(slots=False)
class BrokenProject:
    PACKAGE: Path = field(converter=lambda x: Path(x).parent)
    """Send the importer's __init__.py's __file__ variable"""

    # App information
    APP_NAME: str
    APP_AUTHOR: str
    VERSION: str = Runtime.Version
    ABOUT: str = "No description provided"

    # Standard Broken objects for a project
    DIRECTORIES: _Directories = None
    RESOURCES: _Resources = None

    def __attrs_post_init__(self):
        self.DIRECTORIES = _Directories(PROJECT=self)
        self.RESOURCES = _Resources(PROJECT=self)
        BrokenLogging.set_project(self.APP_NAME)

        # Print version information on "--version/-V"
        if (list_get(sys.argv, 1) in ("--version", "-V")):
            print(f"{self.APP_NAME} {self.VERSION} {BrokenPlatform.Host.value}")
            sys.exit(0)

        # Replace Broken.PROJECT with the first initialized project
        if (project := getattr(Broken, "PROJECT", None)):
            if (project is Broken.BROKEN):
                if (BrokenPlatform.Root and not Runtime.Docker):
                    log.warning("Running as [bold blink red]Administrator or Root[/] is discouraged unless necessary!")
                self._pyapp_management()
                Broken.PROJECT = self

        # Convenience symlink the project's workspace
        if Runtime.Source and Environment.flag("WORKSPACE_SYMLINK", 0):
            BrokenPath.symlink(
                virtual=self.DIRECTORIES.REPOSITORY/"Workspace",
                real=self.DIRECTORIES.WORKSPACE, echo=False
            )

        # Load dotenv files in common directories
        for path in self.DIRECTORIES.REPOSITORY.glob("*.env"):
            dotenv.load_dotenv(path, override=True)

    def chdir(self) -> Self:
        """Change directory to the project's root"""
        return os.chdir(self.PACKAGE.parent.parent) or self

    def welcome(self) -> None:
        import pyfiglet # noqa
        from Broken import BrokenTorch
        torch = BrokenTorch.version()
        ascii = pyfiglet.figlet_format(self.APP_NAME)
        ascii = '\n'.join((x for x in ascii.split('\n') if x.strip()))
        rprint(Panel(
            Align.center(ascii + "\n"),
            subtitle=''.join((
                "[bold dim]📦 "
                f"Version {self.VERSION} ",
                f"• Python {sys.version.split()[0]} ",
                f"• Torch {torch.value} " if torch else "",
                "📦[/]",
            )),
        ))

    def _pyapp_management(self) -> None:

        # Skip if not executing within a binary release
        if not (executable := Environment.get("PYAPP")):
            return None

        # ---------------------------------------------------------------------------------------- #

        import hashlib
        venv_path = Path(Environment.get("VIRTUAL_ENV"))
        hash_file = (venv_path/"version.sha256")
        this_hash = hashlib.sha256(open(executable, "rb").read()).hexdigest()
        old_hash  = (hash_file.read_text() if hash_file.exists() else None)
        hash_file.write_text(this_hash)

        # Fixme (#ntfs): https://superuser.com/questions/488127
        # Fixme (#ntfs): https://unix.stackexchange.com/questions/49299
        ntfs_workaround = venv_path.with_name("0.0.0")

        # "If (not on the first run) and (hash differs)"
        if (old_hash is not None) and (old_hash != this_hash):
            print("-"*shutil.get_terminal_size().columns + "\n")
            log.info(f"Detected different hash for this release version [bold blue]v{self.VERSION}[/], reinstalling..")
            log.info(f"• {venv_path}")

            if BrokenPlatform.OnWindows:
                BrokenPath.remove(ntfs_workaround)
                venv_path.rename(ntfs_workaround)
                try:
                    rprint("\n[bold orange3 blink](Warning)[/] Please, reopen this executable to continue! Press Enter to exit..", end='')
                    input()
                except KeyboardInterrupt:
                    pass
                exit(0)
            else:
                shell(executable, "self", "restore", stdout=subprocess.DEVNULL)
                print("\n" + "-"*shutil.get_terminal_size().columns + "\n")
                try:
                    sys.exit(shell(executable, sys.argv[1:], echo=False).returncode)
                except KeyboardInterrupt:
                    exit(0)

        # Note: Remove before unused version checking
        BrokenPath.remove(ntfs_workaround, echo=False)

        # ---------------------------------------------------------------------------------------- #

        if (not arguments()):
            self.welcome()

        def check_new_version():
            from packaging.version import Version

            # Skip development binaries, as they aren't on PyPI
            if (current := Version(self.VERSION)).is_prerelease:
                return None

            with BrokenCache.requests(
                cache_name=(venv_path/"version.check"),
                expire_after=(3600),
            ) as requests:
                import json

                with contextlib.suppress(Exception):
                    _api   = f"https://pypi.org/pypi/{self.APP_NAME.lower()}/json"
                    latest = Version(json.loads(requests.get(_api).text)["info"]["version"])

                # Newer version available
                if (current < latest):
                    log.minor((
                        f"A newer version of the project [bold blue]v{latest}[/] is available! "
                        f"Get it at https://brokensrc.dev/get/releases/ (Current: v{current})"
                    ))

                # Back to the future!
                elif (current > latest):
                    log.error(f"[bold indian_red]For whatever reason, the current version [bold blue]v{self.VERSION}[/] is newer than the latest [bold blue]v{latest}[/][/]")
                    log.error("[bold indian_red]• This is fine if you're running a development or pre-release version, don't worry;[/]")
                    log.error("[bold indian_red]• Otherwise, it was likely recalled for whatever reason, consider downgrading![/]")

        # Warn: Must not interrupt user if actions are being taken (argv)
        if Environment.flag("VERSION_CHECK", 1) and (not arguments()):
            with contextlib.suppress(Exception):
                check_new_version()

        # ---------------------------------------------------------------------------------------- #

        def manage_install(version: Path):
            tracker = FileTracker(version/"version.tracker")
            tracker.retention.days = 7

            # Running a new version, prune previous cache
            if (tracker.first):
                shell(sys.executable, "-m", "uv", "cache", "prune", "--quiet", echo=False)

            # Skip in-use versions
            if (not tracker.trigger()):
                return None

            # Late-update current tracker
            if (version == venv_path):
                return tracker.update()

            from rich.prompt import Prompt

            log.warning((
                f"The version [bold green]v{version.name}[/] of the projects "
                f"hasn't been used for {tracker.sleeping}, unninstall it to save space!"
                f"\n[bold bright_black]• Files at: {version}[/]"
            ))

            try:
                answer = Prompt.ask(
                    prompt="\n:: Choose an action:",
                    choices=("keep", "delete"),
                    default="delete",
                )
                print()
                if (answer == "delete"):
                    with Halo(f"Deleting unused version v{version.name}.."):
                        shutil.rmtree(version, ignore_errors=True)
                if (answer == "keep"):
                    log.minor("Keeping the version for now, will check again later!")
                    return tracker.update()
            except KeyboardInterrupt:
                exit(0)

        # Note: Avoid interactive prompts if running with arguments
        if Environment.flag("UNUSED_CHECK", 1) and (not arguments()):
            for version in (x for x in venv_path.parent.glob("*") if x.is_dir()):
                with contextlib.suppress(Exception):
                    manage_install(version)

    def uninstall(self) -> None:
        ...