announcer/announcer.py

#!/usr/bin/env python3
"""Announcer – tekst inn, forhåndslytt i nettleseren, les ut på PC-ens lydutgang.

Kjøres på PC-en som er koblet til høyttaleranlegget. Selve avspillingen skjer
alltid lokalt på denne maskinen via pw-play (PipeWire) eller paplay.

Tittel, tekst og hurtigfraser kommer fra config.py, så appen ikke er bundet til
ett arrangement. Innlogging er valgfri og settes opp i auth.py; uten
konfigurasjon er alt åpent, som før.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import logging
import math
import os
import queue
import shutil
import struct
import subprocess
import threading
import time
import wave
from dataclasses import dataclass, field
from pathlib import Path

from flask import Flask, Response, abort, jsonify, request, send_file

import auth
import config
import schedule as sched
from auth import require

HERE = Path(__file__).resolve().parent
CFG = config.load()
VOICE_DIR = CFG.path("voices")
CACHE_DIR = CFG.path("cache")
LOG_PATH = CFG.path("log")
SCHEDULE_PATH = CFG.path("schedule")
SINK = str(CFG.get("audio", "sink", "")).strip()  # pw-play --target
MAX_CHARS = int(CFG.app["max_chars"])

# Stemmene som tilbys i menyen. To motorer:
#   edge  – Microsofts nevrale stemmer via edge-tts (gratis, krever internett,
#           klart mest naturlig; ~2–3 s per melding)
#   piper – lokalt, ingen nett, robotaktig; reserve hvis nettet faller ut.
# Piper-modellene lastes ned fra rhasspy/piper-voices ved første oppstart.
# «speaker» er bare for Piper-flerstemmemodeller (nvcc har ti norske talere).
def _edge(id: str, name: str, group: str, voice: str) -> dict:
    return {"id": id, "name": name, "group": group, "engine": "edge", "model": voice, "speaker": None}


def _piper(id: str, name: str, group: str, model: str, speaker: int | None = None) -> dict:
    return {"id": id, "name": name, "group": group, "engine": "piper", "model": model, "speaker": speaker}


VOICES_LIST: list[dict] = [
    _edge("finn", "Finn (mann)", "Norsk", "nb-NO-FinnNeural"),
    _edge("pernille", "Pernille (kvinne)", "Norsk", "nb-NO-PernilleNeural"),
    _edge("andrew", "Andrew (US, mann)", "English", "en-US-AndrewMultilingualNeural"),
    _edge("ava", "Ava (US, kvinne)", "English", "en-US-AvaMultilingualNeural"),
    _edge("brian", "Brian (US, mann)", "English", "en-US-BrianMultilingualNeural"),
    _edge("emma", "Emma (US, kvinne)", "English", "en-US-EmmaMultilingualNeural"),
    _edge("ryan-uk", "Ryan (UK, mann)", "English", "en-GB-RyanNeural"),
    _edge("sonia", "Sonia (UK, kvinne)", "English", "en-GB-SoniaNeural"),
    _edge("sv", "Svensk – Mattias", "Nabospråk", "sv-SE-MattiasNeural"),
    _edge("da", "Dansk – Christel", "Nabospråk", "da-DK-ChristelNeural"),
    _piper("no-tale", "Talesyntese (nøytral)", "Lokal reserve (uten nett)", "no_NO-talesyntese-medium"),
    _piper("no-knn", "Kvinne – KNN", "Lokal reserve (uten nett)", "no_NO-nvcc-medium", 0),
    _piper("no-msv", "Mann – MSV", "Lokal reserve (uten nett)", "no_NO-nvcc-medium", 5),
    _piper("en-ryan", "Ryan (US)", "Lokal reserve (uten nett)", "en_US-ryan-high"),
    _piper("en-jenny", "Jenny (UK)", "Lokal reserve (uten nett)", "en_GB-jenny_dioco-medium"),
]
VOICES: dict[str, dict] = {v["id"]: v for v in VOICES_LIST}
EDGE_RATE = 24000

# Konfigurerte standardvalg, men bare hvis de finnes – en skrivefeil i
# config.toml skal ikke gjøre sida ubrukelig.
DEFAULT_VOICE = CFG.app["default_voice"] if CFG.app["default_voice"] in VOICES else "finn"
DEFAULT_CHIME = CFG.app["default_chime"]

log = logging.getLogger("announcer")
app = Flask(__name__, static_folder=None)
AUTH = auth.install(app, CFG)


# --------------------------------------------------------------------------
# TTS
# --------------------------------------------------------------------------

_voice_cache: dict[str, object] = {}
_voice_lock = threading.Lock()


def ensure_voices() -> None:
    import piper.download_voices as dl

    VOICE_DIR.mkdir(parents=True, exist_ok=True)
    for name in sorted({v["model"] for v in VOICES_LIST if v["engine"] == "piper"}):
        if not (VOICE_DIR / f"{name}.onnx").exists():
            log.info("laster ned stemme %s", name)
            dl.download_voice(name, VOICE_DIR)


def load_voice(name: str):
    from piper import PiperVoice

    with _voice_lock:
        if name not in _voice_cache:
            _voice_cache[name] = PiperVoice.load(VOICE_DIR / f"{name}.onnx")
        return _voice_cache[name]


# --------------------------------------------------------------------------
# Jingler (chimes). Alle de innebygde genereres i Python; egne lydfiler
# (wav/mp3/ogg/flac) i CHIME_DIR dukker opp automatisk i menyen.
# --------------------------------------------------------------------------

CHIME_DIR = CFG.path("chimes")


def _tone(rate: int, freq: float, dur: float, vol: float = 0.35, wave_: str = "sine",
          decay: float = 5.5, attack: float = 0.005, harm2: float = 0.0, slide: float = 0.0) -> list[float]:
    """Én tone som liste av floats. slide = frekvensendring (multiplikator) over tonen."""
    n = int(rate * dur)
    out = []
    phase = 0.0
    for i in range(n):
        t = i / rate
        f = freq * (slide ** (t / dur)) if slide else freq
        phase += 2 * math.pi * f / rate
        if wave_ == "square":
            v = 1.0 if math.sin(phase) >= 0 else -1.0
        elif wave_ == "saw":
            v = 2.0 * ((phase / (2 * math.pi)) % 1.0) - 1.0
        else:
            v = math.sin(phase) + harm2 * math.sin(2 * phase)
        env = min(1.0, t / attack) * math.exp(-decay * t)
        out.append(v * env * vol)
    return out


def _mix(*parts: list[float], gap: float = 0.0, rate: int = 22050) -> list[float]:
    out: list[float] = []
    for p in parts:
        out += p
        out += [0.0] * int(rate * gap)
    return out


def _overlay(base: list[float], layer: list[float], at: float, rate: int) -> list[float]:
    start = int(at * rate)
    if len(base) < start + len(layer):
        base = base + [0.0] * (start + len(layer) - len(base))
    for i, v in enumerate(layer):
        base[start + i] += v
    return base


def _dingdong(rate: int) -> list[float]:
    return _mix(_tone(rate, 784.0, 0.28, harm2=0.3), _tone(rate, 523.25, 0.42, harm2=0.3), rate=rate)


def _airport(rate: int) -> list[float]:
    """Tre stigende toner med lang klang, som på en flyplass."""
    out: list[float] = []
    for i, f in enumerate((523.25, 659.25, 783.99)):
        out = _overlay(out, _tone(rate, f, 1.2, vol=0.3, decay=2.2, harm2=0.2), i * 0.38, rate)
    return out


def _chiptune(rate: int) -> list[float]:
    """8-bit arpeggio, som et power-up i et gammelt spill."""
    notes = (523.25, 659.25, 783.99, 1046.5, 1318.5, 1567.98)
    parts = [_tone(rate, f, 0.09, vol=0.3, wave_="square", decay=3.0) for f in notes]
    parts.append(_tone(rate, 2093.0, 0.35, vol=0.3, wave_="square", decay=6.0))
    return _mix(*parts, rate=rate)


def _scifi(rate: int) -> list[float]:
    """Sveip opp + to blipp, som en kommunikator."""
    sweep = _tone(rate, 300.0, 0.35, vol=0.3, decay=1.5, slide=5.0)
    blips = _mix(_tone(rate, 1800.0, 0.07, vol=0.25, decay=10), _tone(rate, 2400.0, 0.12, vol=0.25, decay=10), gap=0.03, rate=rate)
    return _mix(sweep, blips, rate=rate)


def _horn(rate: int) -> list[float]:
    """Air horn: to lag av sagtann med litt vibrato, derav grovheten."""
    out: list[float] = []
    for f, v in ((233.08, 0.22), (349.23, 0.16), (466.16, 0.1)):
        out = _overlay(out, _tone(rate, f, 0.9, vol=v, wave_="saw", decay=1.3, attack=0.03), 0.0, rate)
    return out


def _gong(rate: int) -> list[float]:
    """Dyp gong med uharmoniske overtoner."""
    out: list[float] = []
    for f, v, d in ((110.0, 0.3, 1.0), (164.8, 0.18, 1.4), (263.0, 0.12, 2.0), (411.0, 0.07, 3.0), (650.0, 0.04, 4.0)):
        out = _overlay(out, _tone(rate, f, 2.5, vol=v, decay=d, attack=0.004), 0.0, rate)
    return out


BUILTIN_CHIMES: dict[str, tuple[str, object]] = {
    "dingdong": ("Ding-dong", _dingdong),
    "airport": ("Flyplass (tre toner)", _airport),
    "chiptune": ("8-bit power-up", _chiptune),
    "scifi": ("Sci-fi kommunikator", _scifi),
    "horn": ("Air horn", _horn),
    "gong": ("Gong", _gong),
}
CHIME_EXT = {".wav", ".mp3", ".ogg", ".flac", ".m4a"}


def file_chimes() -> dict[str, Path]:
    if not CHIME_DIR.is_dir():
        return {}
    return {f"file:{p.stem}": p for p in sorted(CHIME_DIR.iterdir()) if p.suffix.lower() in CHIME_EXT}


def list_chimes() -> list[dict]:
    out = [{"id": "none", "name": "Ingen"}]
    out += [{"id": k, "name": v[0]} for k, v in BUILTIN_CHIMES.items()]
    out += [{"id": k, "name": f"{p.stem} (egen fil)"} for k, p in file_chimes().items()]
    return out


def chime_pcm(chime: str, rate: int) -> bytes:
    """Jingle som 16-bit mono PCM i gitt samplerate, etterfulgt av en kort pause."""
    if chime == "none":
        return b""
    if chime in BUILTIN_CHIMES:
        samples = BUILTIN_CHIMES[chime][1](rate)
        pcm = b"".join(struct.pack("<h", int(max(-1.0, min(1.0, v)) * 32767)) for v in samples)
    else:
        path = file_chimes().get(chime)
        if path is None:
            raise ValueError("Ukjent jingle")
        pcm = subprocess.run(
            ["ffmpeg", "-v", "error", "-i", str(path), "-t", "6", "-f", "s16le", "-ac", "1", "-ar", str(rate),
             "-af", "loudnorm=I=-18,afade=t=out:st=5:d=1", "-"],
            capture_output=True, check=True,
        ).stdout
    return pcm + b"\x00\x00" * int(rate * 0.35)


def edge_pcm(text: str, voice: str, speed: float) -> bytes:
    """Henter MP3 fra edge-tts og dekoder til 16-bit mono PCM i EDGE_RATE."""
    import asyncio
    import tempfile

    import edge_tts

    rate_pct = int(round((speed - 1.0) * 100))
    rate = f"{rate_pct:+d}%"
    with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False, dir=CACHE_DIR) as tmp:
        mp3 = Path(tmp.name)
    try:
        asyncio.run(edge_tts.Communicate(text, voice, rate=rate).save(str(mp3)))
        out = subprocess.run(
            ["ffmpeg", "-v", "error", "-i", str(mp3), "-f", "s16le", "-ac", "1", "-ar", str(EDGE_RATE), "-"],
            capture_output=True,
            check=True,
        )
        return out.stdout
    finally:
        mp3.unlink(missing_ok=True)


def synth_key(text: str, voice: str, speed: float, chime: str) -> str:
    h = hashlib.sha1(f"{voice}|{speed:.2f}|{chime}|{text}".encode()).hexdigest()
    return h[:20]


def synthesize(text: str, voice_name: str, speed: float, chime: str) -> Path:
    """Returnerer sti til ferdig WAV (bufres på disk etter innhold)."""
    from piper.config import SynthesisConfig

    CACHE_DIR.mkdir(parents=True, exist_ok=True)
    path = CACHE_DIR / f"{synth_key(text, voice_name, speed, chime)}.wav"
    if path.exists():
        return path

    spec = VOICES[voice_name]
    pcm = bytearray()
    if spec["engine"] == "edge":
        rate = EDGE_RATE
        pcm += chime_pcm(chime, rate)
        pcm += edge_pcm(text, spec["model"], speed)
    else:
        voice = load_voice(spec["model"])
        rate = voice.config.sample_rate
        cfg = SynthesisConfig(length_scale=1.0 / speed, speaker_id=spec["speaker"])
        pcm += chime_pcm(chime, rate)
        for chunk in voice.synthesize(text, cfg):
            pcm += chunk.audio_int16_bytes
    pcm += b"\x00\x00" * int(rate * 0.3)

    tmp = path.with_suffix(".tmp")
    with wave.open(str(tmp), "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2)
        w.setframerate(rate)
        w.writeframes(bytes(pcm))
    tmp.rename(path)
    return path


# --------------------------------------------------------------------------
# Avspillingskø
# --------------------------------------------------------------------------


@dataclass
class Item:
    id: int
    text: str
    voice: str
    speed: float
    chime: str
    sender: str                      # IP-adressen forespørselen kom fra
    user: str = ""                   # visningsnavnet til den som sendte
    subject: str = ""                # stabil id, f.eks. "jwt:kent"
    method: str = "none"
    created: float = field(default_factory=time.time)
    status: str = "queued"  # queued | playing | done | stopped | failed
    error: str = ""

    def public(self, *, show_ip: bool = False) -> dict:
        return {
            "id": self.id,
            "text": self.text,
            "voice": VOICES[self.voice]["name"] if self.voice in VOICES else self.voice,
            "user": self.user,
            # IP-en er bare interessant for en admin som skal finne ut hvem
            # som rotet; alle andre ser navnet.
            "sender": self.sender if show_ip else "",
            "created": self.created,
            "status": self.status,
            "error": self.error,
        }


def player_cmd(path: Path) -> list[str]:
    """pw-play når PipeWire kjører, ellers paplay (PulseAudio, f.eks. Linux Mint 21)."""
    runtime = Path(os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}"))
    if shutil.which("pw-play") and (runtime / "pipewire-0").exists():
        cmd = ["pw-play"]
        if SINK:
            cmd += ["--target", SINK]
    elif shutil.which("paplay"):
        cmd = ["paplay"]
        if SINK:
            cmd += ["--device", SINK]
    else:
        raise RuntimeError("fant verken pw-play (pipewire-bin) eller paplay (pulseaudio-utils)")
    return cmd + [str(path)]


class Player:
    def __init__(self) -> None:
        self.q: queue.Queue[Item] = queue.Queue()
        self.history: list[Item] = []
        self.current: Item | None = None
        self.proc: subprocess.Popen | None = None
        self.lock = threading.Lock()
        self.seq = 0
        threading.Thread(target=self._loop, daemon=True, name="player").start()

    def enqueue(self, text: str, voice: str, speed: float, chime: str, sender: str,
                who: auth.Principal | None = None) -> Item:
        with self.lock:
            self.seq += 1
            item = Item(
                self.seq, text, voice, speed, chime, sender,
                user=who.name if who else "",
                subject=who.subject if who else "",
                method=who.method if who else "none",
            )
            self.history.append(item)
            del self.history[:-50]
        self.q.put(item)
        return item

    def stop(self) -> None:
        with self.lock:
            if self.proc and self.proc.poll() is None:
                self.proc.terminate()
                if self.current:
                    self.current.status = "stopped"

    def clear(self) -> None:
        while True:
            try:
                item = self.q.get_nowait()
            except queue.Empty:
                break
            item.status = "stopped"
        self.stop()

    def snapshot(self, *, show_ip: bool = False) -> dict:
        with self.lock:
            return {
                "current": self.current.public(show_ip=show_ip) if self.current else None,
                "queued": [i.public(show_ip=show_ip) for i in self.history if i.status == "queued"],
                "history": [i.public(show_ip=show_ip) for i in reversed(self.history)
                            if i.status not in ("queued", "playing")][:15],
            }

    def _loop(self) -> None:
        while True:
            item = self.q.get()
            if item.status != "queued":
                continue
            item.status = "playing"
            with self.lock:
                self.current = item
            try:
                path = synthesize(item.text, item.voice, item.speed, item.chime)
                cmd = player_cmd(path)
                with self.lock:
                    self.proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
                _, err = self.proc.communicate()
                if item.status == "playing":
                    if self.proc.returncode == 0:
                        item.status = "done"
                    else:
                        item.status = "failed"
                        item.error = err.decode(errors="replace").strip()[-200:]
            except Exception as e:  # noqa: BLE001
                log.exception("avspilling feilet")
                item.status = "failed"
                item.error = str(e)
            finally:
                with self.lock:
                    self.current = None
                    self.proc = None
                audit("played", id=item.id, status=item.status, error=item.error,
                      subject=item.subject, name=item.user, method=item.method,
                      ip=item.sender, text=item.text, voice=item.voice)


player = Player()


# --------------------------------------------------------------------------
# Planlagte sendinger
# --------------------------------------------------------------------------


schedule = sched.Schedule(
    SCHEDULE_PATH,
    grace=float(CFG.get("schedule", "grace_seconds", sched.DEFAULT_GRACE)),
    max_ahead=float(CFG.get("schedule", "max_days_ahead", 30)) * 86400,
)
TICK = max(5.0, float(CFG.get("schedule", "tick_seconds", 20)))


def _scheduler() -> None:
    """Flytter forfalte meldinger inn i den vanlige køen.

    Egen tråd, og den gjør med vilje ingenting annet: avspilling, syntese og
    feilhåndtering er Players jobb, slik at en planlagt melding oppfører seg
    nøyaktig som en sendt for hånd når den først er i kø.
    """
    while True:
        try:
            send, missed = schedule.due()
            for item in missed:
                schedule.mark(item.id, "missed", "maskina var ikke i gang i tide")
                log.warning("planlagt #%s gikk glipp av (%s)", item.id, item.text[:40])
                audit("missed", id=item.id, subject=item.subject, name=item.user,
                      method=item.method, text=item.text, at=item.at)
            for item in send:
                queued = player.enqueue(
                    item.text, item.voice, item.speed, item.chime,
                    sender="planlagt",
                    who=auth.Principal(item.subject, item.user, (), item.method, frozenset()),
                )
                schedule.mark(item.id, "sent", f"kø #{queued.id}")
                log.info("planlagt #%s lagt i kø som #%d", item.id, queued.id)
                audit("scheduled_fire", id=item.id, queue_id=queued.id,
                      subject=item.subject, name=item.user, method=item.method,
                      text=item.text, voice=item.voice, at=item.at)
        except Exception:  # noqa: BLE001
            # En feil her skal ikke drepe tråden; da ville alle senere
            # planlagte sendinger stilltiende blitt borte.
            log.exception("planleggeren feilet")
        time.sleep(TICK)


threading.Thread(target=_scheduler, daemon=True, name="scheduler").start()


# --------------------------------------------------------------------------
# HTTP
# --------------------------------------------------------------------------


MAX_LOG_BYTES = 5_000_000
_log_lock = threading.Lock()


def audit(action: str, **fields) -> None:
    """Én JSON-linje per hendelse, så man i ettertid kan se hvem som sendte hva.

    Append-only og rullerer ved 5 MB. Feiler aldri forespørselen – en logg som
    ikke lar seg skrive skal ikke hindre et utrop.
    """
    rec = {"ts": round(time.time(), 3), "action": action,
           **{k: v for k, v in fields.items() if v not in ("", None)}}
    line = json.dumps(rec, ensure_ascii=False)
    try:
        with _log_lock:
            LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
            if LOG_PATH.exists() and LOG_PATH.stat().st_size > MAX_LOG_BYTES:
                LOG_PATH.replace(LOG_PATH.with_name(LOG_PATH.name + ".1"))
            with LOG_PATH.open("a", encoding="utf-8") as fh:
                fh.write(line + "\n")
    except OSError as e:
        log.warning("fikk ikke skrevet til %s: %s", LOG_PATH, e)


def who(p: auth.Principal) -> dict:
    """Identitetsfeltene audit() vil ha."""
    return {"subject": p.subject, "name": p.name, "method": p.method,
            "ip": request.remote_addr or ""}


def parse_request() -> tuple[str, str, float, str]:
    data = request.get_json(silent=True) or {}
    text = " ".join(str(data.get("text", "")).split())
    if not text:
        abort(400, "Tom melding")
    if len(text) > MAX_CHARS:
        abort(400, f"Maks {MAX_CHARS} tegn")
    voice = str(data.get("voice", DEFAULT_VOICE))
    if voice not in VOICES:
        abort(400, "Ukjent stemme")
    try:
        speed = float(data.get("speed", 1.0))
    except (TypeError, ValueError):
        abort(400, "Ugyldig hastighet")
    speed = max(0.6, min(1.6, speed))
    chime = data.get("chime", DEFAULT_CHIME)
    if chime is True:
        chime = DEFAULT_CHIME
    elif chime is False or chime is None:
        chime = "none"
    chime = str(chime)
    if chime != "none" and chime not in BUILTIN_CHIMES and chime not in file_chimes():
        abort(400, "Ukjent jingle")
    return text, voice, speed, chime


@app.get("/")
def index() -> Response:
    return send_file(HERE / "index.html")


@app.get("/api/config")
def client_config() -> Response:
    """Alt sida trenger for å tegne seg selv: branding, menyer og hvem du er.

    Svarer alltid 200, også med feil PIN – her spør klienten «hvem er jeg?»,
    og da er svaret «ikke innlogget», ikke en feil.
    """
    me = auth.current()
    return jsonify(
        {
            "app": {
                "title": CFG.app["title"],
                "subtitle": CFG.app["subtitle"],
                "accent": CFG.app["accent"],
                "idle_text": CFG.app["idle_text"],
                "message_placeholder": CFG.app["message_placeholder"],
                "quick_phrases": list(CFG.app["quick_phrases"]),
            },
            "voices": [{"id": v["id"], "name": v["name"], "group": v["group"], "engine": v["engine"]} for v in VOICES_LIST],
            "default_voice": DEFAULT_VOICE,
            "chimes": list_chimes(),
            "default_chime": DEFAULT_CHIME,
            "max_chars": MAX_CHARS,
            "schedule": {
                "max_days_ahead": int(CFG.get("schedule", "max_days_ahead", 30)),
                "grace_seconds": int(CFG.get("schedule", "grace_seconds", 120)),
            },
            "auth": {
                **AUTH.describe(),
                "user": None if me.anonymous else me.public(),
                "can": sorted(me.permissions),
                "error": auth.error(),
            },
        }
    )


@app.post("/api/preview")
def preview() -> Response:
    me = require("preview")
    text, voice, speed, chime = parse_request()
    try:
        path = synthesize(text, voice, speed, chime)
    except Exception as e:  # noqa: BLE001
        log.exception("syntese feilet")
        audit("preview", **who(me), text=text, voice=voice, error=str(e)[:200])
        abort(502, f"Syntese feilet ({VOICES[voice]['name']}): {e}. Prøv en lokal reserve-stemme.")
    audit("preview", **who(me), text=text, voice=voice)
    return send_file(path, mimetype="audio/wav", max_age=3600)


@app.post("/api/announce")
def announce() -> Response:
    me = require("announce")
    text, voice, speed, chime = parse_request()
    sender = request.remote_addr or "?"
    item = player.enqueue(text, voice, speed, chime, sender, me)
    log.info("kø #%d fra %s (%s): %s", item.id, item.user or sender, sender, text)
    audit("announce", **who(me), id=item.id, text=text, voice=voice,
          speed=speed, chime=chime)
    return jsonify(item.public(show_ip=me.can("clear")))


@app.post("/api/schedule")
def schedule_add() -> Response:
    me = require("schedule")
    text, voice, speed, chime = parse_request()
    data = request.get_json(silent=True) or {}
    try:
        at = float(data["at"])
    except (KeyError, TypeError, ValueError):
        abort(400, "Mangler eller ugyldig tidspunkt")
    try:
        item = schedule.add(at=at, text=text, voice=voice, speed=speed, chime=chime,
                            user=me.name, subject=me.subject, method=me.method)
    except ValueError as e:
        abort(400, str(e))
    log.info("planlagt #%s kl. %s av %s: %s", item.id,
             time.strftime("%Y-%m-%d %H:%M", time.localtime(at)), me.name or "anonym", text)
    audit("schedule", **who(me), id=item.id, at=at, text=text, voice=voice,
          speed=speed, chime=chime)
    return jsonify(item.public())


@app.post("/api/schedule/cancel")
def schedule_cancel() -> Response:
    me = require("schedule")
    data = request.get_json(silent=True) or {}
    sched_id = str(data.get("id", "")).strip()
    item = schedule.get(sched_id)
    if item is None or item.status != "pending":
        abort(404, "Fant ingen planlagt sending med den id-en")
    # Sin egen kan man alltid avbryte. Andres krever den samme retten som
    # å tømme køen – det er samme slags inngrep i noe andre har satt i gang.
    if item.subject != me.subject and not me.can("clear"):
        abort(403, f"{item.user or 'En annen'} planla denne. Du kan bare avbryte dine egne.")
    schedule.cancel(sched_id)
    audit("schedule_cancel", **who(me), id=sched_id, text=item.text)
    return jsonify({"ok": True, "id": sched_id})


@app.post("/api/stop")
def stop() -> Response:
    me = require("stop")
    player.stop()
    audit("stop", **who(me))
    return jsonify({"ok": True})


@app.post("/api/clear")
def clear() -> Response:
    me = require("clear")
    player.clear()
    audit("clear", **who(me))
    return jsonify({"ok": True})


@app.get("/api/status")
def status() -> Response:
    # Lukket bak «view»: køen inneholder meldingstekstene og IP-ene til dem
    # som har sendt noe, og det er ikke noe alle utenfra skal få se.
    me = require("view")
    return jsonify({
        **player.snapshot(show_ip=me.can("clear")),
        "scheduled": schedule.snapshot(),
        "now": time.time(),   # klienten regner «om 20 min» mot serverens klokke
    })


@app.errorhandler(400)
def bad_request(e):  # noqa: ANN001
    return jsonify({"error": e.description}), 400


@app.errorhandler(502)
def bad_gateway(e):  # noqa: ANN001
    return jsonify({"error": e.description}), 502


@app.errorhandler(401)
def unauthorized(e):  # noqa: ANN001
    return jsonify({"error": e.description, "login": True}), 401


@app.errorhandler(403)
def forbidden(e):  # noqa: ANN001
    return jsonify({"error": e.description}), 403


@app.errorhandler(404)
def not_found(e):  # noqa: ANN001
    # Klienten snakker JSON med /api og /auth, og skal ikke få en HTML-side
    # tilbake og tolke den som en feilmelding.
    if request.path.startswith(("/api/", "/auth/")):
        msg = e.description or ""
        if not msg or "requested URL" in msg:
            msg = "Finnes ikke"
        return jsonify({"error": msg}), 404
    return e


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--host", default=CFG.get("server", "host", "0.0.0.0"))
    ap.add_argument("--port", type=int, default=int(CFG.get("server", "port", 8090)))
    args = ap.parse_args()
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    logging.getLogger("werkzeug").setLevel(logging.WARNING)
    ensure_voices()
    load_voice("no_NO-talesyntese-medium")  # varm opp reserven
    if CFG.source:
        log.info("konfigurasjon fra %s", CFG.source)
    methods = ", ".join(m["method"] for m in AUTH.describe()["methods"]) or "åpen (ingen innlogging)"
    venter = len(schedule.snapshot()["pending"])
    log.info("klar på http://%s:%d  (auth: %s, sink: %s, planlagt: %d)",
             args.host, args.port, methods, SINK or "standard", venter)
    app.run(host=args.host, port=args.port, threaded=True)


if __name__ == "__main__":
    main()