eminux/MatchLiveTv
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Aggiunge replay YouTube temporaneo, pull registrazioni dai CPX e snapshot ingest in admin.

Browse Source 
Così overflow Hetzner e VOD YouTube restano in archivio dopo lo spegnimento del nodo, e la colonna ingest non si svuota.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
eminux
2026-08-28 13:01:41 +02:00
co-authored by Cursor
parent 35dfa923e3
commit 05ef56c56d
54 changed files with 1843 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files
infra/.env.example
+6
View File
@@ -64,3 +64,9 @@ REPLAY_STORAGE_FORCE_PATH_STYLE=true
# Opzionale: durata link firmati per play/download (secondi)
# REPLAY_PRESIGNED_URL_TTL_SECONDS=7200
# REPLAY_DOWNLOAD_URL_TTL_SECONDS=900
# Replay YouTube: copia temporanea di sicurezza (ore, clamp 24–72, default 48)
# YOUTUBE_TEMP_REPLAY_RETENTION_HOURS=48
# YOUTUBE_REPLAY_VERIFY_MAX_ATTEMPTS=12
# YOUTUBE_REPLAY_VERIFY_BASE_INTERVAL_SECS=300
# YOUTUBE_REPLAY_VERIFY_GRACE_SECS=120
infra/.env.production.example
+5
View File
@@ -78,6 +78,11 @@ REPLAY_STORAGE_FORCE_PATH_STYLE=true
# Streaming replay: Rails fa auth, edge serve /media/ → Garage (non passa da Puma)
REPLAY_MEDIA_PUBLIC_BASE_URL=https://www.matchlivetv.it/media
REPLAY_MEDIA_REDIRECT=true
# Copia temporanea post-live YouTube (ore, clamp 24–72)
YOUTUBE_TEMP_REPLAY_RETENTION_HOURS=48
YOUTUBE_REPLAY_VERIFY_MAX_ATTEMPTS=12
YOUTUBE_REPLAY_VERIFY_BASE_INTERVAL_SECS=300
YOUTUBE_REPLAY_VERIFY_GRACE_SECS=120
# Puma: 5 thread (no secondo worker senza più RAM)
RAILS_MAX_THREADS=5
infra/scripts/install_production_cron.sh
+1
View File
@@ -28,6 +28,7 @@ MARKER="# matchlivetv-replay-cron"
CRON_BLOCK="${MARKER}
30 * * * * mkdir -p ${LOG_DIR} && ${RUNNER} recordings:cleanup_local >> ${LOG_FILE} 2>&1
0 3 * * * mkdir -p ${LOG_DIR} && ${RUNNER} recordings:purge_expired >> ${LOG_FILE} 2>&1
30 3 * * * mkdir -p ${LOG_DIR} && ${RUNNER} recordings:purge_temporary >> ${LOG_FILE} 2>&1
0 8 * * * mkdir -p ${LOG_DIR} && ${RUNNER} recordings:expiry_warnings >> ${LOG_FILE} 2>&1
15 * * * * mkdir -p ${LOG_DIR} && ${RUNNER} analytics:aggregate >> ${LOG_DIR}/cron-analytics.log 2>&1
20 4 * * * mkdir -p ${LOG_DIR} && ${RUNNER} analytics:purge >> ${LOG_DIR}/cron-analytics.log 2>&1
infra/stream-node/cloud-init.yaml
+66 -7
View File
@@ -57,6 +57,7 @@ write_files:
STREAM_NODE_LOCAL_HLS_URL=http://127.0.0.1:8888
STREAM_NODE_RELAY_LOG_DIR=/var/log
STREAM_NODE_SLATES_DIR=/slates/custom
STREAM_NODE_RECORDINGS_DIR=/recordings
- path: /opt/stream-node/relay-agent.py
permissions: "0755"
content: |
@@ -68,17 +69,21 @@ write_files:
import json
import os
import shutil
import signal
import subprocess
import tarfile
import tempfile
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse
from urllib.parse import unquote, urlparse
SECRET = os.environ.get("STREAM_NODE_AGENT_SECRET", "")
LISTEN = os.environ.get("STREAM_NODE_AGENT_LISTEN", "0.0.0.0:9100")
HLS_BASE = os.environ.get("STREAM_NODE_LOCAL_HLS_URL", "http://127.0.0.1:8888").rstrip("/")
LOG_DIR = os.environ.get("STREAM_NODE_RELAY_LOG_DIR", "/var/log")
SLATES_DIR = os.environ.get("STREAM_NODE_SLATES_DIR", "/slates/custom")
RECORDINGS_DIR = os.environ.get("STREAM_NODE_RECORDINGS_DIR", "/recordings")
_lock = threading.Lock()
# session_id -> {"pid": int, "path": str, "proc": Popen}
@@ -160,6 +165,27 @@ write_files:
pass
def _recordings_root(path_name):
name = unquote(path_name or "").strip().lstrip("/")
if not name or ".." in name.split("/"):
return None
base = os.path.realpath(RECORDINGS_DIR)
root = os.path.realpath(os.path.join(base, name))
if root != base and not root.startswith(base + os.sep):
return None
return root
def _has_recording_files(root: str) -> bool:
if not os.path.isdir(root):
return False
for dirpath, _dirnames, filenames in os.walk(root):
for name in filenames:
if name.lower().endswith((".mp4", ".fmp4", ".m4s", ".ts")):
return True
return False
def _start_session(session_id: str, path: str, rtmps: str) -> int:
existing = _relays.get(session_id)
if existing:
@@ -214,6 +240,30 @@ write_files:
return
self._json(200, {"running": True, "session_id": session_id, "pid": info["pid"], "path": info["path"]})
return
if parsed.path.startswith("/recordings/"):
name = parsed.path.split("/recordings/", 1)[1].strip("/")
root = _recordings_root(name)
if not root or not _has_recording_files(root):
self._json(404, {"error": "not found"})
return
tmp = tempfile.NamedTemporaryFile(prefix="mltv-rec-", suffix=".tar.gz", delete=False)
tmp.close()
try:
with tarfile.open(tmp.name, "w:gz") as tar:
tar.add(root, arcname=".")
size = os.path.getsize(tmp.name)
self.send_response(200)
self.send_header("Content-Type", "application/gzip")
self.send_header("Content-Length", str(size))
self.end_headers()
with open(tmp.name, "rb") as fh:
shutil.copyfileobj(fh, self.wfile)
finally:
try:
os.unlink(tmp.name)
except OSError:
pass
return
self._json(404, {"error": "not found"})
def do_POST(self) -> None:
@@ -244,13 +294,22 @@ write_files:
self._json(401, {"error": "unauthorized"})
return
parsed = urlparse(self.path)
if not parsed.path.startswith("/relays/"):
self._json(404, {"error": "not found"})
if parsed.path.startswith("/relays/"):
session_id = parsed.path.split("/relays/", 1)[1].strip("/")
with _lock:
_stop_session(session_id)
self._json(200, {"stopped": True, "session_id": session_id})
return
session_id = parsed.path.split("/relays/", 1)[1].strip("/")
with _lock:
_stop_session(session_id)
self._json(200, {"stopped": True, "session_id": session_id})
if parsed.path.startswith("/recordings/"):
name = parsed.path.split("/recordings/", 1)[1].strip("/")
root = _recordings_root(name)
if not root or not os.path.isdir(root):
self._json(404, {"error": "not found"})
return
shutil.rmtree(root, ignore_errors=True)
self._json(200, {"deleted": True, "path": name})
return
self._json(404, {"error": "not found"})
def do_PUT(self) -> None:
if not _authorized(self):
infra/stream-node/relay-agent.py
+65 -7
View File
@@ -6,17 +6,21 @@ from __future__ import annotations
import json
import os
import shutil
import signal
import subprocess
import tarfile
import tempfile
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse
from urllib.parse import unquote, urlparse
SECRET = os.environ.get("STREAM_NODE_AGENT_SECRET", "")
LISTEN = os.environ.get("STREAM_NODE_AGENT_LISTEN", "0.0.0.0:9100")
HLS_BASE = os.environ.get("STREAM_NODE_LOCAL_HLS_URL", "http://127.0.0.1:8888").rstrip("/")
LOG_DIR = os.environ.get("STREAM_NODE_RELAY_LOG_DIR", "/var/log")
SLATES_DIR = os.environ.get("STREAM_NODE_SLATES_DIR", "/slates/custom")
RECORDINGS_DIR = os.environ.get("STREAM_NODE_RECORDINGS_DIR", "/recordings")
_lock = threading.Lock()
# session_id -> {"pid": int, "path": str, "proc": Popen}
@@ -98,6 +102,27 @@ def _stop_session(session_id: str) -> None:
pass
def _recordings_root(path_name):
name = unquote(path_name or "").strip().lstrip("/")
if not name or ".." in name.split("/"):
return None
base = os.path.realpath(RECORDINGS_DIR)
root = os.path.realpath(os.path.join(base, name))
if root != base and not root.startswith(base + os.sep):
return None
return root
def _has_recording_files(root: str) -> bool:
if not os.path.isdir(root):
return False
for dirpath, _dirnames, filenames in os.walk(root):
for name in filenames:
if name.lower().endswith((".mp4", ".fmp4", ".m4s", ".ts")):
return True
return False
def _start_session(session_id: str, path: str, rtmps: str) -> int:
existing = _relays.get(session_id)
if existing:
@@ -152,6 +177,30 @@ class Handler(BaseHTTPRequestHandler):
return
self._json(200, {"running": True, "session_id": session_id, "pid": info["pid"], "path": info["path"]})
return
if parsed.path.startswith("/recordings/"):
name = parsed.path.split("/recordings/", 1)[1].strip("/")
root = _recordings_root(name)
if not root or not _has_recording_files(root):
self._json(404, {"error": "not found"})
return
tmp = tempfile.NamedTemporaryFile(prefix="mltv-rec-", suffix=".tar.gz", delete=False)
tmp.close()
try:
with tarfile.open(tmp.name, "w:gz") as tar:
tar.add(root, arcname=".")
size = os.path.getsize(tmp.name)
self.send_response(200)
self.send_header("Content-Type", "application/gzip")
self.send_header("Content-Length", str(size))
self.end_headers()
with open(tmp.name, "rb") as fh:
shutil.copyfileobj(fh, self.wfile)
finally:
try:
os.unlink(tmp.name)
except OSError:
pass
return
self._json(404, {"error": "not found"})
def do_POST(self) -> None:
@@ -182,13 +231,22 @@ class Handler(BaseHTTPRequestHandler):
self._json(401, {"error": "unauthorized"})
return
parsed = urlparse(self.path)
if not parsed.path.startswith("/relays/"):
self._json(404, {"error": "not found"})
if parsed.path.startswith("/relays/"):
session_id = parsed.path.split("/relays/", 1)[1].strip("/")
with _lock:
_stop_session(session_id)
self._json(200, {"stopped": True, "session_id": session_id})
return
session_id = parsed.path.split("/relays/", 1)[1].strip("/")
with _lock:
_stop_session(session_id)
self._json(200, {"stopped": True, "session_id": session_id})
if parsed.path.startswith("/recordings/"):
name = parsed.path.split("/recordings/", 1)[1].strip("/")
root = _recordings_root(name)
if not root or not os.path.isdir(root):
self._json(404, {"error": "not found"})
return
shutil.rmtree(root, ignore_errors=True)
self._json(200, {"deleted": True, "path": name})
return
self._json(404, {"error": "not found"})
def do_PUT(self) -> None:
if not _authorized(self):