Skip to content

File: Broken/Core/BrokenProject.py

Broken.Core.BrokenProject

BrokenProject

Source code in Broken/Core/BrokenProject.py
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
@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
        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((
                f"[bold dim]📦 Version {self.VERSION} • ",
                f"Python {sys.version.split()[0]} 📦[/]"
            )),
        ))

    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_unused(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_unused(version)

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

PACKAGE

PACKAGE: Path = field(converter=lambda x: Path(x).parent)

Send the importer's init.py's file variable

APP_NAME

APP_NAME: str

APP_AUTHOR

APP_AUTHOR: str

VERSION

VERSION: str = Runtime.Version

ABOUT

ABOUT: str = 'No description provided'

DIRECTORIES

DIRECTORIES: _Directories = None

RESOURCES

RESOURCES: _Resources = None

__attrs_post_init__

__attrs_post_init__()
Source code in Broken/Core/BrokenProject.py
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
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)

chdir

chdir() -> Self

Change directory to the project's root

Source code in Broken/Core/BrokenProject.py
88
89
90
def chdir(self) -> Self:
    """Change directory to the project's root"""
    return os.chdir(self.PACKAGE.parent.parent) or self

welcome

welcome() -> None
Source code in Broken/Core/BrokenProject.py
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def welcome(self) -> None:
    import pyfiglet
    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((
            f"[bold dim]📦 Version {self.VERSION} • ",
            f"Python {sys.version.split()[0]} 📦[/]"
        )),
    ))

uninstall

uninstall() -> None
Source code in Broken/Core/BrokenProject.py
237
238
def uninstall(self) -> None:
    ...

BrokenApp

Bases: ABC, BrokenAttrs

Source code in Broken/Core/BrokenProject.py
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
@define
class BrokenApp(ABC, BrokenAttrs):
    PROJECT: BrokenProject
    cli: BrokenTyper = Factory(BrokenTyper)

    def __post__(self):
        self.cli.should_shell()
        self.cli.description = self.PROJECT.ABOUT

        with BrokenProfiler(self.PROJECT.APP_NAME):
            self.main()

    @abstractmethod
    def main(self) -> None:
        pass

    def find_projects(self, tag: str="Project") -> None:
        """Find Python files in common directories (direct call, cwd) that any class inherits from
        something that contains the substring of `tag` and add as a command to this Typer app"""
        search = deque()

        # Note: Safe get argv[1], pop if valid, else a null path
        if (direct := Path(dict(enumerate(sys.argv)).get(1, "\0"))).exists():
            direct = Path(sys.argv.pop(1))

        # The project file was sent directly
        if (direct.suffix == ".py"):
            search.append(direct)

        # It can be a glob pattern
        elif ("*" in direct.name):
            search.extend(Path.cwd().glob(direct.name))

        # A directory of projects was sent
        elif direct.is_dir():
            search.extend(direct.glob("*.py"))

        # Scan common directories
        else:
            if (Runtime.Source):
                search.extend(self.PROJECT.DIRECTORIES.REPO_PROJECTS.rglob("*.py"))
                search.extend(self.PROJECT.DIRECTORIES.REPO_EXAMPLES.rglob("*.py"))
            search.extend(self.PROJECT.DIRECTORIES.PROJECTS.rglob("*.py"))
            search.extend(Path.cwd().glob("*.py"))

        # Add commands of all files, warn if none was sucessfully added
        if (sum(self.add_project(python=file, tag=tag) for file in search) == 0):
            log.warning(f"No {self.PROJECT.APP_NAME} {tag}s found, searched in:")
            log.warning('\n'.join(f"• {file}" for file in search))

    def _regex(self, tag: str) -> re.Pattern:
        """Generates the self.regex for matching any valid Python class that contains "tag" on the
        inheritance substring, and its optional docstring on the next line"""
        return re.compile(
            r"^class\s+(\w+)\s*\(.*?(?:" + tag + r").*\):\s*(?:\"\"\"((?:\n|.)*?)\"\"\")?",
            re.MULTILINE
        )

    def add_project(self, python: Path, tag: str="Project") -> bool:
        if (not python.exists()):
            return False

        def wrapper(file: Path, name: str, code: str):
            def run(ctx: typer.Context):
                # Note: Point of trust transfer to the file the user is running
                exec(compile(code, file, "exec"), (namespace := {}))
                namespace[name]().cli(*ctx.args)
            return run

        # Match all projects and their optional docstrings
        code = python.read_text("utf-8")
        matches = list(self._regex(tag).finditer(code))

        # Add a command for each match
        for match in matches:
            class_name, docstring = match.groups()
            self.cli.command(
                target=wrapper(python, class_name, code),
                name=class_name.lower(),
                description=(docstring or "No description provided"),
                panel=f"📦 {tag}s at ({python})",
                context=True,
                help=False,
            )

        return bool(matches)

PROJECT

PROJECT: BrokenProject

cli

cli: BrokenTyper = Factory(BrokenTyper)

__post__

__post__()
Source code in Broken/Core/BrokenProject.py
247
248
249
250
251
252
def __post__(self):
    self.cli.should_shell()
    self.cli.description = self.PROJECT.ABOUT

    with BrokenProfiler(self.PROJECT.APP_NAME):
        self.main()

main

main() -> None
Source code in Broken/Core/BrokenProject.py
254
255
256
@abstractmethod
def main(self) -> None:
    pass

find_projects

find_projects(tag: str = 'Project') -> None

Find Python files in common directories (direct call, cwd) that any class inherits from something that contains the substring of tag and add as a command to this Typer app

Source code in Broken/Core/BrokenProject.py
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def find_projects(self, tag: str="Project") -> None:
    """Find Python files in common directories (direct call, cwd) that any class inherits from
    something that contains the substring of `tag` and add as a command to this Typer app"""
    search = deque()

    # Note: Safe get argv[1], pop if valid, else a null path
    if (direct := Path(dict(enumerate(sys.argv)).get(1, "\0"))).exists():
        direct = Path(sys.argv.pop(1))

    # The project file was sent directly
    if (direct.suffix == ".py"):
        search.append(direct)

    # It can be a glob pattern
    elif ("*" in direct.name):
        search.extend(Path.cwd().glob(direct.name))

    # A directory of projects was sent
    elif direct.is_dir():
        search.extend(direct.glob("*.py"))

    # Scan common directories
    else:
        if (Runtime.Source):
            search.extend(self.PROJECT.DIRECTORIES.REPO_PROJECTS.rglob("*.py"))
            search.extend(self.PROJECT.DIRECTORIES.REPO_EXAMPLES.rglob("*.py"))
        search.extend(self.PROJECT.DIRECTORIES.PROJECTS.rglob("*.py"))
        search.extend(Path.cwd().glob("*.py"))

    # Add commands of all files, warn if none was sucessfully added
    if (sum(self.add_project(python=file, tag=tag) for file in search) == 0):
        log.warning(f"No {self.PROJECT.APP_NAME} {tag}s found, searched in:")
        log.warning('\n'.join(f"• {file}" for file in search))

add_project

add_project(python: Path, tag: str = 'Project') -> bool
Source code in Broken/Core/BrokenProject.py
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
def add_project(self, python: Path, tag: str="Project") -> bool:
    if (not python.exists()):
        return False

    def wrapper(file: Path, name: str, code: str):
        def run(ctx: typer.Context):
            # Note: Point of trust transfer to the file the user is running
            exec(compile(code, file, "exec"), (namespace := {}))
            namespace[name]().cli(*ctx.args)
        return run

    # Match all projects and their optional docstrings
    code = python.read_text("utf-8")
    matches = list(self._regex(tag).finditer(code))

    # Add a command for each match
    for match in matches:
        class_name, docstring = match.groups()
        self.cli.command(
            target=wrapper(python, class_name, code),
            name=class_name.lower(),
            description=(docstring or "No description provided"),
            panel=f"📦 {tag}s at ({python})",
            context=True,
            help=False,
        )

    return bool(matches)

mkdir

mkdir(path: Path, resolve: bool = True) -> Path

Make a directory and return it

Source code in Broken/Core/BrokenProject.py
331
332
333
334
335
336
337
def mkdir(path: Path, resolve: bool=True) -> Path:
    """Make a directory and return it"""
    path = Path(path).resolve() if resolve else Path(path)
    if not path.exists():
        log.info(f"Creating directory: {path}")
        path.mkdir(parents=True, exist_ok=True)
    return path