#!/usr/bin/env bash
# ParlamentsTV: private YouTube/Facebook/Vimeo control; no platform credentials included.
set -Eeuo pipefail
umask 077
trap 'printf "Installation unterbrochen. Keine Plattform-Schluessel wurden gespeichert.\n" >&2' ERR
[[ "$EUID" -eq 0 ]] || { printf 'Bitte auf dem Hetzner-Server als root ausfuehren.\n'; exit 1; }
ip -4 addr show | grep -Eq 'inet 2\.28\.103\.14/' || { printf 'Dies ist nicht unser Testserver.\n'; exit 1; }
cd /opt/parlamentstv-stream-test
[[ -f .ptv-relay-owner && -f .env && -f compose.yaml && -f Caddyfile ]] || { printf 'Die vorhandene Player-Installation fehlt.\n'; exit 1; }
grep -qx 'STREAM_DOMAIN=streamtest.stadtparlamentsfernsehen.de' .env || { printf 'Unerwartete Streaming-Domain.\n'; exit 1; }
docker compose version >/dev/null
docker info >/dev/null
install -d -m 700 restream captions
keep_existing_override=0
cat > restream/Caddyfile.original <<'PTV_FILE'
{$STREAM_DOMAIN} {
  @hls path /live/LIVESTREAM/*
  handle @hls {
    reverse_proxy mediamtx:8888 {
      header_up Authorization "Bearer {$STREAM_RELAY_SECRET}"
    }
  }
  handle {
    respond "Not found" 404
  }
}
PTV_FILE
cat > restream/Caddyfile.next <<'PTV_FILE'
{$STREAM_DOMAIN} {
  @control path /control/*
  handle @control {
    reverse_proxy relay-control:9020
  }
  @hls path /live/LIVESTREAM/*
  handle @hls {
    reverse_proxy mediamtx:8888 {
      header_up Authorization "Bearer {$STREAM_RELAY_SECRET}"
    }
  }
  handle {
    respond "Not found" 404
  }
}
PTV_FILE
cat > restream/compose.override.yaml.next <<'PTV_FILE'
services:
  relay-control:
    build: ./restream
    restart: unless-stopped
    init: true
    env_file: control.env
    environment:
      PTV_RELAY_SECRET: ${STREAM_RELAY_SECRET:?Internal relay secret missing}
    volumes:
      - ptv-session-state:/state
      - ./caption-secrets/speechmatics_key:/run/secrets/speechmatics_key:ro
    read_only: true
    cap_drop: [ALL]
    security_opt: [no-new-privileges:true]
    tmpfs:
      - "/tmp:rw,noexec,nosuid,size=8m"
    mem_limit: 1024m
    pids_limit: 128
    stop_grace_period: 30s
    logging:
      driver: local
      options:
        max-size: "5m"
        max-file: "2"
  caption-worker:
    build: ./captions
    restart: unless-stopped
    init: true
    environment:
      PTV_RELAY_SECRET: ${STREAM_RELAY_SECRET:?Internal relay secret missing}
      OMP_NUM_THREADS: '1'
      OPENBLAS_NUM_THREADS: '1'
    volumes:
      - ./caption-secrets/speechmatics_key:/run/secrets/speechmatics_key:ro
    read_only: true
    cap_drop: [ALL]
    security_opt: [no-new-privileges:true]
    tmpfs:
      - "/tmp:rw,noexec,nosuid,size=8m"
    mem_limit: 768m
    cpus: 0.75
    pids_limit: 32
    logging:
      driver: local
      options:
        max-size: "1m"
        max-file: "1"
volumes:
  ptv-session-state:
PTV_FILE
cat > restream/compose.override.yaml.previous <<'PTV_FILE'
services:
  relay-control:
    build: ./restream
    restart: unless-stopped
    init: true
    env_file: control.env
    environment:
      PTV_RELAY_SECRET: ${STREAM_RELAY_SECRET:?Internal relay secret missing}
    read_only: true
    cap_drop: [ALL]
    security_opt: [no-new-privileges:true]
    tmpfs: [/tmp:rw,noexec,nosuid,size=8m]
    mem_limit: 512m
    pids_limit: 64
    stop_grace_period: 30s
    logging:
      driver: local
      options:
        max-size: "5m"
        max-file: "2"
PTV_FILE
cat > restream/compose.override.yaml.v1 <<'PTV_FILE'
services:
  relay-control:
    build: ./restream
    restart: unless-stopped
    init: true
    env_file: control.env
    environment:
      PTV_RELAY_SECRET: ${STREAM_RELAY_SECRET:?Internal relay secret missing}
    read_only: true
    cap_drop: [ALL]
    security_opt: [no-new-privileges:true]
    tmpfs:
      - "/tmp:rw,noexec,nosuid,size=8m"
    mem_limit: 512m
    pids_limit: 64
    stop_grace_period: 30s
    logging:
      driver: local
      options:
        max-size: "5m"
        max-file: "2"
PTV_FILE
cat > restream/compose.override.yaml.v2 <<'PTV_FILE'
services:
  relay-control:
    build: ./restream
    restart: unless-stopped
    init: true
    env_file: control.env
    environment:
      PTV_RELAY_SECRET: ${STREAM_RELAY_SECRET:?Internal relay secret missing}
    read_only: true
    cap_drop: [ALL]
    security_opt: [no-new-privileges:true]
    tmpfs:
      - "/tmp:rw,noexec,nosuid,size=8m"
    mem_limit: 1024m
    pids_limit: 128
    stop_grace_period: 30s
    logging:
      driver: local
      options:
        max-size: "5m"
        max-file: "2"
PTV_FILE
cat > restream/compose.override.yaml.v3 <<'PTV_FILE'
services:
  relay-control:
    build: ./restream
    restart: unless-stopped
    init: true
    env_file: control.env
    environment:
      PTV_RELAY_SECRET: ${STREAM_RELAY_SECRET:?Internal relay secret missing}
    volumes:
      - ptv-session-state:/state
    read_only: true
    cap_drop: [ALL]
    security_opt: [no-new-privileges:true]
    tmpfs:
      - "/tmp:rw,noexec,nosuid,size=8m"
    mem_limit: 1024m
    pids_limit: 128
    stop_grace_period: 30s
    logging:
      driver: local
      options:
        max-size: "5m"
        max-file: "2"
volumes:
  ptv-session-state:
PTV_FILE
cat > restream/compose.override.yaml.v4 <<'PTV_FILE'
services:
  relay-control:
    build: ./restream
    restart: unless-stopped
    init: true
    env_file: control.env
    environment:
      PTV_RELAY_SECRET: ${STREAM_RELAY_SECRET:?Internal relay secret missing}
    volumes:
      - ptv-session-state:/state
    read_only: true
    cap_drop: [ALL]
    security_opt: [no-new-privileges:true]
    tmpfs:
      - "/tmp:rw,noexec,nosuid,size=8m"
    mem_limit: 1024m
    pids_limit: 128
    stop_grace_period: 30s
    logging:
      driver: local
      options:
        max-size: "5m"
        max-file: "2"
  caption-worker:
    build: ./captions
    restart: unless-stopped
    init: true
    environment:
      PTV_RELAY_SECRET: ${STREAM_RELAY_SECRET:?Internal relay secret missing}
      OMP_NUM_THREADS: '1'
      OPENBLAS_NUM_THREADS: '1'
    read_only: true
    cap_drop: [ALL]
    security_opt: [no-new-privileges:true]
    tmpfs:
      - "/tmp:rw,noexec,nosuid,size=8m"
    mem_limit: 768m
    cpus: 0.75
    pids_limit: 32
    logging:
      driver: local
      options:
        max-size: "1m"
        max-file: "1"
volumes:
  ptv-session-state:
PTV_FILE
if [[ ! -f restream/.ptv-restream-owner ]]; then
  cmp -s Caddyfile restream/Caddyfile.original || { printf 'Caddyfile wurde individuell geaendert. Bitte zuerst pruefen lassen.\n'; exit 1; }
  [[ ! -e compose.override.yaml && ! -e compose.override.yml && ! -e control.env ]] || { printf 'Vorhandene fremde Zusatzkonfiguration. Es wird nichts ueberschrieben.\n'; exit 1; }
  cp -p Caddyfile restream/Caddyfile.before
  printf 'PTV_CONTROL_TOKEN=%s\n' "$(openssl rand -hex 32)" > control.env
  printf 'ParlamentsTV restream control v1\n' > restream/.ptv-restream-owner
else
  [[ -f control.env ]] || { printf 'Steuerschluessel-Datei fehlt.\n'; exit 1; }
  if ! cmp -s Caddyfile restream/Caddyfile.next && ! cmp -s Caddyfile restream/Caddyfile.original; then
    printf 'Caddyfile hat fremde Aenderungen. Es wird nichts ueberschrieben.\n'; exit 1
  fi
  if [[ -f compose.override.yaml ]] && ! cmp -s compose.override.yaml restream/compose.override.yaml.next && ! cmp -s compose.override.yaml restream/compose.override.yaml.previous && ! cmp -s compose.override.yaml restream/compose.override.yaml.v1 && ! cmp -s compose.override.yaml restream/compose.override.yaml.v2 && ! cmp -s compose.override.yaml restream/compose.override.yaml.v3 && ! cmp -s compose.override.yaml restream/compose.override.yaml.v4; then
    if python3 - <<'PTV_COMPARE'
import json, subprocess

def load(override):
    result=subprocess.run(['docker','compose','-p','parlamentstv-stream-test','-f','compose.yaml','-f',override,'config','--format','json'],capture_output=True,text=True)
    if result.returncode:raise ValueError('Invalid configuration')
    return json.loads(result.stdout)
try:
    current=load('compose.override.yaml');expected=load('restream/compose.override.yaml.next')
    for config in (current,expected):
        options=config['services']['caption-worker']['logging']['options']
        count=str(options.pop('max-file'))
        if not count.isdigit() or not 1<=int(count)<=100:raise ValueError('Invalid log retention')
    if current!=expected:raise ValueError('Other configuration changes')
except Exception:raise SystemExit(1)
PTV_COMPARE
    then
      keep_existing_override=1
      printf 'Vorhandene Logdatei-Anzahl bleibt erhalten. Servercode wird aktualisiert.\n'
    else
      printf 'Zusatzkonfiguration hat fremde Aenderungen. Es wird nichts ueberschrieben.\n'; exit 1
    fi
  fi
fi
# Change only the reviewed HLS retention field, preserving other server settings.
python3 - <<'PTV_HLS_WINDOW'
from pathlib import Path
import re
path=Path('mediamtx.yml');text=path.read_text()
pattern=r'(?m)^hlsSegmentCount:[ \t]*(\d+)[ \t]*(?:#.*)?$'
matches=list(re.finditer(pattern,text))
if len(matches)!=1:raise SystemExit('HLS-Segmentanzahl nicht eindeutig. Keine Aenderung.')
count=int(matches[0].group(1))
if count==7:
    Path('restream/mediamtx.before-live-buffer.yml').write_text(text)
    path.write_text(re.sub(pattern,'hlsSegmentCount: 30',text))
elif not 30<=count<=120:raise SystemExit('Unerwartete HLS-Segmentanzahl. Keine Aenderung.')
# Preserve the HLS buffer change above and all unrelated configuration.
import json
text=path.read_text()
origin_lines=[line for line in text.splitlines() if line.startswith('hlsAllowOrigins:')]
if len(origin_lines)!=1:raise SystemExit('HLS-Websitefreigabe nicht eindeutig. Keine Aenderung.')
origins=json.loads(origin_lines[0].split(':',1)[1].strip())
if not isinstance(origins,list) or any(not isinstance(value,str) for value in origins):raise SystemExit('Unerwartete HLS-Websitefreigabe.')
for origin in ('https://parlamentstv-player-labor.trickfilmkinder.chatgpt.site','https://mediaplayer.ptvmedienkompetenz.de'):
    if origin not in origins:origins.append(origin)
next_text=text.replace(origin_lines[0],'hlsAllowOrigins: '+json.dumps(origins))
if next_text!=text:
    Path('restream/mediamtx.before-player-domain.yml').write_text(text)
    path.write_text(next_text)
print('HLS-Puffer und Freigabe fuer mediaplayer.ptvmedienkompetenz.de eingerichtet.')
PTV_HLS_WINDOW
chmod 600 control.env
grep -Eq '^PTV_CONTROL_TOKEN=[0-9a-f]{64}$' control.env || { printf 'Ungueltige Steuerschluessel-Datei.\n'; exit 1; }
cat > restream/control.py <<'PTV_SOURCE'
"""Private control API. Never logs request bodies, tokens or destination URLs."""
import hmac
import json
import os
import re
import signal
import subprocess
import threading
import time
from pathlib import Path
import urllib.error
import urllib.parse
import urllib.request
from continuity import Continuity
from audience import Audience, AudienceError

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

PLATFORMS = ("youtube", "facebook", "vimeo")
OUTPUT_PORTS = {"youtube": 15000, "facebook": 15001, "vimeo": 15002}
ALLOWED_ORIGIN = "https://parlamentstv-player-labor.trickfilmkinder.chatgpt.site"
ALLOWED_ORIGINS = (ALLOWED_ORIGIN, "https://mediaplayer.ptvmedienkompetenz.de")


class ControlError(Exception):
    def __init__(self, message, status=400):
        super().__init__(message)
        self.status = status


def destination(platform, server, key):
    if platform not in PLATFORMS:
        raise ControlError("Unbekanntes Ziel.")
    if not isinstance(server, str) or not isinstance(key, str):
        raise ControlError("Serveradresse und Stream-Key fehlen.")
    server, key = server.strip(), key.strip()
    if not key or len(key) > 1024 or re.search(r"[\s\x00-\x1f\x7f?#]", key):
        raise ControlError("Bitte einen gueltigen Stream-Key eintragen.")
    try:
        parsed = urllib.parse.urlsplit(server)
        host = (parsed.hostname or "").lower()
        port = parsed.port
    except ValueError:
        raise ControlError("Bitte die RTMPS-Serveradresse der Plattform eintragen.") from None
    valid_host = {
        "youtube": host.endswith(".rtmps.youtube.com") or host == "rtmps.youtube.com",
        "facebook": host.endswith(".facebook.com"),
        "vimeo": host == "rtmp-global.cloud.vimeo.com",
    }[platform]
    if (parsed.scheme != "rtmps" or not valid_host or port not in (None, 443)
            or parsed.username or parsed.password or parsed.query or parsed.fragment
            or not parsed.path.strip("/") or len(server) > 2048
            or re.search(r"[\s\x00-\x1f\x7f]", server)):
        raise ControlError("Bitte eine RTMPS-Serveradresse der gewaehlten Plattform ohne Stream-Key verwenden.")
    return urllib.parse.urlunsplit(("rtmps", host + ":443", parsed.path.rstrip("/") + "/" + key, "", ""))


def ffmpeg_command(source, target):
    return ["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "quiet", "-progress", "pipe:1",
            "-rw_timeout", "15000000", "-i", source, "-map", "0:v:0", "-map", "0:a:0?",
            "-c", "copy", "-tls_verify", "1", "-verifyhost", urllib.parse.urlsplit(target).hostname,
            "-ca_file", "/etc/ssl/certs/ca-certificates.crt",
            "-rw_timeout", "15000000", "-f", "flv", "-flvflags", "no_duration_filesize", target]


class Relay:
    def __init__(self, source=None, session_file=None):
        self.source = source or "rtmp://mediamtx:1935/live/LIVESTREAM"
        self.lock = threading.RLock()
        self.session_file = Path(session_file) if session_file else None
        self.session = {"state": "ready", "revision": 0}
        if self.session_file and self.session_file.exists():
            try:
                saved = json.loads(self.session_file.read_text())
                if (set(saved) != {"state", "revision"} or saved["state"] not in ("ready", "ended")
                        or type(saved["revision"]) is not int or saved["revision"] < 0):
                    raise ValueError("invalid session")
                self.session = saved
            except (OSError, ValueError, TypeError):
                # Corrupt state must never restart a session that might be ended.
                self.session = {"state": "ended", "revision": 0}
        self.audience = Audience(self.session_file.parent / "audience.sqlite3" if self.session_file else ":memory:")
        self.continuity = None
        self.captions_enabled = False
        self.captions_reviewed = False
        self.caption_revision = 0
        self.caption_started_at = 0
        self.caption_name = ""
        self.caption_vocabulary = []
        self.outputs = {name: {"state": "stopped", "process": None, "seconds": 0} for name in PLATFORMS}

    def input_state(self):
        request = urllib.request.Request("http://mediamtx:8888/live/LIVESTREAM/index.m3u8",
                                         headers={"Authorization": "Bearer " + os.environ.get("PTV_RELAY_SECRET", "")})
        try:
            with urllib.request.urlopen(request, timeout=3) as response:
                return "live" if response.read(2048).lstrip().startswith(b"#EXTM3U") else "unavailable"
        except urllib.error.HTTPError as error:
            return "waiting" if error.code == 404 else "unavailable"
        except (OSError, urllib.error.URLError):
            return "unavailable"

    def snapshot(self, include_input=True):
        with self.lock:
            outputs = {}
            for name, entry in self.outputs.items():
                process = entry["process"]
                if process and process.poll() is not None and entry["state"] in ("starting", "forwarding"):
                    entry["state"] = "ended" if process.returncode == 0 else "error"
                outputs[name] = {"state": entry["state"], "seconds": entry["seconds"]}
        return {"audience_supported": True, "automatic_captions": self.caption_status(), "session": self.viewer_session()["session"], "continuity": self.continuity.state() if self.continuity else "off", "platforms": list(PLATFORMS), "version": 1, "input": self.input_state() if include_input else "unknown", "outputs": outputs}

    def read_progress(self, platform, process):
        try:
            for line in process.stdout:
                if line.startswith("out_time_us="):
                    value = line.partition("=")[2].strip()
                    if value.lstrip("-").isdigit() and int(value) > 0:
                        with self.lock:
                            entry = self.outputs[platform]
                            if entry["process"] is process and entry["state"] in ("starting", "forwarding"):
                                entry.update(state="forwarding", seconds=int(value) // 1000000)
        finally:
            process.stdout.close()

    def start(self, items):
        if not isinstance(items, list) or not 1 <= len(items) <= len(PLATFORMS):
            raise ControlError("Bitte ein bis drei Ziele auswaehlen.")
        prepared = []
        for item in items:
            if not isinstance(item, dict) or set(item) != {"platform", "server", "key"}:
                raise ControlError("Ungueltige Zielkonfiguration.")
            name = item["platform"]
            prepared.append((name, destination(name, item["server"], item["key"])))
        if len({name for name, _ in prepared}) != len(prepared):
            raise ControlError("Jedes Ziel darf nur einmal ausgewaehlt werden.")
        with self.lock:
            if self.session["state"] == "ended":
                raise ControlError("Sitzung beendet. Zuerst eine neue Sitzung starten.", 409)
            if any(self.outputs[name]["process"] and self.outputs[name]["process"].poll() is None for name, _ in prepared):
                raise ControlError("Dieses Ziel laeuft bereits. Zuerst stoppen.", 409)
            if self.input_state() != "live":
                raise ControlError("Noch kein Livesignal. Zuerst Livestream Studio an unseren RTMP-Eingang senden lassen.", 409)
            if Continuity and not self.continuity:
                try:
                    self.continuity = Continuity(self.source)
                    self.continuity.start()
                except (OSError, RuntimeError):
                    if self.continuity:
                        self.continuity.stop()
                    self.continuity = None
                    raise ControlError("Pausenbild-Pipeline konnte nicht gestartet werden. Server-Erweiterung pruefen.", 503) from None
            started = []
            try:
                for name, target in prepared:
                    process = subprocess.Popen(ffmpeg_command(("udp://127.0.0.1:" + str(OUTPUT_PORTS[name]) + "?fifo_size=4096&overrun_nonfatal=1") if self.continuity else self.source, target), stdin=subprocess.DEVNULL,
                                               stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                                               text=True, start_new_session=True)
                    process.args = []
                    self.outputs[name] = {"state": "starting", "process": process, "seconds": 0}
                    started.append(name)
                    threading.Thread(target=self.read_progress, args=(name, process), daemon=True).start()
            except OSError:
                for name in started:
                    self.stop(name)
                if self.continuity and not started:
                    self.continuity.stop(); self.continuity = None
                raise ControlError("Weiterleitung konnte nicht gestartet werden. Server-Erweiterung pruefen.", 503) from None
        return self.snapshot(False)

    def stop(self, platform):
        if platform not in (*PLATFORMS, "all"):
            raise ControlError("Unbekanntes Ziel.")
        with self.lock:
            for name in PLATFORMS if platform == "all" else (platform,):
                process = self.outputs[name]["process"]
                if process and process.poll() is None:
                    process.terminate()
                    try:
                        process.wait(timeout=5)
                    except subprocess.TimeoutExpired:
                        process.kill()
                        process.wait(timeout=2)
                self.outputs[name] = {"state": "stopped", "process": None, "seconds": 0}
            if self.continuity and all(entry["state"] == "stopped" for entry in self.outputs.values()):
                self.continuity.stop()
                self.continuity = None
        return self.snapshot(False)


    def viewer_session(self):
        with self.lock:
            return {"version": 1, "session": dict(self.session), "audience_live_supported": True}

    def save_session(self, state):
        next_session = {"state": state, "revision": self.session["revision"] + 1}
        if self.session_file:
            temporary = self.session_file.with_suffix(".tmp")
            with temporary.open("w") as handle:
                json.dump(next_session, handle)
                handle.flush()
                os.fsync(handle.fileno())
            temporary.chmod(0o600)
            os.replace(temporary, self.session_file)
        self.session = next_session

    def change_session(self, action):
        if action not in ("end", "new"):
            raise ControlError("Ungueltige Sitzungsaktion.")
        with self.lock:
            if self.session["state"] == "ready":
                self.audience.close_session(self.session["revision"])
            self.captions_enabled = False
            self.captions_reviewed = False
            self.caption_revision += 1
            self.caption_name = ""
            self.caption_vocabulary = []
            if action == "end":
                if self.session["state"] != "ended":
                    self.save_session("ended")
                self.stop("all")
            else:
                self.stop("all")
                self.save_session("ready")
            return self.snapshot(False)


    def captions(self):
        with self.lock:
            if not self.captions_enabled or self.session["state"]=="ended":
                return {"version":1,"enabled":False,"state":"off","lag":0,"cues":[]}
        request=urllib.request.Request("http://caption-worker:9030/captions",headers={"Authorization":"Bearer "+os.environ.get("PTV_RELAY_SECRET","")})
        try:
            with urllib.request.urlopen(request,timeout=2) as response:
                raw=response.read(131073)
            if len(raw)>131072:raise ValueError("Too large")
            result=json.loads(raw);result["enabled"]=True
            with self.lock:
                return result if self.captions_enabled and self.session["state"]!="ended" else {"version":1,"enabled":False,"state":"off","lag":0,"cues":[]}
        except (OSError,ValueError,urllib.error.URLError):
            return {"version":1,"enabled":True,"state":"unavailable","lag":0,"cues":[]}

    def caption_transcript(self,format):
        if format not in ('txt','vtt'):raise ControlError('Ungueltiges Exportformat.')
        request=urllib.request.Request('http://caption-worker:9030/transcript?format='+format,headers={'Authorization':'Bearer '+os.environ.get('PTV_RELAY_SECRET','')})
        try:
            with urllib.request.urlopen(request,timeout=10) as response:raw=response.read(32*1024*1024+1)
            if len(raw)>32*1024*1024:raise ValueError('Too large')
            result=json.loads(raw)
            if not result.get('available'):raise ControlError('Noch kein Sitzungstext vorhanden oder bereits geloescht.',404)
            return result
        except (OSError,ValueError,urllib.error.URLError):raise ControlError('Transkript nicht erreichbar. Server-Erweiterung aktualisieren.',503)

    def caption_key_ready(self):
        try:return 16<=Path("/run/secrets/speechmatics_key").stat().st_size<=2049
        except OSError:return False

    def caption_status(self):
        with self.lock:
            return {"supported":True,"transcript_supported":True,"provider":"speechmatics-eu-enhanced","enabled":self.captions_enabled,"configured":self.caption_key_ready(),"name":self.caption_name,"vocabulary":self.caption_vocabulary}

    def caption_config(self):
        with self.lock:
            return {"enabled":self.captions_enabled,"reviewed":self.captions_reviewed,"revision":self.caption_revision,"since":self.caption_started_at,"session":dict(self.session),"vocabulary":list(self.caption_vocabulary)}

    def caption_settings(self,name,vocabulary):
        if not isinstance(name,str) or len(name)>100 or re.search(r"[\x00-\x1f\x7f]",name):raise ControlError("Ungueltiger Woerterbuchname.")
        if not isinstance(vocabulary,list) or len(vocabulary)>100:raise ControlError("Maximal 100 Woerterbucheintraege im Test.")
        normalized=[]
        for item in vocabulary:
            if not isinstance(item,dict) or set(item) not in ({"content"},{"content","sounds_like"}):raise ControlError("Ungueltiger Woerterbucheintrag.")
            texts=[item.get("content")]
            if "sounds_like" in item:
                if not isinstance(item["sounds_like"],list) or not 1<=len(item["sounds_like"])<=3:raise ControlError("Ungueltiger Aussprachehinweis.")
                texts+=item["sounds_like"]
            if any(not isinstance(text,str) or not text.strip() or len(text)>240 or len(text.split())>6 or re.search(r"[\x00-\x1f\x7f]",text) for text in texts):raise ControlError("Je Eintrag maximal sechs Woerter und 240 Zeichen.")
            row={"content":texts[0].strip()}
            if len(texts)>1:row["sounds_like"]=[text.strip() for text in texts[1:]]
            normalized.append(row)
        if len(json.dumps(normalized,ensure_ascii=False).encode())>6500:raise ControlError("Woerterbuch zu gross fuer diesen Test.")
        with self.lock:
            if self.captions_enabled:raise ControlError("Spracherkennung vor dem Woerterbuchwechsel ausschalten.",409)
            self.caption_name=name.strip();self.caption_vocabulary=normalized;self.caption_revision+=1
        return self.snapshot(False)

    def enable_captions(self,enabled,reviewed=False):
        if type(enabled) is not bool or type(reviewed) is not bool:raise ControlError("Ungueltige Untertitel-Einstellung.")
        with self.lock:
            if enabled and self.session["state"]=="ended":raise ControlError("Zuerst neue Sitzung starten.",409)
            if enabled and not reviewed:raise ControlError("Verarbeitung ueber Speechmatics EU fuer diesen Test zuerst freigeben.")
            if enabled and not self.caption_key_ready():raise ControlError("Speechmatics API-Schluessel zuerst geschuetzt im Serverfenster hinterlegen.",409)
            self.captions_enabled=enabled;self.captions_reviewed=enabled and reviewed;self.caption_revision+=1
            self.caption_started_at=time.time() if enabled else 0
        return self.snapshot(False)


def handler_for(relay, token):
    class Handler(BaseHTTPRequestHandler):
        def setup(self):
            super().setup()
            self.connection.settimeout(15)

        def log_message(self, *_):
            pass

        def respond(self, status, payload):
            body = json.dumps(payload, ensure_ascii=True).encode()
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.send_header("Cache-Control", "no-store")
            self.send_header("X-Content-Type-Options", "nosniff")
            origin = self.headers.get("Origin")
            if origin in ALLOWED_ORIGINS:
                self.send_header("Access-Control-Allow-Origin", origin)
            self.send_header("Vary", "Origin")
            self.end_headers()
            self.wfile.write(body)

        def authorize(self):
            origin = self.headers.get("Origin")
            if origin and origin not in ALLOWED_ORIGINS:
                raise ControlError("Zugriff von dieser Website ist nicht erlaubt.", 403)
            if not hmac.compare_digest(self.headers.get("Authorization", "").encode(), ("Bearer " + token).encode()):
                raise ControlError("Steuerschluessel fehlt oder ist ungueltig.", 401)

        def do_OPTIONS(self):
            if self.headers.get("Origin") not in ALLOWED_ORIGINS:
                self.respond(403, {"error": "Zugriff nicht erlaubt."})
                return
            self.send_response(204)
            origin = self.headers.get("Origin")
            if origin in ALLOWED_ORIGINS:
                self.send_header("Access-Control-Allow-Origin", origin)
            self.send_header("Vary", "Origin")
            self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
            self.send_header("Access-Control-Allow-Headers", "Authorization, Content-Type")
            self.send_header("Access-Control-Max-Age", "300")
            self.send_header("Content-Length", "0")
            self.end_headers()

        def do_GET(self):
            try:
                if self.path == "/control/caption-config":
                    secret=os.environ.get("PTV_RELAY_SECRET", "")
                    if not secret or not hmac.compare_digest(self.headers.get("Authorization", "").encode(),("Bearer "+secret).encode()):
                        raise ControlError("Zugriff nicht erlaubt.",403)
                    self.respond(200,relay.caption_config())
                    return
                if self.path == "/control/captions":
                    origin=self.headers.get("Origin")
                    if origin and origin not in ALLOWED_ORIGINS:raise ControlError("Zugriff nicht erlaubt.",403)
                    self.respond(200,relay.captions())
                    return
                if self.path == "/control/session":
                    origin = self.headers.get("Origin")
                    if origin and origin not in ALLOWED_ORIGINS:
                        raise ControlError("Zugriff nicht erlaubt.", 403)
                    self.respond(200, relay.viewer_session())
                    return
                self.authorize()
                if self.path in ("/control/caption-transcript?format=txt","/control/caption-transcript?format=vtt"):
                    self.respond(200,relay.caption_transcript(self.path.rsplit("=",1)[1]))
                    return
                if self.path.startswith("/control/audience?") or self.path == "/control/audience":
                    params=urllib.parse.parse_qs(urllib.parse.urlsplit(self.path).query)
                    self.respond(200, relay.audience.report(params.get("group",[""])[0]))
                    return
                if self.path != "/control/status":
                    raise ControlError("Nicht gefunden.", 404)
                self.respond(200, relay.snapshot())
            except (ControlError, AudienceError) as error:
                self.respond(error.status, {"error": str(error)})

        def do_POST(self):
            try:
                public_event = self.path == "/control/audience/event"
                if public_event:
                    if self.headers.get("Origin") not in ALLOWED_ORIGINS:
                        raise ControlError("Zugriff nicht erlaubt.",403)
                else:
                    self.authorize()
                if self.headers.get("Content-Type", "").split(";")[0] != "application/json":
                    raise ControlError("JSON erforderlich.", 415)
                try:
                    length = int(self.headers.get("Content-Length", "0"))
                except ValueError:
                    raise ControlError("Ungueltige Anfrage.") from None
                if not 0 < length <= 8192:
                    raise ControlError("Anfrage zu gross oder leer.", 413)
                data = json.loads(self.rfile.read(length))
                if not isinstance(data, dict):
                    raise ControlError("Ungueltige Anfrage.")
                if public_event:
                    with relay.lock:
                        result = relay.audience.record(data, relay.session)
                elif self.path == "/control/audience/delete" and set(data) == {"group"}:
                    result = relay.audience.delete(data["group"])
                elif self.path == "/control/start" and set(data) == {"outputs"}:
                    result = relay.start(data["outputs"])
                elif self.path == "/control/stop" and set(data) == {"platform"}:
                    result = relay.stop(data["platform"])
                elif self.path == "/control/captions" and set(data) in ({"enabled"},{"enabled","reviewed"}):
                    result = relay.enable_captions(data["enabled"],data.get("reviewed",False))
                elif self.path == "/control/caption-settings" and set(data) == {"name","vocabulary"}:
                    result = relay.caption_settings(data["name"],data["vocabulary"])
                elif self.path == "/control/session" and set(data) == {"action"}:
                    result = relay.change_session(data["action"])
                else:
                    raise ControlError("Nicht gefunden oder ungueltige Parameter.", 404)
                self.respond(200, result)
            except AudienceError as error:
                self.respond(error.status,{"error":str(error)})
            except (json.JSONDecodeError, UnicodeError):
                self.respond(400, {"error": "Ungueltiges JSON."})
            except ControlError as error:
                self.respond(error.status, {"error": str(error)})
            except Exception:
                self.respond(500, {"error": "Serverfehler. Keine Zugangsdaten wurden protokolliert."})

    return Handler


if __name__ == "__main__":
    control_token = os.environ.get("PTV_CONTROL_TOKEN", "")
    if not re.fullmatch(r"[a-f0-9]{64}", control_token):
        raise SystemExit("Gueltiger Steuerschluessel erforderlich.")
    relay = Relay(session_file=os.environ.get("PTV_SESSION_FILE", "/state/session.json"))
    def clean_statistics():
        while True:
            relay.audience.cleanup()
            time.sleep(3600)
    threading.Thread(target=clean_statistics, daemon=True).start()
    server = ThreadingHTTPServer(("0.0.0.0", 9020), handler_for(relay, control_token))

    def stop_server(*_):
        relay.stop("all")
        raise SystemExit(0)

    signal.signal(signal.SIGTERM, stop_server)
    signal.signal(signal.SIGINT, stop_server)
    server.serve_forever()
PTV_SOURCE
cat > restream/audience.py <<'PTV_SOURCE'
"""Cookieless playback totals. No IP, user agent, referrer or media URL stored."""
import hashlib
import math
import re
import sqlite3
import threading
import time
from datetime import datetime
from zoneinfo import ZoneInfo

class AudienceError(Exception):
    def __init__(self, message, status=400):
        super().__init__(message)
        self.status=status

class Audience:
    def __init__(self, path, clock=time.time):
        self.clock=clock
        self.lock=threading.RLock()
        self.db=sqlite3.connect(str(path),check_same_thread=False)
        self.db.execute('PRAGMA journal_mode=WAL')
        self.db.execute('PRAGMA busy_timeout=5000')
        self.db.execute('PRAGMA secure_delete=ON')
        self.db.executescript('''
        CREATE TABLE IF NOT EXISTS visits (
          id TEXT PRIMARY KEY, source TEXT NOT NULL, label TEXT NOT NULL,
          mode TEXT NOT NULL, revision INTEGER NOT NULL, channel TEXT NOT NULL,
          started REAL NOT NULL, updated REAL NOT NULL, hour INTEGER NOT NULL,
          seconds REAL NOT NULL, reported REAL NOT NULL);
        CREATE INDEX IF NOT EXISTS visits_started ON visits(started);
        CREATE INDEX IF NOT EXISTS visits_group ON visits(source,mode,revision);
        CREATE TABLE IF NOT EXISTS closed_sessions (revision INTEGER PRIMARY KEY, ended REAL NOT NULL);
        CREATE TABLE IF NOT EXISTS tombstones (id TEXT PRIMARY KEY, expires REAL NOT NULL);
        ''')
        if 'active' not in {row[1] for row in self.db.execute('PRAGMA table_info(visits)')}:
            self.db.execute('ALTER TABLE visits ADD COLUMN active INTEGER NOT NULL DEFAULT 0')
        self.db.commit()
        self.pruned=0
        self.rate_time=0
        self.rate_count=0

    def prune(self, now):
        if now-self.pruned<3600:return
        self.db.execute('DELETE FROM visits WHERE started < ?', (now-90*86400,))
        self.db.execute('DELETE FROM tombstones WHERE expires < ?', (now,))
        self.db.execute('DELETE FROM closed_sessions WHERE ended < ?', (now-90*86400,))
        self.pruned=now

    def cleanup(self):
        with self.lock,self.db:
            self.prune(self.clock())

    @staticmethod
    def group(source, mode, revision):return f'{source}:{mode}:{revision}'

    def record(self, data, session):
        if set(data) not in ({'id','source','label','mode','channel','seconds','revision'}, {'id','source','label','mode','channel','seconds','revision','active'}):
            raise AudienceError('Ungueltige Messung.')
        if (not isinstance(data['id'],str) or not re.fullmatch(r'[a-f0-9-]{36}',data['id'])
            or not isinstance(data['source'],str) or not re.fullmatch(r'[a-f0-9]{64}',data['source'])
            or data['mode'] not in ('live','archive') or data['channel'] not in ('direct','embed')
            or not isinstance(data['label'],str) or len(data['label'])>120
            or type(data['revision']) is not int or not 0<=data['revision']<=2147483647
            or type(data['seconds']) not in (int,float) or not math.isfinite(data['seconds'])
            or not 0<=data['seconds']<=86400):raise AudienceError('Ungueltige Messung.')
        if 'active' in data and type(data['active']) is not bool:raise AudienceError('Ungueltiger Wiedergabestatus.')
        now=self.clock()
        revision=data['revision'] if data['mode']=='live' else 0
        # Hash random playback IDs before retaining them. They never identify people.
        key=hashlib.sha256(data['id'].encode()).hexdigest()
        label=''.join(c for c in data['label'] if c>=' ' and c!='\x7f').strip() or 'Ohne Bezeichnung'
        with self.lock, self.db:
            self.prune(now)
            if int(now)!=self.rate_time:self.rate_time=int(now);self.rate_count=0
            self.rate_count+=1
            if self.rate_count>200:raise AudienceError('Zu viele Messmeldungen. Bitte spaeter erneut.',429)
            if self.db.execute('SELECT 1 FROM tombstones WHERE id=?',(key,)).fetchone():
                return {'accepted':True,'deleted':True}
            row=self.db.execute('SELECT source,mode,revision,started,updated,seconds,reported FROM visits WHERE id=?',(key,)).fetchone()
            active=data.get('active',not row or float(data['seconds'])>row[6])
            effective_now=now
            if data['mode']=='live' and (session['state']=='ended' or revision!=session['revision']):
                closed=self.db.execute('SELECT ended FROM closed_sessions WHERE revision=?',(revision,)).fetchone()
                if not row or not closed or now>closed[0]+60:
                    raise AudienceError('Sitzung gewechselt oder beendet.',409)
                effective_now=closed[0]
                active=False
            if row:
                if (row[0],row[1],row[2])!=(data['source'],data['mode'],revision):raise AudienceError('Messkennung passt nicht.')
                # Cumulative updates are idempotent and tolerate delayed/out-of-order requests.
                reported=max(row[6],float(data['seconds']))
                delta=min(max(0,reported-row[6]),max(0,effective_now-row[4])+2,60)
                seconds=min(row[5]+delta,max(0,effective_now-row[3])+2)
                self.db.execute('UPDATE visits SET updated=?,seconds=?,reported=?,active=? WHERE id=?',(effective_now,seconds,reported,int(active),key))
            else:
                if self.db.execute('SELECT COUNT(*) FROM visits').fetchone()[0]>=200000:raise AudienceError('Messspeicher voll.',503)
                hour=datetime.fromtimestamp(now,ZoneInfo('Europe/Berlin')).hour
                self.db.execute('INSERT INTO visits (id,source,label,mode,revision,channel,started,updated,hour,seconds,reported,active) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)',
                    (key,data['source'],label,data['mode'],revision,data['channel'],now,now,hour,0,float(data['seconds']),int(active)))
        return {'accepted':True}

    def close_session(self,revision):
        with self.lock,self.db:
            self.db.execute('INSERT OR IGNORE INTO closed_sessions VALUES (?,?)',(revision,self.clock()))

    def report(self, group=''):
        if group and not re.fullmatch(r'[a-f0-9]{64}:(live|archive):[0-9]{1,10}',group):raise AudienceError('Ungueltiger Filter.')
        with self.lock, self.db:
            self.prune(self.clock())
            where='';args=()
            if group:
                source,mode,revision=group.split(':');where=' WHERE source=? AND mode=? AND revision=?';args=(source,mode,int(revision))
            groups=[{'id':self.group(s,m,v),'label':l,'mode':m,'revision':v,'started':t,'count':c}
                for s,m,v,l,t,c in self.db.execute('SELECT source,mode,revision,MAX(label),MIN(started),COUNT(*) FROM visits GROUP BY source,mode,revision ORDER BY MIN(started) DESC')]
            rows={h:(c,s) for h,c,s in self.db.execute('SELECT hour,COUNT(*),SUM(seconds) FROM visits'+where+' GROUP BY hour',args)}
            hours=[{'hour':h,'count':rows.get(h,(0,0))[0],'seconds':rows.get(h,(0,0))[1],
                    'average':rows[h][1]/rows[h][0] if h in rows else 0} for h in range(24)]
            count=sum(h['count'] for h in hours);seconds=sum(h['seconds'] for h in hours)
            channels=dict(self.db.execute('SELECT channel,COUNT(*) FROM visits'+where+' GROUP BY channel',args))
            live_where=(where+' AND ' if where else ' WHERE ')+"mode='live' AND active=1 AND updated>=? AND revision NOT IN (SELECT revision FROM closed_sessions)"
            live_channels=dict(self.db.execute('SELECT channel,COUNT(*) FROM visits'+live_where+' GROUP BY channel',(*args,self.clock()-45)))
            current_live={'total':sum(live_channels.values()),'direct':live_channels.get('direct',0),'embed':live_channels.get('embed',0),'timeoutSeconds':45}
            return {'supported':True,'currentLive':current_live,'groups':groups,'summary':{'hours':hours,'count':count,'totalSeconds':seconds,'average':seconds/count if count else 0},'channels':channels,'retentionDays':90,'updated':self.clock()}

    def delete(self, group):
        if not isinstance(group,str) or (group and not re.fullmatch(r'[a-f0-9]{64}:(live|archive):[0-9]{1,10}',group)):raise AudienceError('Ungueltiger Filter.')
        where='';args=()
        if group:
            s,m,v=group.split(':');where=' WHERE source=? AND mode=? AND revision=?';args=(s,m,int(v))
        with self.lock,self.db:
            # Do not recreate deleted playback records on their next heartbeat.
            self.db.execute('INSERT OR REPLACE INTO tombstones SELECT id,? FROM visits'+where,(self.clock()+86400,*args))
            self.db.execute('DELETE FROM visits'+where,args)
        return {'deleted':True}
PTV_SOURCE
cat > restream/continuity.py <<'PTV_SOURCE'
"""Continuous clocked A/V output. Source decoder may restart; encoder never does on source loss."""
import collections
import os
import select
import subprocess
import threading
import time

WIDTH, HEIGHT, FPS = 1280, 720, 25
FRAME_BYTES = WIDTH * HEIGHT * 3 // 2
AUDIO_BYTES = 48000 * 2 * 2 // 50  # 20 ms stereo PCM


def terminate(process):
    if process and process.poll() is None:
        process.terminate()
        try:
            process.wait(timeout=3)
        except subprocess.TimeoutExpired:
            process.kill()
            process.wait(timeout=2)


def read_exact(pipe, size):
    parts = bytearray()
    while len(parts) < size:
        part = pipe.read(size - len(parts))
        if not part:
            return None
        parts.extend(part)
    return bytes(parts)


class Continuity:
    def __init__(self, source):
        self.source = source
        self.lock = threading.RLock()
        self.closed = threading.Event()
        self.frame = None
        self.last_frame = 0
        self.audio = collections.deque(maxlen=15)
        self.decoder = None
        self.encoder = None
        self.threads = []
        with open(os.environ.get('PTV_PAUSE_FRAME', '/app/pause.yuv'), 'rb') as pause_file:
            self.pause_frame = pause_file.read()
        if len(self.pause_frame) != FRAME_BYTES:
            raise RuntimeError('Invalid pause frame')

    def source_live(self):
        with self.lock:
            return self.frame is not None and time.monotonic() - self.last_frame < 2

    def state(self):
        if not self.encoder or self.encoder.poll() is not None:
            return 'error'
        return 'live' if self.source_live() else 'pause'

    def launch(self, function, *args):
        thread = threading.Thread(target=function, args=args, daemon=True)
        self.threads.append(thread)
        thread.start()

    def start(self):
        ar, aw = os.pipe()
        command = ['ffmpeg', '-nostdin', '-hide_banner', '-loglevel', 'quiet',
                   '-thread_queue_size', '8', '-f', 'rawvideo', '-pixel_format', 'yuv420p',
                   '-video_size', f'{WIDTH}x{HEIGHT}', '-framerate', str(FPS), '-i', 'pipe:0',
                   '-thread_queue_size', '32', '-f', 's16le', '-ar', '48000', '-ac', '2', '-i', f'pipe:{ar}',
                   '-map', '0:v', '-map', '1:a', '-c:v', 'libx264', '-preset', 'ultrafast', '-tune', 'zerolatency',
                   '-threads', '2', '-b:v', '2500k', '-maxrate', '2500k', '-bufsize', '5000k',
                   '-g', '50', '-keyint_min', '50', '-sc_threshold', '0', '-pix_fmt', 'yuv420p',
                   '-c:a', 'aac', '-b:a', '128k', '-ar', '48000', '-f', 'tee',
                   '[f=mpegts]udp://127.0.0.1:15000?pkt_size=1316|[f=mpegts]udp://127.0.0.1:15001?pkt_size=1316|[f=mpegts]udp://127.0.0.1:15002?pkt_size=1316']
        try:
            self.encoder = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL,
                                            stderr=subprocess.DEVNULL, pass_fds=(ar,), bufsize=0)
        except OSError:
            os.close(aw)
            raise
        finally:
            os.close(ar)
        self.launch(self.feed, self.encoder.stdin.fileno(), False)
        self.launch(self.feed, aw, True)
        self.launch(self.decode)

    def feed(self, fd, audio):
        os.set_blocking(fd, False)
        period = .02 if audio else 1 / FPS
        deadline = time.monotonic()
        try:
            while not self.closed.is_set():
                with self.lock:
                    live = self.frame is not None and time.monotonic() - self.last_frame < 2
                    if audio:
                        data = self.audio.popleft() if live and self.audio else bytes(AUDIO_BYTES)
                        if not live:
                            self.audio.clear()
                    else:
                        data = self.frame if live else self.pause_frame
                view = memoryview(data)
                write_deadline = time.monotonic() + 3
                while view and not self.closed.is_set():
                    if time.monotonic() > write_deadline:
                        # A blocked encoder is a real pipeline failure, not a source interruption.
                        self.closed.set()
                        terminate(self.encoder)
                        break
                    if select.select([], [fd], [], .1)[1]:
                        try:
                            view = view[os.write(fd, view):]
                        except BlockingIOError:
                            continue
                deadline += period
                if deadline < time.monotonic() - period:
                    deadline = time.monotonic()
                self.closed.wait(max(0, deadline - time.monotonic()))
        except (BrokenPipeError, OSError):
            self.closed.set()
        finally:
            if audio:
                os.close(fd)

    def decode(self):
        while not self.closed.is_set():
            ar, aw = os.pipe()
            command = ['ffmpeg', '-nostdin', '-hide_banner', '-loglevel', 'quiet', '-rw_timeout', '3000000',
                       '-analyzeduration', '1000000', '-probesize', '262144', '-i', self.source, '-map', '0:v:0', '-vf',
                       f'scale={WIDTH}:{HEIGHT}:force_original_aspect_ratio=decrease,pad={WIDTH}:{HEIGHT}:(ow-iw)/2:(oh-ih)/2,fps={FPS}',
                       '-threads', '2', '-pix_fmt', 'yuv420p', '-f', 'rawvideo', 'pipe:1',
                       '-map', '0:a:0?', '-ac', '2', '-ar', '48000', '-f', 's16le', f'pipe:{aw}']
            try:
                process = subprocess.Popen(command, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
                                           stderr=subprocess.DEVNULL, pass_fds=(aw,), bufsize=0)
            except OSError:
                os.close(ar)
                os.close(aw)
                self.closed.wait(2)
                continue
            os.close(aw)
            self.decoder = process
            with self.lock:
                self.frame = None
                self.last_frame = 0
                self.audio.clear()
            epoch = object()
            self.epoch = epoch
            astream = os.fdopen(ar, 'rb', buffering=0)
            def video_reader():
                while not self.closed.is_set():
                    frame = read_exact(process.stdout, FRAME_BYTES)
                    if frame is None:
                        break
                    with self.lock:
                        if self.epoch is epoch:
                            self.frame, self.last_frame = frame, time.monotonic()
            def audio_reader():
                while not self.closed.is_set():
                    chunk = read_exact(astream, AUDIO_BYTES)
                    if chunk is None:
                        break
                    with self.lock:
                        if self.epoch is epoch:
                            self.audio.append(chunk)
            vt = threading.Thread(target=video_reader, daemon=True)
            at = threading.Thread(target=audio_reader, daemon=True)
            vt.start(); at.start()
            born = time.monotonic()
            while not self.closed.wait(.2) and process.poll() is None:
                with self.lock:
                    last = self.last_frame or born
                if time.monotonic() - last > 5:
                    break
            terminate(process)
            vt.join(2); at.join(2)
            process.stdout.close(); astream.close()
            self.decoder = None
            with self.lock:
                self.frame = None
                self.last_frame = 0
                self.audio.clear()
            self.closed.wait(2)

    def stop(self):
        self.closed.set()
        terminate(self.decoder)
        terminate(self.encoder)
        for thread in self.threads:
            thread.join(timeout=4)
        if self.encoder and self.encoder.stdin:
            self.encoder.stdin.close()
PTV_SOURCE
cat > restream/Dockerfile <<'PTV_SOURCE'
FROM python:3.12-slim-bookworm
RUN apt-get update && apt-get install -y --no-install-recommends ffmpeg ca-certificates fonts-dejavu-core tzdata && rm -rf /var/lib/apt/lists/*
WORKDIR /app
COPY control.py continuity.py audience.py /app/
RUN ffmpeg -hide_banner -loglevel error -f lavfi -i color=c=0x142337:s=1280x720 -vf "drawtext=fontfile=/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf:text='ParlamentsTV':fontcolor=white:fontsize=58:x=(w-text_w)/2:y=230,drawtext=fontfile=/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf:text='Die Übertragung wird gleich fortgesetzt':fontcolor=white:fontsize=34:x=(w-text_w)/2:y=360" -frames:v 1 -pix_fmt yuv420p -f rawvideo /app/pause.yuv
RUN chmod 0444 /app/audience.py /app/control.py /app/continuity.py /app/pause.yuv
RUN mkdir /state && chown 10001:10001 /state && chmod 0700 /state
USER 10001:10001
CMD ["python", "-u", "/app/control.py"]
PTV_SOURCE
cat > captions/worker.py <<'PTV_SOURCE'
"""Speechmatics EU real-time bridge. Opt-in only; no recording or transcript files.
HLS wall-clock timestamps map final words to the viewer's actual video timeline.
"""
import collections
import datetime
import hmac
import json
import logging
import math
import os
import pathlib
import re
import ssl
import subprocess
import threading
import time
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

SECRET=os.environ.get('PTV_RELAY_SECRET','')
BASE='http://mediamtx:8888/live/LIVESTREAM/'
ENDPOINT='wss://eu.rt.speechmatics.com/v2/'
KEY_FILE=pathlib.Path('/run/secrets/speechmatics_key')
RATE=16000
RETRY_DELAY=10
# No library debug logs: they could contain authentication headers or transcripts.
LOGGER=logging.getLogger('ptv.speechmatics');LOGGER.addHandler(logging.NullHandler());LOGGER.propagate=False;LOGGER.disabled=True

def read_url(url,limit):
    request=urllib.request.Request(url,headers={'Authorization':'Bearer '+SECRET})
    with urllib.request.urlopen(request,timeout=2) as response:data=response.read(limit+1)
    if len(data)>limit:raise ValueError('Oversized media')
    return data

def safe_name(value):return bool(re.fullmatch(r'[A-Za-z0-9_-]+\.(?:m3u8|ts)',value))

def parse_segments(body):
    result=[];stamp=None;duration=None
    for line in body.splitlines():
        line=line.strip()
        if line.startswith('#EXT-X-PROGRAM-DATE-TIME:'):
            date=datetime.datetime.fromisoformat(line.split(':',1)[1].replace('Z','+00:00'))
            stamp=date.timestamp() if date.tzinfo else None
        elif line.startswith('#EXTINF:'):duration=float(line.split(':',1)[1].split(',')[0])
        elif line and not line.startswith('#'):
            if stamp is not None and math.isfinite(stamp) and duration is not None and math.isfinite(duration) and 0<duration<=12 and safe_name(line) and line.endswith('.ts'):
                result.append((line,stamp,duration));stamp+=duration
            duration=None
    return result

def read_segments():
    body=read_url(BASE+'index.m3u8',65536).decode()
    if '#EXT-X-STREAM-INF:' in body:
        variants=[line.strip() for line in body.splitlines() if line.strip() and not line.startswith('#')]
        if not variants or not safe_name(variants[0]):raise ValueError('Invalid HLS variant')
        body=read_url(BASE+variants[0],65536).decode()
    return parse_segments(body)

def decode_segment(name):
    if not safe_name(name) or not name.endswith('.ts'):raise ValueError('Invalid segment')
    media=read_url(BASE+name,12*1024*1024)
    try:
        process=subprocess.run(['ffmpeg','-nostdin','-hide_banner','-loglevel','quiet','-threads','1','-i','pipe:0','-map','0:a:0','-vn','-t','12','-ac','1','-ar',str(RATE),'-f','s16le','pipe:1'],input=media,capture_output=True,timeout=8)
    finally:del media
    if process.returncode or not process.stdout or len(process.stdout)>RATE*2*13:raise ValueError('Audio unavailable')
    return process.stdout

def recognition_config(vocabulary):
    return {'message':'StartRecognition','audio_format':{'type':'raw','encoding':'pcm_s16le','sample_rate':RATE},'transcription_config':{'language':'de','model':'enhanced','additional_vocab':vocabulary,'max_delay':2.5,'max_delay_mode':'fixed','enable_partials':False,'diarization':'none','enable_entities':False}}

def wall_time(offset,timeline):
    for start,end,stamp in timeline:
        if start<=offset<=end:return stamp+offset-start
    return None

def transcript_cues(message,timeline):
    """Accept only final word results. Punctuation is retained, never arbitrary HTML."""
    if message.get('message')!='AddTranscript':return []
    groups=[];group=[]
    for row in message.get('results',[])[:200]:
        alternatives=row.get('alternatives') or []
        text=alternatives[0].get('content','') if alternatives else ''
        if not isinstance(text,str) or not 0<len(text)<=200:continue
        if row.get('type')=='punctuation':
            if group:group[-1]=(group[-1][0]+text,group[-1][1],group[-1][2])
            continue
        start=row.get('start_time');end=row.get('end_time')
        if row.get('type')!='word' or type(start) not in (int,float) or type(end) not in (int,float) or not math.isfinite(start) or not math.isfinite(end) or not 0<=start<end:continue
        first=wall_time(start,timeline);last=wall_time(end,timeline)
        if first is None or last is None or not 0<last-first<=12:continue
        if group and (len(group)>=7 or last-group[0][1]>8 or first-group[-1][2]>1):groups.append(group);group=[]
        group.append((text,first,last))
    if group:groups.append(group)
    return [{'start':part[0][1]*1000,'end':(part[-1][2]+.5)*1000,'text':' '.join(word[0] for word in part)[:600]} for part in groups]

class ProviderError(Exception):
    def __init__(self,state):self.state=state

def safe_provider_error(message):
    kind=message.get('type','')
    return 'quota' if kind=='quota_exceeded' else 'auth-error' if kind in ('not_authorised','not_authorized') else 'provider-error'

class Bridge:
    """One bounded recognition session. Reader never logs provider responses."""
    def __init__(self,worker,config,key):
        from websockets.sync.client import connect
        self.worker=worker;self.timeline=collections.deque(maxlen=60);self.offset=0.;self.seq=0;self.ack=0;self.ended=threading.Event();self.error=None
        self.socket=connect(ENDPOINT,additional_headers={'Authorization':'Bearer '+key},ssl=ssl.create_default_context(),proxy=None,open_timeout=8,close_timeout=1,max_size=131072,max_queue=8,compression=None,logger=LOGGER)
        try:
            self.socket.send(json.dumps(recognition_config(config.get('vocabulary',[]))))
            deadline=time.monotonic()+20
            while time.monotonic()<deadline:
                if not worker.allowed(config):raise ProviderError('off')
                try:message=json.loads(self.socket.recv(timeout=.5))
                except TimeoutError:continue
                if message.get('message')=='RecognitionStarted':break
                if message.get('message')=='Error':raise ProviderError(safe_provider_error(message))
            else:raise ProviderError('provider-error')
        except Exception:self.socket.close();raise
        self.thread=threading.Thread(target=self.receive,daemon=True);self.thread.start()
    def receive(self):
        try:
            while not self.ended.is_set():
                try:message=json.loads(self.socket.recv(timeout=.5))
                except TimeoutError:continue
                if message.get('message')=='AudioAdded':self.ack=max(self.ack,message.get('seq_no',0))
                elif message.get('message')=='Error':self.error=safe_provider_error(message);break
                elif message.get('message')=='AddTranscript':
                    with self.worker.lock:
                        if self.worker.bridge is self and self.worker.authorized:
                            cues=transcript_cues(message,list(self.timeline));self.worker.cues=(self.worker.cues+cues)[-200:]
                            self.worker.collect_transcript(cues)
                            if cues:self.worker.lag=max(0,time.time()-cues[-1]['end']/1000)
                            self.worker.state='slow' if self.worker.lag>10 else 'listening'
                    message=None;cues=None
        except Exception:
            if not self.ended.is_set():self.error='provider-error'
        finally:self.ended.set()
    def send(self,pcm,stamp,config):
        duration=len(pcm)/(RATE*2)
        with self.worker.lock:self.timeline.append((self.offset,self.offset+duration,stamp))
        self.offset+=duration
        for offset in range(0,len(pcm),RATE//2):
            if not self.worker.allowed(config):raise ProviderError('off')
            deadline=time.monotonic()+5
            while self.seq-self.ack>=8:
                if self.ended.wait(.1) or self.error:raise ProviderError(self.error or 'provider-error')
                if not self.worker.allowed(config):raise ProviderError('off')
                if time.monotonic()>deadline:raise ProviderError('slow')
            if self.error or self.ended.is_set():raise ProviderError(self.error or 'provider-error')
            self.socket.send(pcm[offset:offset+RATE//2]);self.seq+=1
    def close(self):
        self.ended.set()
        # Close immediately on operator stop. No extra audio or silent padding.
        try:self.socket.close()
        except Exception:pass
        self.thread.join(timeout=2)
        with self.worker.lock:self.timeline.clear()

class Worker:
    def __init__(self):
        self.lock=threading.RLock();self.closed=threading.Event();self.cues=[];self.state='off';self.lag=0;self.bridge=None;self.authorized=False;self.identity=None
        self.transcript=[];self.transcript_keys=set();self.transcript_session=None;self.transcript_origin=None;self.transcript_created=None;self.transcript_bytes=0;self.transcript_partial=False;self.transcript_expired=False
    def reset_transcript(self,revision):
        self.transcript=[];self.transcript_keys=set();self.transcript_session=revision;self.transcript_origin=None;self.transcript_created=None;self.transcript_bytes=0;self.transcript_partial=False;self.transcript_expired=False
    def sync_transcript(self,config):
        with self.lock:
            session=config['session']
            if session['state']=='ready' and session['revision']!=self.transcript_session:
                self.authorized=False;self.reset_transcript(session['revision'])
            if config.get('enabled') and self.transcript_origin is None and not self.transcript_expired:
                self.transcript_origin=config.get('since',time.time())*1000;self.transcript_created=time.monotonic()
            self.expire_transcript()
    def expire_transcript(self):
        if self.transcript_created is not None and time.monotonic()-self.transcript_created>=86400:
            self.transcript=[];self.transcript_keys=set();self.transcript_bytes=0;self.transcript_expired=True
    def collect_transcript(self,cues):
        # Called only for authorized final results, independently of viewers.
        self.expire_transcript()
        if self.transcript_origin is None or self.transcript_expired:return
        for cue in cues:
            key=(cue['start'],cue['end'],cue['text'])
            if key in self.transcript_keys:continue
            size=len(cue['text'].encode('utf-8'))+128
            if len(self.transcript)>=100000 or self.transcript_bytes+size>16*1024*1024:
                self.transcript_partial=True;return
            self.transcript.append(dict(cue));self.transcript_keys.add(key);self.transcript_bytes+=size
    def transcript_export(self,format):
        if format not in ('txt','vtt'):raise ValueError('Invalid export format')
        self.sync_transcript(self.config())
        with self.lock:
            self.expire_transcript()
            cues=sorted(self.transcript,key=lambda cue:(cue['start'],cue['end']))
            if not cues:return {'available':False,'expired':self.transcript_expired}
            if format=='txt':text=' '.join(cue['text'] for cue in cues)+'\n'
            else:
                def stamp(value):
                    ms=max(0,round(value-self.transcript_origin));seconds,ms=divmod(ms,1000);minutes,seconds=divmod(seconds,60);hours,minutes=divmod(minutes,60)
                    return f'{hours:02}:{minutes:02}:{seconds:02}.{ms:03}'
                rows=['WEBVTT','', 'NOTE Zeitnull: Beginn der Untertitelsammlung, nicht zwingend Beginn einer Videoaufzeichnung.', '']
                for index,cue in enumerate(cues,1):
                    clean=cue['text'].replace('-->','→').replace('\r',' ').replace('\n',' ').replace('&','&amp;').replace('<','&lt;').replace('>','&gt;')
                    rows.extend([str(index),stamp(cue['start'])+' --> '+stamp(cue['end']),clean,''])
                text='\n'.join(rows)+'\n'
            date=datetime.datetime.fromtimestamp(self.transcript_origin/1000,datetime.timezone.utc).strftime('%Y-%m-%d_%H-%M')
            return {'available':True,'format':format,'text':text,'filename':f'ParlamentsTV_Untertitel_{date}.{format}','partial':self.transcript_partial}
    def snapshot(self):
        with self.lock:
            self.cues=[cue for cue in self.cues if cue['end']>time.time()*1000-90000][-200:]
            return {'version':1,'state':self.state,'lag':round(self.lag,1),'cues':list(self.cues)}
    def expire(self):
        while not self.closed.wait(1):
            with self.lock:
                self.cues=[cue for cue in self.cues if cue['end']>time.time()*1000-89000][-200:]
                self.expire_transcript()
    def clear(self,state):
        with self.lock:self.cues=[];self.state=state;self.lag=0
    def config(self):return json.loads(read_url('http://relay-control:9020/control/caption-config',65536))
    def allowed(self,config):
        try:
            current=self.config()
            return current.get('enabled') is True and current.get('reviewed') is True and current['session']['state']!='ended' and current['session']['revision']==config['session']['revision'] and current.get('revision')==config.get('revision')
        except Exception:return False
    def disconnect(self,state):
        with self.lock:self.authorized=False;bridge=self.bridge;self.bridge=None
        if bridge:bridge.close()
        self.clear(state)
    def run(self):
        seen=collections.deque(maxlen=100);last_stamp=None;last_new=time.monotonic();retry_at=0;fatal=False
        while not self.closed.wait(.5):
            pcm=None
            try:
                config=self.config()
                self.sync_transcript(config)
                try:key_version=KEY_FILE.stat().st_mtime_ns
                except OSError:key_version=0
                identity=(config.get('revision'),config['session']['revision'],key_version)
                if identity!=self.identity:
                    self.identity=identity;self.disconnect('off');seen.clear();last_stamp=None;fatal=False;retry_at=0
                if not config.get('enabled') or not config.get('reviewed') or config['session']['state']=='ended':
                    self.disconnect('off');seen.clear();last_stamp=None;continue
                if fatal or time.monotonic()<retry_at:continue
                key=KEY_FILE.read_text().strip() if KEY_FILE.exists() else ''
                if not key:self.disconnect('not-configured');continue
                segments=read_segments()
                if not segments:self.disconnect('no-timing');continue
                fresh=[row for row in segments if row[1]>=config.get('since',time.time()) and (row[0],row[1]) not in seen and -2<time.time()-(row[1]+row[2])<15]
                if not fresh:
                    if time.monotonic()-last_new>8:self.disconnect('waiting');last_stamp=None
                    continue
                # Start at the live edge; don't send old pre-activation speech to the cloud.
                segment=fresh[-1] if last_stamp is None else fresh[0]
                name,stamp,duration=segment
                if last_stamp is not None and abs(stamp-last_stamp)>.5:self.disconnect('waiting');last_stamp=None;segment=fresh[-1];name,stamp,duration=segment
                pcm=decode_segment(name)
                if not self.allowed(config):self.disconnect('off');continue
                if self.bridge is None:
                    self.clear('loading');bridge=Bridge(self,config,key)
                    with self.lock:self.bridge=bridge;self.authorized=True;self.state='listening'
                if not self.allowed(config):self.disconnect('off');continue
                self.bridge.send(pcm,stamp,config)
                seen.append((name,stamp));last_stamp=stamp+duration;last_new=time.monotonic()
            except ProviderError as error:
                self.disconnect(error.state);last_stamp=None
                fatal=error.state in ('auth-error','quota');retry_at=time.monotonic()+RETRY_DELAY
            except Exception as error:
                # Classify HTTP handshake errors without emitting key/response/URLs.
                status=getattr(getattr(error,'response',None),'status_code',None)
                state='auth-error' if status in (401,403) else 'quota' if status==429 else 'waiting'
                self.disconnect(state);last_stamp=None;fatal=state in ('auth-error','quota');retry_at=time.monotonic()+RETRY_DELAY
            finally:pcm=None;key=None
        self.disconnect('off')

class Handler(BaseHTTPRequestHandler):
    def log_message(self,*_):pass
    def do_GET(self):
        if not SECRET or self.path not in ('/captions','/transcript?format=txt','/transcript?format=vtt') or not hmac.compare_digest(self.headers.get('Authorization','').encode(),('Bearer '+SECRET).encode()):self.send_error(403);return
        try:result=worker.snapshot() if self.path=='/captions' else worker.transcript_export(self.path.rsplit('=',1)[1])
        except Exception:self.send_error(503);return
        data=json.dumps(result).encode();self.send_response(200);self.send_header('Content-Type','application/json');self.send_header('Cache-Control','no-store');self.send_header('Content-Length',str(len(data)));self.end_headers();self.wfile.write(data)

if __name__=='__main__':
    if not SECRET:raise SystemExit('Internal secret missing')
    worker=Worker();threading.Thread(target=worker.expire,daemon=True).start();threading.Thread(target=worker.run,daemon=True).start()
    ThreadingHTTPServer(('0.0.0.0',9030),Handler).serve_forever()
PTV_SOURCE
cat > captions/Dockerfile <<'PTV_SOURCE'
FROM python:3.11-slim-bookworm
RUN apt-get update && apt-get install -y --no-install-recommends ffmpeg ca-certificates && rm -rf /var/lib/apt/lists/*
RUN pip install --no-cache-dir websockets==15.0.1
WORKDIR /app
COPY worker.py /app/worker.py
RUN chmod 0444 /app/worker.py
USER 10001:10001
CMD ["python","-u","/app/worker.py"]
PTV_SOURCE
cat > captions/set-key.py <<'PTV_SOURCE'
"""Run directly in SSH, never in a piped installation. Input is hidden."""
import getpass
import os
import pathlib
import re
import sys
if os.geteuid()!=0 or not sys.stdin.isatty():raise SystemExit('Bitte direkt im SSH-Fenster als root starten.')
path=pathlib.Path('/opt/parlamentstv-stream-test/caption-secrets/speechmatics_key')
if not path.parent.is_dir():raise SystemExit('Zuerst die Server-Erweiterung installieren.')
key=getpass.getpass('Speechmatics API-Schluessel (Eingabe unsichtbar): ').strip()
if not re.fullmatch(r'[A-Za-z0-9_.=-]{16,2048}',key):raise SystemExit('Schluessel-Format ungueltig. Es wurde nichts geaendert.')
# Preserve inode: the running read-only Docker bind mount sees the updated key.
fd=os.open(path,os.O_WRONLY|os.O_CREAT|os.O_TRUNC|os.O_NOFOLLOW,0o400)
with os.fdopen(fd,'w') as handle:
    os.fchown(handle.fileno(),10001,10001);os.fchmod(handle.fileno(),0o400)
    handle.write(key+'\n');handle.flush();os.fsync(handle.fileno())
key=None
print('Speechmatics-Schluessel geschuetzt gespeichert. Im Player Serverstatus aktualisieren. Noch keine Spracherkennung gestartet.')
PTV_SOURCE
install -d -m 700 caption-secrets
[[ -e caption-secrets/speechmatics_key ]] || : > caption-secrets/speechmatics_key
[[ -f caption-secrets/speechmatics_key && ! -L caption-secrets/speechmatics_key ]] || { printf 'Unerwartete Schluesseldatei. Abgebrochen.\n'; exit 1; }
chown 10001:10001 caption-secrets/speechmatics_key
chmod 400 caption-secrets/speechmatics_key
if [[ "$keep_existing_override" -eq 0 ]]; then cp restream/compose.override.yaml.next compose.override.yaml; fi
docker compose -p parlamentstv-stream-test config --quiet
printf 'Baue und starte die geschuetzte Streaming-Erweiterung ...\n'
docker compose -p parlamentstv-stream-test up -d --build
docker compose -p parlamentstv-stream-test up -d --force-recreate mediamtx
docker compose -p parlamentstv-stream-test exec -T relay-control python -c 'import json,os,urllib.request; r=urllib.request.Request("http://127.0.0.1:9020/control/status",headers={"Authorization":"Bearer "+os.environ["PTV_CONTROL_TOKEN"]}); d=json.load(urllib.request.urlopen(r,timeout=10)); assert d["version"]==1; assert d.get("audience_supported") is True; assert d["automatic_captions"].get("transcript_supported") is True; print("Serversteuerung, Admin-Textdownload und zentrale Zuschauerstatistik bereit.")' </dev/null
docker compose -p parlamentstv-stream-test exec -T caddy caddy validate --config /etc/caddy/Caddyfile </dev/null >/dev/null
cp restream/Caddyfile.next Caddyfile
if ! docker compose -p parlamentstv-stream-test exec -T caddy caddy reload --config /etc/caddy/Caddyfile </dev/null >/dev/null; then
  cp restream/Caddyfile.before Caddyfile
  printf 'Proxy konnte nicht geladen werden. Vorherige Konfiguration wiederhergestellt.\n'; exit 1
fi
docker compose -p parlamentstv-stream-test ps
printf '\nErweiterung gestartet. YouTube und Facebook sind noch NICHT gestartet.\n'
printf 'Steuerschluessel nur im privaten Player eintragen, nicht im Chat teilen:\n'
cut -d= -f2- control.env
printf '\nRTMP aus Livestream Studio 6: rtmp://streamtest.stadtparlamentsfernsehen.de:1935/live\n'
printf 'Test-Key: LIVESTREAM\n'
printf 'Vor dem Weiterleiten muss Livestream Studio ein H.264/AAC-Signal senden.\n'
printf 'Speechmatics: API-Schluessel anschliessend DIREKT im SSH-Fenster hinterlegen mit:\n'
printf 'python3 /opt/parlamentstv-stream-test/captions/set-key.py\n'
printf 'Eingabe bleibt unsichtbar. Noch keine Spracherkennung gestartet.\n'