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:
...
|