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 @@