# -*- coding: utf-8 -*- """VPS agent: server-side downloads, show recordings and cookies. Runs INSIDE the remote-access relay (started by relay.serve() via ensure_worker()). Owner features live here (2026-10-01): * yt_download : yt-dlp a video URL on the VPS (audio mp3 or video) * radio_record: capture a radio show with ffmpeg at a scheduled time * cookies : per-account YouTube cookie store, used by yt_download Finished downloads land in the requesting session's Cloud Library (store.lib_add - the same store the phone plays from with the PC off), and the PC hears about every state change through the account notification queue (accounts.notify_user -> drained by /app/sync). Music Anywhere (cloud_sync) is deliberately PC-side: the desktop app uploads the chosen playlists with its EXISTING cloudlib client. Design constraints mirrored from _AudioVaultWorker and _Store: * standard library only (urllib, subprocess; no third-party imports); * auth is done by the relay handlers (token -> th -> account email); this module only ever sees validated job fields; * every persisted JSON write is atomic (tmp + os.replace) so a power cut can never leave a corrupt state file behind; * un-finished jobs at startup are marked "failed" honestly (like the AudioVault history), never silently forgotten; * the worker NEVER receives or uses the account password - the only secret it holds is the optional YouTube cookie file, which stays inside the agent data dir with 0600 permissions; * cleanup is bounded: history caps per account, downloads per day, and rows older than JOBS_MAX_AGE_S are pruned. """ import calendar import json import os import re import subprocess import threading import time try: # dual-mode import, same trick as relay.py from . import accounts as _accounts except ImportError: # pragma: no cover - standalone deployment import os as _os import sys as _sys _sys.path.insert(0, _os.path.dirname(_os.path.abspath(__file__))) import accounts as _accounts # --------------------------------------------------------------------------- # Tunables (env-overridable so the self-hosted relay can cap its box) # --------------------------------------------------------------------------- AGENT_INTERVAL = float(os.environ.get("WAVELINE_AGENT_INTERVAL", "5")) JOBS_MAX_PER_ACCT = 40 # history rows kept per account JOBS_MAX_AGE_S = 14 * 86400.0 # finished rows older than this are pruned DL_MAX_PER_DAY = 40 # download starts per account per day REC_MAX_PER_ACCT = 8 # scheduled show definitions per account REC_MAX_PER_DAY = 12 # recording starts per account per day COOKIES_MAX_BYTES = 512 * 1024 # a cookies.txt is a few KB; be generous # yt-dlp download is the only long job: cap it so one bad link cannot # monopolise the agent all day. DL_TIMEOUT_S = 45 * 60 # Live progress (2026-10-05 "same progress as a PC download"): parse # yt-dlp's --newline percent lines ("[download] 45.3% of 10.00MiB ...", # the exact format the PC's downloader parses), keep the percent in the # job row for the jobs dialog, carry it in the agent's active label, and # announce the quarter milestones through the notification queue so the # owner hears progress without opening anything. _PROGRESS_RE = re.compile(r"\[download\]\s+(\d+(?:\.\d+)?)%") _PROGRESS_MILESTONES = (25, 50, 75) DL_PROGRESS_SAVE_S = 5.0 # throttle progress-triggered state writes _DAYS = ("mon", "tue", "wed", "thu", "fri", "sat", "sun") def _agent_dir() -> str: return os.environ.get( "WAVELINE_AGENT_DIR", os.path.join(os.path.expanduser("~"), ".waveline-relay", "agent")) def _atomic_write(path: str, text: str) -> None: os.makedirs(os.path.dirname(path), exist_ok=True) tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as fh: fh.write(text) os.replace(tmp, path) def _load_json(path: str, default): try: with open(path, "r", encoding="utf-8") as fh: out = json.load(fh) return out if isinstance(out, type(default)) else default except (OSError, ValueError): return default def _today() -> str: return time.strftime("%Y-%m-%d", time.gmtime()) def _to_netscape(text: str) -> str: """Normalise pasted cookies to Netscape format (server-side port of the PC store's converter, trimmed to what YouTube needs). Accepts: a real Netscape file (passed through line by line), a browser "Copy as cookie" header line (``a=b; c=d``), or a single ``name=value`` pair. Quote-aware semicolon splitting so values containing ';' survive.""" text = (text or "").strip() if not text: return "" lines = text.splitlines() header_like = sum( 1 for ln in lines if ln.strip() and not ln.startswith("#") and "\t" not in ln and "=" in ln and ";" in ln) looks_netscape = any("\t" in ln for ln in lines) if looks_netscape or not header_like: # Netscape passthrough: keep comment/header lines and real rows. # (entry 87 review fix: some exporters write the 7 fields # SPACE-separated with no tabs - recognise those rows and # rebuild them with tabs instead of dropping everything.) out = [] for ln in lines: s = ln.rstrip() if not s: continue if s.startswith("#HttpOnly_"): out.append(s) # yt-dlp understands these elif s.startswith("#"): out.append(s) elif "\t" in s: out.append(s) else: # Some exporters write the 7 Netscape fields separated # by plain spaces. Fields 0..5 are positional; anything # after the name is the value (it may contain spaces). f = s.split() if len(f) >= 7: out.append("\t".join(( f[0], "TRUE" if f[1].upper() == "TRUE" else "FALSE", f[2], "TRUE" if f[3].upper() == "TRUE" else "FALSE", f[4], f[5], " ".join(f[6:])))) out = [ln for ln in out if not ln.startswith("#") or ln.startswith("#HttpOnly_") or ln.startswith("# Netscape") or ln.startswith("# HTTP")] if not any("\t" in ln for ln in out): return "" return "\n".join(out) + "\n" # Header line (or a bare pair): build one Netscape row per cookie. parts, buf, in_q = [], "", False for ch in text: if ch == '"': in_q = not in_q buf += ch elif ch == ";" and not in_q: parts.append(buf.strip()) buf = "" else: buf += ch if buf.strip(): parts.append(buf.strip()) rows = [] exp = str(int(time.time()) + 365 * 86400) for part in parts: if "=" not in part: continue name, _, value = part.partition("=") name = name.strip() value = value.strip().strip('"') if not name or name.lower() in ("path", "domain", "secure", "httponly", "samesite"): continue rows.append("\t".join((".youtube.com", "TRUE", "/", "TRUE", exp, name, value))) if not rows: return "" return ("# Netscape HTTP Cookie File\n" + "\n".join(rows) + "\n") def _tool_works(cmd) -> bool: """True when the tool both exists and can actually run here. `which` finds PATH entries the process cannot EXECUTE: on the relay host /usr/local/bin/yt-dlp and /usr/local/bin/deno were symlinks into /root, unreadable by the service user under ProtectHome=yes, so execvp silently skipped them and downloads ran with a fossil fallback version (owner report 2026-10-02, handbook 91). Probe by running the tool's --version and requiring exit 0.""" try: p = subprocess.run(list(cmd) + ["--version"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=20) return p.returncode == 0 except Exception: return False def _find_working_tool(name: str) -> str: """First PATH entry named `name` that actually RUNS, not merely exists. Falls back to plain `which` behaviour when nothing verifies (a probe is best-effort hardening, never a hard gate).""" from shutil import which for d in os.environ.get("PATH", "").split(os.pathsep): if not d: continue cand = os.path.join(d, name) if os.path.isfile(cand) and os.access(cand, os.X_OK) \ and _tool_works([cand]): return cand return which(name) or "" class _VpsAgent: """One instance per relay process (relay.serve / reset_store).""" def __init__(self, store): self._store = store self._lock = threading.Lock() # account email -> {"dl": [rows newest first], "recs": {id: rec}} self._state = {} self._loaded = False self._active = "" # human label of the running job self._notify = None # fn(email, kind, title, detail) self._yt_bin = "" self._ffmpeg_bin = "" self._probe() self._load() threading.Thread(target=self._loop, name="waveline-vps-agent", daemon=True).start() # -- tools -------------------------------------------------------------- def _probe(self): """Pin the tool paths ONCE at startup - but only tools that verified as runnable (see _find_working_tool).""" self._yt_bin = (os.environ.get("WAVELINE_AGENT_YTDLP") or _find_working_tool("yt-dlp") or _find_working_tool("yt-dlp.exe")) self._ffmpeg_bin = (os.environ.get("WAVELINE_AGENT_FFMPEG") or _find_working_tool("ffmpeg")) def _state_path(self) -> str: return os.path.join(_agent_dir(), "agent-state.json") def _load(self): if self._loaded: return self._loaded = True raw = _load_json(self._state_path(), {}) if not isinstance(raw, dict): return today = _today() for email, st in raw.items(): if not isinstance(st, dict) or not isinstance(email, str) \ or not email: continue st.setdefault("dl", []) st.setdefault("recs", {}) if st.get("day") != today: # daily counters reset st["day"], st["dl_count"], st["rec_count"] = today, 0, 0 st.setdefault("dl_count", 0) st.setdefault("rec_count", 0) for row in st["dl"]: if row.get("status") == "queued": row["status"], row["detail"] = "failed", ( "server restarted before the download finished") row["updated"] = time.time() self._state[email] = st self._recover_stuck_locked() def _save_locked(self): _atomic_write(self._state_path(), json.dumps(self._state)) def _acct(self, email: str) -> dict: st = self._state.get(email) if st is None: st = {"dl": [], "recs": {}, "day": _today(), "dl_count": 0, "rec_count": 0} self._state[email] = st if st.get("day") != _today(): st["day"], st["dl_count"], st["rec_count"] = _today(), 0, 0 return st def set_notify(self, fn): """Injected by the relay: fn(email, kind, title, detail).""" self._notify = fn def _tell(self, email: str, kind: str, title: str, detail: str = ""): if not email or not self._notify: return try: self._notify(email, kind, title, detail) except Exception: pass # -- status (relay handler support) -------------------------------------- def status(self) -> dict: """Agent health for the app's Cloud panel: what the server can do and how busy it is right now.""" with self._lock: dl_total = sum(len(st["dl"]) for st in self._state.values()) rec_total = sum(len(st["recs"]) for st in self._state.values()) return {"ok": True, "yt_dlp": bool(self._yt_bin), "ffmpeg": bool(self._ffmpeg_bin), # Diagnostics: which binary each pin resolved to, so # an environment problem (unreadable/stale tool) is # visible from the app instead of invisible. "yt_dlp_path": str(self._yt_bin or ""), "ffmpeg_path": str(self._ffmpeg_bin or ""), "active": self._active, "downloads_total": dl_total, "recordings_total": rec_total} # -- yt downloads --------------------------------------------------------- def dl_enqueue(self, email: str, th: str, url: str, mode: str, name: str = "") -> str: """Queue one yt-dlp job. Returns 'ok' or an error reason.""" if not self._yt_bin: return "no-ytdlp" mode = "video" if mode == "video" else "audio" url = str(url)[:500] if not url.startswith(("http://", "https://")): return "url" with self._lock: st = self._acct(email) if st["dl_count"] >= DL_MAX_PER_DAY: return "rate" already = any(r["url"] == url and r["status"] in ("queued", "active") for r in st["dl"]) row = {"id": "%d-%s" % (int(time.time()), os.urandom(3).hex()), "url": url, "mode": mode, "th": str(th)[:64], "status": "queued", "detail": "", "name": str(name or "")[:180], "requested": time.time(), "updated": time.time()} st["dl"].insert(0, row) del st["dl"][JOBS_MAX_PER_ACCT:] self._prune_locked() if not already: st["dl_count"] += 1 self._save_locked() if not already: self._tell(email, "vps", "Download queued on the server", "The Waveline server will download %s" % (str(name) or url)[:120]) return "ok" def dl_list(self, email: str): with self._lock: return [dict(r) for r in self._acct(email)["dl"][:JOBS_MAX_PER_ACCT]] def dl_clear(self, email: str) -> None: """Drop finished/failed rows; queued and active jobs stay (the user asked to clear HISTORY, not to cancel running work).""" with self._lock: st = self._acct(email) st["dl"] = [r for r in st["dl"] if r.get("status") in ("queued", "active")] self._save_locked() # -- radio recordings --------------------------------------------------- def rec_add(self, email: str, th: str, url: str, name: str, days: str, hhmm: str, minutes: int, tz_offset_min: int = 0) -> str: """Add one scheduled show. *days* like 'mon,thu' or 'daily'. *hhmm* is the OWNER's local wall time; *tz_offset_min* is the offset the OWNER's PC reported (minutes east of UTC, e.g. 120 for CEST) - the server fires when that local wall time arrives, so "21:30" means the owner's evening, not the server's. Returns 'ok' or an error reason.""" if not self._ffmpeg_bin: return "no-ffmpeg" if not str(url).startswith(("http://", "https://")): return "url" try: minutes = max(1, min(int(minutes or 30), 360)) tz_offset_min = max(-720, min(840, int(tz_offset_min or 0))) except (TypeError, ValueError): return "time" hhmm = str(hhmm or "") if not re.match(r"^([01]\d|2[0-3]):[0-5]\d$", hhmm): return "time" dset = set(d.strip().lower()[:3] for d in str(days).split(",") if d.strip()) if "dai" in dset: dset = set(_DAYS) dset = dset & set(_DAYS) if not dset: return "days" rid = os.urandom(6).hex() rec = {"id": rid, "url": str(url)[:500], "name": str(name or "Recording")[:120], "days": sorted(dset), "time": hhmm, "minutes": minutes, "tz": tz_offset_min, "th": str(th)[:64], "status": "scheduled", "detail": "", "last_run": 0.0} with self._lock: st = self._acct(email) if len(st["recs"]) >= REC_MAX_PER_ACCT: return "full" st["recs"][rid] = rec self._save_locked() return "ok" def rec_list(self, email: str): with self._lock: st = self._acct(email) return sorted((dict(r) for r in st["recs"].values()), key=lambda r: r["time"]) def rec_del(self, email: str, rid: str) -> bool: with self._lock: st = self._acct(email) gone = st["recs"].pop(str(rid)[:32], None) is not None if gone: self._save_locked() return gone # -- cookies (per-account YouTube store) --------------------------------- def _cookies_path(self, email: str) -> str: import hashlib h = hashlib.sha256(email.encode("utf-8", "replace")).hexdigest()[:24] return os.path.join(_agent_dir(), "cookies", h + ".txt") def cookies_put(self, email: str, text: str) -> str: """Store the owner's pasted YouTube cookies for server downloads. Accepts the same shapes as the PC-side store: a Netscape file or a raw 'name=value; name2=value2' header line.""" try: netscape = _to_netscape(str(text or "")) except Exception: return "bad" if not netscape.strip(): return "bad" if len(netscape.encode("utf-8")) > COOKIES_MAX_BYTES: return "size" path = self._cookies_path(email) _atomic_write(path, netscape) try: os.chmod(path, 0o600) except OSError: pass self._tell(email, "vps", "Cookies saved on the server", "Server downloads can now use your YouTube account") return "ok" def cookies_delete(self, email: str) -> bool: try: os.remove(self._cookies_path(email)) return True except OSError: return False def cookies_status(self, email: str) -> dict: try: mt = os.path.getmtime(self._cookies_path(email)) except OSError: return {"saved": False} return {"saved": True, "when": time.strftime("%Y-%m-%d", time.gmtime(mt))} def write_cookies_for(self, email: str) -> str: """Copy the account cookies to a 0600 temp file for one yt-dlp call ('' when none are stored). Caller deletes it after.""" try: with open(self._cookies_path(email), "r", encoding="utf-8") as fh: data = fh.read() except OSError: return "" import tempfile fd, dst = tempfile.mkstemp(prefix="wl-cookies-", suffix=".txt") with os.fdopen(fd, "w", encoding="utf-8") as fh: fh.write(data) try: os.chmod(dst, 0o600) except OSError: pass return dst # -- worker loop ---------------------------------------------------------- def _loop(self): while True: time.sleep(AGENT_INTERVAL) try: self._tick_downloads() except Exception: pass try: self._tick_recordings() except Exception: pass def _prune_locked(self): cutoff = time.time() - JOBS_MAX_AGE_S for st in self._state.values(): st["dl"] = [r for r in st["dl"] if r.get("status") in ("queued", "active") or r.get("updated", 0) >= cutoff] def _tick_downloads(self): job, email = None, "" with self._lock: for em, st in self._state.items(): for row in st["dl"]: if row.get("status") == "queued": job, email = row, em break if job: break if job is None: return job["status"] = "active" self._active = "downloading %s" % (job.get("name") or job["url"][:60]) self._run_download(email, job) self._active = "" with self._lock: job["updated"] = time.time() self._save_locked() def _recover_stuck_locked(self): """Startup/restart hygiene: a recording left in 'recording' (relay killed mid-capture) becomes an honest error - otherwise the show would never fire again.""" for st in self._state.values(): for rec in st.get("recs", {}).values(): if rec.get("status") == "recording": rec["status"], rec["detail"] = "error", ( "server restarted during the recording") def _run_download(self, email: str, job: dict) -> None: """One yt-dlp run; the result lands in the session library.""" try: self._download_to_lib(email, job) except subprocess.TimeoutExpired: job["status"], job["detail"] = "failed", ( "download took longer than %d minutes" % (DL_TIMEOUT_S // 60)) except Exception as exc: if job.get("status") not in ("done", "failed"): job["status"], job["detail"] = "failed", str(exc)[:160] if job["status"] == "done": self._tell(email, "vps", "Download finished on the server", "%s is in the cloud library" % (job.get("name") or "The file")) else: self._tell(email, "vps", "Download failed on the server", job.get("detail") or job["url"][:120]) def _download_to_lib(self, email: str, job: dict) -> None: """Run yt-dlp in a temp dir, move the file into the session library. Sets job status/detail/name; raises on hard errors.""" import shutil import tempfile workdir = tempfile.mkdtemp(prefix="wl-agent-") outtmpl = os.path.join(workdir, "%(title).120s.%(ext)s") # _yt_bin may be a string (PATH probe) or a list (tests inject # [python, script]); both must produce a flat argv. base = list(self._yt_bin) if isinstance(self._yt_bin, (list, tuple)) \ else [self._yt_bin] cmd = base + ["--no-playlist", "--no-warnings", "--newline", "-o", outtmpl, job["url"]] if job["mode"] == "audio": cmd += ["-f", "bestaudio/best", "-x", "--audio-format", "mp3", "--audio-quality", "0"] else: cmd += ["-f", "best[ext=mp4]/best", "--merge-output-format", "mp4"] cookies = self.write_cookies_for(email) if cookies: cmd += ["--cookies", cookies] try: proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, errors="replace", cwd=workdir) deadline = time.time() + DL_TIMEOUT_S last_err = "" last_save = time.time() seen_mile = set() label = job.get("name") or job["url"][:60] for line in proc.stdout: line = line.strip() if line: last_err = line[:300] m = _PROGRESS_RE.search(line) if m: try: pct = min(100, int(float(m.group(1)))) except ValueError: pct = 0 if pct > int(job.get("pct", 0) or 0): job["pct"] = pct self._active = "downloading %s - %d%%" % (label, pct) crossed = [p for p in _PROGRESS_MILESTONES if p not in seen_mile and pct >= p] if crossed: # one announcement per update: the highest # milestone crossed (a fast start that jumps # 0 -> 55 says 50%, not 25% then 50%) seen_mile.update(crossed) self._tell(email, "vps", "Server download: %d%% done" % max(crossed), label) if time.time() - last_save >= DL_PROGRESS_SAVE_S: with self._lock: self._save_locked() last_save = time.time() if time.time() > deadline: proc.kill() raise subprocess.TimeoutExpired(cmd, DL_TIMEOUT_S) rc = proc.wait(timeout=60) finally: if cookies: try: os.remove(cookies) except OSError: pass if rc != 0: # Keep yt-dlp's real error line: the old fixed text ("the # link may be private...") hid the actual cause and made # the 2026-10-02 environment failure undiagnosable. job["status"], job["detail"] = "failed", ( "yt-dlp failed (exit %d): %s" % (rc, last_err or "no output captured")) raise RuntimeError(job["detail"]) files = sorted(f for f in os.listdir(workdir) if os.path.isfile(os.path.join(workdir, f))) if not files: job["status"], job["detail"] = "failed", ( "no file was produced") raise RuntimeError(job["detail"]) path = os.path.join(workdir, files[0]) size = os.path.getsize(path) ext = os.path.splitext(files[0])[1] or ".bin" base = job.get("name") or os.path.splitext(files[0])[0] or "download" name = (base if base.lower().endswith(ext) else base + ext)[:180] if size <= 0 or size > self._store.LIB_MAX_FILE: job["status"], job["detail"] = "failed", ( "file is empty or over the 2 GB cloud-library cap") raise RuntimeError(job["detail"]) lid = "agent" + os.urandom(10).hex() add = self._store.lib_add(job["th"], lid, name, size) if add != "ok": job["status"], job["detail"] = "failed", { "full": "cloud library is full (20 GB / 200 files)", "disk": "server is low on disk space", "missing": "session expired - reopen Waveline to re-pair", }.get(add, add) raise RuntimeError(job["detail"]) dst = self._store._lib_file(job["th"], lid) os.makedirs(os.path.dirname(dst), exist_ok=True) with open(path, "rb") as fh, open(dst, "wb") as out: while True: chunk = fh.read(65536) if not chunk: break out.write(chunk) shutil.rmtree(workdir, ignore_errors=True) job["status"], job["detail"], job["name"] = "done", "", name job["pct"] = 100 def _tick_recordings(self): due = [] with self._lock: for email, st in self._state.items(): if not st.get("recs"): continue for rec in st["recs"].values(): if rec.get("status") not in ("scheduled", "error"): continue # Fire on the OWNER's local wall clock: utc + the # offset their PC reported at schedule time. Weekday # follows the owner's local day too (a 23:30 Monday # show in owner time must not fire on the server's # UTC day). The fire window is 150 s so a late or # busy tick never silently skips a show. off = int(rec.get("tz", 0) or 0) local = time.gmtime(time.time() + off * 60) day = _DAYS[local.tm_wday] if day not in rec["days"]: continue try: hh, mm = rec["time"].split(":") due_epoch = calendar.timegm( (local.tm_year, local.tm_mon, local.tm_mday, int(hh), int(mm), 0)) - off * 60 except (KeyError, ValueError): continue now = time.time() if not (due_epoch <= now < due_epoch + 150): continue if now - rec.get("last_run", 0) < 6 * 3600: continue if st["rec_count"] >= REC_MAX_PER_DAY: continue rec["last_run"] = time.time() rec["status"] = "recording" st["rec_count"] += 1 due.append((email, dict(rec))) for email, rec in due: self._tell(email, "vps", "Recording started on the server", "%s (%d minutes)" % (rec["name"], rec["minutes"])) self._run_recording(email, rec) with self._lock: self._save_locked() def _run_recording(self, email: str, rec: dict) -> None: """ffmpeg capture -> mp3 -> cloud library (runs on the agent thread; state transitions are re-checked under the lock).""" import shutil import tempfile def fail(detail): with self._lock: live = self._acct(email)["recs"].get(rec["id"]) if live: live["status"], live["detail"] = "error", detail self._tell(email, "vps", "Recording failed on the server", "%s: %s" % (rec["name"], detail)) workdir = tempfile.mkdtemp(prefix="wl-rec-") dump = os.path.join(workdir, "dump.ts") seconds = int(rec["minutes"]) * 60 ff = list(self._ffmpeg_bin) if isinstance(self._ffmpeg_bin, (list, tuple)) \ else [self._ffmpeg_bin] try: proc = subprocess.Popen( ff + ["-y", "-hide_banner", "-loglevel", "error", "-i", rec["url"], "-t", str(seconds), "-c", "copy", dump], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) # Poll instead of wait(): wait(timeout) raises but can leave # the tick thread blocked on a wedged child forever (seen # live 2026-10-01). poll() + hard kill can never hang. deadline = time.time() + seconds + 120 while proc.poll() is None and time.time() < deadline: time.sleep(2) if proc.poll() is None: proc.kill() try: proc.wait(timeout=10) except subprocess.TimeoutExpired: pass except Exception as exc: fail(str(exc)[:120] or "capture failed") shutil.rmtree(workdir, ignore_errors=True) return if not os.path.isfile(dump) or os.path.getsize(dump) <= 0: fail("the station sent no audio") shutil.rmtree(workdir, ignore_errors=True) return final = os.path.join(workdir, "out.mp3") if subprocess.call( ff + ["-y", "-hide_banner", "-loglevel", "error", "-i", dump, "-c:a", "libmp3lame", "-q:a", "0", final]) != 0 \ or not os.path.isfile(final) \ or os.path.getsize(final) <= 0: fail("conversion failed") shutil.rmtree(workdir, ignore_errors=True) return size = os.path.getsize(final) name = ("%s %s.mp3" % (rec["name"], time.strftime("%Y-%m-%d %H%M", time.gmtime(rec["last_run"])) )).strip()[:180] if size > self._store.LIB_MAX_FILE: fail("recording is over the 2 GB cloud-library cap") shutil.rmtree(workdir, ignore_errors=True) return lid = "agent" + os.urandom(10).hex() add = self._store.lib_add(rec["th"], lid, name, size) if add != "ok": fail("cloud library refused the file (%s)" % add) shutil.rmtree(workdir, ignore_errors=True) return dst = self._store._lib_file(rec["th"], lid) os.makedirs(os.path.dirname(dst), exist_ok=True) with open(final, "rb") as fh, open(dst, "wb") as out: while True: chunk = fh.read(65536) if not chunk: break out.write(chunk) shutil.rmtree(workdir, ignore_errors=True) with self._lock: live = self._acct(email)["recs"].get(rec["id"]) if live: live["status"], live["detail"] = "scheduled", "" self._tell(email, "vps", "Recording finished on the server", "%s is in the cloud library" % name) def ensure_worker(store) -> "_VpsAgent": """Create/replace the module-level worker (relay.serve + tests).""" global _worker _worker = _VpsAgent(store) return _worker _worker: "_VpsAgent" = None def worker() -> "_VpsAgent": return _worker