feat(tower): transcode episodes + async push
Episodes are 18-24 MB and the box's link tops out near 100 KB/s, so full-size pushes time out. Transcode to mono 48 kbps MP3 (a few MB, plenty for spoken-word kids podcasts); the _a48 tag gives a fresh name so the box re-pulls the light version. Ack /register immediately and push the delta on a background thread (one batch per box) so the box's short register timeout never trips.
This commit is contained in:
+21
-2
@@ -57,11 +57,30 @@ def main():
|
||||
# Unique per episode (enclosure URL is stable+unique) so a new episode
|
||||
# = a new filename = the box detects it as missing. Positional names
|
||||
# (oli_01) would hide a new "latest" episode behind the same name.
|
||||
fname = f"{key}_{hashlib.sha1(url.encode()).hexdigest()[:8]}.{ext_for(url, enc.get('type',''))}"
|
||||
# The "_a48" tag marks the transcode profile: a 20+ MB original would
|
||||
# take minutes over the box's ~100 KB/s link, so we re-encode to mono
|
||||
# 48 kbps MP3 (plenty for spoken-word kids podcasts) — files drop to a
|
||||
# few MB and each push completes well under the server timeout. The tag
|
||||
# also gives the file a fresh name, so the box re-pulls the light
|
||||
# version and the old heavy original becomes an unreferenced orphan.
|
||||
h = hashlib.sha1(url.encode()).hexdigest()[:8]
|
||||
fname = f"{key}_{h}_a48.mp3"
|
||||
raw = os.path.join(tmp, f".raw_{key}_{h}")
|
||||
try:
|
||||
open(os.path.join(tmp, fname), "wb").write(fetch(url))
|
||||
open(raw, "wb").write(fetch(url))
|
||||
except Exception as e:
|
||||
print("dl fail", fname, e); continue
|
||||
out = os.path.join(tmp, fname)
|
||||
rc = subprocess.run(
|
||||
["ffmpeg", "-y", "-loglevel", "error", "-i", raw,
|
||||
"-ac", "1", "-b:a", "48k", "-c:a", "libmp3lame", out],
|
||||
capture_output=True).returncode
|
||||
try:
|
||||
os.remove(raw)
|
||||
except OSError:
|
||||
pass
|
||||
if rc != 0 or not os.path.exists(out):
|
||||
print("transcode fail", fname); continue
|
||||
eps.append({"file": fname, "title": clean(it.findtext("title") or f"{show} {i}")})
|
||||
if not eps:
|
||||
continue
|
||||
|
||||
+29
-9
@@ -1,7 +1,7 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Always-on: POST /register {id, have} -> push the staging delta to the box's
|
||||
file-server (:8080). The box IP is the request source address."""
|
||||
import json, os, urllib.request
|
||||
import json, os, threading, urllib.request
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
|
||||
STAGING = os.environ.get("LISAEL_STAGING", "/home/clems/lisael-content/podcasts")
|
||||
@@ -24,10 +24,29 @@ def compute_delta(staging, have):
|
||||
def push_file(ip, name):
|
||||
data = open(os.path.join(STAGING, name), "rb").read()
|
||||
url = f"http://{ip}:{BOX_PORT}/put?name=podcasts/{name}"
|
||||
# Generous timeout: a multi-MB episode written to the box's slow microSD can
|
||||
# take well over a minute, especially during boot when the box is also doing
|
||||
# its first weather TLS fetch.
|
||||
with urllib.request.urlopen(urllib.request.Request(url, data=data, method="POST"),
|
||||
timeout=40) as r:
|
||||
timeout=300) as r:
|
||||
return r.status
|
||||
|
||||
# Pushes run off the request thread (one batch per box at a time) so the box's
|
||||
# short register timeout never waits on a multi-file, multi-minute transfer.
|
||||
_busy = set()
|
||||
_busy_lock = threading.Lock()
|
||||
|
||||
def push_batch(ip, delta):
|
||||
ok = 0
|
||||
for f in delta:
|
||||
try:
|
||||
push_file(ip, f); ok += 1
|
||||
except Exception as e:
|
||||
print("push fail", f, e, flush=True)
|
||||
print(f"push done {ip} {ok}/{len(delta)}", flush=True)
|
||||
with _busy_lock:
|
||||
_busy.discard(ip)
|
||||
|
||||
class H(BaseHTTPRequestHandler):
|
||||
def do_POST(self):
|
||||
if self.path.split("?")[0] != "/register":
|
||||
@@ -41,14 +60,15 @@ class H(BaseHTTPRequestHandler):
|
||||
delta = compute_delta(staging_files(), body.get("have", []) or [])
|
||||
print(f"register {ip} id={body.get('id')} have={len(body.get('have',[]) or [])} push={len(delta)}",
|
||||
flush=True)
|
||||
ok = 0
|
||||
for f in delta:
|
||||
try:
|
||||
push_file(ip, f); ok += 1
|
||||
except Exception as e:
|
||||
print("push fail", f, e, flush=True)
|
||||
with _busy_lock:
|
||||
already = ip in _busy
|
||||
if delta and not already:
|
||||
_busy.add(ip)
|
||||
if delta and not already:
|
||||
threading.Thread(target=push_batch, args=(ip, delta), daemon=True).start()
|
||||
# Respond immediately; the (possibly slow) push runs in the background.
|
||||
self.send_response(200); self.send_header("Content-Type", "application/json"); self.end_headers()
|
||||
self.wfile.write(json.dumps({"pushed": ok, "of": len(delta)}).encode())
|
||||
self.wfile.write(json.dumps({"queued": len(delta), "busy": already}).encode())
|
||||
|
||||
def log_message(self, *a):
|
||||
pass
|
||||
|
||||
Reference in New Issue
Block a user