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)