stream-recorder/bin/backend/app.py

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)