From 58da6b63e47e00e67e62ad737563df70ee27a5e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?L=27=C3=A9lectron=20rare?= <108685187+electron-rare@users.noreply.github.com> Date: Sun, 21 Jun 2026 09:28:45 +0200 Subject: [PATCH] feat(tower): transfers journal begin mark end --- tower/lisael_content.py | 83 ++++++++++++++++++++++++++++++++++++++++- tower/test_transfers.py | 42 +++++++++++++++++++++ 2 files changed, 124 insertions(+), 1 deletion(-) create mode 100644 tower/test_transfers.py diff --git a/tower/lisael_content.py b/tower/lisael_content.py index 5c71ea1..703a0b9 100644 --- a/tower/lisael_content.py +++ b/tower/lisael_content.py @@ -2,7 +2,7 @@ """Shared content logic for the Lisael Box on Tower: feeds config, RSS, transcode, covers, staging/manifest ops, box state, push. Imported by lisael_refresh.py (cron) and lisael_server.py (web).""" -import hashlib, io, json, os, ssl, subprocess, sys, tempfile, threading, urllib.request +import hashlib, io, json, os, ssl, subprocess, sys, tempfile, threading, time, urllib.request import xml.etree.ElementTree as ET from PIL import Image @@ -433,3 +433,84 @@ def try_push(ip, delta): _busy.add(ip) threading.Thread(target=push_batch, args=(ip, delta), daemon=True).start() return True + +# ---- transfers journal ---- +TRANSFERS_JSON = os.environ.get("LISAEL_TRANSFERS", os.path.join(BASE, "transfers.json")) +TRANSFERS_MAX = 200 + +_active = None # active batch dict or None +_active_lock = threading.Lock() + +def _now_iso(): + return time.strftime("%Y-%m-%dT%H:%M:%S") + +def transfers_begin(box_ip, files): + """files: list of (name, subdir[, size]); size resolved via getsize if missing.""" + global _active + recs = [] + for it in files: + name, subdir = it[0], it[1] + size = it[2] if len(it) > 2 else None + if size is None: + src = STAGING_ROUTINES if subdir == "routines" else STAGING + try: + size = os.path.getsize(os.path.join(src, name)) + except OSError: + size = 0 + recs.append({"name": name, "dir": subdir, "size": size, "status": "pending", + "started": None, "ended": None, "duration_s": None, + "rate_bps": None, "error": None, "box_ip": box_ip, + "_t0": None}) + with _active_lock: + _active = {"box_ip": box_ip, "started": _now_iso(), + "total": len(recs), "current": 0, "files": recs} + +def transfers_mark(name, status, error=None): + with _active_lock: + if not _active: + return + for f in _active["files"]: + if f["name"] != name: + continue + if status == "sending": + f["status"] = "sending"; f["started"] = _now_iso(); f["_t0"] = time.monotonic() + _active["current"] += 1 + elif status in ("done", "failed"): + f["status"] = status; f["ended"] = _now_iso(); f["error"] = error + if f.get("_t0") is not None: + dur = max(time.monotonic() - f["_t0"], 0.001) + f["duration_s"] = round(dur, 3) + f["rate_bps"] = int((f["size"] or 0) / dur) + return + +def transfers_active(): + with _active_lock: + if not _active: + return None + # strip internal _t0 for the API view + files = [{k: v for k, v in f.items() if k != "_t0"} for f in _active["files"]] + return {**_active, "files": files} + +def _transfers_history_load(): + if os.path.exists(TRANSFERS_JSON): + try: + return json.load(open(TRANSFERS_JSON, encoding="utf-8")) + except Exception: + pass + return [] + +def transfers_history(limit=TRANSFERS_MAX): + return _transfers_history_load()[-limit:] + +def transfers_end(): + global _active + with _active_lock: + if not _active: + return + done = [{k: v for k, v in f.items() if k != "_t0"} + for f in _active["files"] if f["status"] in ("done", "failed")] + _active = None + hist = _transfers_history_load() + hist.extend(done) + hist = hist[-TRANSFERS_MAX:] + _atomic_write_json(TRANSFERS_JSON, hist) diff --git a/tower/test_transfers.py b/tower/test_transfers.py new file mode 100644 index 0000000..afd9252 --- /dev/null +++ b/tower/test_transfers.py @@ -0,0 +1,42 @@ +import os, tempfile, time, json +os.environ["LISAEL_BASE"] = tempfile.mkdtemp(prefix="transfers_test_") +os.environ["LISAEL_STAGING"] = os.path.join(os.environ["LISAEL_BASE"], "podcasts") +os.environ["LISAEL_STAGING_ROUTINES"] = os.path.join(os.environ["LISAEL_BASE"], "routines") +os.environ["LISAEL_TRANSFERS"] = os.path.join(os.environ["LISAEL_BASE"], "transfers.json") +import lisael_content as C + +def test_begin_mark_end_cycle(): + C.transfers_begin("192.168.0.250", [("a.mp3", "routines", 1000), + ("routines.json", "routines", 50)]) + a = C.transfers_active() + assert a["box_ip"] == "192.168.0.250" and a["total"] == 2 and a["current"] == 0 + assert all(f["status"] == "pending" for f in a["files"]) + C.transfers_mark("a.mp3", "sending") + assert C.transfers_active()["current"] == 1 + time.sleep(0.05) + C.transfers_mark("a.mp3", "done") + f = next(x for x in C.transfers_active()["files"] if x["name"] == "a.mp3") + assert f["status"] == "done" and f["duration_s"] > 0 and f["rate_bps"] > 0 + C.transfers_mark("routines.json", "sending") + C.transfers_mark("routines.json", "failed", "boom") + C.transfers_end() + assert C.transfers_active() is None + hist = C.transfers_history() + names = {h["name"]: h for h in hist} + assert names["a.mp3"]["status"] == "done" + assert names["routines.json"]["status"] == "failed" and names["routines.json"]["error"] == "boom" + +def test_history_truncates_200(): + for i in range(250): + C.transfers_begin("1.2.3.4", [(f"f{i}.mp3", "routines", 10)]) + C.transfers_mark(f"f{i}.mp3", "sending"); C.transfers_mark(f"f{i}.mp3", "done") + C.transfers_end() + assert len(C.transfers_history()) == 200 + # the most recent survive + assert any(h["name"] == "f249.mp3" for h in C.transfers_history()) + +if __name__ == "__main__": + for name, fn in sorted(globals().items()): + if name.startswith("test_"): + fn(); print("ok", name) + print("ALL OK")