feat: add --dev flag to watch and auto-reload sfx archive changes

when --dev is passed, starts a background thread that monitors the sfx archive file for modifications and automatically re-unpacks when changes are detected.
Skips checksum validation during repack to allow for content updates.
This commit is contained in:
grifinas 2025-11-22 16:55:02 +02:00
parent 77f74ddb2f
commit 546aca71b2
2 changed files with 52 additions and 8 deletions

View file

@ -397,6 +397,12 @@ build the sfx using any of the following examples:
./scripts/make-sfx.sh gz no-cm # gzip-compressed + no fancy markdown editor ./scripts/make-sfx.sh gz no-cm # gzip-compressed + no fancy markdown editor
``` ```
from there, you can run with:
```sh
./dist/copyparty-sfx.py # regular way, probably what the user will do
./dist/copyparty-sfx.py --dev # with a file watcher for auto-reloading JS/HTML changes
```
## build from release tarball ## build from release tarball

View file

@ -33,8 +33,28 @@ PY37 = sys.version_info > (3, 7)
WINDOWS = sys.platform in ["win32", "msys"] WINDOWS = sys.platform in ["win32", "msys"]
sys.dont_write_bytecode = True sys.dont_write_bytecode = True
me = os.path.abspath(os.path.realpath(__file__)) me = os.path.abspath(os.path.realpath(__file__))
_watcher_started = False
unpackLocation = None
def _watch_me():
msg("watcher started")
lastUnpackTimestamp = os.path.getmtime(me)
while True:
try:
mt = os.path.getmtime(me)
if mt > lastUnpackTimestamp:
msg("archive changed, re-unpacking")
try:
unpack(True)
except Exception:
traceback.print_exc()
lastUnpackTimestamp = mt
except Exception:
traceback.print_exc()
time.sleep(1.0)
def eprint(*a, **ka): def eprint(*a, **ka):
ka["file"] = sys.stderr ka["file"] = sys.stderr
print(*a, **ka) print(*a, **ka)
@ -248,8 +268,9 @@ def hashfile(fn):
return h.hexdigest()[:24] return h.hexdigest()[:24]
def unpack(): def unpack(repack=False):
"""unpacks the tar yielded by `data`""" """unpacks the tar yielded by `data`"""
global unpackLocation
name = "pe-copyparty" name = "pe-copyparty"
try: try:
name += "." + str(os.geteuid()) name += "." + str(os.geteuid())
@ -268,26 +289,33 @@ def unpack():
if not ofe(mine): if not ofe(mine):
break break
if repack:
mine = unpackLocation
else:
unpackLocation = mine
os.mkdir(mine)
tar = opj(mine, "tar") tar = opj(mine, "tar")
try: try:
if tag in os.listdir(final) and ofe(san): if repack == False and tag in os.listdir(final) and ofe(san):
msg("found early") msg("found early")
return final return final
except: except:
pass pass
sz = 0 sz = 0
os.mkdir(mine)
with open(tar, "wb") as f: with open(tar, "wb") as f:
for buf in get_payload(): for buf in get_payload():
sz += len(buf) sz += len(buf)
f.write(buf) f.write(buf)
ck = hashfile(tar) # If we are repacking with a listener hashfiles will probably not match
if ck != CKSUM: if repack == False:
t = "\n\nexpected %s (%d byte)\nobtained %s (%d byte)\nsfx corrupt" ck = hashfile(tar)
raise Exception(t % (CKSUM, SIZE, ck, sz)) if ck != CKSUM:
t = "\n\nexpected %s (%d byte)\nobtained %s (%d byte)\nsfx corrupt"
raise Exception(t % (CKSUM, SIZE, ck, sz))
with tarfile.open(tar, "r:bz2") as tf: with tarfile.open(tar, "r:bz2") as tf:
# this is safe against traversal # this is safe against traversal
@ -307,7 +335,7 @@ def unpack():
f.write(b"h\n") f.write(b"h\n")
try: try:
if tag in os.listdir(final) and ofe(san): if repack == False and tag in os.listdir(final) and ofe(san):
msg("found late") msg("found late")
return final return final
except: except:
@ -339,6 +367,7 @@ def unpack():
except: except:
msg("reloc fail,", mine) msg("reloc fail,", mine)
msg("unpacked to", mine)
return mine return mine
@ -392,6 +421,15 @@ def run(tmp, j2, ftp):
ld = (("", ""), (j2, "j2"), (ftp, "ftp"), (not PY2, "py2"), (PY37, "py37")) ld = (("", ""), (j2, "j2"), (ftp, "ftp"), (not PY2, "py2"), (PY37, "py37"))
ld = [os.path.join(tmp, b) for a, b in ld if not a] ld = [os.path.join(tmp, b) for a, b in ld if not a]
global _watcher_started
if any(arg == "--dev" for arg in sys.argv) and not _watcher_started:
sys.argv.remove('--dev')
import threading
_watcher_started = True
t = threading.Thread(target=_watch_me)
t.daemon = True
t.start()
# skip 1 # skip 1
# enable this to dynamically remove type hints at startup, # enable this to dynamically remove type hints at startup,
# in case a future python version can use them for performance # in case a future python version can use them for performance