diff --git a/bin/backend/app.py b/bin/backend/app.py new file mode 100644 index 0000000..1245bbc --- /dev/null +++ b/bin/backend/app.py @@ -0,0 +1,369 @@ +import base64 +import contextlib +import datetime +import os +import secrets +import sys + +from fastapi import FastAPI, HTTPException, Request, Query +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import HTMLResponse, JSONResponse +from fastapi.staticfiles import StaticFiles +from pydantic import BaseModel +from typing import Optional + +from config import Config +from models import init_db, get_session, Job, Recording as RecordingModel, LogEntry +from recorder import RecordingManager, format_elapsed +from scheduler import StreamScheduler + +DATA_DIR = os.environ.get("DATA_DIR", "/app/data") +STATIC_DIR = os.environ.get("STATIC_DIR", "/app/frontend/static") + +os.makedirs(DATA_DIR, exist_ok=True) + +config = Config(os.path.join(DATA_DIR, "config.yml")) +init_db(os.path.join(DATA_DIR, "stream-recorder.db")) + +recorder = RecordingManager(config) +scheduler = StreamScheduler(recorder) + + +@contextlib.asynccontextmanager +async def lifespan(app): + recorder.cleanup_stale() + scheduler.start() + yield + scheduler.stop() + + +app = FastAPI(title="stream-recorder", version="3.0.0", lifespan=lifespan) + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_methods=["*"], + allow_headers=["*"], +) + + +@app.middleware("http") +async def auth_middleware(request: Request, call_next): + if not config.get("auth", "enabled", default=False): + return await call_next(request) + + if request.url.path.startswith("/api/health"): + return await call_next(request) + + auth_header = request.headers.get("Authorization", "") + if auth_header.startswith("Basic "): + try: + decoded = base64.b64decode(auth_header[6:]).decode("utf-8") + username, password = decoded.split(":", 1) + expected_user = config.get("auth", "username", default="admin") + expected_pass = config.get("auth", "password", default="stream-recorder") + if secrets.compare_digest(username, expected_user) and secrets.compare_digest(password, expected_pass): + return await call_next(request) + except Exception: + pass + + return JSONResponse( + status_code=401, + content={"detail": "Unauthorized"}, + headers={"WWW-Authenticate": 'Basic realm="stream-recorder"'}, + ) + + +# --- Health --- + +@app.get("/api/health") +async def health(): + return {"status": "ok", "version": "3.0.0"} + + +# --- Jobs CRUD --- + +class JobCreate(BaseModel): + 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 + schedule_enabled: bool = False + schedule_once: bool = False + schedule_date: str = "" + schedule_days: str = "*" + schedule_start: str = "" + schedule_stop: str = "" + ntfy_enabled: bool = False + plik_enabled: bool = False + delete_after_upload: bool = False + metadata_monitor_enabled: bool = False + metadata_pattern: str = "" + metadata_grace_period: int = 300 + metadata_poll_interval: int = 30 + + +class JobUpdate(JobCreate): + pass + + +@app.get("/api/jobs") +async def list_jobs(): + session = get_session() + try: + jobs = session.query(Job).order_by(Job.id).all() + result = [] + for j in jobs: + is_recording = session.query( + session.query(RecordingModel).filter_by(job_id=j.id, status="running").exists() + ).scalar() + + result.append({ + "id": j.id, + "name": j.name, + "stream_url": j.stream_url, + "stream_type": j.stream_type, + "output_format": j.output_format, + "max_duration": j.max_duration, + "extra_ffmpeg_args": j.extra_ffmpeg_args, + "segment_duration": j.segment_duration, + "schedule_enabled": j.schedule_enabled, + "schedule_once": j.schedule_once, + "schedule_date": j.schedule_date, + "schedule_days": j.schedule_days, + "schedule_start": j.schedule_start, + "schedule_stop": j.schedule_stop, + "ntfy_enabled": j.ntfy_enabled, + "plik_enabled": j.plik_enabled, + "delete_after_upload": j.delete_after_upload, + "metadata_monitor_enabled": j.metadata_monitor_enabled, + "metadata_pattern": j.metadata_pattern, + "metadata_grace_period": j.metadata_grace_period, + "metadata_poll_interval": j.metadata_poll_interval, + "is_recording": is_recording, + "created_at": j.created_at.isoformat() if j.created_at else None, + }) + return result + finally: + session.close() + + +@app.post("/api/jobs") +async def create_job(data: JobCreate): + session = get_session() + try: + job = Job(**data.model_dump()) + session.add(job) + session.commit() + return {"id": job.id, "message": "Job erstellt"} + finally: + session.close() + + +@app.get("/api/jobs/{job_id}") +async def get_job(job_id: int): + session = get_session() + try: + job = session.query(Job).filter_by(id=job_id).first() + if not job: + raise HTTPException(404, "Job nicht gefunden") + return { + "id": job.id, "name": job.name, "stream_url": job.stream_url, + "stream_type": job.stream_type, "output_format": job.output_format, + "max_duration": job.max_duration, "extra_ffmpeg_args": job.extra_ffmpeg_args, + "segment_duration": job.segment_duration, + "schedule_enabled": job.schedule_enabled, "schedule_once": job.schedule_once, + "schedule_date": job.schedule_date, "schedule_days": job.schedule_days, + "schedule_start": job.schedule_start, "schedule_stop": job.schedule_stop, + "ntfy_enabled": job.ntfy_enabled, + "plik_enabled": job.plik_enabled, "delete_after_upload": job.delete_after_upload, + "metadata_monitor_enabled": job.metadata_monitor_enabled, + "metadata_pattern": job.metadata_pattern, + "metadata_grace_period": job.metadata_grace_period, + "metadata_poll_interval": job.metadata_poll_interval, + } + finally: + session.close() + + +@app.put("/api/jobs/{job_id}") +async def update_job(job_id: int, data: JobUpdate): + session = get_session() + try: + job = session.query(Job).filter_by(id=job_id).first() + if not job: + raise HTTPException(404, "Job nicht gefunden") + for k, v in data.model_dump().items(): + setattr(job, k, v) + job.updated_at = datetime.datetime.utcnow() + session.commit() + return {"message": "Job aktualisiert"} + finally: + session.close() + + +@app.delete("/api/jobs/{job_id}") +async def delete_job(job_id: int): + session = get_session() + try: + job = session.query(Job).filter_by(id=job_id).first() + if not job: + raise HTTPException(404, "Job nicht gefunden") + session.delete(job) + session.commit() + return {"message": "Job gelöscht"} + finally: + session.close() + + +@app.post("/api/jobs/{job_id}/record") +async def start_job_recording(job_id: int): + session = get_session() + try: + job = session.query(Job).filter_by(id=job_id).first() + if not job: + raise HTTPException(404, "Job nicht gefunden") + job_data = { + "name": job.name, "stream_url": job.stream_url, + "stream_type": job.stream_type, "output_format": job.output_format, + "max_duration": job.max_duration, "extra_ffmpeg_args": job.extra_ffmpeg_args, + "segment_duration": job.segment_duration, + "metadata_monitor_enabled": job.metadata_monitor_enabled, + "metadata_pattern": job.metadata_pattern, + "metadata_grace_period": job.metadata_grace_period or 300, + "metadata_poll_interval": job.metadata_poll_interval or 30, + "ntfy_enabled": job.ntfy_enabled, + "plik_enabled": job.plik_enabled, + "delete_after_upload": job.delete_after_upload, + } + finally: + session.close() + + result = recorder.start_recording(job_id=job_id, **job_data) + return result + + +# --- Quick Recording --- + +class QuickRecordRequest(BaseModel): + url: str + name: str = "" + stream_type: str = "auto" + max_duration: Optional[int] = None + + +@app.post("/api/quick-record") +async def quick_record(data: QuickRecordRequest): + name = data.name or f"Schnellaufnahme_{datetime.datetime.now().strftime('%H%M%S')}" + result = recorder.start_recording( + job_id=None, + name=name, + stream_url=data.url, + stream_type=data.stream_type, + max_duration=data.max_duration, + ) + return result + + +# --- Recordings --- + +@app.get("/api/recordings") +async def list_recordings(active_only: bool = False, limit: int = 50): + if active_only: + return recorder.get_all_active() + return recorder.get_all_recordings(limit=limit) + + +@app.get("/api/recordings/{rec_id}") +async def get_recording(rec_id: int): + status = recorder.get_status(rec_id) + if not status: + raise HTTPException(404, "Aufnahme nicht gefunden") + return status + + +@app.post("/api/recordings/{rec_id}/stop") +async def stop_recording(rec_id: int): + recorder.stop_recording(rec_id) + return {"message": "Aufnahme gestoppt"} + + +@app.post("/api/recordings/{rec_id}/extend") +async def extend_recording(rec_id: int, minutes: int = Query(default=30)): + ok = recorder.extend_recording(rec_id, minutes * 60) + if not ok: + raise HTTPException(400, "Aufnahme kann nicht verlängert werden") + return {"message": f"Aufnahme um {minutes} Minuten verlängert"} + + +@app.post("/api/recordings/{rec_id}/unlimited") +async def remove_limit(rec_id: int): + ok = recorder.remove_duration_limit(rec_id) + if not ok: + raise HTTPException(400, "Zeitlimit kann nicht entfernt werden") + return {"message": "Zeitlimit entfernt"} + + +# --- Logs --- + +@app.get("/api/logs") +async def get_logs(limit: int = 100, level: str = ""): + session = get_session() + try: + q = session.query(LogEntry).order_by(LogEntry.id.desc()) + if level: + q = q.filter(LogEntry.level == level.upper()) + logs = q.limit(limit).all() + return [ + { + "id": l.id, + "timestamp": l.timestamp.isoformat() if l.timestamp else None, + "level": l.level, + "message": l.message, + "job_name": l.job_name, + } + for l in logs + ] + finally: + session.close() + + +# --- Config --- + +@app.get("/api/config") +async def get_config(): + safe_config = {**config.data} + if "auth" in safe_config: + safe_auth = {**safe_config["auth"]} + if safe_auth.get("password"): + safe_auth["password"] = "***" + safe_config["auth"] = safe_auth + return safe_config + + +class ConfigUpdate(BaseModel): + data: dict + + +@app.put("/api/config") +async def update_config(payload: ConfigUpdate): + from config import _deep_merge + _deep_merge(config.data, payload.data) + config.save() + return {"message": "Konfiguration gespeichert"} + + +# --- Static files (Frontend) --- + +if os.path.isdir(STATIC_DIR): + app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="static") + + +if __name__ == "__main__": + import uvicorn + host = config.get("server", "host", default="0.0.0.0") + port = config.get("server", "port", default=8484) + uvicorn.run("app:app", host=host, port=port, reload=False) diff --git a/bin/backend/config.py b/bin/backend/config.py new file mode 100644 index 0000000..3a5e881 --- /dev/null +++ b/bin/backend/config.py @@ -0,0 +1,78 @@ +import os +import yaml + +DEFAULT_CONFIG = { + "recording": { + "download_path": "/app/data/recordings", + "max_retries": 5, + "retry_delay": 5, + }, + "server": { + "host": "0.0.0.0", + "port": 8484, + }, + "auth": { + "enabled": False, + "username": "admin", + "password": "stream-recorder", + }, + "ntfy": { + "url": "", + "token": "", + "events": "error", + }, + "plik": { + "url": "", + "api_key": "", + "ttl": "30d", + }, + "logging": { + "level": "INFO", + }, +} + + +class Config: + def __init__(self, config_path: str = "/app/data/config.yml"): + self.config_path = config_path + self.data = {} + self.load() + + def load(self): + self.data = _deep_copy(DEFAULT_CONFIG) + if os.path.exists(self.config_path): + with open(self.config_path, "r") as f: + user_config = yaml.safe_load(f) or {} + _deep_merge(self.data, user_config) + else: + self.save() + + def save(self): + os.makedirs(os.path.dirname(self.config_path), exist_ok=True) + with open(self.config_path, "w") as f: + yaml.dump(self.data, f, default_flow_style=False, allow_unicode=True, sort_keys=False) + + def get(self, *keys, default=None): + val = self.data + for key in keys: + if isinstance(val, dict) and key in val: + val = val[key] + else: + return default + return val + + +def _deep_copy(d): + if isinstance(d, dict): + return {k: _deep_copy(v) for k, v in d.items()} + if isinstance(d, list): + return [_deep_copy(v) for v in d] + return d + + +def _deep_merge(base, override): + for key, value in override.items(): + if key in base and isinstance(base[key], dict) and isinstance(value, dict): + _deep_merge(base[key], value) + else: + base[key] = value diff --git a/bin/backend/models.py b/bin/backend/models.py new file mode 100644 index 0000000..4066909 --- /dev/null +++ b/bin/backend/models.py @@ -0,0 +1,83 @@ +import datetime +from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime, Text +from sqlalchemy.orm import declarative_base, sessionmaker + +Base = declarative_base() + + +class Job(Base): + __tablename__ = "jobs" + + id = Column(Integer, primary_key=True, autoincrement=True) + name = Column(String(200), nullable=False) + stream_url = Column(String(1000), nullable=False) + stream_type = Column(String(20), default="auto") + output_format = Column(String(10), default="") + max_duration = Column(Integer, nullable=True) + extra_ffmpeg_args = Column(String(500), default="") + segment_duration = Column(Integer, nullable=True) + + schedule_enabled = Column(Boolean, default=False) + schedule_once = Column(Boolean, default=False) + schedule_date = Column(String(10), default="") + schedule_days = Column(String(50), default="*") + schedule_start = Column(String(5), default="") + schedule_stop = Column(String(5), default="") + + plik_enabled = Column(Boolean, default=False) + delete_after_upload = Column(Boolean, default=False) + + ntfy_enabled = Column(Boolean, default=False) + + metadata_monitor_enabled = Column(Boolean, default=False) + metadata_pattern = Column(String(200), default="") + metadata_grace_period = Column(Integer, default=300) + metadata_poll_interval = Column(Integer, default=30) + + created_at = Column(DateTime, default=datetime.datetime.utcnow) + updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow) + + +class Recording(Base): + __tablename__ = "recordings" + + id = Column(Integer, primary_key=True, autoincrement=True) + job_id = Column(Integer, nullable=True) + job_name = Column(String(200), nullable=False) + stream_url = Column(String(1000), nullable=False) + stream_type = Column(String(20), default="") + output_dir = Column(String(500), default="") + output_file = Column(String(500), default="") + pid = Column(Integer, nullable=True) + status = Column(String(20), default="running") # running, stopped, completed, error + started_at = Column(DateTime, default=datetime.datetime.utcnow) + stopped_at = Column(DateTime, nullable=True) + scheduled_stop = Column(DateTime, nullable=True) + segment_mode = Column(Boolean, default=False) + is_scheduled = Column(Boolean, default=False) + error_message = Column(Text, nullable=True) + + +class LogEntry(Base): + __tablename__ = "logs" + + id = Column(Integer, primary_key=True, autoincrement=True) + timestamp = Column(DateTime, default=datetime.datetime.utcnow) + level = Column(String(10), default="INFO") + message = Column(Text, nullable=False) + job_name = Column(String(200), nullable=True) + + +_engine = None +_SessionLocal = None + + +def init_db(db_path: str): + global _engine, _SessionLocal + _engine = create_engine(f"sqlite:///{db_path}", connect_args={"check_same_thread": False}) + Base.metadata.create_all(_engine) + _SessionLocal = sessionmaker(bind=_engine) + + +def get_session(): + return _SessionLocal() diff --git a/bin/backend/plugins/__init__.py b/bin/backend/plugins/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/bin/backend/plugins/euer_radio.py b/bin/backend/plugins/euer_radio.py new file mode 100644 index 0000000..976f16f --- /dev/null +++ b/bin/backend/plugins/euer_radio.py @@ -0,0 +1,138 @@ +""" +Euer-Radio Metadata-Monitor Plugin. + +Polls RTMP stream metadata via ffprobe to detect whether a specific show +is currently live. When the show ends (pattern disappears from metadata), +the recording is stopped after a configurable grace period. +""" + +import json +import subprocess +import threading +import time +from typing import Optional, Callable + + +def probe_stream_title(stream_url: str, timeout: int = 10) -> Optional[str]: + """Query the current title tag from a live stream via ffprobe.""" + cmd = [ + "ffprobe", "-v", "quiet", + "-print_format", "json", + "-show_entries", "format_tags", + stream_url, + ] + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout, + stdin=subprocess.DEVNULL, + ) + if result.returncode != 0: + return None + data = json.loads(result.stdout) + return data.get("format", {}).get("tags", {}).get("title") + except (subprocess.TimeoutExpired, json.JSONDecodeError, Exception): + return None + + +def title_matches_pattern(title: str, pattern: str) -> bool: + """ + Check whether a stream title matches the configured show pattern. + + TODO: Finalize once we've seen real live-show metadata from euer.tv. + Current assumption: the title contains the pattern string as a prefix + or substring, e.g. "Donnerwolke live - Artist - Song". + """ + if not title or not pattern: + return False + return pattern.lower() in title.lower() + + +class MetadataMonitor: + """Monitors a stream's metadata and stops the recording when a show ends.""" + + def __init__(self, log_fn: Callable, stop_fn: Callable): + self._log = log_fn + self._stop_recording = stop_fn + self._monitors: dict[int, threading.Thread] = {} + self._cancel_events: dict[int, threading.Event] = {} + + def start_monitoring( + self, + rec_id: int, + stream_url: str, + pattern: str, + grace_period: int = 300, + poll_interval: int = 30, + job_name: str = "", + ): + if rec_id in self._monitors: + return + + cancel = threading.Event() + self._cancel_events[rec_id] = cancel + + thread = threading.Thread( + target=self._monitor_loop, + args=(rec_id, stream_url, pattern, grace_period, poll_interval, cancel, job_name), + daemon=True, + ) + self._monitors[rec_id] = thread + thread.start() + self._log("INFO", f"Metadata-Monitor gestartet für '{job_name}' (Pattern: '{pattern}')", job_name) + + def stop_monitoring(self, rec_id: int): + cancel = self._cancel_events.pop(rec_id, None) + if cancel: + cancel.set() + self._monitors.pop(rec_id, None) + + def is_monitoring(self, rec_id: int) -> bool: + return rec_id in self._monitors + + def _monitor_loop( + self, + rec_id: int, + stream_url: str, + pattern: str, + grace_period: int, + poll_interval: int, + cancel: threading.Event, + job_name: str, + ): + grace_start = None + + while not cancel.is_set(): + title = probe_stream_title(stream_url) + + if title is None: + if cancel.wait(timeout=poll_interval): + break + continue + + if title_matches_pattern(title, pattern): + if grace_start is not None: + self._log("INFO", + f"Show '{pattern}' wieder erkannt, Karenz abgebrochen [{job_name}]", + job_name) + grace_start = None + else: + if grace_start is None: + grace_start = time.time() + self._log("INFO", + f"Show '{pattern}' nicht mehr erkannt, Karenzzeit läuft ({grace_period}s) [{job_name}]", + job_name) + elif time.time() - grace_start >= grace_period: + self._log("INFO", + f"Karenzzeit abgelaufen, Aufnahme wird gestoppt [{job_name}]", + job_name) + self._stop_recording(rec_id) + break + + if cancel.wait(timeout=poll_interval): + break + + self._monitors.pop(rec_id, None) + self._cancel_events.pop(rec_id, None) diff --git a/bin/backend/plugins/ntfy.py b/bin/backend/plugins/ntfy.py new file mode 100644 index 0000000..839d95a --- /dev/null +++ b/bin/backend/plugins/ntfy.py @@ -0,0 +1,65 @@ +"""NTFY push notification plugin.""" + +import subprocess +import threading + +EVENT_TAGS = { + "start": "red_circle", + "stop": "white_check_mark", + "error": "rotating_light", + "upload": "outbox_tray", +} + + +def send_notification( + url: str, + title: str, + message: str, + token: str = "", + priority: str = "default", + tags: str = "", +): + if not url: + return + + cmd = [ + "curl", "-s", "-o", "/dev/null", "--max-time", "10", + "-H", f"Title: {title}", + "-H", f"Priority: {priority}", + ] + if tags: + cmd += ["-H", f"Tags: {tags}"] + cmd += ["-d", message] + if token: + cmd += ["-H", f"Authorization: Bearer {token}"] + cmd.append(url) + + def _send(): + try: + subprocess.run(cmd, capture_output=True, timeout=15, stdin=subprocess.DEVNULL) + except Exception: + pass + + threading.Thread(target=_send, daemon=True).start() + + +class NtfyNotifier: + def __init__(self, config): + self.config = config + + def _should_notify(self, event: str, job_ntfy_enabled: bool) -> bool: + if not job_ntfy_enabled: + return False + url = self.config.get("ntfy", "url", default="") + if not url: + return False + events = self.config.get("ntfy", "events", default="error") + return event in [e.strip() for e in events.split(",")] + + def notify(self, event: str, title: str, message: str, job_ntfy_enabled: bool = False, priority: str = "default"): + if not self._should_notify(event, job_ntfy_enabled): + return + url = self.config.get("ntfy", "url", default="") + token = self.config.get("ntfy", "token", default="") + tag = EVENT_TAGS.get(event, "") + send_notification(url, title, message, token=token, priority=priority, tags=tag) diff --git a/bin/backend/plugins/plik.py b/bin/backend/plugins/plik.py new file mode 100644 index 0000000..6097b41 --- /dev/null +++ b/bin/backend/plugins/plik.py @@ -0,0 +1,168 @@ +"""Plik file upload plugin.""" + +import json +import os +import re +import subprocess +import threading +from typing import Optional, Callable + + +def _parse_ttl_seconds(ttl: str) -> Optional[int]: + if not ttl: + return None + m = re.match(r"^(\d+)d$", ttl) + if m: + return int(m.group(1)) * 86400 + m = re.match(r"^(\d+)h$", ttl) + if m: + return int(m.group(1)) * 3600 + m = re.match(r"^(\d+)$", ttl) + if m: + return int(m.group(1)) + return None + + +def upload_file( + file_path: str, + plik_url: str, + api_key: str = "", + ttl: str = "30d", +) -> Optional[dict]: + """Upload a file to Plik. Returns dict with 'browser_url' and 'download_url', or None on failure.""" + if not plik_url or not os.path.isfile(file_path): + return None + if os.path.getsize(file_path) == 0: + return None + + plik_base = plik_url.rstrip("/") + filename = os.path.basename(file_path) + + cmd = ["curl", "-sS", "--max-time", "7200"] + if api_key: + cmd += ["-H", f"X-PlikToken: {api_key}"] + + cmd += ["-F", f"file=@{file_path};filename={filename}"] + + ttl_secs = _parse_ttl_seconds(ttl) + if ttl_secs is not None: + cmd += ["-F", f"ttl={ttl_secs}"] + + cmd.append(plik_base) + + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=7200, stdin=subprocess.DEVNULL) + if result.returncode != 0: + return None + + try: + data = json.loads(result.stdout) + upload_id = data.get("id", "") + if not upload_id: + return None + + browser_url = f"{plik_base}/#/?id={upload_id}" + + download_url = None + files = data.get("files", []) + if files: + file_id = files[0].get("id", "") + file_name = files[0].get("fileName", filename) + if file_id: + download_url = f"{plik_base}/file/{upload_id}/{file_id}/{file_name}" + + return {"browser_url": browser_url, "download_url": download_url} + except (json.JSONDecodeError, KeyError): + urls = re.findall(r'https?://\S+', result.stdout) + if urls: + download_url = urls[-1] + prefix = f"{plik_base}/file/" + if download_url.startswith(prefix): + rest = download_url[len(prefix):] + uid = rest.split("/")[0] + if uid: + return {"browser_url": f"{plik_base}/#/?id={uid}", "download_url": download_url} + return {"browser_url": download_url, "download_url": download_url} + return None + except Exception: + return None + + +class PlikUploader: + def __init__(self, config, log_fn: Callable, notify_fn: Callable = None): + self.config = config + self._log = log_fn + self._notify = notify_fn + + def upload_recording(self, output_dir: str, output_file: str, job_name: str, + segment_mode: bool, delete_after: bool, ntfy_enabled: bool): + plik_url = self.config.get("plik", "url", default="") + if not plik_url: + self._log("WARN", f"Plik-Upload übersprungen: Keine URL konfiguriert [{job_name}]", job_name) + return + + api_key = self.config.get("plik", "api_key", default="") + ttl = self.config.get("plik", "ttl", default="30d") + + def _do_upload(): + try: + if segment_mode: + self._upload_segments(output_dir, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled) + else: + self._upload_single(output_file, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled) + except Exception as e: + self._log("ERROR", f"Plik-Upload Fehler: {e} [{job_name}]", job_name) + + threading.Thread(target=_do_upload, daemon=True).start() + + def _upload_single(self, file_path, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled): + if not os.path.isfile(file_path) or os.path.getsize(file_path) == 0: + self._log("WARN", f"Plik-Upload übersprungen: Datei fehlt oder leer [{job_name}]", job_name) + return + + self._log("INFO", f"Plik-Upload: {os.path.basename(file_path)} [{job_name}]", job_name) + result = upload_file(file_path, plik_url, api_key, ttl) + + if result: + self._log("INFO", f"Plik-Upload abgeschlossen: {result['browser_url']} [{job_name}]", job_name) + if result.get("download_url"): + self._log("INFO", f"Plik-Download: {result['download_url']} [{job_name}]", job_name) + if self._notify: + self._notify("upload", "Plik-Upload", f"{job_name}: {result['browser_url']}", ntfy_enabled) + if delete_after: + os.remove(file_path) + self._log("INFO", f"Lokale Datei gelöscht: {file_path} [{job_name}]", job_name) + else: + self._log("ERROR", f"Plik-Upload fehlgeschlagen [{job_name}]", job_name) + + def _upload_segments(self, output_dir, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled): + if not os.path.isdir(output_dir): + return + + files = sorted(f for f in os.listdir(output_dir) if not f.startswith(".") and os.path.isfile(os.path.join(output_dir, f))) + if not files: + return + + uploaded = 0 + last_browser_url = None + for fname in files: + fpath = os.path.join(output_dir, fname) + if os.path.getsize(fpath) == 0: + continue + self._log("INFO", f"Plik Segment-Upload: {fname} [{job_name}]", job_name) + result = upload_file(fpath, plik_url, api_key, ttl) + if result: + last_browser_url = result["browser_url"] + self._log("INFO", f"Plik Segment hochgeladen: {last_browser_url} [{job_name}]", job_name) + uploaded += 1 + if delete_after: + os.remove(fpath) + self._log("INFO", f"Segment gelöscht: {fname} [{job_name}]", job_name) + else: + self._log("ERROR", f"Plik Segment-Upload fehlgeschlagen: {fname} [{job_name}]", job_name) + + if uploaded > 0 and self._notify: + msg = f"{job_name}: {uploaded} Segment(e) hochgeladen" + if last_browser_url: + msg += f"\n{last_browser_url}" + self._notify("upload", "Plik-Upload", msg, ntfy_enabled) diff --git a/bin/backend/recorder.py b/bin/backend/recorder.py new file mode 100644 index 0000000..08a2663 --- /dev/null +++ b/bin/backend/recorder.py @@ -0,0 +1,501 @@ +import asyncio +import datetime +import os +import re +import signal +import subprocess +import threading +import time +from pathlib import Path +from typing import Optional + +from models import Recording, get_session, LogEntry + + +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._metadata_monitor = None + self._ntfy = None + self._plik = None + self._rec_options: dict[int, dict] = {} + + def _get_metadata_monitor(self): + if self._metadata_monitor is None: + from plugins.euer_radio import MetadataMonitor + self._metadata_monitor = MetadataMonitor( + log_fn=self._log, + stop_fn=self.stop_recording, + ) + return self._metadata_monitor + + 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) + 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): + session = get_session() + try: + entry = LogEntry(level=level, message=message, job_name=job_name) + session.add(entry) + session.commit() + finally: + session.close() + + 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, + metadata_monitor_enabled: bool = False, + metadata_pattern: str = "", + metadata_grace_period: int = 300, + metadata_poll_interval: int = 30, + ntfy_enabled: bool = False, + plik_enabled: bool = False, + delete_after_upload: bool = False, + ) -> dict: + if 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.config.get("recording", "download_path", default="/app/data/recordings") + 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 = datetime.datetime.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() + + 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._rec_options[rec_id] = { + "ntfy_enabled": ntfy_enabled, + "plik_enabled": plik_enabled, + "delete_after_upload": delete_after_upload, + "segment_mode": segment_mode, + } + + self._log("INFO", f"Aufnahme gestartet: {name} [{stream_type}]", name) + self._notify("start", "Aufnahme gestartet", f"{name} [{stream_type}]", rec_id) + + if metadata_monitor_enabled and metadata_pattern: + self._get_metadata_monitor().start_monitoring( + rec_id=rec_id, + stream_url=stream_url, + pattern=metadata_pattern, + grace_period=metadata_grace_period, + poll_interval=metadata_poll_interval, + job_name=name, + ) + + 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 + + while not stop_event.is_set(): + cmd = self._build_ffmpeg_cmd( + stream_url, stream_type, output_file, + extra_ffmpeg_args, segment_mode, segment_duration, + ) + + try: + 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: + self._update_recording(rec_id, status="error", error_message=str(e)) + 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) + + 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" + self._update_recording(rec_id, status=status) + self._log("INFO", f"Aufnahme {status}: {job_name}", job_name) + self._notify("stop", f"Aufnahme {status}", 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") + 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 + + 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}") + 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") + 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._update_recording(rec_id, status="stopped") + + def _build_ffmpeg_cmd(self, url, stream_type, output_file, extra_args, segment_mode, segment_duration): + cmd = ["ffmpeg", "-nostdin", "-y", "-hide_banner", "-loglevel", "warning"] + + 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"] + + if extra_args: + cmd += extra_args.split() + + cmd.append(output_file) + return cmd + + def stop_recording(self, rec_id: int) -> bool: + if self._metadata_monitor: + self._metadata_monitor.stop_monitoring(rec_id) + + 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 + + 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) -> bool: + session = get_session() + try: + rec = session.query(Recording).filter_by(id=rec_id).first() + if not rec or rec.status != "running": + return False + + if rec.scheduled_stop: + rec.scheduled_stop = rec.scheduled_stop + datetime.timedelta(seconds=extra_seconds) + else: + rec.scheduled_stop = datetime.datetime.utcnow() + datetime.timedelta(seconds=extra_seconds) + + session.commit() + self._log("INFO", f"Aufnahme verlängert um {format_elapsed(extra_seconds)}: {rec.job_name}", rec.job_name) + return True + finally: + session.close() + + 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 = datetime.datetime.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): + session = get_session() + try: + recs = session.query(Recording).filter(Recording.status.in_(["running", "starting"])).all() + for rec in recs: + if rec.pid: + try: + os.kill(rec.pid, 0) + except (ProcessLookupError, OSError): + rec.status = "error" + rec.error_message = "Process vanished" + rec.stopped_at = datetime.datetime.utcnow() + session.commit() + finally: + session.close() + + def _recording_to_dict(self, rec: Recording) -> dict: + elapsed = 0 + if rec.started_at: + end = rec.stopped_at or datetime.datetime.utcnow() + elapsed = int((end - rec.started_at).total_seconds()) + + total_size = 0 + segment_count = 0 + if rec.output_dir and os.path.isdir(rec.output_dir): + for f in os.listdir(rec.output_dir): + if f.startswith("."): + continue + fp = os.path.join(rec.output_dir, f) + if os.path.isfile(fp): + total_size += os.path.getsize(fp) + segment_count += 1 + elif rec.output_file and os.path.isfile(rec.output_file): + total_size = os.path.getsize(rec.output_file) + + remaining = None + if rec.scheduled_stop and rec.status == "running": + diff = (rec.scheduled_stop - datetime.datetime.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": rec.started_at.isoformat() if rec.started_at else None, + "stopped_at": rec.stopped_at.isoformat() if rec.stopped_at else None, + "scheduled_stop": rec.scheduled_stop.isoformat() if rec.scheduled_stop else None, + "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, + } + + 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 = datetime.datetime.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"): + self._get_plik().upload_recording( + output_dir=output_dir, + output_file=output_file, + job_name=job_name, + segment_mode=segment_mode, + delete_after=opts.get("delete_after_upload", False), + ntfy_enabled=opts.get("ntfy_enabled", False), + ) + + 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) diff --git a/bin/backend/requirements.txt b/bin/backend/requirements.txt new file mode 100644 index 0000000..58675a2 --- /dev/null +++ b/bin/backend/requirements.txt @@ -0,0 +1,8 @@ +fastapi==0.115.12 +uvicorn[standard]==0.34.2 +sqlalchemy==2.0.40 +pydantic==2.11.3 +apscheduler==3.11.0 +pyyaml==6.0.2 +python-multipart==0.0.20 +httpx==0.28.1 diff --git a/bin/backend/scheduler.py b/bin/backend/scheduler.py new file mode 100644 index 0000000..f4b0886 --- /dev/null +++ b/bin/backend/scheduler.py @@ -0,0 +1,187 @@ +import datetime +from apscheduler.schedulers.background import BackgroundScheduler +from models import Job, Recording, get_session + +DAY_MAP = { + "mo": 1, "mon": 1, "montag": 1, + "di": 2, "tue": 2, "dienstag": 2, + "mi": 3, "wed": 3, "mittwoch": 3, + "do": 4, "thu": 4, "donnerstag": 4, + "fr": 5, "fri": 5, "freitag": 5, + "sa": 6, "sat": 6, "samstag": 6, + "so": 7, "sun": 7, "sonntag": 7, +} + + +def _day_to_num(day: str) -> int: + return DAY_MAP.get(day.lower().strip(), 0) + + +def _matches_day(today_num: int, schedule: str) -> bool: + if schedule.strip() == "*": + return True + for entry in schedule.split(","): + entry = entry.strip() + if "-" in entry: + parts = entry.split("-", 1) + start = _day_to_num(parts[0]) + end = _day_to_num(parts[1]) + if start <= end: + if start <= today_num <= end: + return True + else: + if today_num >= start or today_num <= end: + return True + else: + if _day_to_num(entry) == today_num: + return True + return False + + +def _time_to_minutes(t: str) -> int: + parts = t.split(":") + return int(parts[0]) * 60 + int(parts[1]) + + +def _in_time_window(now_time: str, start: str, stop: str) -> bool: + now_min = _time_to_minutes(now_time) + start_min = _time_to_minutes(start) + stop_min = _time_to_minutes(stop) + if start_min <= stop_min: + return start_min <= now_min < stop_min + else: + return now_min >= start_min or now_min < stop_min + + +class StreamScheduler: + def __init__(self, recorder): + self.recorder = recorder + self._scheduler = BackgroundScheduler() + self._started_keys: set[str] = set() + + def start(self): + self._scheduler.add_job(self._tick, "interval", seconds=30, id="scheduler_tick", + replace_existing=True, max_instances=1) + self._scheduler.add_job(self.recorder.check_scheduled_stops, "interval", seconds=10, + id="check_stops", replace_existing=True, max_instances=1) + self._scheduler.start() + + def stop(self): + self._scheduler.shutdown(wait=False) + + def _occurrence_key(self, job: Job) -> str: + today = datetime.date.today().isoformat() + if job.schedule_once: + today = job.schedule_date or today + return f"{job.id}|{job.schedule_once}|{today}|{job.schedule_start}" + + def _is_job_recording(self, job_id: int) -> bool: + session = get_session() + try: + return session.query(Recording).filter( + Recording.job_id == job_id, + Recording.status.in_(["running", "starting"]), + ).first() is not None + finally: + session.close() + + def _get_active_recording_for_job(self, job_id: int): + session = get_session() + try: + rec = session.query(Recording).filter( + Recording.job_id == job_id, + Recording.status.in_(["running", "starting"]), + ).first() + return rec.id if rec else None + finally: + session.close() + + def _tick(self): + now = datetime.datetime.now() + today_num = now.isoweekday() + today_date = now.strftime("%Y-%m-%d") + now_time = now.strftime("%H:%M") + + session = get_session() + try: + jobs = session.query(Job).filter_by(schedule_enabled=True).all() + job_list = [] + for j in jobs: + job_list.append({ + "id": j.id, "name": j.name, "stream_url": j.stream_url, + "stream_type": j.stream_type, "output_format": j.output_format, + "max_duration": j.max_duration, "extra_ffmpeg_args": j.extra_ffmpeg_args, + "segment_duration": j.segment_duration, + "schedule_once": j.schedule_once, "schedule_date": j.schedule_date, + "schedule_days": j.schedule_days, "schedule_start": j.schedule_start, + "schedule_stop": j.schedule_stop, + "metadata_monitor_enabled": j.metadata_monitor_enabled, + "metadata_pattern": j.metadata_pattern, + "metadata_grace_period": j.metadata_grace_period, + "metadata_poll_interval": j.metadata_poll_interval, + "ntfy_enabled": j.ntfy_enabled, + "plik_enabled": j.plik_enabled, + "delete_after_upload": j.delete_after_upload, + }) + finally: + session.close() + + for job in job_list: + if not job["schedule_start"]: + continue + + is_today = False + if job["schedule_once"]: + if job["schedule_date"] == today_date: + is_today = True + else: + if _matches_day(today_num, job["schedule_days"] or "*"): + is_today = True + + in_window = False + if is_today: + if job["schedule_stop"]: + in_window = _in_time_window(now_time, job["schedule_start"], job["schedule_stop"]) + else: + now_min = _time_to_minutes(now_time) + start_min = _time_to_minutes(job["schedule_start"]) + in_window = now_min >= start_min + + occ_key = f"{job['id']}|{job['schedule_once']}|{today_date if not job['schedule_once'] else job['schedule_date']}|{job['schedule_start']}" + + if in_window: + if occ_key not in self._started_keys and not self._is_job_recording(job["id"]): + duration = job["max_duration"] + if not duration and job["schedule_stop"]: + start_min = _time_to_minutes(job["schedule_start"]) + stop_min = _time_to_minutes(job["schedule_stop"]) + if stop_min > start_min: + duration = (stop_min - start_min) * 60 + else: + duration = (1440 - start_min + stop_min) * 60 + + self.recorder.start_recording( + job_id=job["id"], + name=job["name"], + stream_url=job["stream_url"], + stream_type=job["stream_type"], + output_format=job["output_format"] or "", + max_duration=duration, + extra_ffmpeg_args=job["extra_ffmpeg_args"] or "", + segment_duration=job["segment_duration"], + is_scheduled=True, + metadata_monitor_enabled=job.get("metadata_monitor_enabled", False), + metadata_pattern=job.get("metadata_pattern", ""), + metadata_grace_period=job.get("metadata_grace_period") or 300, + metadata_poll_interval=job.get("metadata_poll_interval") or 30, + ntfy_enabled=job.get("ntfy_enabled", False), + plik_enabled=job.get("plik_enabled", False), + delete_after_upload=job.get("delete_after_upload", False), + ) + self._started_keys.add(occ_key) + else: + if job["schedule_stop"] and self._is_job_recording(job["id"]): + rec_id = self._get_active_recording_for_job(job["id"]) + if rec_id: + self.recorder.stop_recording(rec_id) + self._started_keys.discard(occ_key) diff --git a/bin/frontend/static/index.html b/bin/frontend/static/index.html new file mode 100644 index 0000000..045d05d --- /dev/null +++ b/bin/frontend/static/index.html @@ -0,0 +1,1095 @@ + + +
+ + +Aktive Aufnahmen
+ +Letzte Aufnahmen
+ +Geplante Jobs
+ +Protokoll
+Konfiguration
+ +