Release: v3.1.0

bin/backend/app.py

@@ -17,6 +17,8 @@ from models import init_db, get_session, Job, Station, Recording as RecordingMod
 from recorder import RecordingManager, format_elapsed
 from scheduler import StreamScheduler
 
+APP_VERSION = "3.0.0"
+
 DATA_DIR = os.environ.get("DATA_DIR", "/app/data")
 STATIC_DIR = os.environ.get("STATIC_DIR", "/app/frontend/static")
 
@@ -37,7 +39,7 @@ async def lifespan(app):
     scheduler.stop()
 
 
-app = FastAPI(title="stream-recorder", version="3.0.0", lifespan=lifespan)
+app = FastAPI(title="stream-recorder", version=APP_VERSION, lifespan=lifespan)
 
 app.add_middleware(
     CORSMiddleware,
@@ -78,7 +80,7 @@ async def auth_middleware(request: Request, call_next):
 
 @app.get("/api/health")
 async def health():
-    return {"status": "ok", "version": "3.0.0"}
+    return {"status": "ok", "version": APP_VERSION}
 
 
 # --- Jobs CRUD ---
@@ -198,14 +200,13 @@ class JobCreate(BaseModel):
     schedule_start: str = ""
     schedule_stop: str = ""
     schedule_end_date: str = ""
+    pre_buffer: int = 0
+    post_buffer: int = 0
     description: 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
+    station_id: Optional[int] = None
 
 
 class JobUpdate(JobCreate):
@@ -217,6 +218,13 @@ async def list_jobs():
     session = get_session()
     try:
         jobs = session.query(Job).order_by(Job.id).all()
+
+        station_ids = {j.station_id for j in jobs if j.station_id}
+        station_names = {}
+        if station_ids:
+            for s in session.query(Station).filter(Station.id.in_(station_ids)).all():
+                station_names[s.id] = s.name
+
         result = []
         for j in jobs:
             is_recording = session.query(
@@ -239,14 +247,14 @@ async def list_jobs():
                 "schedule_start": j.schedule_start,
                 "schedule_stop": j.schedule_stop,
                 "schedule_end_date": j.schedule_end_date,
+                "pre_buffer": j.pre_buffer or 0,
+                "post_buffer": j.post_buffer or 0,
                 "description": j.description,
                 "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,
+                "station_id": j.station_id,
+                "station_name": station_names.get(j.station_id) if j.station_id else None,
                 "is_recording": is_recording,
                 "created_at": j.created_at.isoformat() if j.created_at else None,
             })
@@ -282,13 +290,12 @@ async def get_job(job_id: int):
             "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,
-            "schedule_end_date": job.schedule_end_date, "description": job.description,
+            "schedule_end_date": job.schedule_end_date,
+            "pre_buffer": job.pre_buffer or 0, "post_buffer": job.post_buffer or 0,
+            "description": job.description,
             "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,
+            "station_id": job.station_id,
         }
     finally:
         session.close()
@@ -354,10 +361,6 @@ async def start_job_recording(job_id: int):
             "max_duration": duration, "extra_ffmpeg_args": job.extra_ffmpeg_args,
             "segment_duration": job.segment_duration,
             "description": job.description,
-            "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,

bin/backend/config.py

@@ -26,6 +26,10 @@ DEFAULT_CONFIG = {
         "api_key": "",
         "ttl": "30d",
     },
+    "schedule": {
+        "default_pre_buffer": 5,
+        "default_post_buffer": 5,
+    },
     "logging": {
         "level": "INFO",
     },

bin/backend/models.py

@@ -25,6 +25,9 @@ class Job(Base):
     schedule_stop = Column(String(5), default="")
     schedule_end_date = Column(String(10), default="")
 
+    pre_buffer = Column(Integer, default=0)
+    post_buffer = Column(Integer, default=0)
+
     description = Column(Text, default="")
 
     plik_enabled = Column(Boolean, default=False)
@@ -32,10 +35,7 @@ class Job(Base):
 
     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)
+    station_id = Column(Integer, nullable=True)
 
     created_at = Column(DateTime, default=datetime.datetime.utcnow)
     updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
@@ -72,6 +72,7 @@ class Recording(Base):
     is_scheduled = Column(Boolean, default=False)
     error_message = Column(Text, nullable=True)
     file_size = Column(Integer, nullable=True)
+    plik_url = Column(String(1000), nullable=True)
 
 
 class LogEntry(Base):
@@ -98,11 +99,18 @@ def init_db(db_path: str):
 
 def _migrate(engine):
     with engine.connect() as conn:
-        try:
-            conn.execute(text("ALTER TABLE recordings ADD COLUMN file_size INTEGER"))
-            conn.commit()
-        except Exception:
-            pass
+        for stmt in [
+            "ALTER TABLE recordings ADD COLUMN file_size INTEGER",
+            "ALTER TABLE jobs ADD COLUMN station_id INTEGER",
+            "ALTER TABLE jobs ADD COLUMN pre_buffer INTEGER DEFAULT 0",
+            "ALTER TABLE jobs ADD COLUMN post_buffer INTEGER DEFAULT 0",
+            "ALTER TABLE recordings ADD COLUMN plik_url VARCHAR(1000)",
+        ]:
+            try:
+                conn.execute(text(stmt))
+                conn.commit()
+            except Exception:
+                pass
 
 
 def get_session():

bin/backend/plugins/euer_radio.py

@@ -1,138 +0,0 @@
-"""
-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)

bin/backend/plugins/plik.py

@@ -95,7 +95,8 @@ class PlikUploader:
         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):
+                         segment_mode: bool, delete_after: bool, ntfy_enabled: bool,
+                         save_url_fn: Callable = None):
         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)
@@ -107,15 +108,15 @@ class PlikUploader:
         def _do_upload():
             try:
                 if segment_mode:
-                    self._upload_segments(output_dir, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled)
+                    self._upload_segments(output_dir, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled, save_url_fn)
                 else:
-                    self._upload_single(output_file, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled)
+                    self._upload_single(output_file, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled, save_url_fn)
             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):
+    def _upload_single(self, file_path, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled, save_url_fn=None):
         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
@@ -127,6 +128,8 @@ class PlikUploader:
             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 save_url_fn:
+                save_url_fn(result["browser_url"])
             if self._notify:
                 self._notify("upload", "Plik-Upload", f"{job_name}: {result['browser_url']}", ntfy_enabled)
             if delete_after:
@@ -135,7 +138,7 @@ class PlikUploader:
         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):
+    def _upload_segments(self, output_dir, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled, save_url_fn=None):
         if not os.path.isdir(output_dir):
             return
 
@@ -161,8 +164,11 @@ class PlikUploader:
             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)
+        if uploaded > 0:
+            if save_url_fn and last_browser_url:
+                save_url_fn(last_browser_url)
+            if 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)

bin/backend/recorder.py

@@ -65,21 +65,11 @@ class RecordingManager:
         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] = {}
         self._output_files: dict[int, list[str]] = {}
 
-    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
@@ -126,10 +116,6 @@ class RecordingManager:
         segment_duration: Optional[int] = None,
         is_scheduled: bool = False,
         description: str = "",
-        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,
@@ -199,16 +185,6 @@ class RecordingManager:
         self._log("INFO", f"Aufnahme gestartet: {name}", name)
         self._notify("start", "Aufnahme gestartet", name, 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(
@@ -331,9 +307,6 @@ class RecordingManager:
         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()
@@ -487,6 +460,7 @@ class RecordingManager:
             "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,
         }
 
     def _calc_recording_size(self, output_files: list[str], segment_mode: bool) -> tuple[int, int]:
@@ -536,6 +510,7 @@ class RecordingManager:
                 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):

bin/backend/scheduler.py

@@ -54,6 +54,23 @@ def _in_time_window(now_time: str, start: str, stop: str) -> bool:
         return now_min >= start_min or now_min < stop_min
 
 
+def _apply_buffers(job: dict) -> dict:
+    eff = dict(job)
+    pre = job.get("pre_buffer", 0) or 0
+    post = job.get("post_buffer", 0) or 0
+    if pre > 0 and eff.get("schedule_start"):
+        m = _time_to_minutes(eff["schedule_start"]) - pre
+        if m < 0:
+            m += 1440
+        eff["schedule_start"] = f"{m // 60:02d}:{m % 60:02d}"
+    if post > 0 and eff.get("schedule_stop"):
+        m = _time_to_minutes(eff["schedule_stop"]) + post
+        if m >= 1440:
+            m -= 1440
+        eff["schedule_stop"] = f"{m // 60:02d}:{m % 60:02d}"
+    return eff
+
+
 class StreamScheduler:
     def __init__(self, recorder):
         self.recorder = recorder
@@ -76,6 +93,12 @@ class StreamScheduler:
             today = job.schedule_date or today
         return f"{job.id}|{job.schedule_once}|{today}|{job.schedule_start}"
 
+    def _original_occ_key(self, job: dict, now: datetime.datetime | None = None) -> str:
+        now = now or datetime.datetime.now()
+        today_date = now.strftime("%Y-%m-%d")
+        date = today_date if not job["schedule_once"] else job["schedule_date"]
+        return f"{job['id']}|{job['schedule_once']}|{date}|{job['schedule_start']}"
+
     def _schedule_state(self, job: dict, now: datetime.datetime | None = None) -> dict:
         now = now or datetime.datetime.now()
         today_num = now.isoweekday()
@@ -140,15 +163,19 @@ class StreamScheduler:
                 "schedule_start": j.schedule_start,
                 "schedule_stop": j.schedule_stop,
                 "schedule_end_date": j.schedule_end_date,
+                "pre_buffer": j.pre_buffer or 0,
+                "post_buffer": j.post_buffer or 0,
             }
         finally:
             session.close()
 
-        state = self._schedule_state(job)
+        occ_key = self._original_occ_key(job)
+        eff_job = _apply_buffers(job)
+        state = self._schedule_state(eff_job)
         if not state["in_window"]:
             return False
 
-        self._started_keys.add(state["occ_key"])
+        self._started_keys.add(occ_key)
         return True
 
     def _occurrence_start_utc(self, job: dict, now: datetime.datetime) -> datetime.datetime | None:
@@ -219,11 +246,8 @@ class StreamScheduler:
                     "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, "schedule_end_date": j.schedule_end_date,
+                    "pre_buffer": j.pre_buffer or 0, "post_buffer": j.post_buffer or 0,
                     "description": j.description,
-                    "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,
@@ -235,9 +259,10 @@ class StreamScheduler:
             if not job["schedule_start"]:
                 continue
 
-            state = self._schedule_state(job, now)
+            occ_key = self._original_occ_key(job, now)
+            eff_job = _apply_buffers(job)
+            state = self._schedule_state(eff_job, now)
             in_window = state["in_window"]
-            occ_key = state["occ_key"]
 
             if in_window:
                 if (
@@ -246,16 +271,16 @@ class StreamScheduler:
                     and not self._has_recording_for_occurrence(job, now)
                 ):
                     duration = job["max_duration"]
-                    if not duration and job["schedule_stop"]:
+                    if not duration and eff_job["schedule_stop"]:
                         if state["multi_day"]:
                             start_dt = datetime.datetime.strptime(
-                                f"{job['schedule_date']} {job['schedule_start']}", "%Y-%m-%d %H:%M")
+                                f"{job['schedule_date']} {eff_job['schedule_start']}", "%Y-%m-%d %H:%M")
                             end_dt = datetime.datetime.strptime(
-                                f"{job['schedule_end_date']} {job['schedule_stop']}", "%Y-%m-%d %H:%M")
+                                f"{job['schedule_end_date']} {eff_job['schedule_stop']}", "%Y-%m-%d %H:%M")
                             duration = max(0, int((end_dt - start_dt).total_seconds()))
                         else:
-                            start_min = _time_to_minutes(job["schedule_start"])
-                            stop_min = _time_to_minutes(job["schedule_stop"])
+                            start_min = _time_to_minutes(eff_job["schedule_start"])
+                            stop_min = _time_to_minutes(eff_job["schedule_stop"])
                             if stop_min > start_min:
                                 duration = (stop_min - start_min) * 60
                             else:
@@ -272,10 +297,6 @@ class StreamScheduler:
                         segment_duration=job["segment_duration"],
                         is_scheduled=True,
                         description=job.get("description", ""),
-                        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),
@@ -285,7 +306,7 @@ class StreamScheduler:
                 if (
                     not state["multi_day"]
                     and not state["is_cross_midnight"]
-                    and job["schedule_stop"]
+                    and eff_job["schedule_stop"]
                     and self._is_job_recording(job["id"])
                 ):
                     rec_id = self._get_active_scheduled_recording_for_job(job["id"])