648 lines
25 KiB
Python
648 lines
25 KiB
Python
import datetime
|
||
import glob as globmod
|
||
import logging
|
||
import os
|
||
import re
|
||
import shlex
|
||
import signal
|
||
import subprocess
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
from urllib.parse import urlsplit
|
||
|
||
from config import DEFAULT_DOWNLOAD_PATH, validate_download_path
|
||
from models import Recording, get_session, LogEntry, iso_utc, utcnow
|
||
|
||
logger = logging.getLogger("recorder")
|
||
|
||
ALLOWED_URL_SCHEMES = ("http", "https", "rtmp", "rtmps", "rtsp", "rtsps")
|
||
|
||
PROTOCOL_WHITELIST = "http,https,tcp,tls,crypto,hls,httpproxy,rtmp,rtmps,rtmpt,rtmpts,ffrtmphttp,rtp,udp"
|
||
|
||
FFMPEG_FLAGS = {"-vn", "-an", "-sn", "-dn", "-shortest", "-copyts"}
|
||
FFMPEG_VALUE_OPTIONS = {
|
||
"-c", "-c:a", "-c:v", "-codec", "-codec:a", "-codec:v", "-acodec", "-vcodec",
|
||
"-b:a", "-b:v", "-ar", "-ac", "-map", "-metadata", "-metadata:s:a", "-t", "-to",
|
||
"-q:a", "-aq", "-af", "-filter:a", "-bsf:a", "-bsf:v", "-f", "-movflags",
|
||
"-segment_format", "-fflags", "-avoid_negative_ts", "-max_muxing_queue_size",
|
||
}
|
||
FORBIDDEN_FILTER_WORDS = ("movie", "sendcmd", "zmq", "file")
|
||
MAX_FFMPEG_ARGS_LEN = 500
|
||
MIN_STABLE_RUNTIME = 60
|
||
|
||
STALE_ERROR_MESSAGE = "Abgebrochen durch Neustart"
|
||
|
||
|
||
def validate_ffmpeg_args(args: str) -> list[str]:
|
||
"""Prüft zusätzliche ffmpeg-Argumente gegen eine Allowlist. Wirft ValueError."""
|
||
if not args:
|
||
return []
|
||
if len(args) > MAX_FFMPEG_ARGS_LEN:
|
||
raise ValueError(f"Zusätzliche ffmpeg-Argumente sind zu lang (max. {MAX_FFMPEG_ARGS_LEN} Zeichen)")
|
||
try:
|
||
tokens = shlex.split(args)
|
||
except ValueError as e:
|
||
raise ValueError(f"Zusätzliche ffmpeg-Argumente nicht lesbar: {e}")
|
||
|
||
i = 0
|
||
while i < len(tokens):
|
||
tok = tokens[i]
|
||
if tok in FFMPEG_FLAGS:
|
||
i += 1
|
||
continue
|
||
if tok in FFMPEG_VALUE_OPTIONS:
|
||
if i + 1 >= len(tokens):
|
||
raise ValueError(f"ffmpeg-Option '{tok}' benötigt einen Wert")
|
||
val = tokens[i + 1]
|
||
if not val or val.startswith("-"):
|
||
raise ValueError(f"Ungültiger Wert für ffmpeg-Option '{tok}': '{val}'")
|
||
if tok in ("-af", "-filter:a"):
|
||
low = val.lower()
|
||
for word in FORBIDDEN_FILTER_WORDS:
|
||
if word in low:
|
||
raise ValueError(f"Filter '{val}' ist nicht erlaubt ('{word}')")
|
||
i += 2
|
||
continue
|
||
raise ValueError(f"ffmpeg-Argument nicht erlaubt: '{tok}'")
|
||
return tokens
|
||
|
||
|
||
def validate_stream_url(url: str) -> str:
|
||
"""Prüft eine Stream-URL. Wirft ValueError."""
|
||
if not isinstance(url, str):
|
||
raise ValueError("Stream-URL fehlt")
|
||
url = url.strip()
|
||
if not url:
|
||
raise ValueError("Stream-URL fehlt")
|
||
if len(url) > 1000:
|
||
raise ValueError("Stream-URL ist zu lang (max. 1000 Zeichen)")
|
||
if any(ch.isspace() or ord(ch) < 32 or ord(ch) == 127 for ch in url):
|
||
raise ValueError("Stream-URL enthält Leer- oder Steuerzeichen")
|
||
try:
|
||
parts = urlsplit(url)
|
||
hostname = parts.hostname
|
||
except ValueError:
|
||
raise ValueError("Stream-URL ist ungültig")
|
||
if parts.scheme.lower() not in ALLOWED_URL_SCHEMES:
|
||
raise ValueError(f"Stream-URL muss mit {', '.join(s + '://' for s in ALLOWED_URL_SCHEMES)} beginnen")
|
||
if not parts.netloc or not hostname:
|
||
raise ValueError("Stream-URL enthält keinen Host")
|
||
return url
|
||
|
||
|
||
def detect_stream_type(url: str) -> str:
|
||
url_lower = url.lower().split("?")[0]
|
||
if url_lower.startswith(("rtmp://", "rtmps://")):
|
||
return "rtmp"
|
||
if url_lower.endswith(".m3u8"):
|
||
return "hls"
|
||
if url_lower.endswith(".mp3"):
|
||
return "mp3"
|
||
if url_lower.endswith(".aac"):
|
||
return "aac"
|
||
if url_lower.endswith(".ogg"):
|
||
return "ogg"
|
||
if url_lower.endswith(".flac"):
|
||
return "flac"
|
||
return "http"
|
||
|
||
|
||
def default_extension(stream_type: str) -> str:
|
||
return {"mp3": "mp3", "aac": "aac", "ogg": "ogg", "flac": "flac"}.get(stream_type, "mp4")
|
||
|
||
|
||
def safe_name(name: str) -> str:
|
||
return re.sub(r"[^A-Za-z0-9_-]", "_", name).strip("_")
|
||
|
||
|
||
def format_elapsed(seconds: int) -> str:
|
||
d, rem = divmod(seconds, 86400)
|
||
h, rem = divmod(rem, 3600)
|
||
m, s = divmod(rem, 60)
|
||
if d > 0:
|
||
return f"{d}d {h}h {m}m"
|
||
if h > 0:
|
||
return f"{h}h {m}m {s}s"
|
||
return f"{m}m {s}s"
|
||
|
||
|
||
def format_size(size_bytes: int) -> str:
|
||
if size_bytes >= 1073741824:
|
||
return f"{size_bytes / 1073741824:.1f} GB"
|
||
if size_bytes >= 1048576:
|
||
return f"{size_bytes / 1048576:.1f} MB"
|
||
if size_bytes >= 1024:
|
||
return f"{size_bytes / 1024:.1f} KB"
|
||
return f"{size_bytes} B"
|
||
|
||
|
||
class RecordingManager:
|
||
def __init__(self, config):
|
||
self.config = config
|
||
self._processes: dict[int, subprocess.Popen] = {}
|
||
self._stop_events: dict[int, threading.Event] = {}
|
||
self._threads: dict[int, threading.Thread] = {}
|
||
self._ntfy = None
|
||
self._plik = None
|
||
self._rec_options: dict[int, dict] = {}
|
||
self._output_files: dict[int, list[str]] = {}
|
||
|
||
def _get_ntfy(self):
|
||
if self._ntfy is None:
|
||
from plugins.ntfy import NtfyNotifier
|
||
self._ntfy = NtfyNotifier(self.config)
|
||
return self._ntfy
|
||
|
||
def _get_plik(self):
|
||
if self._plik is None:
|
||
from plugins.plik import PlikUploader
|
||
self._plik = PlikUploader(
|
||
self.config,
|
||
log_fn=self._log,
|
||
notify_fn=lambda event, title, msg, ntfy_en: self._get_ntfy().notify(event, title, msg, ntfy_en),
|
||
)
|
||
return self._plik
|
||
|
||
def _notify(self, event: str, title: str, message: str, rec_id: int = None, priority: str = "default"):
|
||
ntfy_enabled = False
|
||
if rec_id and rec_id in self._rec_options:
|
||
ntfy_enabled = self._rec_options[rec_id].get("ntfy_enabled", False)
|
||
desc = self._rec_options[rec_id].get("description", "")
|
||
if desc:
|
||
message = f"{message}\n{desc}"
|
||
self._get_ntfy().notify(event, title, message, job_ntfy_enabled=ntfy_enabled, priority=priority)
|
||
|
||
def _log(self, level: str, message: str, job_name: str = None):
|
||
py_level = logging.WARNING if level == "WARN" else getattr(logging, level, logging.INFO)
|
||
logger.log(py_level, message)
|
||
session = get_session()
|
||
try:
|
||
entry = LogEntry(level=level, message=message, job_name=job_name)
|
||
session.add(entry)
|
||
session.commit()
|
||
except Exception as e:
|
||
session.rollback()
|
||
logger.error("Log-Eintrag konnte nicht gespeichert werden: %s", e)
|
||
finally:
|
||
session.close()
|
||
|
||
def _download_path(self) -> str:
|
||
configured = self.config.get("recording", "download_path", default=DEFAULT_DOWNLOAD_PATH)
|
||
try:
|
||
return validate_download_path(configured)
|
||
except ValueError as e:
|
||
self._log("ERROR", f"Ungültiger Download-Pfad '{configured}': {e} – verwende {DEFAULT_DOWNLOAD_PATH}")
|
||
return DEFAULT_DOWNLOAD_PATH
|
||
|
||
def start_recording(
|
||
self,
|
||
job_id: Optional[int],
|
||
name: str,
|
||
stream_url: str,
|
||
stream_type: str = "auto",
|
||
output_format: str = "",
|
||
max_duration: Optional[int] = None,
|
||
extra_ffmpeg_args: str = "",
|
||
segment_duration: Optional[int] = None,
|
||
is_scheduled: bool = False,
|
||
description: str = "",
|
||
ntfy_enabled: bool = False,
|
||
plik_enabled: bool = False,
|
||
delete_after_upload: bool = False,
|
||
) -> dict:
|
||
stream_url = validate_stream_url(stream_url)
|
||
validate_ffmpeg_args(extra_ffmpeg_args or "")
|
||
|
||
if not stream_type or stream_type == "auto":
|
||
stream_type = detect_stream_type(stream_url)
|
||
|
||
ext = output_format if output_format and output_format.lower() != "auto" else default_extension(stream_type)
|
||
sname = safe_name(name)
|
||
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
download_path = self._download_path()
|
||
output_dir = os.path.join(download_path, sname)
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
segment_mode = segment_duration is not None and segment_duration > 0
|
||
|
||
if segment_mode:
|
||
output_file = os.path.join(output_dir, f"{sname}_{timestamp}_seg%03d.{ext}")
|
||
else:
|
||
output_file = os.path.join(output_dir, f"{sname}_{timestamp}.{ext}")
|
||
|
||
scheduled_stop = None
|
||
if max_duration and max_duration > 0:
|
||
scheduled_stop = utcnow() + datetime.timedelta(seconds=max_duration)
|
||
|
||
session = get_session()
|
||
try:
|
||
recording = Recording(
|
||
job_id=job_id,
|
||
job_name=name,
|
||
stream_url=stream_url,
|
||
stream_type=stream_type,
|
||
output_dir=output_dir,
|
||
output_file=output_file,
|
||
status="starting",
|
||
scheduled_stop=scheduled_stop,
|
||
segment_mode=segment_mode,
|
||
is_scheduled=is_scheduled,
|
||
)
|
||
session.add(recording)
|
||
session.commit()
|
||
rec_id = recording.id
|
||
finally:
|
||
session.close()
|
||
|
||
self._rec_options[rec_id] = {
|
||
"ntfy_enabled": ntfy_enabled,
|
||
"plik_enabled": plik_enabled,
|
||
"delete_after_upload": delete_after_upload,
|
||
"segment_mode": segment_mode,
|
||
"description": description,
|
||
}
|
||
self._output_files[rec_id] = [output_file]
|
||
|
||
stop_event = threading.Event()
|
||
self._stop_events[rec_id] = stop_event
|
||
|
||
thread = threading.Thread(
|
||
target=self._recording_loop,
|
||
args=(rec_id, stream_url, stream_type, output_file, output_dir,
|
||
extra_ffmpeg_args, segment_mode, segment_duration, stop_event, name),
|
||
daemon=True,
|
||
)
|
||
self._threads[rec_id] = thread
|
||
thread.start()
|
||
|
||
self._log("INFO", f"Aufnahme gestartet: {name}", name)
|
||
self._notify("start", "Aufnahme gestartet", name, rec_id)
|
||
|
||
return {"recording_id": rec_id, "output_dir": output_dir, "output_file": output_file}
|
||
|
||
def _recording_loop(
|
||
self,
|
||
rec_id: int,
|
||
stream_url: str,
|
||
stream_type: str,
|
||
output_file: str,
|
||
output_dir: str,
|
||
extra_ffmpeg_args: str,
|
||
segment_mode: bool,
|
||
segment_duration: Optional[int],
|
||
stop_event: threading.Event,
|
||
job_name: str,
|
||
):
|
||
max_retries = self.config.get("recording", "max_retries", default=5)
|
||
retry_delay = self.config.get("recording", "retry_delay", default=5)
|
||
retry = 0
|
||
|
||
try:
|
||
while not stop_event.is_set():
|
||
log_fh = None
|
||
try:
|
||
cmd = self._build_ffmpeg_cmd(
|
||
stream_url, stream_type, output_file,
|
||
extra_ffmpeg_args, segment_mode, segment_duration,
|
||
)
|
||
log_path = os.path.join(output_dir, f".ffmpeg_{rec_id}.log")
|
||
log_fh = open(log_path, "a")
|
||
process = subprocess.Popen(
|
||
cmd,
|
||
stdout=log_fh,
|
||
stderr=log_fh,
|
||
stdin=subprocess.DEVNULL,
|
||
)
|
||
except Exception as e:
|
||
if log_fh:
|
||
log_fh.close()
|
||
self._update_recording(rec_id, status="error", error_message=str(e),
|
||
file_size=self._finalize_file_size(rec_id, segment_mode))
|
||
self._log("ERROR", f"ffmpeg konnte nicht gestartet werden: {e}", job_name)
|
||
self._notify("error", "Aufnahme fehlgeschlagen", f"{job_name}: {e}", rec_id, priority="high")
|
||
return
|
||
|
||
self._processes[rec_id] = process
|
||
self._update_recording(rec_id, status="running", pid=process.pid)
|
||
proc_started = time.monotonic()
|
||
|
||
process.wait()
|
||
retcode = process.returncode
|
||
log_fh.close()
|
||
self._processes.pop(rec_id, None)
|
||
|
||
if stop_event.is_set() or retcode == 0:
|
||
status = "completed" if retcode == 0 else "stopped"
|
||
status_de = "abgeschlossen" if status == "completed" else "gestoppt"
|
||
self._update_recording(rec_id, status=status,
|
||
file_size=self._finalize_file_size(rec_id, segment_mode))
|
||
self._log("INFO", f"Aufnahme {status_de}: {job_name}", job_name)
|
||
self._notify("stop", f"Aufnahme {status_de}", job_name, rec_id)
|
||
self._handle_post_recording(rec_id, output_dir, output_file, job_name, segment_mode)
|
||
return
|
||
|
||
if retcode in (-2, 130, 143, 255):
|
||
self._update_recording(rec_id, status="stopped",
|
||
file_size=self._finalize_file_size(rec_id, segment_mode))
|
||
self._log("INFO", f"Aufnahme gestoppt: {job_name}", job_name)
|
||
self._notify("stop", "Aufnahme gestoppt", job_name, rec_id)
|
||
self._handle_post_recording(rec_id, output_dir, output_file, job_name, segment_mode)
|
||
return
|
||
|
||
if time.monotonic() - proc_started >= MIN_STABLE_RUNTIME:
|
||
retry = 0
|
||
retry += 1
|
||
if max_retries > 0 and retry >= max_retries:
|
||
self._update_recording(rec_id, status="error",
|
||
error_message=f"Max retries ({max_retries}) reached, exit code {retcode}",
|
||
file_size=self._finalize_file_size(rec_id, segment_mode))
|
||
self._log("ERROR", f"Max. Versuche ({max_retries}) erreicht: {job_name}", job_name)
|
||
self._notify("error", "Aufnahme fehlgeschlagen", f"{job_name}: Max. Versuche erreicht", rec_id, priority="high")
|
||
return
|
||
|
||
self._log("WARN", f"Verbindung verloren (Code: {retcode}), Reconnect in {retry_delay}s... [{job_name}]", job_name)
|
||
self._notify("error", "Stream abgerissen", f"{job_name}: Reconnect in {retry_delay}s...", rec_id)
|
||
if stop_event.wait(timeout=retry_delay):
|
||
self._update_recording(rec_id, status="stopped",
|
||
file_size=self._finalize_file_size(rec_id, segment_mode))
|
||
self._log("INFO", f"Aufnahme gestoppt: {job_name}", job_name)
|
||
self._notify("stop", "Aufnahme gestoppt", job_name, rec_id)
|
||
self._handle_post_recording(rec_id, output_dir, output_file, job_name, segment_mode)
|
||
return
|
||
|
||
sname = safe_name(job_name)
|
||
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
ext = Path(output_file).suffix.lstrip(".")
|
||
if segment_mode:
|
||
output_file = os.path.join(output_dir, f"{sname}_{ts}_seg%03d.{ext}")
|
||
else:
|
||
output_file = os.path.join(output_dir, f"{sname}_{ts}.{ext}")
|
||
self._output_files.setdefault(rec_id, []).append(output_file)
|
||
|
||
self._update_recording(rec_id, status="stopped",
|
||
file_size=self._finalize_file_size(rec_id, segment_mode))
|
||
self._log("INFO", f"Aufnahme gestoppt: {job_name}", job_name)
|
||
self._notify("stop", "Aufnahme gestoppt", job_name, rec_id)
|
||
self._handle_post_recording(rec_id, output_dir, output_file, job_name, segment_mode)
|
||
finally:
|
||
self._cleanup(rec_id)
|
||
|
||
def _build_ffmpeg_cmd(self, url, stream_type, output_file, extra_args, segment_mode, segment_duration):
|
||
url = validate_stream_url(url)
|
||
extra_tokens = validate_ffmpeg_args(extra_args or "")
|
||
|
||
cmd = ["ffmpeg", "-nostdin", "-y", "-hide_banner", "-loglevel", "warning",
|
||
"-protocol_whitelist", PROTOCOL_WHITELIST]
|
||
|
||
if stream_type == "hls":
|
||
cmd += ["-i", url, "-c", "copy", "-bsf:a", "aac_adtstoasc"]
|
||
else:
|
||
cmd += ["-i", url, "-c", "copy"]
|
||
|
||
if segment_mode and segment_duration:
|
||
cmd += ["-f", "segment", "-segment_time", str(segment_duration), "-reset_timestamps", "1"]
|
||
|
||
cmd += extra_tokens
|
||
|
||
cmd.append(output_file)
|
||
return cmd
|
||
|
||
def stop_recording(self, rec_id: int) -> bool:
|
||
stop_event = self._stop_events.get(rec_id)
|
||
if stop_event:
|
||
stop_event.set()
|
||
|
||
process = self._processes.get(rec_id)
|
||
if process:
|
||
try:
|
||
process.send_signal(signal.SIGINT)
|
||
try:
|
||
process.wait(timeout=10)
|
||
except subprocess.TimeoutExpired:
|
||
process.kill()
|
||
except (ProcessLookupError, OSError):
|
||
pass
|
||
|
||
if not process and rec_id not in self._threads:
|
||
self._update_recording(rec_id, status="stopped")
|
||
self._log("INFO", f"Aufnahme gestoppt (ID: {rec_id})")
|
||
self._cleanup(rec_id)
|
||
return True
|
||
|
||
def extend_recording(self, rec_id: int, extra_seconds: int) -> str:
|
||
"""Ergebnis: "ok", "not_found", "not_running" oder "unlimited"."""
|
||
session = get_session()
|
||
try:
|
||
rec = session.query(Recording).filter_by(id=rec_id).first()
|
||
if not rec:
|
||
return "not_found"
|
||
if rec.status != "running":
|
||
return "not_running"
|
||
if not rec.scheduled_stop:
|
||
return "unlimited"
|
||
|
||
rec.scheduled_stop = rec.scheduled_stop + datetime.timedelta(seconds=extra_seconds)
|
||
job_name = rec.job_name
|
||
session.commit()
|
||
finally:
|
||
session.close()
|
||
self._log("INFO", f"Aufnahme verlängert um {format_elapsed(extra_seconds)}: {job_name}", job_name)
|
||
return "ok"
|
||
|
||
def remove_duration_limit(self, rec_id: int) -> bool:
|
||
session = get_session()
|
||
try:
|
||
rec = session.query(Recording).filter_by(id=rec_id).first()
|
||
if not rec or rec.status != "running":
|
||
return False
|
||
rec.scheduled_stop = None
|
||
session.commit()
|
||
self._log("INFO", f"Zeitlimit entfernt: {rec.job_name}", rec.job_name)
|
||
return True
|
||
finally:
|
||
session.close()
|
||
|
||
def get_status(self, rec_id: int) -> Optional[dict]:
|
||
session = get_session()
|
||
try:
|
||
rec = session.query(Recording).filter_by(id=rec_id).first()
|
||
if not rec:
|
||
return None
|
||
return self._recording_to_dict(rec)
|
||
finally:
|
||
session.close()
|
||
|
||
def get_all_active(self) -> list[dict]:
|
||
session = get_session()
|
||
try:
|
||
recs = session.query(Recording).filter(Recording.status.in_(["running", "starting"])).all()
|
||
return [self._recording_to_dict(r) for r in recs]
|
||
finally:
|
||
session.close()
|
||
|
||
def get_all_recordings(self, limit: int = 50) -> list[dict]:
|
||
session = get_session()
|
||
try:
|
||
recs = session.query(Recording).order_by(Recording.id.desc()).limit(limit).all()
|
||
return [self._recording_to_dict(r) for r in recs]
|
||
finally:
|
||
session.close()
|
||
|
||
def check_scheduled_stops(self):
|
||
session = get_session()
|
||
try:
|
||
now = utcnow()
|
||
recs = session.query(Recording).filter(
|
||
Recording.status == "running",
|
||
Recording.scheduled_stop.isnot(None),
|
||
Recording.scheduled_stop <= now,
|
||
).all()
|
||
rec_ids = [r.id for r in recs]
|
||
finally:
|
||
session.close()
|
||
|
||
for rec_id in rec_ids:
|
||
self._log("INFO", f"Geplanter Stopp erreicht (ID: {rec_id})")
|
||
self.stop_recording(rec_id)
|
||
|
||
def cleanup_stale(self):
|
||
# Nach einem Neustart kann kein ffmpeg-Kindprozess mehr laufen
|
||
session = get_session()
|
||
try:
|
||
recs = session.query(Recording).filter(Recording.status.in_(["running", "starting"])).all()
|
||
now = utcnow()
|
||
for rec in recs:
|
||
rec.status = "error"
|
||
rec.error_message = STALE_ERROR_MESSAGE
|
||
rec.stopped_at = now
|
||
try:
|
||
files = [rec.output_file] if rec.output_file else []
|
||
rec.file_size, _ = self._calc_recording_size(files, bool(rec.segment_mode))
|
||
except Exception:
|
||
pass
|
||
session.commit()
|
||
if recs:
|
||
logger.warning("%d verwaiste Aufnahme(n) als abgebrochen markiert", len(recs))
|
||
finally:
|
||
session.close()
|
||
|
||
def _recording_to_dict(self, rec: Recording) -> dict:
|
||
elapsed = 0
|
||
if rec.started_at:
|
||
end = rec.stopped_at or utcnow()
|
||
elapsed = int((end - rec.started_at).total_seconds())
|
||
|
||
total_size = 0
|
||
segment_count = 0
|
||
if rec.file_size is not None:
|
||
total_size = rec.file_size
|
||
else:
|
||
files = self._output_files.get(rec.id, [rec.output_file] if rec.output_file else [])
|
||
total_size, segment_count = self._calc_recording_size(files, rec.segment_mode)
|
||
|
||
remaining = None
|
||
if rec.scheduled_stop and rec.status == "running":
|
||
diff = (rec.scheduled_stop - utcnow()).total_seconds()
|
||
remaining = max(0, int(diff))
|
||
|
||
return {
|
||
"id": rec.id,
|
||
"job_id": rec.job_id,
|
||
"job_name": rec.job_name,
|
||
"stream_url": rec.stream_url,
|
||
"stream_type": rec.stream_type,
|
||
"output_dir": rec.output_dir,
|
||
"output_file": rec.output_file,
|
||
"pid": rec.pid,
|
||
"status": rec.status,
|
||
"started_at": iso_utc(rec.started_at),
|
||
"stopped_at": iso_utc(rec.stopped_at),
|
||
"scheduled_stop": iso_utc(rec.scheduled_stop),
|
||
"segment_mode": rec.segment_mode,
|
||
"is_scheduled": rec.is_scheduled,
|
||
"error_message": rec.error_message,
|
||
"elapsed": elapsed,
|
||
"elapsed_display": format_elapsed(elapsed),
|
||
"total_size": total_size,
|
||
"total_size_display": format_size(total_size),
|
||
"segment_count": segment_count if rec.segment_mode else None,
|
||
"remaining": remaining,
|
||
"remaining_display": format_elapsed(remaining) if remaining is not None else None,
|
||
"plik_url": rec.plik_url,
|
||
}
|
||
|
||
@staticmethod
|
||
def _segment_files(pattern: str) -> list[str]:
|
||
glob_pattern = globmod.escape(pattern).replace("%03d", "*")
|
||
return sorted(
|
||
f for f in globmod.glob(glob_pattern)
|
||
if os.path.isfile(f) and not os.path.basename(f).startswith(".")
|
||
)
|
||
|
||
def _resolve_output_files(self, patterns: list[str], segment_mode: bool) -> list[str]:
|
||
files = []
|
||
for of in patterns:
|
||
if segment_mode:
|
||
files.extend(self._segment_files(of))
|
||
elif os.path.isfile(of) and os.path.getsize(of) > 0:
|
||
files.append(of)
|
||
return [f for f in files if not os.path.basename(f).startswith(".")]
|
||
|
||
def _calc_recording_size(self, output_files: list[str], segment_mode: bool) -> tuple[int, int]:
|
||
total_size = 0
|
||
segment_count = 0
|
||
for of in output_files:
|
||
if segment_mode:
|
||
for f in self._segment_files(of):
|
||
total_size += os.path.getsize(f)
|
||
segment_count += 1
|
||
else:
|
||
if os.path.isfile(of):
|
||
total_size += os.path.getsize(of)
|
||
segment_count += 1
|
||
return total_size, segment_count
|
||
|
||
def _finalize_file_size(self, rec_id: int, segment_mode: bool) -> int:
|
||
files = self._output_files.get(rec_id, [])
|
||
if not files:
|
||
return 0
|
||
total_size, _ = self._calc_recording_size(files, segment_mode)
|
||
return total_size
|
||
|
||
def _update_recording(self, rec_id: int, **kwargs):
|
||
session = get_session()
|
||
try:
|
||
rec = session.query(Recording).filter_by(id=rec_id).first()
|
||
if rec:
|
||
for k, v in kwargs.items():
|
||
setattr(rec, k, v)
|
||
if kwargs.get("status") in ("stopped", "completed", "error"):
|
||
rec.stopped_at = utcnow()
|
||
session.commit()
|
||
finally:
|
||
session.close()
|
||
|
||
def _handle_post_recording(self, rec_id: int, output_dir: str, output_file: str,
|
||
job_name: str, segment_mode: bool):
|
||
opts = self._rec_options.get(rec_id, {})
|
||
if opts.get("plik_enabled"):
|
||
patterns = list(self._output_files.get(rec_id, [output_file]))
|
||
files = self._resolve_output_files(patterns, segment_mode)
|
||
self._get_plik().upload_recording(
|
||
files=files,
|
||
job_name=job_name,
|
||
segment_mode=segment_mode,
|
||
delete_after=opts.get("delete_after_upload", False),
|
||
ntfy_enabled=opts.get("ntfy_enabled", False),
|
||
save_url_fn=lambda url: self._update_recording(rec_id, plik_url=url),
|
||
)
|
||
|
||
def _cleanup(self, rec_id: int):
|
||
self._processes.pop(rec_id, None)
|
||
self._stop_events.pop(rec_id, None)
|
||
self._threads.pop(rec_id, None)
|
||
self._rec_options.pop(rec_id, None)
|
||
self._output_files.pop(rec_id, None)
|