57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def _runtime_dir(base_dir: str | Path | None = None) -> Path:
|
|
root = Path(base_dir or os.environ.get("LIVE_PROJECT_ROOT") or Path(__file__).resolve().parent)
|
|
path = root / "runtime" / "youtube_relay"
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
|
|
|
|
def heartbeat_path(pm2: str, *, base_dir: str | Path | None = None) -> Path:
|
|
safe = "".join(ch if ch.isalnum() or ch in "._-" else "_" for ch in (pm2 or "youtube").strip()) or "youtube"
|
|
return _runtime_dir(base_dir) / f"{safe}.json"
|
|
|
|
|
|
def write_relay_heartbeat(
|
|
pm2: str,
|
|
payload: dict[str, Any],
|
|
*,
|
|
base_dir: str | Path | None = None,
|
|
) -> None:
|
|
data = dict(payload)
|
|
data["updated_at"] = time.time()
|
|
data["pm2"] = (pm2 or "").strip() or "youtube"
|
|
path = heartbeat_path(data["pm2"], base_dir=base_dir)
|
|
tmp = path.with_suffix(".json.tmp")
|
|
try:
|
|
tmp.write_text(json.dumps(data, ensure_ascii=False, separators=(",", ":")), encoding="utf-8")
|
|
tmp.replace(path)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def read_relay_heartbeat(pm2: str, *, base_dir: str | Path | None = None) -> dict[str, Any]:
|
|
path = heartbeat_path(pm2, base_dir=base_dir)
|
|
if not path.is_file():
|
|
return {}
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
|
|
def clear_relay_heartbeat(pm2: str, *, base_dir: str | Path | None = None) -> None:
|
|
path = heartbeat_path(pm2, base_dir=base_dir)
|
|
try:
|
|
if path.is_file():
|
|
path.unlink()
|
|
except OSError:
|
|
pass
|