L'électron rare
2fffcfe9a3
Make the end-of-routine "Bravo !" lively: the star now spins
(continuous full turn via lv_image_set_rotation) and bounces
(translate_y ping-pong). Both use layer-free paths — image
rotation transforms against the flush band, translate is not a
layer-forcing style — so the render task stays light (no repeat
of the full-screen-layer task_wdt). Auto-return stretched to
3.6 s so the spin shows and the festive cue finishes.
Tower: seed a more festive bravo.mp3 line ("Youpi, tu as
réussi ! Tu es super !").
582 lines
23 KiB
Python
582 lines
23 KiB
Python
#!/usr/bin/env python3
|
||
"""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, time, urllib.request
|
||
import xml.etree.ElementTree as ET
|
||
from PIL import Image
|
||
|
||
BASE = os.environ.get("LISAEL_BASE", "/home/clems/lisael-content")
|
||
STAGING = os.environ.get("LISAEL_STAGING", os.path.join(BASE, "podcasts"))
|
||
STAGING_ROUTINES = os.environ.get("LISAEL_STAGING_ROUTINES", os.path.join(BASE, "routines"))
|
||
LVGLIMG = os.environ.get("LISAEL_LVGLIMG", os.path.join(BASE, "LVGLImage.py"))
|
||
FEEDS_JSON = os.path.join(BASE, "feeds.json")
|
||
ROUTINES_JSON = os.path.join(BASE, "routines.json")
|
||
BOX_JSON = os.path.join(BASE, "box-state.json")
|
||
BOX_PORT = int(os.environ.get("LISAEL_BOX_PORT", "8080"))
|
||
TTS_URL = os.environ.get("LISAEL_TTS_URL", "http://100.78.6.122:9300/v1/audio/speech")
|
||
TTS_VOICE = os.environ.get("LISAEL_TTS_VOICE", "fr")
|
||
TWEMOJI = os.environ.get("LISAEL_TWEMOJI", "https://raw.githubusercontent.com/jdecked/twemoji/main/assets/72x72")
|
||
UA = {"User-Agent": "Mozilla/5.0"}
|
||
NS = {"itunes": "http://www.itunes.com/dtds/podcast-1.0.dtd"}
|
||
SUBS = {"’": "'", "“": '"', "”": '"', "–": "-",
|
||
"—": "-", "…": "...", "«": '"', "»": '"'}
|
||
|
||
DEFAULT_FEEDS = [
|
||
{"key": "oli", "name": "Une histoire et Oli", "rss": "https://radiofrance-podcast.net/podcast09/rss_19721.xml", "n": 3},
|
||
{"key": "odyssees", "name": "Les Odyssees", "rss": "https://radiofrance-podcast.net/podcast09/rss_20108.xml", "n": 3},
|
||
{"key": "salutinfo", "name": "Salut l'info", "rss": "https://radiofrance-podcast.net/podcast09/rss_20689.xml", "n": 2},
|
||
{"key": "bestioles", "name": "Bestioles", "rss": "https://radiofrance-podcast.net/podcast09/35099478-7c72-4f9e-a6de-1b928400e9e5/podcast_a80ecbd5-df3d-4c9d-bee7-4e3d9efc1974.xml", "n": 2},
|
||
]
|
||
|
||
def fetch(u, t=180):
|
||
return urllib.request.urlopen(urllib.request.Request(u, headers=UA), timeout=t,
|
||
context=ssl.create_default_context()).read()
|
||
|
||
def clean(s):
|
||
for a, b in SUBS.items():
|
||
s = s.replace(a, b)
|
||
return " ".join(s.split()).strip()
|
||
|
||
def _atomic_write_bytes(path, data):
|
||
d = os.path.dirname(path) or "."
|
||
os.makedirs(d, exist_ok=True)
|
||
fd, tmp = tempfile.mkstemp(dir=d)
|
||
with os.fdopen(fd, "wb") as f:
|
||
f.write(data)
|
||
os.replace(tmp, path)
|
||
|
||
def _atomic_write_json(path, obj):
|
||
_atomic_write_bytes(path, json.dumps(obj, ensure_ascii=False, indent=2).encode("utf-8"))
|
||
|
||
# ---- feeds config ----
|
||
def load_feeds():
|
||
if os.path.exists(FEEDS_JSON):
|
||
try:
|
||
return json.load(open(FEEDS_JSON, encoding="utf-8"))
|
||
except Exception:
|
||
pass
|
||
save_feeds(DEFAULT_FEEDS)
|
||
return [dict(f) for f in DEFAULT_FEEDS]
|
||
|
||
def save_feeds(feeds):
|
||
_atomic_write_json(FEEDS_JSON, feeds)
|
||
|
||
def add_feed(key, name, rss, n=2):
|
||
feeds = load_feeds()
|
||
feeds = [f for f in feeds if f["key"] != key]
|
||
feeds.append({"key": key, "name": name, "rss": rss, "n": int(n)})
|
||
save_feeds(feeds)
|
||
return feeds
|
||
|
||
def remove_feed(key):
|
||
save_feeds([f for f in load_feeds() if f["key"] != key])
|
||
|
||
# ---- routines (truth) ----
|
||
DEFAULT_ROUTINES = [
|
||
{"id": "matin", "title": "Le matin", "emoji": "☀️", "steps": [
|
||
{"text": "Habille-toi", "emoji": "\U0001F455"},
|
||
{"text": "Petit-déjeuner", "emoji": "\U0001F950"},
|
||
{"text": "Brosse tes dents", "emoji": "\U0001FAA5"},
|
||
{"text": "Prépare ton cartable", "emoji": "\U0001F392"}]},
|
||
{"id": "soir", "title": "Le soir", "emoji": "\U0001F319", "steps": [
|
||
{"text": "Mets ton pyjama", "emoji": "\U0001F476"},
|
||
{"text": "Brosse tes dents", "emoji": "\U0001FAA5"},
|
||
{"text": "Lis une histoire", "emoji": "\U0001F4D6"}]},
|
||
]
|
||
|
||
def routines_load():
|
||
if os.path.exists(ROUTINES_JSON):
|
||
try:
|
||
return json.load(open(ROUTINES_JSON, encoding="utf-8")).get("routines", [])
|
||
except Exception:
|
||
pass
|
||
routines_save([dict(r) for r in DEFAULT_ROUTINES])
|
||
return [dict(r) for r in DEFAULT_ROUTINES]
|
||
|
||
def routines_save(routines):
|
||
_atomic_write_json(ROUTINES_JSON, {"routines": routines})
|
||
|
||
# ---- routine asset manifest ----
|
||
ROUTINES_CACHE = lambda: os.path.join(STAGING_ROUTINES, ".cache.json")
|
||
|
||
def asset_key(rid, idx):
|
||
return f"{rid}_{idx:02d}"
|
||
|
||
def _routine_cache_load():
|
||
p = ROUTINES_CACHE()
|
||
if os.path.exists(p):
|
||
try:
|
||
return json.load(open(p, encoding="utf-8"))
|
||
except Exception:
|
||
pass
|
||
return {}
|
||
|
||
def build_routine_manifest(gen_bin=None, gen_audio=None):
|
||
"""Regenerate per-step .bin (emoji) + .mp3 (TTS) into STAGING_ROUTINES,
|
||
skipping assets whose (emoji|text) is unchanged, then write the SD manifest.
|
||
Also seeds ok.mp3 and bravo.mp3 (fixed firmware cues) if missing.
|
||
gen_bin(emoji, key)->bin_name ; gen_audio(text, key)->mp3_name (injectable).
|
||
Fixed sounds are only generated on the real path (when gen_audio is not
|
||
overridden), so injected-fake test runs stay offline."""
|
||
_real_audio = gen_audio is None # True => real network path
|
||
gen_bin = gen_bin or emoji_bin
|
||
gen_audio = gen_audio or _gen_step_audio
|
||
os.makedirs(STAGING_ROUTINES, exist_ok=True)
|
||
cache = _routine_cache_load()
|
||
new_cache = {}
|
||
generated = skipped = 0
|
||
manifest = []
|
||
for r in routines_load():
|
||
rid = r["id"]
|
||
# routine icon (emoji -> .bin), key = rid
|
||
icon_key = rid
|
||
c = cache.get(icon_key)
|
||
if c and c.get("emoji") == r.get("emoji") and os.path.exists(
|
||
os.path.join(STAGING_ROUTINES, icon_key + ".bin")):
|
||
icon = icon_key + ".bin"; skipped += 1
|
||
else:
|
||
icon = gen_bin(r.get("emoji", ""), icon_key) or (icon_key + ".bin"); generated += 1
|
||
new_cache[icon_key] = {"emoji": r.get("emoji")}
|
||
steps = []
|
||
for i, st in enumerate(r.get("steps", []), start=1):
|
||
k = asset_key(rid, i)
|
||
prev = cache.get(k) or {}
|
||
bin_name = k + ".bin"; mp3_name = k + ".mp3"
|
||
# picto
|
||
if prev.get("emoji") == st.get("emoji") and os.path.exists(
|
||
os.path.join(STAGING_ROUTINES, bin_name)):
|
||
skipped += 1
|
||
else:
|
||
gen_bin(st.get("emoji", ""), k); generated += 1
|
||
# audio
|
||
if prev.get("text") == st.get("text") and os.path.exists(
|
||
os.path.join(STAGING_ROUTINES, mp3_name)):
|
||
skipped += 1
|
||
else:
|
||
gen_audio(st.get("text", ""), k); generated += 1
|
||
new_cache[k] = {"emoji": st.get("emoji"), "text": st.get("text")}
|
||
steps.append({"text": st.get("text", ""), "icon": bin_name, "audio": mp3_name})
|
||
manifest.append({"id": rid, "title": r.get("title", rid),
|
||
"icon": icon, "steps": steps})
|
||
_atomic_write_json(os.path.join(STAGING_ROUTINES, "routines.json"),
|
||
{"routines": manifest})
|
||
_atomic_write_json(ROUTINES_CACHE(), new_cache)
|
||
# Seed fixed firmware sound cues (only on the real TTS path, not in test fakes)
|
||
if _real_audio:
|
||
_seed_fixed_sound("ok.mp3", "Et hop !")
|
||
_seed_fixed_sound("bravo.mp3", "Bravo ! Youpi, tu as réussi ! Tu es super !")
|
||
return {"generated": generated, "skipped": skipped, "routines": manifest}
|
||
|
||
def _gen_step_audio(text, key):
|
||
"""Default audio generator: TTS -> mono mp3 in STAGING_ROUTINES. '' on failure."""
|
||
try:
|
||
au = tts_speak(text)
|
||
_atomic_write_bytes(os.path.join(STAGING_ROUTINES, key + ".mp3"), au)
|
||
return key + ".mp3"
|
||
except Exception as e:
|
||
print("step audio fail", key, e, flush=True)
|
||
return ""
|
||
|
||
def _seed_fixed_sound(name, text):
|
||
"""Write a fixed firmware cue (ok.mp3 / bravo.mp3) into STAGING_ROUTINES if
|
||
absent or empty. Non-fatal: logs and returns on TTS failure."""
|
||
p = os.path.join(STAGING_ROUTINES, name)
|
||
if os.path.exists(p) and os.path.getsize(p) > 0:
|
||
return
|
||
try:
|
||
_atomic_write_bytes(p, tts_speak(text))
|
||
except Exception as e:
|
||
print("fixed sound fail", name, e, flush=True)
|
||
|
||
# ---- RSS ----
|
||
def cover_url(root):
|
||
el = root.find(".//channel/itunes:image", NS)
|
||
if el is not None and el.get("href"):
|
||
return el.get("href")
|
||
el = root.find(".//channel/image/url")
|
||
return el.text if el is not None else None
|
||
|
||
def fetch_episodes(rss, limit=None):
|
||
"""Return ([{url,title,duration,guid}], cover_href)."""
|
||
root = ET.fromstring(fetch(rss))
|
||
items = root.findall(".//item")
|
||
if limit:
|
||
items = items[:limit]
|
||
out = []
|
||
for it in items:
|
||
enc = it.find("enclosure")
|
||
if enc is None or not enc.get("url"):
|
||
continue
|
||
out.append({
|
||
"url": enc.get("url"),
|
||
"title": clean(it.findtext("title") or "Episode"),
|
||
"duration": (it.findtext("itunes:duration", namespaces=NS) or "").strip(),
|
||
"guid": (it.findtext("guid") or enc.get("url")),
|
||
})
|
||
return out, cover_url(root)
|
||
|
||
# ---- transcode / cover ----
|
||
def episode_fname(key, url):
|
||
return f"{key}_{hashlib.sha1(url.encode()).hexdigest()[:8]}_a48.mp3"
|
||
|
||
def transcode(raw_path, out_path):
|
||
rc = subprocess.run(["ffmpeg", "-y", "-loglevel", "error", "-i", raw_path,
|
||
"-ac", "1", "-b:a", "48k", "-c:a", "libmp3lame", out_path],
|
||
capture_output=True).returncode
|
||
return rc == 0 and os.path.exists(out_path)
|
||
|
||
def tts_speak(text):
|
||
"""Generate a mono mp3 announcement via the tailnet gateway. Returns bytes.
|
||
Raises on network/HTTP failure (caller decides degraded behaviour)."""
|
||
body = json.dumps({"model": "tts-1", "input": text, "voice": TTS_VOICE,
|
||
"response_format": "mp3"}).encode()
|
||
req = urllib.request.Request(TTS_URL, data=body,
|
||
headers={"Content-Type": "application/json"})
|
||
with urllib.request.urlopen(req, timeout=30) as r:
|
||
return r.read()
|
||
|
||
def emoji_cp(emoji):
|
||
"""Twemoji filename codepoints, VS16 (U+FE0F) stripped or twemoji 404s."""
|
||
return "-".join(f"{ord(c):x}" for c in emoji if c != "️")
|
||
|
||
def emoji_bin(emoji, key):
|
||
"""Download the Twemoji for `emoji`, resize 160x160 RGBA, write `<key>.bin`
|
||
(RGB565A8, keeps alpha) into STAGING_ROUTINES. Returns '<key>.bin' or ''."""
|
||
try:
|
||
url = f"{TWEMOJI}/{emoji_cp(emoji)}.png"
|
||
raw = urllib.request.urlopen(
|
||
urllib.request.Request(url, headers={"User-Agent": "curl/8"}),
|
||
timeout=15, context=ssl.create_default_context()).read()
|
||
im = Image.open(io.BytesIO(raw)).convert("RGBA").resize((160, 160), Image.LANCZOS)
|
||
os.makedirs(STAGING_ROUTINES, exist_ok=True)
|
||
with tempfile.TemporaryDirectory() as td:
|
||
png = os.path.join(td, f"{key}.png"); im.save(png)
|
||
subprocess.run([sys.executable, LVGLIMG, "--ofmt", "BIN", "--cf", "RGB565A8",
|
||
"-o", td, png], check=True, capture_output=True)
|
||
binp = os.path.join(td, f"{key}.bin")
|
||
if os.path.exists(binp):
|
||
os.replace(binp, os.path.join(STAGING_ROUTINES, f"{key}.bin"))
|
||
return f"{key}.bin"
|
||
except Exception as e:
|
||
print("emoji_bin fail", key, emoji, e, flush=True)
|
||
return ""
|
||
|
||
def make_cover(img_bytes, key):
|
||
"""160x160 RGB565 .bin into STAGING via LVGLImage. Returns '<key>.bin' or ''."""
|
||
try:
|
||
im = Image.open(io.BytesIO(img_bytes)).convert("RGB").resize((160, 160))
|
||
os.makedirs(STAGING, exist_ok=True)
|
||
with tempfile.TemporaryDirectory() as td:
|
||
png = os.path.join(td, f"{key}.png"); im.save(png)
|
||
subprocess.run([sys.executable, LVGLIMG, "--ofmt", "BIN", "--cf", "RGB565",
|
||
"-o", td, png], check=True, capture_output=True)
|
||
binp = os.path.join(td, f"{key}.bin")
|
||
if os.path.exists(binp):
|
||
os.replace(binp, os.path.join(STAGING, f"{key}.bin"))
|
||
return f"{key}.bin"
|
||
except Exception as e:
|
||
print("cover fail", key, e, flush=True)
|
||
return ""
|
||
|
||
# ---- staging / manifest ----
|
||
def staging_files():
|
||
if not os.path.isdir(STAGING):
|
||
return []
|
||
return sorted(f for f in os.listdir(STAGING) if not f.startswith("."))
|
||
|
||
def manifest_load():
|
||
p = os.path.join(STAGING, "manifest.json")
|
||
if os.path.exists(p):
|
||
try:
|
||
return json.load(open(p, encoding="utf-8"))
|
||
except Exception:
|
||
pass
|
||
return []
|
||
|
||
def manifest_save(m):
|
||
_atomic_write_json(os.path.join(STAGING, "manifest.json"), m)
|
||
|
||
def _manifest_upsert(m, show, cover, file, title, src="auto"):
|
||
# src: "auto" (weekly refresh), "pick" (hand-selected), "upload" (custom file).
|
||
# Only "auto" episodes are eligible for pruning by the cap.
|
||
for s in m:
|
||
if s["show"] == show:
|
||
if cover and not s.get("cover"):
|
||
s["cover"] = cover
|
||
for e in s["episodes"]:
|
||
if e["file"] == file:
|
||
e["title"] = title
|
||
if src != "auto": # a manual pick/upload wins over auto
|
||
e["src"] = src
|
||
return
|
||
s["episodes"].append({"file": file, "title": title, "src": src})
|
||
return
|
||
m.append({"show": show, "cover": cover or "",
|
||
"episodes": [{"file": file, "title": title, "src": src}]})
|
||
|
||
def add_episode(key, show, cover_bytes, url, title, src="auto"):
|
||
"""Download + transcode an episode into STAGING + upsert manifest. Idempotent by fname.
|
||
cover_bytes: raw image bytes for the show cover, or None. src: auto|pick."""
|
||
os.makedirs(STAGING, exist_ok=True)
|
||
fname = episode_fname(key, url)
|
||
out = os.path.join(STAGING, fname)
|
||
if not os.path.exists(out):
|
||
with tempfile.TemporaryDirectory() as td:
|
||
raw = os.path.join(td, "raw"); open(raw, "wb").write(fetch(url))
|
||
tmp_out = os.path.join(td, fname)
|
||
if not transcode(raw, tmp_out):
|
||
raise RuntimeError(f"transcode failed {fname}")
|
||
os.replace(tmp_out, out)
|
||
cover = make_cover(cover_bytes, key) if cover_bytes else ""
|
||
m = manifest_load()
|
||
_manifest_upsert(m, show, cover, fname, clean(title), src)
|
||
manifest_save(m)
|
||
return fname
|
||
|
||
def add_upload(show, src_bytes, title, key="perso"):
|
||
os.makedirs(STAGING, exist_ok=True)
|
||
h = hashlib.sha1(src_bytes[:65536] + str(len(src_bytes)).encode()).hexdigest()[:8]
|
||
fname = f"{key}_{h}_a48.mp3"
|
||
out = os.path.join(STAGING, fname)
|
||
with tempfile.TemporaryDirectory() as td:
|
||
raw = os.path.join(td, "raw"); open(raw, "wb").write(src_bytes)
|
||
tmp_out = os.path.join(td, fname)
|
||
if not transcode(raw, tmp_out):
|
||
raise RuntimeError("transcode failed")
|
||
os.replace(tmp_out, out)
|
||
m = manifest_load()
|
||
_manifest_upsert(m, show, "", fname, clean(title), "upload")
|
||
manifest_save(m)
|
||
return fname
|
||
|
||
def prune_show(show, keep_files):
|
||
"""Cap a feed show: drop its 'auto' episodes whose file is not in keep_files
|
||
(the current latest-N from the feed). 'pick'/'upload' episodes are always kept.
|
||
Deletes the staged file too. Returns the number removed."""
|
||
m = manifest_load()
|
||
removed = 0
|
||
for s in m:
|
||
if s["show"] != show:
|
||
continue
|
||
kept = []
|
||
for e in s["episodes"]:
|
||
if e.get("src", "auto") != "auto" or e["file"] in keep_files:
|
||
kept.append(e)
|
||
else:
|
||
p = os.path.join(STAGING, e["file"])
|
||
if os.path.exists(p):
|
||
os.remove(p)
|
||
removed += 1
|
||
s["episodes"] = kept
|
||
m = [s for s in m if s["episodes"]]
|
||
manifest_save(m)
|
||
return removed
|
||
|
||
def remove_file(name):
|
||
if "/" in name or ".." in name or name.startswith("."):
|
||
raise ValueError("bad name")
|
||
p = os.path.join(STAGING, name)
|
||
if os.path.exists(p):
|
||
os.remove(p)
|
||
m = manifest_load()
|
||
for s in m:
|
||
s["episodes"] = [e for e in s["episodes"] if e["file"] != name]
|
||
m = [s for s in m if s["episodes"]]
|
||
manifest_save(m)
|
||
|
||
# ---- box state + push ----
|
||
def box_state_load():
|
||
if os.path.exists(BOX_JSON):
|
||
try:
|
||
return json.load(open(BOX_JSON, encoding="utf-8"))
|
||
except Exception:
|
||
pass
|
||
return {}
|
||
|
||
def box_state_save(st):
|
||
_atomic_write_json(BOX_JSON, st)
|
||
|
||
def compute_delta(staging, have):
|
||
haveset = set(have)
|
||
files = [f for f in staging if f != "manifest.json" and f not in haveset]
|
||
if files and "manifest.json" in staging:
|
||
files.append("manifest.json")
|
||
return files
|
||
|
||
def compute_delta_named(staging, have, manifest_name):
|
||
haveset = set(have)
|
||
files = [f for f in staging if f != manifest_name and f not in haveset]
|
||
if files and manifest_name in staging:
|
||
files.append(manifest_name)
|
||
return files
|
||
|
||
def routines_staging_files():
|
||
if not os.path.isdir(STAGING_ROUTINES):
|
||
return []
|
||
return sorted(f for f in os.listdir(STAGING_ROUTINES)
|
||
if not f.startswith(".") and f != ".cache.json")
|
||
|
||
def compute_routines_delta(have):
|
||
return compute_delta_named(routines_staging_files(), have, "routines.json")
|
||
|
||
def push_file(ip, name, subdir="podcasts"):
|
||
src_dir = STAGING_ROUTINES if subdir == "routines" else STAGING
|
||
data = open(os.path.join(src_dir, name), "rb").read()
|
||
url = f"http://{ip}:{BOX_PORT}/put?name={subdir}/{name}"
|
||
with urllib.request.urlopen(urllib.request.Request(url, data=data, method="POST"),
|
||
timeout=300) as r:
|
||
return r.status
|
||
|
||
_busy = set()
|
||
_busy_lock = threading.Lock()
|
||
|
||
# ---- transfers journal lock ----
|
||
_history_lock = threading.Lock()
|
||
|
||
def push_batch(ip, delta, subdir="podcasts"):
|
||
transfers_begin(ip, [(f, subdir) for f in delta])
|
||
ok = 0
|
||
try:
|
||
for f in delta:
|
||
transfers_mark(f, "sending", dir=subdir)
|
||
try:
|
||
push_file(ip, f, subdir=subdir); ok += 1
|
||
transfers_mark(f, "done", dir=subdir)
|
||
except Exception as e:
|
||
transfers_mark(f, "failed", str(e)[:200], dir=subdir)
|
||
print("push fail", f, e, flush=True)
|
||
finally:
|
||
transfers_end()
|
||
print(f"push done {ip} {subdir} {ok}/{len(delta)}", flush=True)
|
||
with _busy_lock:
|
||
_busy.discard(ip)
|
||
|
||
def try_push(ip, delta, subdir="podcasts"):
|
||
"""Start a background push unless one already runs for ip. Returns started?"""
|
||
if not delta:
|
||
return False
|
||
with _busy_lock:
|
||
if ip in _busy:
|
||
return False
|
||
_busy.add(ip)
|
||
threading.Thread(target=push_batch, args=(ip, delta, subdir), daemon=True).start()
|
||
return True
|
||
|
||
def failed_names(box_ip):
|
||
"""Failed (name, dir) for box_ip: active batch first, else recent history.
|
||
Skips entries already re-sent successfully later in history.
|
||
Keyed by (dir, name) so same basename in different subdirs are tracked separately."""
|
||
a = transfers_active()
|
||
pool = (a["files"] if a and a.get("box_ip") == box_ip else []) \
|
||
+ [h for h in transfers_history() if h.get("box_ip") == box_ip]
|
||
# later success overrides earlier failure: iterate chronological, keep last status
|
||
status_by = {}
|
||
for h in pool:
|
||
d = h.get("dir", "podcasts")
|
||
status_by[(d, h["name"])] = h["status"]
|
||
out = []
|
||
for (d, name), st in status_by.items():
|
||
if st == "failed":
|
||
out.append((name, d))
|
||
return out
|
||
|
||
def retry_failed(box_ip):
|
||
"""Re-push failed files for box_ip, grouped by subdir. Returns {started, count}."""
|
||
fails = failed_names(box_ip)
|
||
if not fails:
|
||
return {"started": False, "count": 0}
|
||
by_dir = {}
|
||
for name, d in fails:
|
||
by_dir.setdefault(d, []).append(name)
|
||
started = False
|
||
for d, names in by_dir.items():
|
||
if try_push(box_ip, names, subdir=d):
|
||
started = True
|
||
return {"started": started, "count": len(fails)}
|
||
|
||
# ---- 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, dir=None):
|
||
with _active_lock:
|
||
if not _active:
|
||
return
|
||
for f in _active["files"]:
|
||
if f["name"] != name:
|
||
continue
|
||
if dir is not None and f.get("dir") != dir:
|
||
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:
|
||
with open(TRANSFERS_JSON, encoding="utf-8") as fh:
|
||
return json.load(fh)
|
||
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
|
||
with _history_lock:
|
||
hist = _transfers_history_load()
|
||
hist.extend(done)
|
||
hist = hist[-TRANSFERS_MAX:]
|
||
_atomic_write_json(TRANSFERS_JSON, hist)
|