910 lines
29 KiB
Python
910 lines
29 KiB
Python
import base64
|
||
import binascii
|
||
import contextlib
|
||
import datetime
|
||
import hashlib
|
||
import hmac
|
||
import logging
|
||
import os
|
||
import re
|
||
import sys
|
||
import threading
|
||
import time
|
||
from typing import Annotated, Literal, Optional
|
||
from urllib.parse import urlsplit
|
||
|
||
from fastapi import FastAPI, HTTPException, Request, Query
|
||
from fastapi.responses import JSONResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
from pydantic import BaseModel, ConfigDict, Field, StringConstraints, field_validator
|
||
from starlette.concurrency import run_in_threadpool
|
||
from starlette.middleware.base import BaseHTTPMiddleware
|
||
|
||
from config import Config, hash_password, verify_password, validate_download_path
|
||
from models import (init_db, get_session, iso_utc, utcnow,
|
||
Job, Station, Recording as RecordingModel, LogEntry)
|
||
from recorder import RecordingManager, validate_ffmpeg_args, validate_stream_url
|
||
from scheduler import StreamScheduler, compute_duration, validate_schedule
|
||
|
||
APP_VERSION = os.environ.get("APP_VERSION", "dev")
|
||
|
||
DATA_DIR = os.environ.get("DATA_DIR", "/app/data")
|
||
STATIC_DIR = os.environ.get("STATIC_DIR", "/app/frontend/static")
|
||
ENABLE_API_DOCS = os.environ.get("ENABLE_API_DOCS", "") == "1"
|
||
|
||
LOG_FORMAT = "%(asctime)s %(levelname)s %(name)s: %(message)s"
|
||
LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR")
|
||
|
||
os.makedirs(DATA_DIR, exist_ok=True)
|
||
|
||
config = Config(os.path.join(DATA_DIR, "config.yml"))
|
||
|
||
|
||
def _apply_log_level(level: str):
|
||
level = str(level or "INFO").upper()
|
||
logging.getLogger().setLevel(getattr(logging, level, logging.INFO) if level in LOG_LEVELS else logging.INFO)
|
||
|
||
|
||
logging.basicConfig(format=LOG_FORMAT, stream=sys.stdout, force=True)
|
||
_apply_log_level(config.get("logging", "level", default="INFO"))
|
||
logging.getLogger("apscheduler").setLevel(logging.WARNING)
|
||
logging.getLogger("httpx").setLevel(logging.WARNING)
|
||
logger = logging.getLogger("app")
|
||
|
||
if config.password_migrated:
|
||
logger.warning("Klartext-Passwort in config.yml wurde in einen Hash migriert")
|
||
|
||
if config.default_password_active:
|
||
logger.warning("Basic-Auth nutzt das bekannte Standardpasswort – bitte in den Einstellungen ändern!")
|
||
|
||
if config.generated_password:
|
||
logger.warning("=" * 60)
|
||
logger.warning("Erster Start: Basic-Auth wurde aktiviert.")
|
||
logger.warning("Benutzer: %s", config.get("auth", "username", default="admin"))
|
||
logger.warning("Passwort: %s", config.generated_password)
|
||
logger.warning("Bitte nach dem ersten Login in den Einstellungen ändern.")
|
||
logger.warning("=" * 60)
|
||
config.generated_password = None
|
||
|
||
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.prune_logs()
|
||
scheduler.start()
|
||
yield
|
||
scheduler.stop()
|
||
|
||
|
||
app = FastAPI(
|
||
title="stream-recorder",
|
||
version=APP_VERSION,
|
||
lifespan=lifespan,
|
||
docs_url="/docs" if ENABLE_API_DOCS else None,
|
||
redoc_url="/redoc" if ENABLE_API_DOCS else None,
|
||
openapi_url="/openapi.json" if ENABLE_API_DOCS else None,
|
||
)
|
||
|
||
|
||
# --- Auth ---
|
||
|
||
AUTH_CACHE_MAX = 256
|
||
AUTH_FAIL_LIMIT = 10
|
||
AUTH_FAIL_WINDOW = 300
|
||
EXEMPT_PATHS = {"/api/health"}
|
||
|
||
_auth_lock = threading.Lock()
|
||
_auth_cache: dict[str, str] = {}
|
||
_auth_failures: dict[str, list[float]] = {}
|
||
|
||
|
||
def clear_auth_cache():
|
||
with _auth_lock:
|
||
_auth_cache.clear()
|
||
|
||
|
||
def reset_auth_failures():
|
||
with _auth_lock:
|
||
_auth_failures.clear()
|
||
|
||
|
||
def _recent_failures(ip: str, now: float) -> list[float]:
|
||
fails = [t for t in _auth_failures.get(ip, []) if now - t < AUTH_FAIL_WINDOW]
|
||
if fails:
|
||
_auth_failures[ip] = fails
|
||
else:
|
||
_auth_failures.pop(ip, None)
|
||
return fails
|
||
|
||
|
||
def _register_failure(ip: str, now: float):
|
||
with _auth_lock:
|
||
if len(_auth_failures) > 10000:
|
||
for key in list(_auth_failures):
|
||
_recent_failures(key, now)
|
||
_auth_failures.setdefault(ip, []).append(now)
|
||
|
||
|
||
def _check_credentials(auth_header: str) -> bool:
|
||
expected_user = config.get("auth", "username", default="admin")
|
||
pw_hash = config.get("auth", "password_hash", default="")
|
||
if not pw_hash:
|
||
return False
|
||
cache_key = hashlib.sha256(auth_header.encode("utf-8", "surrogateescape")).hexdigest()
|
||
marker = f"{expected_user}\x00{pw_hash}"
|
||
with _auth_lock:
|
||
if _auth_cache.get(cache_key) == marker:
|
||
return True
|
||
|
||
if not auth_header.startswith("Basic "):
|
||
return False
|
||
try:
|
||
decoded = base64.b64decode(auth_header[6:], validate=True).decode("utf-8")
|
||
username, password = decoded.split(":", 1)
|
||
except (binascii.Error, UnicodeDecodeError, ValueError):
|
||
return False
|
||
|
||
user_ok = hmac.compare_digest(username.encode("utf-8"), str(expected_user).encode("utf-8"))
|
||
pass_ok = verify_password(password, pw_hash)
|
||
if not (user_ok and pass_ok):
|
||
return False
|
||
|
||
with _auth_lock:
|
||
if len(_auth_cache) >= AUTH_CACHE_MAX:
|
||
_auth_cache.pop(next(iter(_auth_cache)))
|
||
_auth_cache[cache_key] = marker
|
||
return True
|
||
|
||
|
||
def _unauthorized():
|
||
return JSONResponse(
|
||
status_code=401,
|
||
content={"detail": "Unauthorized"},
|
||
headers={"WWW-Authenticate": 'Basic realm="stream-recorder"'},
|
||
)
|
||
|
||
|
||
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 in EXEMPT_PATHS:
|
||
return await call_next(request)
|
||
|
||
ip = request.client.host if request.client else "unknown"
|
||
now = time.monotonic()
|
||
with _auth_lock:
|
||
blocked = len(_recent_failures(ip, now)) >= AUTH_FAIL_LIMIT
|
||
if blocked:
|
||
return JSONResponse(status_code=429, content={"detail": "Zu viele fehlgeschlagene Anmeldeversuche"})
|
||
|
||
auth_header = request.headers.get("Authorization", "")
|
||
if not auth_header:
|
||
return _unauthorized()
|
||
|
||
try:
|
||
ok = await run_in_threadpool(_check_credentials, auth_header)
|
||
except Exception:
|
||
ok = False
|
||
if ok:
|
||
return await call_next(request)
|
||
|
||
_register_failure(ip, now)
|
||
return _unauthorized()
|
||
|
||
|
||
# --- CSRF ---
|
||
|
||
_DEFAULT_PORTS = {"http": "80", "https": "443"}
|
||
|
||
|
||
def _norm_host(netloc: str, scheme: str) -> str:
|
||
netloc = netloc.strip().lower()
|
||
if not netloc:
|
||
return ""
|
||
if netloc.startswith("["):
|
||
host, _, rest = netloc.partition("]")
|
||
host += "]"
|
||
port = rest[1:] if rest.startswith(":") else ""
|
||
elif ":" in netloc:
|
||
host, port = netloc.rsplit(":", 1)
|
||
else:
|
||
host, port = netloc, ""
|
||
if not port:
|
||
port = _DEFAULT_PORTS.get(scheme, "")
|
||
return f"{host}:{port}"
|
||
|
||
|
||
async def csrf_middleware(request: Request, call_next):
|
||
if request.method in ("POST", "PUT", "PATCH", "DELETE"):
|
||
if request.headers.get("sec-fetch-site", "").lower() == "cross-site":
|
||
return JSONResponse(status_code=403, content={"detail": "Cross-Site-Anfrage abgelehnt"})
|
||
origin = request.headers.get("origin")
|
||
if origin is not None:
|
||
try:
|
||
o = urlsplit(origin)
|
||
origin_host = _norm_host(o.netloc, o.scheme.lower())
|
||
except ValueError:
|
||
origin_host = ""
|
||
candidates = set()
|
||
for header in ("host", "x-forwarded-host"):
|
||
value = request.headers.get(header)
|
||
if value:
|
||
candidates.add(_norm_host(value.split(",")[0], o.scheme.lower() if origin_host else "http"))
|
||
if not origin_host or origin_host not in candidates:
|
||
return JSONResponse(status_code=403, content={"detail": "Ungültiger Origin"})
|
||
return await call_next(request)
|
||
|
||
|
||
# --- Security-Header ---
|
||
|
||
SECURITY_HEADERS = {
|
||
"X-Content-Type-Options": "nosniff",
|
||
"X-Frame-Options": "DENY",
|
||
"Referrer-Policy": "no-referrer",
|
||
"Content-Security-Policy": (
|
||
"default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; "
|
||
"img-src 'self' data:; connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; "
|
||
"form-action 'self'; object-src 'none'"
|
||
),
|
||
}
|
||
|
||
|
||
async def security_headers_middleware(request: Request, call_next):
|
||
response = await call_next(request)
|
||
for k, v in SECURITY_HEADERS.items():
|
||
response.headers[k] = v
|
||
return response
|
||
|
||
|
||
# Zuletzt hinzugefügt = äußerste Schicht
|
||
app.add_middleware(BaseHTTPMiddleware, dispatch=auth_middleware)
|
||
app.add_middleware(BaseHTTPMiddleware, dispatch=csrf_middleware)
|
||
app.add_middleware(BaseHTTPMiddleware, dispatch=security_headers_middleware)
|
||
|
||
|
||
# --- Health ---
|
||
|
||
@app.get("/api/health")
|
||
async def health():
|
||
return {"status": "ok", "version": APP_VERSION}
|
||
|
||
|
||
# --- Validierung ---
|
||
|
||
StreamType = Literal["auto", "http", "hls", "mp3", "aac", "ogg", "flac", "rtmp"]
|
||
OutputFormat = Literal["", "auto", "mp3", "aac", "m4a", "mp4", "mkv", "ogg", "opus", "flac", "ts", "wav"]
|
||
Name = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=200)]
|
||
Description = Annotated[str, StringConstraints(max_length=2000)]
|
||
MaxDuration = Optional[Annotated[int, Field(ge=1, le=2592000)]]
|
||
Buffer = Annotated[int, Field(ge=0, le=120)]
|
||
|
||
|
||
def _bad_request(e: Exception):
|
||
raise HTTPException(400, str(e))
|
||
|
||
|
||
def _check_url(url: str) -> str:
|
||
try:
|
||
return validate_stream_url(url)
|
||
except ValueError as e:
|
||
_bad_request(e)
|
||
|
||
|
||
class StationCreate(BaseModel):
|
||
name: Name
|
||
stream_url: Annotated[str, StringConstraints(strip_whitespace=True)]
|
||
stream_type: StreamType = "auto"
|
||
description: Description = ""
|
||
|
||
|
||
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": iso_utc(station.created_at),
|
||
"updated_at": iso_utc(station.updated_at),
|
||
}
|
||
|
||
|
||
def _clean_station_data(data: StationCreate) -> dict:
|
||
values = data.model_dump()
|
||
values["stream_url"] = _check_url(values["stream_url"])
|
||
values["description"] = (values.get("description") or "").strip()
|
||
return values
|
||
|
||
|
||
# --- Stations ---
|
||
|
||
@app.get("/api/stations")
|
||
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")
|
||
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}")
|
||
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}")
|
||
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 = utcnow()
|
||
session.commit()
|
||
return {"message": "Sender aktualisiert"}
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
@app.delete("/api/stations/{station_id}")
|
||
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.query(Job).filter(Job.station_id == station_id).update(
|
||
{Job.station_id: None}, synchronize_session=False)
|
||
session.delete(station)
|
||
session.commit()
|
||
return {"message": "Sender gelöscht"}
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
# --- Jobs ---
|
||
|
||
class JobCreate(BaseModel):
|
||
name: Name
|
||
stream_url: Annotated[str, StringConstraints(strip_whitespace=True)]
|
||
stream_type: StreamType = "auto"
|
||
output_format: OutputFormat = ""
|
||
max_duration: MaxDuration = None
|
||
extra_ffmpeg_args: Annotated[str, StringConstraints(strip_whitespace=True)] = ""
|
||
segment_duration: Optional[int] = None
|
||
schedule_enabled: bool = False
|
||
schedule_once: bool = False
|
||
schedule_date: Annotated[str, StringConstraints(strip_whitespace=True, max_length=10)] = ""
|
||
schedule_days: Annotated[str, StringConstraints(strip_whitespace=True, max_length=50)] = "*"
|
||
schedule_start: Annotated[str, StringConstraints(strip_whitespace=True, max_length=5)] = ""
|
||
schedule_stop: Annotated[str, StringConstraints(strip_whitespace=True, max_length=5)] = ""
|
||
schedule_end_date: Annotated[str, StringConstraints(strip_whitespace=True, max_length=10)] = ""
|
||
pre_buffer: Buffer = 0
|
||
post_buffer: Buffer = 0
|
||
description: Description = ""
|
||
ntfy_enabled: bool = False
|
||
plik_enabled: bool = False
|
||
delete_after_upload: bool = False
|
||
station_id: Optional[int] = None
|
||
|
||
@field_validator("segment_duration")
|
||
@classmethod
|
||
def _segment_duration(cls, v):
|
||
if not v:
|
||
return None
|
||
if v < 10 or v > 86400:
|
||
raise ValueError("Segmentdauer muss zwischen 10 und 86400 Sekunden liegen")
|
||
return v
|
||
|
||
@field_validator("schedule_days")
|
||
@classmethod
|
||
def _schedule_days(cls, v):
|
||
return v or "*"
|
||
|
||
|
||
class JobUpdate(JobCreate):
|
||
pass
|
||
|
||
|
||
def _validate_job(data: JobCreate, session) -> dict:
|
||
values = data.model_dump()
|
||
values["stream_url"] = _check_url(values["stream_url"])
|
||
try:
|
||
validate_ffmpeg_args(values["extra_ffmpeg_args"])
|
||
validate_schedule(values)
|
||
except ValueError as e:
|
||
_bad_request(e)
|
||
if values["station_id"] is not None:
|
||
if not session.query(Station).filter_by(id=values["station_id"]).first():
|
||
raise HTTPException(400, "Sender nicht gefunden")
|
||
return values
|
||
|
||
|
||
def _job_active(session, job_id: int) -> bool:
|
||
return session.query(
|
||
session.query(RecordingModel).filter(
|
||
RecordingModel.job_id == job_id,
|
||
RecordingModel.status.in_(["running", "starting"]),
|
||
).exists()
|
||
).scalar()
|
||
|
||
|
||
def _job_to_dict(j: Job) -> dict:
|
||
return {
|
||
"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,
|
||
"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,
|
||
"station_id": j.station_id,
|
||
"created_at": iso_utc(j.created_at),
|
||
"updated_at": iso_utc(j.updated_at),
|
||
}
|
||
|
||
|
||
@app.get("/api/jobs")
|
||
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
|
||
|
||
active_job_ids = {
|
||
r[0] for r in session.query(RecordingModel.job_id).filter(
|
||
RecordingModel.job_id.isnot(None),
|
||
RecordingModel.status.in_(["running", "starting"]),
|
||
).all()
|
||
}
|
||
|
||
result = []
|
||
for j in jobs:
|
||
d = _job_to_dict(j)
|
||
d["station_name"] = station_names.get(j.station_id) if j.station_id else None
|
||
d["is_recording"] = j.id in active_job_ids
|
||
result.append(d)
|
||
return result
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
@app.post("/api/jobs")
|
||
def create_job(data: JobCreate):
|
||
session = get_session()
|
||
try:
|
||
values = _validate_job(data, session)
|
||
job = Job(**values)
|
||
session.add(job)
|
||
session.commit()
|
||
return {"id": job.id, "message": "Job erstellt"}
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
@app.get("/api/jobs/{job_id}")
|
||
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 _job_to_dict(job)
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
@app.put("/api/jobs/{job_id}")
|
||
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")
|
||
values = _validate_job(data, session)
|
||
for k, v in values.items():
|
||
setattr(job, k, v)
|
||
job.updated_at = utcnow()
|
||
session.commit()
|
||
return {"message": "Job aktualisiert"}
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
@app.delete("/api/jobs/{job_id}")
|
||
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")
|
||
if _job_active(session, job_id):
|
||
raise HTTPException(400, "Job nimmt gerade auf – bitte zuerst stoppen")
|
||
session.delete(job)
|
||
session.commit()
|
||
return {"message": "Job gelöscht"}
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
_job_start_lock = threading.Lock()
|
||
|
||
|
||
@app.post("/api/jobs/{job_id}/record")
|
||
def start_job_recording(job_id: int):
|
||
with _job_start_lock:
|
||
session = get_session()
|
||
try:
|
||
job = session.query(Job).filter_by(id=job_id).first()
|
||
if not job:
|
||
raise HTTPException(404, "Job nicht gefunden")
|
||
if _job_active(session, job_id):
|
||
raise HTTPException(409, "Job nimmt bereits auf")
|
||
job_dict = _job_to_dict(job)
|
||
finally:
|
||
session.close()
|
||
|
||
try:
|
||
duration = compute_duration(job_dict, apply_buffers=False)
|
||
except (ValueError, TypeError) as e:
|
||
raise HTTPException(400, f"Ungültige Zeitplandaten: {e}")
|
||
|
||
try:
|
||
return recorder.start_recording(
|
||
job_id=job_id,
|
||
name=job_dict["name"],
|
||
stream_url=job_dict["stream_url"],
|
||
stream_type=job_dict["stream_type"],
|
||
output_format=job_dict["output_format"] or "",
|
||
max_duration=duration,
|
||
extra_ffmpeg_args=job_dict["extra_ffmpeg_args"] or "",
|
||
segment_duration=job_dict["segment_duration"],
|
||
description=job_dict["description"] or "",
|
||
ntfy_enabled=job_dict["ntfy_enabled"],
|
||
plik_enabled=job_dict["plik_enabled"],
|
||
delete_after_upload=job_dict["delete_after_upload"],
|
||
)
|
||
except ValueError as e:
|
||
_bad_request(e)
|
||
|
||
|
||
# --- Quick Recording ---
|
||
|
||
class QuickRecordRequest(BaseModel):
|
||
url: Annotated[str, StringConstraints(strip_whitespace=True)]
|
||
name: Annotated[str, StringConstraints(strip_whitespace=True, max_length=200)] = ""
|
||
stream_type: StreamType = "auto"
|
||
max_duration: MaxDuration = None
|
||
|
||
|
||
@app.post("/api/quick-record")
|
||
def quick_record(data: QuickRecordRequest):
|
||
url = _check_url(data.url)
|
||
name = data.name or f"Schnellaufnahme_{datetime.datetime.now().strftime('%H%M%S')}"
|
||
try:
|
||
return recorder.start_recording(
|
||
job_id=None,
|
||
name=name,
|
||
stream_url=url,
|
||
stream_type=data.stream_type,
|
||
max_duration=data.max_duration,
|
||
)
|
||
except ValueError as e:
|
||
_bad_request(e)
|
||
|
||
|
||
# --- Recordings ---
|
||
|
||
@app.get("/api/recordings")
|
||
def list_recordings(active_only: bool = False, limit: int = Query(default=50, ge=1, le=500)):
|
||
if active_only:
|
||
return recorder.get_all_active()
|
||
return recorder.get_all_recordings(limit=limit)
|
||
|
||
|
||
@app.get("/api/recordings/{rec_id}")
|
||
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")
|
||
def stop_recording(rec_id: int):
|
||
status = recorder.get_status(rec_id)
|
||
if not status:
|
||
raise HTTPException(404, "Aufnahme nicht gefunden")
|
||
if status.get("status") not in ("running", "starting"):
|
||
raise HTTPException(400, "Aufnahme läuft nicht")
|
||
if status.get("job_id"):
|
||
scheduler.suppress_current_occurrence(status["job_id"])
|
||
recorder.stop_recording(rec_id)
|
||
return {"message": "Aufnahme gestoppt"}
|
||
|
||
|
||
@app.post("/api/recordings/{rec_id}/extend")
|
||
def extend_recording(rec_id: int, minutes: int = Query(default=30, ge=1, le=1440)):
|
||
result = recorder.extend_recording(rec_id, minutes * 60)
|
||
if result == "not_found":
|
||
raise HTTPException(404, "Aufnahme nicht gefunden")
|
||
if result == "unlimited":
|
||
raise HTTPException(400, "Aufnahme ist bereits unbegrenzt")
|
||
if result != "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")
|
||
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")
|
||
def get_logs(
|
||
limit: int = Query(default=100, ge=1, le=1000),
|
||
level: str = "",
|
||
):
|
||
level = level.strip().upper()
|
||
if level not in ("", "DEBUG", "INFO", "WARN", "WARNING", "ERROR"):
|
||
raise HTTPException(400, "Ungültiges Log-Level")
|
||
session = get_session()
|
||
try:
|
||
q = session.query(LogEntry).order_by(LogEntry.id.desc())
|
||
if level in ("WARN", "WARNING"):
|
||
q = q.filter(LogEntry.level.in_(["WARN", "WARNING"]))
|
||
elif level:
|
||
q = q.filter(LogEntry.level == level)
|
||
logs = q.limit(limit).all()
|
||
return [
|
||
{
|
||
"id": entry.id,
|
||
"timestamp": iso_utc(entry.timestamp),
|
||
"level": entry.level,
|
||
"message": entry.message,
|
||
"job_name": entry.job_name,
|
||
}
|
||
for entry in logs
|
||
]
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
# --- Config ---
|
||
|
||
class _StrictModel(BaseModel):
|
||
model_config = ConfigDict(extra="forbid")
|
||
|
||
|
||
class RecordingConfigIn(_StrictModel):
|
||
download_path: Optional[str] = None
|
||
max_retries: Optional[Annotated[int, Field(ge=0, le=1000)]] = None
|
||
retry_delay: Optional[Annotated[int, Field(ge=1, le=3600)]] = None
|
||
|
||
|
||
class AuthConfigIn(_StrictModel):
|
||
enabled: Optional[bool] = None
|
||
username: Optional[Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=64)]] = None
|
||
password: Optional[str] = None
|
||
|
||
|
||
class NtfyConfigIn(_StrictModel):
|
||
url: Optional[Annotated[str, StringConstraints(strip_whitespace=True, max_length=1000)]] = None
|
||
events: Optional[Annotated[str, StringConstraints(strip_whitespace=True, max_length=100)]] = None
|
||
token: Optional[Annotated[str, StringConstraints(max_length=500)]] = None
|
||
|
||
|
||
class PlikConfigIn(_StrictModel):
|
||
url: Optional[Annotated[str, StringConstraints(strip_whitespace=True, max_length=1000)]] = None
|
||
ttl: Optional[Annotated[str, StringConstraints(strip_whitespace=True, max_length=20)]] = None
|
||
api_key: Optional[Annotated[str, StringConstraints(max_length=500)]] = None
|
||
|
||
|
||
class ScheduleConfigIn(_StrictModel):
|
||
default_pre_buffer: Optional[Buffer] = None
|
||
default_post_buffer: Optional[Buffer] = None
|
||
|
||
|
||
class LoggingConfigIn(_StrictModel):
|
||
level: Optional[Literal["DEBUG", "INFO", "WARNING", "ERROR"]] = None
|
||
|
||
|
||
class ConfigDataIn(_StrictModel):
|
||
recording: Optional[RecordingConfigIn] = None
|
||
auth: Optional[AuthConfigIn] = None
|
||
ntfy: Optional[NtfyConfigIn] = None
|
||
plik: Optional[PlikConfigIn] = None
|
||
schedule: Optional[ScheduleConfigIn] = None
|
||
logging: Optional[LoggingConfigIn] = None
|
||
|
||
|
||
class ConfigUpdate(_StrictModel):
|
||
data: ConfigDataIn
|
||
|
||
|
||
NTFY_EVENTS = {"start", "stop", "error", "upload"}
|
||
|
||
|
||
def _check_http_url(url: str, label: str):
|
||
if not url:
|
||
return
|
||
try:
|
||
parts = urlsplit(url)
|
||
hostname = parts.hostname
|
||
except ValueError:
|
||
hostname = None
|
||
parts = None
|
||
if not parts or parts.scheme not in ("http", "https") or not hostname or any(c.isspace() for c in url):
|
||
raise HTTPException(400, f"{label}-URL muss eine gültige http(s)-URL sein")
|
||
|
||
|
||
@app.get("/api/config")
|
||
def get_config():
|
||
with config.lock:
|
||
g = config.get
|
||
return {
|
||
"recording": {
|
||
"download_path": g("recording", "download_path"),
|
||
"max_retries": g("recording", "max_retries"),
|
||
"retry_delay": g("recording", "retry_delay"),
|
||
},
|
||
"server": {
|
||
"host": g("server", "host"),
|
||
"port": g("server", "port"),
|
||
},
|
||
"auth": {
|
||
"enabled": bool(g("auth", "enabled", default=False)),
|
||
"username": g("auth", "username", default="admin"),
|
||
"password_set": bool(g("auth", "password_hash", default="")),
|
||
},
|
||
"ntfy": {
|
||
"url": g("ntfy", "url", default=""),
|
||
"events": g("ntfy", "events", default=""),
|
||
"token_set": bool(g("ntfy", "token", default="")),
|
||
},
|
||
"plik": {
|
||
"url": g("plik", "url", default=""),
|
||
"ttl": g("plik", "ttl", default=""),
|
||
"api_key_set": bool(g("plik", "api_key", default="")),
|
||
},
|
||
"schedule": {
|
||
"default_pre_buffer": g("schedule", "default_pre_buffer"),
|
||
"default_post_buffer": g("schedule", "default_post_buffer"),
|
||
},
|
||
"logging": {
|
||
"level": g("logging", "level", default="INFO"),
|
||
},
|
||
}
|
||
|
||
|
||
@app.put("/api/config")
|
||
def update_config(payload: ConfigUpdate):
|
||
sections = {
|
||
name: {k: v for k, v in section.model_dump(exclude_unset=True).items() if v is not None}
|
||
for name, section in payload.data
|
||
if section is not None
|
||
}
|
||
|
||
rec = sections.get("recording", {})
|
||
if "download_path" in rec:
|
||
try:
|
||
rec["download_path"] = validate_download_path(rec["download_path"])
|
||
except ValueError as e:
|
||
_bad_request(e)
|
||
|
||
ntfy = sections.get("ntfy", {})
|
||
if "url" in ntfy:
|
||
_check_http_url(ntfy["url"], "NTFY")
|
||
if "events" in ntfy:
|
||
events = [e.strip() for e in ntfy["events"].split(",") if e.strip()]
|
||
invalid = [e for e in events if e not in NTFY_EVENTS]
|
||
if invalid:
|
||
raise HTTPException(400, f"Unbekannte NTFY-Ereignisse: {', '.join(invalid)}")
|
||
ntfy["events"] = ",".join(events)
|
||
|
||
plik = sections.get("plik", {})
|
||
if "url" in plik:
|
||
_check_http_url(plik["url"], "Plik")
|
||
if "ttl" in plik and plik["ttl"] and not re.fullmatch(r"\d+[dh]?", plik["ttl"]):
|
||
raise HTTPException(400, "Plik-TTL muss z.B. 30d, 12h oder eine Sekundenzahl sein")
|
||
|
||
auth = sections.get("auth", {})
|
||
new_password = auth.pop("password", None)
|
||
if new_password is not None:
|
||
if new_password == "":
|
||
raise HTTPException(400, "Passwort darf nicht leer sein")
|
||
if len(new_password) < 8:
|
||
raise HTTPException(400, "Passwort muss mindestens 8 Zeichen lang sein")
|
||
if len(new_password) > 1024:
|
||
raise HTTPException(400, "Passwort ist zu lang")
|
||
|
||
with config.lock:
|
||
will_enable = auth.get("enabled", config.get("auth", "enabled", default=False))
|
||
has_password = bool(new_password) or bool(config.get("auth", "password_hash", default=""))
|
||
if will_enable and not has_password:
|
||
raise HTTPException(400, "Zum Aktivieren der Anmeldung muss ein Passwort gesetzt sein")
|
||
|
||
auth_changed = new_password is not None or "username" in auth
|
||
if new_password is not None:
|
||
auth["password_hash"] = hash_password(new_password)
|
||
|
||
for name, values in sections.items():
|
||
config.data.setdefault(name, {}).update(values)
|
||
config.save()
|
||
|
||
if auth_changed:
|
||
clear_auth_cache()
|
||
if "level" in sections.get("logging", {}):
|
||
_apply_log_level(sections["logging"]["level"])
|
||
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)
|