From 851ff086ac86b609d6fd694af55f3724e77ce128 Mon Sep 17 00:00:00 2001 From: Patrick Asmus Date: Wed, 16 Sep 2026 11:46:18 +0200 Subject: [PATCH] =?UTF-8?q?Dateigr=C3=B6=C3=9Fe=20pro=20Aufnahme=20entkopp?= =?UTF-8?q?elt=20-=20feste=20Gr=C3=B6=C3=9Fe=20nach=20Abschluss?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Die Dateigröße wird jetzt pro Aufnahme korrekt berechnet statt das gesamte Verzeichnis zu summieren. Bei abgeschlossenen Aufnahmen wird die Größe einmalig gespeichert und nicht mehr durch laufende Aufnahmen verändert. Co-Authored-By: Claude Opus 4.6 --- bin/backend/app.py | 105 ++++++++++++++- bin/backend/models.py | 25 +++- bin/backend/recorder.py | 199 +++++++++++++++++------------ bin/backend/scheduler.py | 159 +++++++++++++++++------ bin/frontend/static/index.html | 226 ++++++++++++++++++++++++++++++++- docs/api.md | 10 ++ 6 files changed, 600 insertions(+), 124 deletions(-) diff --git a/bin/backend/app.py b/bin/backend/app.py index 4cbac4e..54a56c5 100644 --- a/bin/backend/app.py +++ b/bin/backend/app.py @@ -13,7 +13,7 @@ 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 models import init_db, get_session, Job, Station, Recording as RecordingModel, LogEntry from recorder import RecordingManager, format_elapsed from scheduler import StreamScheduler @@ -83,6 +83,106 @@ async def health(): # --- Jobs CRUD --- +class StationCreate(BaseModel): + name: str + stream_url: str + stream_type: str = "auto" + description: str = "" + + +class StationUpdate(StationCreate): + pass + + +def _station_to_dict(station: Station) -> dict: + return { + "id": station.id, + "name": station.name, + "stream_url": station.stream_url, + "stream_type": station.stream_type, + "description": station.description, + "created_at": station.created_at.isoformat() if station.created_at else None, + "updated_at": station.updated_at.isoformat() if station.updated_at else None, + } + + +def _clean_station_data(data: StationCreate | StationUpdate) -> dict: + values = data.model_dump() + values["name"] = values["name"].strip() + values["stream_url"] = values["stream_url"].strip() + values["stream_type"] = (values.get("stream_type") or "auto").strip() or "auto" + values["description"] = (values.get("description") or "").strip() + if not values["name"] or not values["stream_url"]: + raise HTTPException(400, "Name und URL sind Pflichtfelder") + return values + + +@app.get("/api/stations") +async def list_stations(): + session = get_session() + try: + stations = session.query(Station).order_by(Station.name.collate("NOCASE"), Station.id).all() + return [_station_to_dict(s) for s in stations] + finally: + session.close() + + +@app.post("/api/stations") +async def create_station(data: StationCreate): + values = _clean_station_data(data) + session = get_session() + try: + station = Station(**values) + session.add(station) + session.commit() + return {"id": station.id, "message": "Sender gespeichert"} + finally: + session.close() + + +@app.get("/api/stations/{station_id}") +async def get_station(station_id: int): + session = get_session() + try: + station = session.query(Station).filter_by(id=station_id).first() + if not station: + raise HTTPException(404, "Sender nicht gefunden") + return _station_to_dict(station) + finally: + session.close() + + +@app.put("/api/stations/{station_id}") +async def update_station(station_id: int, data: StationUpdate): + values = _clean_station_data(data) + session = get_session() + try: + station = session.query(Station).filter_by(id=station_id).first() + if not station: + raise HTTPException(404, "Sender nicht gefunden") + for k, v in values.items(): + setattr(station, k, v) + station.updated_at = datetime.datetime.utcnow() + session.commit() + return {"message": "Sender aktualisiert"} + finally: + session.close() + + +@app.delete("/api/stations/{station_id}") +async def delete_station(station_id: int): + session = get_session() + try: + station = session.query(Station).filter_by(id=station_id).first() + if not station: + raise HTTPException(404, "Sender nicht gefunden") + session.delete(station) + session.commit() + return {"message": "Sender gelöscht"} + finally: + session.close() + + class JobCreate(BaseModel): name: str stream_url: str @@ -310,6 +410,9 @@ async def get_recording(rec_id: int): @app.post("/api/recordings/{rec_id}/stop") async def stop_recording(rec_id: int): + status = recorder.get_status(rec_id) + if status and status.get("job_id"): + scheduler.suppress_current_occurrence(status["job_id"]) recorder.stop_recording(rec_id) return {"message": "Aufnahme gestoppt"} diff --git a/bin/backend/models.py b/bin/backend/models.py index 305e385..98102c6 100644 --- a/bin/backend/models.py +++ b/bin/backend/models.py @@ -1,5 +1,5 @@ import datetime -from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime, Text +from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime, Text, text from sqlalchemy.orm import declarative_base, sessionmaker Base = declarative_base() @@ -41,6 +41,18 @@ class Job(Base): updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow) +class Station(Base): + __tablename__ = "stations" + + 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") + description = Column(Text, default="") + 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" @@ -59,6 +71,7 @@ class Recording(Base): segment_mode = Column(Boolean, default=False) is_scheduled = Column(Boolean, default=False) error_message = Column(Text, nullable=True) + file_size = Column(Integer, nullable=True) class LogEntry(Base): @@ -80,6 +93,16 @@ def init_db(db_path: str): _engine = create_engine(f"sqlite:///{db_path}", connect_args={"check_same_thread": False}) Base.metadata.create_all(_engine) _SessionLocal = sessionmaker(bind=_engine) + _migrate(_engine) + + +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 def get_session(): diff --git a/bin/backend/recorder.py b/bin/backend/recorder.py index edec349..dfae8f2 100644 --- a/bin/backend/recorder.py +++ b/bin/backend/recorder.py @@ -1,5 +1,6 @@ import asyncio import datetime +import glob as globmod import os import re import signal @@ -68,6 +69,7 @@ class RecordingManager: 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: @@ -173,6 +175,15 @@ class RecordingManager: 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 @@ -185,14 +196,6 @@ class RecordingManager: 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, - "description": description, - } - self._log("INFO", f"Aufnahme gestartet: {name}", name) self._notify("start", "Aufnahme gestartet", name, rec_id) @@ -225,74 +228,90 @@ class RecordingManager: 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, + try: + while not stop_event.is_set(): + cmd = self._build_ffmpeg_cmd( + stream_url, stream_type, output_file, + extra_ffmpeg_args, segment_mode, segment_duration, ) - 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) + 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), + 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 - process.wait() - retcode = process.returncode - log_fh.close() - self._processes.pop(rec_id, None) + self._processes[rec_id] = process + self._update_recording(rec_id, status="running", pid=process.pid) - 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) - 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 + process.wait() + retcode = process.returncode + log_fh.close() + self._processes.pop(rec_id, None) - 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 + 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 - 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 + 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 - 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 + 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 - 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._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 - self._update_recording(rec_id, status="stopped") + 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): cmd = ["ffmpeg", "-nostdin", "-y", "-hide_banner", "-loglevel", "warning"] @@ -330,9 +349,10 @@ class RecordingManager: except (ProcessLookupError, OSError): pass - self._update_recording(rec_id, status="stopped") - self._log("INFO", f"Aufnahme gestoppt (ID: {rec_id})") - self._cleanup(rec_id) + 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) -> bool: @@ -433,16 +453,11 @@ class RecordingManager: 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) + 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": @@ -474,6 +489,29 @@ class RecordingManager: "remaining_display": format_elapsed(remaining) if remaining is not None else None, } + 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: + pattern = of.replace('%03d', '*') + for f in globmod.glob(pattern): + if os.path.isfile(f): + 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: @@ -505,3 +543,4 @@ class RecordingManager: 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) diff --git a/bin/backend/scheduler.py b/bin/backend/scheduler.py index fc855e7..5ff55b1 100644 --- a/bin/backend/scheduler.py +++ b/bin/backend/scheduler.py @@ -1,5 +1,6 @@ import datetime from apscheduler.schedulers.background import BackgroundScheduler +from sqlalchemy import or_ from models import Job, Recording, get_session DAY_MAP = { @@ -75,6 +76,111 @@ class StreamScheduler: today = job.schedule_date or today return f"{job.id}|{job.schedule_once}|{today}|{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() + today_date = now.strftime("%Y-%m-%d") + now_time = now.strftime("%H:%M") + + occ_key = ( + f"{job['id']}|{job['schedule_once']}|" + f"{today_date if not job['schedule_once'] else job['schedule_date']}|" + f"{job['schedule_start']}" + ) + + is_today = False + multi_day = bool(job.get("schedule_end_date") and job["schedule_once"]) + + 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 + is_cross_midnight = False + if is_today: + if multi_day: + now_min = _time_to_minutes(now_time) + start_min = _time_to_minutes(job["schedule_start"]) + in_window = now_min >= start_min + elif job["schedule_stop"]: + start_min = _time_to_minutes(job["schedule_start"]) + stop_min = _time_to_minutes(job["schedule_stop"]) + if stop_min <= start_min: + is_cross_midnight = True + now_min = _time_to_minutes(now_time) + in_window = now_min >= start_min + else: + 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 + + return { + "occ_key": occ_key, + "in_window": in_window, + "multi_day": multi_day, + "is_cross_midnight": is_cross_midnight, + } + + def suppress_current_occurrence(self, job_id: int) -> bool: + session = get_session() + try: + j = session.query(Job).filter_by(id=job_id).first() + if not j or not j.schedule_enabled or not j.schedule_start: + return False + job = { + "id": j.id, + "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, + } + finally: + session.close() + + state = self._schedule_state(job) + if not state["in_window"]: + return False + + self._started_keys.add(state["occ_key"]) + return True + + def _occurrence_start_utc(self, job: dict, now: datetime.datetime) -> datetime.datetime | None: + date_text = job["schedule_date"] if job["schedule_once"] else now.strftime("%Y-%m-%d") + try: + local_start = datetime.datetime.strptime( + f"{date_text} {job['schedule_start']}", + "%Y-%m-%d %H:%M", + ) + except (TypeError, ValueError): + return None + + local_utc_offset = datetime.datetime.now() - datetime.datetime.utcnow() + return local_start - local_utc_offset + + def _has_recording_for_occurrence(self, job: dict, now: datetime.datetime) -> bool: + occurrence_start = self._occurrence_start_utc(job, now) + if occurrence_start is None: + return False + + session = get_session() + try: + return session.query(Recording).filter( + Recording.job_id == job["id"], + or_( + Recording.started_at >= occurrence_start, + Recording.stopped_at >= occurrence_start, + ), + ).first() is not None + finally: + session.close() + def _is_job_recording(self, job_id: int) -> bool: session = get_session() try: @@ -99,9 +205,6 @@ class StreamScheduler: 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: @@ -132,44 +235,19 @@ class StreamScheduler: if not job["schedule_start"]: continue - is_today = False - multi_day = bool(job.get("schedule_end_date") and job["schedule_once"]) - - 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 - is_cross_midnight = False - if is_today: - if multi_day: - now_min = _time_to_minutes(now_time) - start_min = _time_to_minutes(job["schedule_start"]) - in_window = now_min >= start_min - elif job["schedule_stop"]: - start_min = _time_to_minutes(job["schedule_start"]) - stop_min = _time_to_minutes(job["schedule_stop"]) - if stop_min <= start_min: - is_cross_midnight = True - now_min = _time_to_minutes(now_time) - in_window = now_min >= start_min - else: - 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']}" + state = self._schedule_state(job, now) + in_window = state["in_window"] + occ_key = state["occ_key"] if in_window: - if occ_key not in self._started_keys and not self._is_job_recording(job["id"]): + if ( + occ_key not in self._started_keys + and not self._is_job_recording(job["id"]) + and not self._has_recording_for_occurrence(job, now) + ): duration = job["max_duration"] if not duration and job["schedule_stop"]: - if multi_day: + if state["multi_day"]: start_dt = datetime.datetime.strptime( f"{job['schedule_date']} {job['schedule_start']}", "%Y-%m-%d %H:%M") end_dt = datetime.datetime.strptime( @@ -204,7 +282,12 @@ class StreamScheduler: ) self._started_keys.add(occ_key) else: - if not multi_day and not is_cross_midnight and job["schedule_stop"] and self._is_job_recording(job["id"]): + if ( + not state["multi_day"] + and not state["is_cross_midnight"] + and job["schedule_stop"] + and self._is_job_recording(job["id"]) + ): rec_id = self._get_active_scheduled_recording_for_job(job["id"]) if rec_id: self.recorder.stop_recording(rec_id) diff --git a/bin/frontend/static/index.html b/bin/frontend/static/index.html index 9dfba34..109b8aa 100644 --- a/bin/frontend/static/index.html +++ b/bin/frontend/static/index.html @@ -3,7 +3,7 @@ - stream-recorder + Stream-Recorder
-

stream-recorder

+

Stream-Recorder

+
@@ -373,6 +397,15 @@
+ +
+
+

Senderbibliothek

+ +
+
+
+
@@ -402,6 +435,12 @@