diora-web/podcasts/views.py

import hashlib
import json
import logging
import math
import re
import time
import urllib.parse
import uuid
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeout, as_completed
from dataclasses import dataclass
from datetime import timedelta, timezone as dt_timezone
import xml.etree.ElementTree as ET

import feedparser
import requests
from django.conf import settings
from django.db import IntegrityError, transaction
from django.db.models import (
    Case, Count, Exists, F, FilteredRelation, IntegerField, Max, OuterRef, Q, Sum, Value, When,
)
from django.db.models.functions import Coalesce, Lower
from django.http import HttpResponse, JsonResponse, StreamingHttpResponse
from django.utils import timezone
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods

from . import chapters as chapters_mod
from . import silence as silence_mod
from . import feedext
from . import transcripts as transcripts_mod
from . import youtube as youtube_mod
from .models import (
    EpisodeProgress, EpisodeTranscript, PlayerDevice, PlayerState, PodcastEpisode, PodcastFeed, PodcastQueue,
    MAX_SPEED, MIN_SPEED, PodcastSettings, counts_as_played,
)
from .shownotes import extract_timestamps, sanitize_html

feedext.install()

log = logging.getLogger(__name__)

FEED_TIMEOUT = 20      # seconds per feed download
NEW_ON_SUBSCRIBE = 1   # episodes of a new subscription that count as new
FEED_WORKERS = 16      # feeds downloaded at the same time
REFRESH_WAIT = 40      # seconds a refresh-all (⟳) waits; feeds not done by then wait for the hourly job
MAX_DESCRIPTION = 50_000  # characters of shownotes HTML kept per episode
MAX_TRANSCRIPTS = 10
MAX_DURATION = 7 * 24 * 3600   # seconds: a longer one reported is nonsense


# ---------------------------------------------------------------------------
# Queue positions
# ---------------------------------------------------------------------------

QUEUE_POSITIONS = {key for key, _ in PodcastSettings.POSITIONS}
KEEP_SORTED = {key for key, _ in PodcastSettings.SORT_ORDERS}


def _queue_order(user):
    """Episode ids in the user's queue, first to last."""
    return list(
        PodcastQueue.objects.filter(user=user)
        .order_by('position', 'added_at', 'id')
        .values_list('episode_id', flat=True)
    )


def _write_queue_order(user, order):
    """Renumber the queue 0..n-1 in the given order."""
    rows = {q.episode_id: q for q in PodcastQueue.objects.filter(user=user)}
    changed = []
    for pos, ep_id in enumerate(order):
        row = rows.get(ep_id)
        if row is not None and row.position != pos:
            row.position = pos
            changed.append(row)
    if changed:
        PodcastQueue.objects.bulk_update(changed, ['position'])


def _current_episode_id(user, queued, hint=None):
    """The episode being listened to: the client's word for it, else the
    queued episode most recently listened to and not finished."""
    if hint in queued:
        return hint
    p = (
        EpisodeProgress.objects
        .filter(user=user, episode_id__in=queued, played=False, position_seconds__gt=0)
        .order_by('-updated_at')
        .values_list('episode_id', flat=True)
        .first()
    )
    return p


def _enqueue(user, episode_ids, where, current_hint=None):
    """Put episodes that aren't queued yet into the queue at `where`
    (back, front or after_current), keeping their given order.
    Returns the ids that were added."""
    queued = _queue_order(user)
    present = set(queued)
    new = []
    for ep_id in episode_ids:
        if ep_id not in present and ep_id not in new:
            new.append(ep_id)
    if not new:
        return []
    if where == PodcastSettings.FRONT:
        at = 0
    elif where == PodcastSettings.AFTER_CURRENT:
        current = _current_episode_id(user, queued, current_hint)
        at = queued.index(current) + 1 if current in queued else 0
    else:
        at = len(queued)
    PodcastQueue.objects.bulk_create(
        [PodcastQueue(user=user, episode_id=ep_id, position=len(queued) + i) for i, ep_id in enumerate(new)],
        ignore_conflicts=True,
    )
    _write_queue_order(user, queued[:at] + new + queued[at:])
    _leave_inbox(user, new)
    # Wanted again after "Folge löschen": devices keep (or fetch) their download.
    EpisodeProgress.objects.filter(user=user, episode_id__in=new, discarded_at__isnull=False).update(discarded_at=None)
    _keep_sorted(user)
    return new


def _dequeue(user, episode_ids):
    """Take episodes out of the user's queue. The others keep their order
    (a queue kept sorted stays sorted: taking some out doesn't change that)."""
    PodcastQueue.objects.filter(user=user, episode_id__in=episode_ids).delete()


def _sort_key(rule):
    """Sort key for one queue row ({pub_date, duration, position, feed_title})
    by a PodcastSettings.SORT_ORDERS rule. Ties keep their current order
    (the sort is stable)."""
    def ts(row):
        return row['pub_date'].timestamp() if row['pub_date'] else None

    def remaining(row):
        return max((row['duration'] or 0) - (row['position'] or 0), 0)

    if rule == 'new':  # no date: last
        return lambda r: (ts(r) is None, -(ts(r) or 0))
    if rule == 'old':  # no date: last as well
        return lambda r: (ts(r) is None, ts(r) or 0)
    if rule == 'short':
        return remaining
    if rule == 'long':
        return lambda r: -remaining(r)
    if rule == 'feed':  # by podcast title, then newest first
        return lambda r: ((r['feed_title'] or '').casefold(), ts(r) is None, -(ts(r) or 0))
    return None


def _keep_sorted(user, rule=None):
    """Sort the user's whole queue by their keep_sorted rule (or `rule`), if
    there is one. Returns True if it is kept sorted."""
    if rule is None:
        rule = PodcastSettings.objects.filter(user=user).values_list('keep_sorted', flat=True).first() or ''
    key = _sort_key(rule)
    if key is None:
        return False
    rows = list(
        PodcastQueue.objects.filter(user=user)
        .order_by('position', 'added_at', 'id')
        .values('episode_id', pub_date=F('episode__pub_date'), duration=F('episode__duration_seconds'),
                feed_title=F('episode__feed__title'))
    )
    positions = dict(
        EpisodeProgress.objects.filter(user=user, episode_id__in=[r['episode_id'] for r in rows])
        .values_list('episode_id', 'position_seconds')
    )
    for r in rows:
        r['position'] = positions.get(r['episode_id'], 0)
    rows.sort(key=key)
    _write_queue_order(user, [r['episode_id'] for r in rows])
    return True


def _leave_inbox(user, episode_ids):
    """Queued or started episodes are no longer new (AntennaPod's rule): out
    of the inbox, without counting as played."""
    EpisodeProgress.objects.filter(user=user, episode_id__in=episode_ids, dismissed=False).update(dismissed=True)
    have = set(EpisodeProgress.objects.filter(user=user, episode_id__in=episode_ids).values_list('episode_id', flat=True))
    EpisodeProgress.objects.bulk_create(
        [EpisodeProgress(user=user, episode_id=ep_id, dismissed=True) for ep_id in episode_ids if ep_id not in have],
        ignore_conflicts=True,
    )


# ---------------------------------------------------------------------------
# Duration helper
# ---------------------------------------------------------------------------

def _parse_duration(raw):
    """Return duration in seconds from itunes:duration string or int."""
    if not raw:
        return 0
    if isinstance(raw, int):
        return raw
    raw = str(raw).strip()
    parts = raw.split(':')
    try:
        if len(parts) == 3:
            return int(parts[0]) * 3600 + int(parts[1]) * 60 + int(parts[2])
        if len(parts) == 2:
            return int(parts[0]) * 60 + int(parts[1])
        return int(raw)
    except (ValueError, TypeError):
        return 0


# ---------------------------------------------------------------------------
# Shownotes, chapters and transcripts of a feed entry
# ---------------------------------------------------------------------------

def _entry_description(entry):
    """Shownotes HTML: content:encoded when the feed has it (that's where the
    full notes usually are), else the description/summary."""
    for c in entry.get('content') or []:
        value = (c.get('value') or '').strip()
        if value:
            return value[:MAX_DESCRIPTION]
    return (entry.get('summary') or entry.get('description') or '')[:MAX_DESCRIPTION]


def _entry_chapters_url(entry):
    url = ((entry.get('podcast_chapters') or {}).get('url') or '').strip()
    return url[:1000] if url.lower().startswith(('http://', 'https://')) else ''


def _entry_transcripts(entry):
    raw = entry.get('podcast_transcripts')
    if raw is None and entry.get('podcast_transcript'):
        raw = [entry['podcast_transcript']]  # feedparser without feedext
    transcripts = []
    for t in raw or []:
        url = (t.get('url') or '').strip()
        if not url.lower().startswith(('http://', 'https://')) or len(url) > 1000:
            continue
        transcripts.append({
            'url': url,
            'type': (t.get('type') or '')[:100],
            'language': (t.get('language') or '')[:35],
        })
        if len(transcripts) >= MAX_TRANSCRIPTS:
            break
    return transcripts


def _apply_extras(episode, description, chapters_url, psc_chapters, transcripts):
    """Copy what the feed says now onto `episode`; True if anything changed.

    Inline (Podlove) chapters win. Without them, fetched chapters (from
    chapters_url, or from the audio file when there's no URL) stay cached for
    as long as the URL stays the same; inline chapters going away means a
    fresh fetch.
    """
    changed = []
    if episode.description != description:
        episode.description = description
        changed.append('description')
    if episode.transcripts != transcripts:
        episode.transcripts = transcripts
        changed.append('transcripts')
    if psc_chapters:
        chapters, fetched_at = psc_chapters, None
    elif chapters_url == episode.chapters_url and episode.chapters_fetched_at:
        chapters, fetched_at = episode.chapters, episode.chapters_fetched_at
    else:
        chapters, fetched_at = [], None
    if episode.chapters_url != chapters_url:
        episode.chapters_url = chapters_url
        changed.append('chapters_url')
    if episode.chapters != chapters:
        episode.chapters = chapters
        changed.append('chapters')
    if episode.chapters_fetched_at != fetched_at:
        episode.chapters_fetched_at = fetched_at
        changed.append('chapters_fetched_at')
    return bool(changed)


EXTRA_FIELDS = ['description', 'transcripts', 'chapters_url', 'chapters', 'chapters_fetched_at']


# ---------------------------------------------------------------------------
# Shared feed refresh helper
# ---------------------------------------------------------------------------

class FeedFetchError(Exception):
    """A feed that could not be fetched or is not a feed; str() is the short
    message stored in PodcastFeed.last_error ("HTTP 404", "timeout", …)."""


@dataclass
class FeedFetch:
    """What _fetch_feed got. status 200: `parsed` is the feedparser result and
    etag/last_modified are the new validators ('' when the server sent none).
    status 304: unchanged since the validators that were sent, parsed is None.
    final_url is where the content came from after all redirects;
    permanent_url is the requested URL after following only the 301/308 hops
    at the start of the redirect chain (== the requested URL without one)."""
    status: int
    parsed: object = None
    etag: str = ''
    last_modified: str = ''
    final_url: str = ''
    permanent_url: str = ''
    body_hash: str = ''


# What changes in a feed on every request without anything in it changing:
# the build date, comments with a generation time.
_VOLATILE = re.compile(rb'<lastBuildDate>.*?</lastBuildDate>|<!--.*?-->', re.S)


def _body_hash(content):
    """Fingerprint of a feed body, without the parts that change on every request."""
    return hashlib.sha256(_VOLATILE.sub(b'', content)).hexdigest()


PERMANENT_REDIRECTS = (301, 308)


def _fetch_feed(url, etag='', last_modified='', body_hash='', user_agent='diora podcast refresh'):
    """Download and parse a feed (with a timeout, unlike feedparser's own
    fetch). With etag/last_modified it is a conditional GET; with body_hash a
    body like the last one counts as unchanged too (a 304), unparsed —
    parsing is what takes the time (seconds for a big feed). Returns a
    FeedFetch; raises FeedFetchError when there is nothing usable."""
    headers = {'User-Agent': user_agent}
    if etag:
        headers['If-None-Match'] = etag
    if last_modified:
        headers['If-Modified-Since'] = last_modified
    try:
        r = requests.get(url, timeout=FEED_TIMEOUT, headers=headers)
    except requests.Timeout:
        raise FeedFetchError('timeout')
    except requests.TooManyRedirects:
        raise FeedFetchError('too many redirects')
    except requests.ConnectionError:
        raise FeedFetchError('connection failed')
    except requests.RequestException as e:
        raise FeedFetchError(f'request failed ({type(e).__name__})')

    # Each hop's target is the next hop's URL (the last one's is r.url).
    permanent_url = url
    targets = [h.url for h in r.history[1:]] + [r.url]
    for hop, target in zip(r.history, targets):
        if hop.status_code not in PERMANENT_REDIRECTS:
            break
        permanent_url = target

    if r.status_code == 304:
        if not (etag or last_modified):
            raise FeedFetchError('HTTP 304')  # nothing to be unchanged from
        return FeedFetch(304, None, etag, last_modified, r.url, permanent_url)
    if r.status_code >= 400:
        raise FeedFetchError(f'HTTP {r.status_code}')

    new_hash = _body_hash(r.content)
    if body_hash and new_hash == body_hash:
        return FeedFetch(304, None, etag, last_modified, r.url, permanent_url, body_hash)

    parsed = feedparser.parse(r.content, response_headers={'content-location': r.url, **r.headers})
    if parsed.get('bozo') and not parsed.entries and not parsed.feed.get('title'):
        raise FeedFetchError('not a feed')
    new_etag = r.headers.get('ETag', '')
    new_modified = r.headers.get('Last-Modified', '')
    return FeedFetch(
        r.status_code, parsed,
        # A validator too long to store is dropped (a truncated one never matches).
        new_etag if len(new_etag) <= 500 else '',
        new_modified if len(new_modified) <= 100 else '',
        r.url, permanent_url, new_hash,
    )


def _validators(feeds):
    """If-None-Match / If-Modified-Since for one download that serves all
    `feeds` (several users' subscriptions of the same URL): only when every
    one of them was fetched before and stored the same validators, because a
    304 then skips all of them. Otherwise {} (a full response)."""
    pairs = {(f.http_etag, f.http_last_modified, f.http_body_hash) for f in feeds}
    if len(pairs) != 1 or any(f.last_refreshed_at is None for f in feeds):
        return {}
    etag, last_modified, body_hash = pairs.pop()
    out = {'etag': etag, 'last_modified': last_modified} if etag or last_modified else {}
    if body_hash:
        out['body_hash'] = body_hash
    return out


def _error_text(e):
    if isinstance(e, FeedFetchError):
        return str(e)[:500]
    return (str(e) or type(e).__name__)[:500]


def _record_error(feed_obj, e):
    feed_obj.last_error = _error_text(e)
    feed_obj.last_error_at = timezone.now()
    feed_obj.error_count += 1
    try:
        feed_obj.save(update_fields=['last_error', 'last_error_at', 'error_count'])
    except Exception:
        log.exception('could not store the refresh error of feed %s', feed_obj.pk)


ERROR_FIELDS = ['last_error', 'last_error_at', 'error_count']


def _clear_error(feed_obj):
    feed_obj.last_error = ''
    feed_obj.last_error_at = None
    feed_obj.error_count = 0


BACKOFF_AFTER = 3        # failures in a row before the hourly job backs off
BACKOFF_MAX_HOURS = 24


def _backoff_hours(error_count):
    """Hours between hourly-job attempts for a feed failing error_count
    times in a row: 2, 4, 8, 16, 24 … from the third failure on."""
    if error_count < BACKOFF_AFTER:
        return 0
    return min(2 ** (error_count - 2), BACKOFF_MAX_HOURS)


def _in_backoff_q(now=None):
    """Q for feeds that failed too often to be tried again yet."""
    now = now or timezone.now()
    n = BACKOFF_AFTER
    while _backoff_hours(n) < BACKOFF_MAX_HOURS:
        n += 1
    # From n failures on the wait is capped; below, one term per count.
    q = Q(error_count__gte=n, last_error_at__gt=now - timedelta(hours=BACKOFF_MAX_HOURS))
    for count in range(BACKOFF_AFTER, n):
        q |= Q(error_count=count, last_error_at__gt=now - timedelta(hours=_backoff_hours(count)))
    return q


# File endings of an enclosure without a usable type.
AUDIO_EXTENSIONS = ('.mp3', '.m4a', '.m4b', '.ogg', '.oga', '.opus', '.aac', '.flac', '.wav')
VIDEO_EXTENSIONS = ('.mp4', '.m4v', '.mov', '.webm')


def _refresh_feeds(feeds, full=False, deadline=None):
    """Refresh many feeds at once: the downloads run in parallel and each
    feed URL is fetched once, even when several users subscribed to it; each
    download is written as soon as it's there (the writes one by one, while
    the others still load). Returns (refreshed, failed, new).
    `full`: download every feed again (no 304) and take in episodes the
    reader skipped before (see _apply_feed). `deadline` (time.monotonic()):
    stop there; feeds not done by then stay as they are."""
    feeds = list(feeds)
    refreshed = failed = new = 0
    by_url = {}
    for f in feeds:
        by_url.setdefault(f.rss_url, []).append(f)
    if not by_url:
        return 0, 0, 0

    def fetch(url):
        try:
            return url, _fetch_feed(url, **({} if full else _validators(by_url[url]))), None
        except Exception as e:  # network, HTTP or parser trouble: report and go on
            return url, None, e

    pool = ThreadPoolExecutor(max_workers=min(FEED_WORKERS, len(by_url)))
    futures = [pool.submit(fetch, url) for url in sorted(by_url)]
    try:
        wait = None if deadline is None else max(0.0, deadline - time.monotonic())
        for future in as_completed(futures, timeout=wait):
            url, fetched, err = future.result()
            for feed in by_url[url]:
                if err is not None:
                    failed += 1
                    log.warning('refreshing %s failed: %s', feed.rss_url, err)
                    _record_error(feed, err)
                    continue
                try:
                    new += _refresh_feed(feed, fetched, full=full)
                    refreshed += 1
                except Exception as e:
                    failed += 1
                    log.warning('updating %s failed: %s', feed.rss_url, e)
    except FuturesTimeout:
        log.info('refresh stopped at its deadline: %d of %d feeds done', refreshed + failed, len(feeds))
    finally:
        pool.shutdown(wait=False, cancel_futures=True)
    return refreshed, failed, new


def _refresh_feed(feed_obj, fetched=None, full=False):
    """Fetch rss_url (unless `fetched`, a FeedFetch, is given) and upsert
    episodes. Returns count of new episodes. A failure is stored on the feed
    (last_error, error_count) and raised again."""
    if fetched is None:
        try:
            fetched = _fetch_feed(feed_obj.rss_url, **_validators([feed_obj]))
        except Exception as e:
            _record_error(feed_obj, e)
            raise
    try:
        return _apply_feed(feed_obj, fetched, full=full)
    except Exception as e:
        _record_error(feed_obj, e)
        raise


def _move_to_permanent_url(feed_obj, new_url):
    """The feed answered with a permanent redirect: subscribe to the new URL,
    unless the user already has that one too (then nothing changes)."""
    if not new_url or new_url == feed_obj.rss_url or len(new_url) > 1000:
        return
    try:
        with transaction.atomic():
            PodcastFeed.objects.filter(pk=feed_obj.pk).update(rss_url=new_url)
    except IntegrityError:  # the user's other subscription has it
        return
    log.info('feed %s moved permanently: %s -> %s', feed_obj.pk, feed_obj.rss_url, new_url)
    feed_obj.rss_url = new_url


def _entry_media_url(entry):
    """The episode's file: its audio enclosure, else a video one (the players
    play only its sound). '' when there is neither."""
    video = ''
    for enc in entry.get('enclosures', []):
        url = enc.get('url', '')
        mime = (enc.get('type') or '').lower()
        path = urllib.parse.urlsplit(url).path.lower()
        if mime.startswith('audio/') or path.endswith(AUDIO_EXTENSIONS):
            return url
        if not video and (mime.startswith('video/') or path.endswith(VIDEO_EXTENSIONS)):
            video = url
    # A YouTube feed has no files: the video's page, whose sound the app finds itself.
    return video or youtube_mod.video_url(entry)


def _apply_feed(feed_obj, fetched, full=False):
    """Store a fetched feed. `full` (a re-read after the reader learned to
    take more episodes, e.g. video ones): episodes it finds that came out
    before the last refresh go into the podcast's list, not into "Neu"."""
    now = timezone.now()
    previous_refresh = feed_obj.last_refreshed_at
    if fetched.status == 304:
        # Unchanged since the last full download: nothing to parse or write.
        feed_obj.last_refreshed_at = now
        _clear_error(feed_obj)
        feed_obj.save(update_fields=['last_refreshed_at', *ERROR_FIELDS])
        return 0

    parsed = fetched.parsed
    channel = parsed.feed
    # The first fetch of a new subscription brings the whole back catalogue.
    first_fetch = feed_obj.last_refreshed_at is None

    feed_obj.title = channel.get('title', feed_obj.title or feed_obj.rss_url)[:300]
    feed_obj.description = channel.get('subtitle', channel.get('description', ''))[:2000]
    feed_obj.author = channel.get('author', channel.get('itunes_author', ''))[:300]
    feed_obj.link = channel.get('link', '')[:1000]

    # Artwork
    image = channel.get('image', {})
    feed_obj.artwork_url = (
        channel.get('itunes_image', {}).get('href', '')
        or image.get('href', '')
        or image.get('url', '')
        or feed_obj.artwork_url   # a YouTube feed has none: the channel's, found when subscribing
    )[:1000]
    if not feed_obj.artwork_url and youtube_mod.feed_channel_id(channel):
        feed_obj.artwork_url = youtube_mod.channel_artwork(youtube_mod.feed_channel_id(channel))[:1000]

    feed_obj.last_refreshed_at = now
    feed_obj.save()

    max_ep = getattr(settings, 'PODCAST_MAX_EPISODES_PER_FEED', 200)
    new_count = 0
    new_episodes_list = []
    entries = parsed.entries[:max_ep]

    # Episodes we already have, in one query, so shownotes/chapters/transcripts
    # that changed in the feed can be brought up to date (only those rows).
    guids = [str(e.get('id') or e.get('guid') or '')[:1000] for e in entries]
    existing = {
        ep.guid: ep
        for ep in feed_obj.episodes.filter(guid__in=[g for g in guids if g]).only('id', 'guid', *EXTRA_FIELDS)
    }
    changed_episodes = {}

    for entry in entries:
        audio_url = _entry_media_url(entry)
        if not audio_url:
            continue

        guid = entry.get('id') or entry.get('guid') or audio_url
        guid = str(guid)[:1000]
        title = entry.get('title', 'Untitled')[:500]
        video = youtube_mod.is_video(audio_url)
        description = youtube_mod.description_html(entry.get('summary')) if video else _entry_description(entry)
        chapters_url = _entry_chapters_url(entry)
        psc_chapters = chapters_mod.from_psc(entry)
        transcripts = _entry_transcripts(entry)

        known = existing.get(guid)
        if known is not None:
            if _apply_extras(known, description, chapters_url, psc_chapters, transcripts):
                changed_episodes[known.id] = known
            continue

        duration_raw = entry.get('itunes_duration', 0)
        duration_seconds = _parse_duration(duration_raw)

        pub_date = None
        if entry.get('published_parsed'):
            import datetime as dt
            t = entry.published_parsed
            pub_date = dt.datetime(*t[:6], tzinfo=dt.timezone.utc)

        ep_number = None
        try:
            ep_number = int(entry.get('itunes_episode', ''))
        except (ValueError, TypeError):
            pass

        season_number = None
        try:
            season_number = int(entry.get('itunes_season', ''))
        except (ValueError, TypeError):
            pass

        artwork_url = (youtube_mod.entry_artwork(entry) if video else entry.get('itunes_image', {}).get('href', ''))[:1000]

        episode_obj, created = PodcastEpisode.objects.get_or_create(
            feed=feed_obj,
            guid=guid,
            defaults={
                'title': title,
                'description': description,
                'audio_url': audio_url[:1000],
                'duration_seconds': duration_seconds,
                'pub_date': pub_date,
                'episode_number': ep_number,
                'season_number': season_number,
                'artwork_url': artwork_url,
                'chapters_url': chapters_url,
                'chapters': psc_chapters,
                'transcripts': transcripts,
            },
        )
        existing[guid] = episode_obj  # the same guid twice in one feed
        if created:
            new_count += 1
            new_episodes_list.append(episode_obj)
        elif _apply_extras(episode_obj, description, chapters_url, psc_chapters, transcripts):
            changed_episodes[episode_obj.id] = episode_obj

    if changed_episodes:
        PodcastEpisode.objects.bulk_update(list(changed_episodes.values()), EXTRA_FIELDS, batch_size=100)

    if full and not first_fetch and new_episodes_list:
        # Found only now, but out before the last refresh: not news.
        late = [e for e in new_episodes_list if e.pub_date is None or e.pub_date < previous_refresh]
        if late:
            EpisodeProgress.objects.bulk_create(
                [EpisodeProgress(user=feed_obj.user, episode=e, dismissed=True) for e in late],
                ignore_conflicts=True,
            )
            gone = {e.id for e in late}
            new_episodes_list = [e for e in new_episodes_list if e.id not in gone]

    if first_fetch and len(new_episodes_list) > NEW_ON_SUBSCRIBE:
        # Only the newest episodes count as new; the rest stay in the
        # podcast's episode list, out of the inbox (dismissed, not played).
        newest_first = sorted(new_episodes_list, key=lambda e: e.pub_date or e.discovered_at, reverse=True)
        older = newest_first[NEW_ON_SUBSCRIBE:]
        EpisodeProgress.objects.bulk_create(
            [EpisodeProgress(user=feed_obj.user, episode=e, dismissed=True) for e in older],
            ignore_conflicts=True,
        )
        new_episodes_list = newest_first[:NEW_ON_SUBSCRIBE]

    # The feed's own rules for its new episodes: those its filter turns down,
    # and all of them for 'ignore', go straight out of the inbox (not played);
    # 'queue' puts them in the queue (which also takes them out of the inbox);
    # 'inbox' leaves them there.
    if new_episodes_list:
        unwanted = [e for e in new_episodes_list if not feed_obj.matches_filter(e.title, e.duration_seconds)]
        if feed_obj.new_episodes == PodcastFeed.NEW_IGNORE or feed_obj.kind == PodcastFeed.AUDIOBOOK:
            unwanted = new_episodes_list   # an audiobook's new chapters wait in the library
        if unwanted:
            gone = {e.id for e in unwanted}
            _leave_inbox(feed_obj.user, list(gone))
            new_episodes_list = [e for e in new_episodes_list if e.id not in gone]

    if feed_obj.new_episodes == PodcastFeed.NEW_QUEUE and new_episodes_list and feed_obj.kind != PodcastFeed.AUDIOBOOK:
        where = PodcastSettings.for_user(feed_obj.user).auto_queue_position
        # Newest first in front of what's there; oldest first when appended,
        # so the end of the queue plays in the order the episodes came out.
        newest_first = sorted(
            new_episodes_list,
            key=lambda e: e.pub_date or e.discovered_at,
            reverse=True,
        )
        ordered = newest_first if where != PodcastSettings.BACK else newest_first[::-1]
        _enqueue(feed_obj.user, [e.id for e in ordered], where)

    # Only now, with every episode written: the validators (a 304 next time
    # skips all of the above) and a clean error state.
    feed_obj.http_etag = fetched.etag
    feed_obj.http_last_modified = fetched.last_modified
    feed_obj.http_body_hash = fetched.body_hash
    _clear_error(feed_obj)
    feed_obj.save(update_fields=['http_etag', 'http_last_modified', 'http_body_hash', *ERROR_FIELDS])
    _move_to_permanent_url(feed_obj, fetched.permanent_url)
    return new_count


# ---------------------------------------------------------------------------
# Search
# ---------------------------------------------------------------------------

def podcast_search(request):
    q = request.GET.get('q', '').strip()
    if not q:
        return JsonResponse({'results': []})
    if youtube_mod.is_youtube(q):
        try:
            found = youtube_mod.resolve(q)
        except youtube_mod.NotAChannel:
            return JsonResponse({'results': []})
        return JsonResponse({'results': [{
            'id': found['feed'], 'title': found['title'] or 'YouTube', 'author': 'YouTube',
            'artwork_url': found['artwork'], 'rss_url': found['feed'], 'genre': 'YouTube',
        }]})

    try:
        url = f'https://itunes.apple.com/search?term={urllib.parse.quote(q)}&media=podcast&limit=20'
        resp = requests.get(url, timeout=getattr(settings, 'ITUNES_TIMEOUT', 6))
        resp.raise_for_status()
        raw = resp.json().get('results', [])
    except Exception as e:
        return JsonResponse({'error': str(e)}, status=502)

    results = []
    for r in raw:
        results.append({
            'id': r.get('collectionId'),
            'title': r.get('collectionName', ''),
            'author': r.get('artistName', ''),
            'artwork_url': r.get('artworkUrl100', ''),
            'rss_url': r.get('feedUrl', ''),
            'genre': r.get('primaryGenreName', ''),
        })

    return JsonResponse({'results': results})


# ---------------------------------------------------------------------------
# Feed list
# ---------------------------------------------------------------------------

@csrf_exempt
def feed_list(request):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    return JsonResponse({'feeds': _feeds_data(request.user)})


FEED_SETTING_FIELDS = (
    'playback_speed', 'skip_intro', 'skip_outro', 'volume',
    'new_episodes', 'include_words', 'exclude_words', 'min_duration', 'auto_queue', 'auto_download',
    'notify', 'episode_sort', 'episode_filter', 'skip_silence', 'kind',
)


def _feed_settings(feed):
    return {'feed_id': feed.id, **{name: getattr(feed, name) for name in FEED_SETTING_FIELDS}}


def _feeds_data(user):
    """The user's feeds with their settings and, per feed, how many episodes
    it has, how many of them are unplayed and how many new (as in the inbox:
    neither played nor dismissed), and when the user last listened to one.
    Two queries, however many feeds: the feeds with their episode count, and
    the user's progress rows counted per feed."""
    feeds = list(
        user.podcast_feeds
        .annotate(latest_episode_at=Max('episodes__pub_date'), episode_count=Count('episodes'))
        .values('id', 'title', 'artwork_url', 'rss_url', 'link', 'last_refreshed_at', 'author', 'added_at',
                'latest_episode_at', 'episode_count', 'last_error', 'last_error_at', 'error_count',
                *FEED_SETTING_FIELDS)
    )
    listened = {
        row['episode__feed_id']: row
        for row in EpisodeProgress.objects.filter(user=user, episode__feed__user=user)
        .order_by()
        .values('episode__feed_id')
        .annotate(
            played_count=Count('id', filter=Q(played=True)),
            hidden_count=Count('id', filter=Q(played=True) | Q(dismissed=True)),
            last_played_at=Max('updated_at', filter=Q(played=True) | Q(position_seconds__gt=0)),
        )
    }

    for f in feeds:
        f['feed_id'] = f['id']  # as in feeds/<id>/settings/
        mine = listened.get(f['id'], {})
        f['unplayed_count'] = max(f['episode_count'] - mine.get('played_count', 0), 0)
        f['new_count'] = max(f['episode_count'] - mine.get('hidden_count', 0), 0)
        last = mine.get('last_played_at')
        f['last_played_at'] = last.isoformat() if last else None
        if f['last_refreshed_at']:
            f['last_refreshed_at'] = f['last_refreshed_at'].isoformat()
        if f['last_error_at']:
            f['last_error_at'] = f['last_error_at'].isoformat()
        if f['added_at']:
            f['added_at'] = f['added_at'].isoformat()
        if f['latest_episode_at']:
            f['latest_episode_at'] = f['latest_episode_at'].isoformat()
    return feeds


# ---------------------------------------------------------------------------
# Add feed
# ---------------------------------------------------------------------------

@csrf_exempt
@require_http_methods(['POST'])
def add_feed(request):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    try:
        body = json.loads(request.body)
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)

    rss_url = body.get('rss_url', '').strip()
    if not rss_url:
        return JsonResponse({'error': 'rss_url required'}, status=400)
    title = body.get('title', rss_url)
    artwork = ''
    if youtube_mod.is_youtube(rss_url) and not youtube_mod.is_feed(rss_url):
        # A channel, playlist or video address: its channel's feed (the server reads only that).
        try:
            found = youtube_mod.resolve(rss_url)
        except youtube_mod.NotAChannel as e:
            return JsonResponse({'error': f'No YouTube channel there ({e})'}, status=400)
        rss_url, artwork = found['feed'], found['artwork']
        title = found['title'] or title

    feed, created = PodcastFeed.objects.get_or_create(
        user=request.user,
        rss_url=rss_url,
        defaults={'title': title[:300], 'artwork_url': artwork[:1000],
                  **PodcastSettings.new_feed_defaults(request.user)},
    )

    new_episodes = 0
    if created:
        try:
            new_episodes = _refresh_feed(feed)
        except Exception as e:
            feed.delete()
            return JsonResponse({'error': f'Failed to parse feed: {e}'}, status=400)

    return JsonResponse({
        'ok': True,
        'created': created,
        'feed_id': feed.id,
        'title': feed.title,
        'artwork_url': feed.artwork_url,
        'new_episodes': new_episodes,
    })


# ---------------------------------------------------------------------------
# Remove feed
# ---------------------------------------------------------------------------

@csrf_exempt
@require_http_methods(['POST'])
def remove_feed(request, pk):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    try:
        feed = PodcastFeed.objects.get(pk=pk, user=request.user)
        feed.delete()
        return JsonResponse({'ok': True})
    except PodcastFeed.DoesNotExist:
        return JsonResponse({'error': 'not found'}, status=404)


# ---------------------------------------------------------------------------
# Set auto-queue
# ---------------------------------------------------------------------------

@csrf_exempt
@require_http_methods(['POST'])
def set_auto_queue(request, pk):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    try:
        body = json.loads(request.body)
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)
    try:
        feed = PodcastFeed.objects.get(pk=pk, user=request.user)
    except PodcastFeed.DoesNotExist:
        return JsonResponse({'error': 'not found'}, status=404)
    # On: new episodes go to the queue; off: back to the inbox (a feed set
    # to 'ignore' stays so). PodcastFeed.save() sets new_episodes to match.
    feed.auto_queue = bool(body.get('auto_queue', not feed.auto_queue))
    feed.save(update_fields=['auto_queue'])
    return JsonResponse({'ok': True, 'auto_queue': feed.auto_queue})


# ---------------------------------------------------------------------------
# Per-feed settings
# ---------------------------------------------------------------------------

def _number(value):
    return isinstance(value, (int, float)) and not isinstance(value, bool)


def _whole(value, low, high):
    if not _number(value) or value != int(value) or not low <= value <= high:
        raise ValueError
    return int(value)


def _speed(value):
    if value is None:
        return None
    if not _number(value) or not MIN_SPEED <= value <= MAX_SPEED:
        raise ValueError
    return float(value)


def _episode_sort(value):
    if value in (None, ''):
        return ''
    if value not in PodcastFeed.EPISODE_SORTS:
        raise ValueError
    return value


def _episode_filter(value):
    """A list of filter ids (each once, in the given order); [] = all episodes."""
    if value is None:
        return []
    if not isinstance(value, list) or any(v not in PodcastFeed.EPISODE_FILTERS for v in value):
        raise ValueError
    return list(dict.fromkeys(value))


def _optional_boolean(value):
    if value is not None and not isinstance(value, bool):
        raise ValueError
    return value


def _volume(value):
    if not _number(value) or float(value) not in PodcastFeed.VOLUMES:
        raise ValueError
    return float(value)


def _kind(value):
    if value not in {key for key, _ in PodcastFeed.KINDS}:
        raise ValueError
    return value


def _new_episodes(value):
    if value not in {key for key, _ in PodcastFeed.NEW_EPISODE_ACTIONS}:
        raise ValueError
    return value


def _boolean(value):
    if not isinstance(value, bool):
        raise ValueError
    return value


def _word_list(value):
    if value is None:
        return ''
    if not isinstance(value, str) or len(value.strip()) > 500:
        raise ValueError
    return value.strip()


FEED_SETTING_PARSERS = {
    'playback_speed': _speed,
    'skip_intro': lambda v: _whole(v, 0, 600),
    'skip_outro': lambda v: _whole(v, 0, 600),
    'volume': _volume,
    'new_episodes': _new_episodes,
    'include_words': _word_list,
    'exclude_words': _word_list,
    'min_duration': lambda v: _whole(v, 0, 600),
    'auto_download': _boolean,
    'notify': _boolean,
    'episode_sort': _episode_sort,
    'episode_filter': _episode_filter,
    'kind': _kind,
    'skip_silence': _optional_boolean,
}


@csrf_exempt
@require_http_methods(['POST'])
def move_feed(request, pk):
    """Give a subscription a new address, keeping it: its settings, episodes, places,
    played and favourite marks, queue entries. For a podcast that moved, or one that was
    a YouTube channel fed through some other server (a podsync-like one): moved to the
    channel itself (a YouTube address of any kind), each episode whose file was named
    after its video becomes that video (docs/youtube.md), so the channel's feed finds it
    again instead of adding it twice. {"ok", "rss_url", "videos": episodes made videos,
    "new_episodes"}."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    try:
        body = json.loads(request.body)
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)
    feed = PodcastFeed.objects.filter(pk=pk, user=request.user).first()
    if feed is None:
        return JsonResponse({'error': 'not found'}, status=404)
    rss_url = str(body.get('rss_url') or '').strip()
    if not rss_url.startswith(('http://', 'https://')) or len(rss_url) > 1000:
        return JsonResponse({'error': 'rss_url required'}, status=400)
    artwork = ''
    if youtube_mod.is_youtube(rss_url) and not youtube_mod.is_feed(rss_url):
        try:
            found = youtube_mod.resolve(rss_url)
        except youtube_mod.NotAChannel as e:
            return JsonResponse({'error': f'No YouTube channel there ({e})'}, status=400)
        rss_url, artwork = found['feed'], found['artwork']
    if rss_url != feed.rss_url and PodcastFeed.objects.filter(user=request.user, rss_url=rss_url).exists():
        return JsonResponse({'error': 'already subscribed at that address'}, status=409)

    videos = 0
    with transaction.atomic():
        if youtube_mod.is_feed(rss_url):
            taken = set(feed.episodes.values_list('guid', flat=True))
            changed = []
            for ep in feed.episodes.only('id', 'guid', 'audio_url'):
                vid = youtube_mod.video_id_in(ep.audio_url)
                guid = f'yt:video:{vid}' if vid else ''
                if not vid or (guid in taken and ep.guid != guid):
                    continue   # not named after a video, or that video is here already
                taken.add(guid)
                ep.guid, ep.audio_url = guid, youtube_mod.WATCH + vid
                changed.append(ep)
            PodcastEpisode.objects.bulk_update(changed, ['guid', 'audio_url'], batch_size=200)
            videos = len(changed)
        feed.rss_url = rss_url
        if artwork:
            feed.artwork_url = artwork[:1000]
        # Whatever the old address answered says nothing about the new one.
        feed.http_etag = feed.http_last_modified = feed.http_body_hash = ''
        feed.save()
    try:
        # Videos the old address didn't carry that came out before its last refresh aren't news.
        new_episodes = _refresh_feed(feed, full=True)
    except Exception as e:  # moved all the same: the next refresh tries again
        log.warning('feed %s moved, first refresh failed: %s', feed.pk, e)
        new_episodes = 0
    return JsonResponse({'ok': True, 'rss_url': feed.rss_url, 'videos': videos, 'new_episodes': new_episodes})


@csrf_exempt
@require_http_methods(['POST'])
def feed_settings(request, pk):
    """Change some of one feed's settings (any of FEED_SETTING_PARSERS);
    returns all of them. Nothing is changed if one value is invalid."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    try:
        body = json.loads(request.body or b'{}')
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)
    if not isinstance(body, dict):
        return JsonResponse({'error': 'invalid JSON'}, status=400)
    try:
        feed = PodcastFeed.objects.get(pk=pk, user=request.user)
    except PodcastFeed.DoesNotExist:
        return JsonResponse({'error': 'not found'}, status=404)

    changes = {}
    for name, parse in FEED_SETTING_PARSERS.items():
        if name in body:
            try:
                changes[name] = parse(body[name])
            except (ValueError, TypeError, OverflowError):
                return JsonResponse({'error': f'invalid {name}', 'field': name}, status=400)
    became_book = changes.get('kind') == PodcastFeed.AUDIOBOOK and feed.kind != PodcastFeed.AUDIOBOOK
    if changes:
        for name, value in changes.items():
            setattr(feed, name, value)
        feed.save(update_fields=list(changes))
    if became_book:   # an audiobook's chapters aren't news (docs/audiobooks.md)
        _leave_inbox(request.user, list(feed.episodes.values_list('id', flat=True)))
    return JsonResponse(_feed_settings(feed))


# ---------------------------------------------------------------------------
# Feed episodes
# ---------------------------------------------------------------------------

@csrf_exempt
def feed_episodes(request, pk):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    try:
        feed = PodcastFeed.objects.get(pk=pk, user=request.user)
    except PodcastFeed.DoesNotExist:
        return JsonResponse({'error': 'not found'}, status=404)

    episodes = list(
        feed.episodes.values(
            'id', 'title', 'description', 'audio_url', 'duration_seconds',
            'pub_date', 'artwork_url', 'episode_number', 'season_number',
        )
    )

    # Batch-fetch progress
    progress_map = {
        ep['episode_id']: ep
        for ep in EpisodeProgress.objects.filter(
            user=request.user,
            episode__feed=feed,
        ).values('episode_id', 'position_seconds', 'played', 'favorite')
    }

    # Batch-fetch queue membership
    queued_ids = set(
        PodcastQueue.objects.filter(
            user=request.user,
            episode__feed=feed,
        ).values_list('episode_id', flat=True)
    )

    for ep in episodes:
        if ep['pub_date']:
            ep['pub_date'] = ep['pub_date'].isoformat()
        prog = progress_map.get(ep['id'], {})
        ep['position_seconds'] = prog.get('position_seconds', 0)
        ep['played'] = prog.get('played', False)
        ep['favorite'] = prog.get('favorite', False)
        ep['in_queue'] = ep['id'] in queued_ids

    return JsonResponse({
        'feed': {
            'id': feed.id,
            'title': feed.title,
            'artwork_url': feed.artwork_url,
            'author': feed.author,
            'auto_queue': feed.auto_queue,
            # For the podcast's info page.
            'description': feed.description,
            'link': feed.link,
            'rss_url': feed.rss_url,
            'last_refreshed_at': feed.last_refreshed_at.isoformat() if feed.last_refreshed_at else None,
        },
        'episodes': episodes,
    })


# ---------------------------------------------------------------------------
# One episode in detail: shownotes, chapters, transcripts
# ---------------------------------------------------------------------------

def _load_chapters(episode):
    """Fetch the chapters the feed doesn't carry inline, once, and cache them
    on `episode`: the JSON at chapters_url, else (no URL, or nothing there)
    the CHAP frames in the ID3v2 tag at the start of the audio file.

    Problems that may go away (timeouts, server errors) aren't cached, so the
    next request tries again; anything else is remembered as "no chapters".
    """
    url = episode.chapters_url
    chapters = []
    try:
        if url:
            try:
                chapters = chapters_mod.fetch(url)
            except chapters_mod.ChaptersInvalid as e:
                log.info('chapters of episode %s unusable: %s', episode.id, e)
        if not chapters and episode.audio_url and not youtube_mod.is_video(episode.audio_url):
            try:
                chapters = chapters_mod.fetch_id3(episode.audio_url)
            except chapters_mod.ChaptersInvalid as e:
                log.info('ID3 chapters of episode %s unusable: %s', episode.id, e)
    except chapters_mod.ChaptersUnavailable as e:
        log.info('chapters of episode %s not available: %s', episode.id, e)
        return []
    now = timezone.now()
    # Only if a refresh hasn't changed the URL in the meantime.
    PodcastEpisode.objects.filter(pk=episode.pk, chapters_url=url).update(
        chapters=chapters, chapters_fetched_at=now,
    )
    episode.chapters, episode.chapters_fetched_at = chapters, now
    return chapters


@csrf_exempt
@require_http_methods(['GET'])
def episode_detail(request, pk):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    try:
        ep = PodcastEpisode.objects.select_related('feed').get(pk=pk, feed__user=request.user)
    except PodcastEpisode.DoesNotExist:
        return JsonResponse({'error': 'not found'}, status=404)

    chapters = ep.chapters
    if not chapters and ep.chapters_fetched_at is None:
        chapters = _load_chapters(ep)

    progress = EpisodeProgress.objects.filter(user=request.user, episode=ep).first()
    return JsonResponse({
        'id': ep.id,
        'title': ep.title,
        'description': sanitize_html(ep.description),
        'timestamps': extract_timestamps(ep.description, ep.duration_seconds),
        'chapters': chapters,
        'transcripts': ep.transcripts,
        'has_transcript': _has_transcript(ep),
        'feed': {'id': ep.feed.id, 'title': ep.feed.title, 'artwork_url': ep.feed.artwork_url},
        'audio_url': ep.audio_url,
        'artwork_url': ep.artwork_url,
        'duration_seconds': ep.duration_seconds,
        'pub_date': ep.pub_date.isoformat() if ep.pub_date else None,
        'episode_number': ep.episode_number,
        'season_number': ep.season_number,
        'position_seconds': progress.position_seconds if progress else 0,
        'played': progress.played if progress else False,
        'favorite': progress.favorite if progress else False,
        'last_played_at': progress.updated_at.isoformat() if progress else None,
        'in_queue': PodcastQueue.objects.filter(user=request.user, episode=ep).exists(),
    })


def _transcript_sources(ep):
    return [t['url'] for t in transcripts_mod.candidates(ep.transcripts)]


def _has_transcript(ep):
    """True if the episode offers a transcript we can read, and it hasn't
    turned out to be unusable."""
    sources = _transcript_sources(ep)
    if not sources:
        return False
    cached = EpisodeTranscript.objects.filter(episode=ep).values_list('sources', 'error').first()
    return not (cached and cached[0] == sources and cached[1])


@csrf_exempt
@require_http_methods(['GET'])
def episode_silences(request, pk):
    """What to cut out of the episode's pauses for skipping silence in the
    browser: {"ready": true, "v": 2, "spans": [[from, to], …]} once found (the
    parts to leave out, podcasts/silence.py); the first request starts the search
    on the server ({"ready": false}, ask again later)."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    ep = PodcastEpisode.objects.filter(pk=pk, feed__user=request.user).only('id', 'silences', 'audio_url').first()
    if ep is None:
        return JsonResponse({'error': 'not found'}, status=404)
    if youtube_mod.is_video(ep.audio_url):   # the server never fetches a video's sound
        return JsonResponse({'ready': True, 'v': silence_mod.VERSION, 'spans': [], 'error': 'youtube'})
    if not silence_mod.current(ep.silences):   # none yet, or found by older rules
        silence_mod.ensure(ep.pk)
        return JsonResponse({'ready': False})
    return JsonResponse({'ready': True, 'v': silence_mod.VERSION, 'spans': ep.silences.get('spans', []),
                         **({'error': ep.silences['error']} if ep.silences.get('error') else {})})


@csrf_exempt
@require_http_methods(['GET'])
def episode_transcript(request, pk):
    """The episode's transcript as {"segments": [{start, end, speaker, text}],
    "source": {url, type, language}}, fetched and parsed on first use, then
    kept. 404 {"error": "no transcript" | "unsupported" | "invalid"} when
    there's none to show; 502 {"error": "unavailable"} when it couldn't be
    loaded right now."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    try:
        ep = PodcastEpisode.objects.get(pk=pk, feed__user=request.user)
    except PodcastEpisode.DoesNotExist:
        return JsonResponse({'error': 'not found'}, status=404)
    if not ep.transcripts:
        return JsonResponse({'error': 'no transcript'}, status=404)
    sources = _transcript_sources(ep)
    if not sources:
        return JsonResponse({'error': 'unsupported'}, status=404)

    cached = EpisodeTranscript.objects.filter(episode=ep).first()
    if cached is not None and cached.sources == sources:
        data, error = cached.data, cached.error
    else:
        try:
            data, final = transcripts_mod.load(ep.transcripts)
            error = ''
        except chapters_mod.FetchUnavailable as e:
            log.info('transcript of episode %s not available: %s', ep.id, e)
            return JsonResponse({'error': 'unavailable'}, status=502)
        except transcripts_mod.Unsupported as e:
            log.info('transcript of episode %s unsupported: %s', ep.id, e)
            data, error, final = None, 'unsupported', True
        except chapters_mod.FetchInvalid as e:
            log.info('transcript of episode %s unusable: %s', ep.id, e)
            data, error, final = None, 'invalid', True
        if final:
            EpisodeTranscript.objects.update_or_create(
                episode=ep, defaults={'sources': sources, 'data': data, 'error': error},
            )
    if error:
        return JsonResponse({'error': error}, status=404)
    return JsonResponse(data)


# ---------------------------------------------------------------------------
# Favorites
# ---------------------------------------------------------------------------

@csrf_exempt
@require_http_methods(['POST'])
def set_favorite(request, pk):
    """Star or unstar an episode: {"favorite": true|false}, no value toggles."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    try:
        body = json.loads(request.body or b'{}')
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)
    if not isinstance(body, dict):
        return JsonResponse({'error': 'invalid JSON'}, status=400)
    try:
        episode = PodcastEpisode.objects.get(pk=pk, feed__user=request.user)
    except PodcastEpisode.DoesNotExist:
        return JsonResponse({'error': 'not found'}, status=404)

    progress, _ = EpisodeProgress.objects.get_or_create(user=request.user, episode=episode)
    wanted = body.get('favorite')
    favorite = (not progress.favorite) if wanted is None else bool(wanted)
    if favorite != progress.favorite:
        # update() leaves updated_at alone: a star isn't listening, and the
        # history and the statistics go by updated_at.
        EpisodeProgress.objects.filter(pk=progress.pk).update(favorite=favorite)
    return JsonResponse({'ok': True, 'favorite': favorite})


@csrf_exempt
@require_http_methods(['POST'])
def set_favorites(request):
    """Star or unstar several episodes at once (choosing several in a list):
    {"episode_ids": [...], "favorite": true}. Like set_favorite, it leaves
    the listening time (updated_at) of existing rows alone."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    try:
        body = json.loads(request.body or b'{}')
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)
    if not isinstance(body, dict):
        return JsonResponse({'error': 'invalid JSON'}, status=400)
    episode_ids = body.get('episode_ids')
    favorite = body.get('favorite')
    if not isinstance(episode_ids, list) or not episode_ids or not isinstance(favorite, bool):
        return JsonResponse({'error': 'episode_ids (a list) and favorite (true/false) required'}, status=400)
    try:
        wanted = {int(i) for i in episode_ids[:EPISODE_IDS_MAX]}
    except (TypeError, ValueError):
        return JsonResponse({'error': 'episode_ids must be numbers'}, status=400)
    valid = set(
        PodcastEpisode.objects.filter(id__in=wanted, feed__user=request.user).values_list('id', flat=True)
    )
    with transaction.atomic():
        rows = EpisodeProgress.objects.filter(user=request.user, episode_id__in=valid)
        rows.exclude(favorite=favorite).update(favorite=favorite)
        if favorite:
            have = set(rows.values_list('episode_id', flat=True))
            EpisodeProgress.objects.bulk_create(
                [EpisodeProgress(user=request.user, episode_id=i, favorite=True) for i in sorted(valid - have)],
                ignore_conflicts=True,
            )
    return JsonResponse({'ok': True, 'favorite': favorite, 'count': len(valid)})


# ---------------------------------------------------------------------------
# Episodes of all feeds: filters, history, search
# ---------------------------------------------------------------------------

EPISODE_FILTERS = {
    'all': Q(),
    'unplayed': Q(played=False),
    'in_progress': Q(position_seconds__gt=0, played=False),
    'played': Q(played=True),
    'favorites': Q(favorite=True),
    'history': Q(played=True) | Q(position_seconds__gt=0),
    'queued': Q(in_queue=True),
}
# Episodes of unknown length (0) come last when sorting by length.
_UNKNOWN_LENGTH_LAST = Case(When(duration_seconds__gt=0, then=Value(0)), default=Value(1), output_field=IntegerField())
EPISODE_SORTS = {
    'new': (F('pub_date').desc(nulls_last=True), '-id'),
    'old': (F('pub_date').asc(nulls_last=True), 'id'),
    'long': (_UNKNOWN_LENGTH_LAST, '-duration_seconds', F('pub_date').desc(nulls_last=True), '-id'),
    'short': (_UNKNOWN_LENGTH_LAST, 'duration_seconds', F('pub_date').desc(nulls_last=True), '-id'),
    'title': (Lower('title'), 'id'),
    'played': (F('last_played_at').desc(nulls_last=True), '-id'),
}
EPISODE_IDS_MAX = 1000
EPISODE_LIST_LIMIT = 100
EPISODE_LIST_MAX = 500
EPISODE_LIST_FIELDS = (
    'id', 'title', 'audio_url', 'duration_seconds', 'pub_date', 'artwork_url',
    'feed__id', 'feed__title', 'feed__artwork_url',
    'position_seconds', 'played', 'favorite', 'in_queue', 'last_played_at',
)


def _int_param(request, name, default):
    raw = request.GET.get(name, '').strip()
    return default if raw == '' else int(raw)


def _episode_list_data(user, filter_name='all', feed_id=None, q='', offset=0, limit=EPISODE_LIST_LIMIT,
                       sort=None, ids=None, notes=False):
    """(episodes, total) for the episode list. Two queries, whatever the
    page size: the count and the page. The user's progress row comes in by
    a join, queue membership by a subquery. No shownotes (up to 50 KB each).

    filter_name: one of EPISODE_FILTERS, or several (a list, or joined by
    commas), which all have to match. sort: one of EPISODE_SORTS; by default
    the newest first, in the history the last listened first. ids: only these
    episodes (the ones downloaded on a device). notes: `q` also matches the
    shownotes; title matches come first."""
    names = filter_name.split(',') if isinstance(filter_name, str) else list(filter_name)
    names = [n for n in names if n and n != 'all']
    qs = (
        PodcastEpisode.objects.filter(feed__user=user)
        .annotate(mine=FilteredRelation('progress', condition=Q(progress__user=user)))
        .annotate(
            position_seconds=Coalesce('mine__position_seconds', Value(0)),
            played=Coalesce('mine__played', Value(False)),
            favorite=Coalesce('mine__favorite', Value(False)),
            last_played_at=F('mine__updated_at'),
            in_queue=Exists(PodcastQueue.objects.filter(user=user, episode=OuterRef('pk'))),
        )
    )
    for name in names:
        qs = qs.filter(EPISODE_FILTERS[name])
    if feed_id is not None:
        qs = qs.filter(feed_id=feed_id)
    if ids is not None:
        qs = qs.filter(id__in=ids)
    if feed_id is None and ids is None:   # audiobooks are listened to in the library
        qs = qs.exclude(feed__kind=PodcastFeed.AUDIOBOOK)
    order = EPISODE_SORTS[sort or ('played' if 'history' in names else 'new')]
    if q and notes:
        qs = qs.filter(Q(title__icontains=q) | Q(description__icontains=q)).annotate(
            title_hit=Case(When(title__icontains=q, then=Value(0)), default=Value(1), output_field=IntegerField()),
        )
        order = ('title_hit', *order)
    elif q:
        qs = qs.filter(title__icontains=q)
    qs = qs.order_by(*order)

    total = qs.count()
    episodes = list(qs.values(*EPISODE_LIST_FIELDS)[offset:offset + limit]) if limit else []
    for ep in episodes:
        ep['pub_date'] = ep['pub_date'].isoformat() if ep['pub_date'] else None
        ep['last_played_at'] = ep['last_played_at'].isoformat() if ep['last_played_at'] else None
        ep['played'] = bool(ep['played'])
        ep['favorite'] = bool(ep['favorite'])
        ep['in_queue'] = bool(ep['in_queue'])
    return episodes, total


@csrf_exempt
@require_http_methods(['GET'])
def episode_list(request):
    """Episodes of all the user's feeds, filtered (all, unplayed, in_progress,
    played, favorites, history, queued; several joined by commas all have to
    match), optionally of one feed, of some episodes (`ids`, comma-separated)
    or matching `q` (in the title; with notes=1 also in the shownotes), sorted
    by `sort` (new, old, long, short, title, played)."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    names = [n.strip() for n in request.GET.get('filter', '').split(',') if n.strip()]
    if any(n not in EPISODE_FILTERS for n in names):
        return JsonResponse({'error': 'unknown filter'}, status=400)
    sort = request.GET.get('sort', '').strip() or None
    if sort is not None and sort not in EPISODE_SORTS:
        return JsonResponse({'error': 'unknown sort'}, status=400)
    try:
        offset = max(_int_param(request, 'offset', 0), 0)
        limit = min(max(_int_param(request, 'limit', EPISODE_LIST_LIMIT), 0), EPISODE_LIST_MAX)
        feed_id = _int_param(request, 'feed_id', None)
        raw_ids = request.GET.get('ids')
        ids = None if raw_ids is None else [int(i) for i in raw_ids.split(',') if i.strip()][:EPISODE_IDS_MAX]
    except ValueError:
        return JsonResponse({'error': 'offset, limit, feed_id and ids must be numbers'}, status=400)
    episodes, total = _episode_list_data(
        request.user, names, feed_id, request.GET.get('q', '').strip(), offset, limit,
        sort=sort, ids=ids, notes=request.GET.get('notes') == '1',
    )
    return JsonResponse({'episodes': episodes, 'offset': offset, 'limit': limit, 'total': total})


# ---------------------------------------------------------------------------
# Listening statistics
# ---------------------------------------------------------------------------

@csrf_exempt
@require_http_methods(['GET'])
def podcast_stats(request):
    """How much the user listened, in total and per feed; `days` limits it to
    progress rows updated within the last N days."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    rows = EpisodeProgress.objects.filter(user=request.user, episode__feed__user=request.user)
    since = None
    raw_days = request.GET.get('days', '').strip()
    if raw_days:
        try:
            days = int(raw_days)
        except ValueError:
            return JsonResponse({'error': 'days must be a number'}, status=400)
        if days < 0:
            return JsonResponse({'error': 'days must not be negative'}, status=400)
        since = timezone.now() - timedelta(days=days)
        rows = rows.filter(updated_at__gte=since)

    # A played episode counts with its full length (or as far as it got, if
    # the feed gave no length); anything else as far as it got.
    listened = Case(
        When(played=True, episode__duration_seconds__gt=0, then=F('episode__duration_seconds')),
        default=F('position_seconds'),
        output_field=IntegerField(),
    )
    per_feed = (
        rows.order_by()
        .values('episode__feed_id', 'episode__feed__title', 'episode__feed__artwork_url')
        .annotate(
            seconds=Sum(listened),
            played_count=Count('id', filter=Q(played=True)),
            started_count=Count('id', filter=Q(played=False, position_seconds__gt=0)),
        )
    )
    total_seconds = episodes_played = episodes_started = 0
    feeds = []
    for row in per_feed:
        seconds = row['seconds'] or 0
        total_seconds += seconds
        episodes_played += row['played_count']
        episodes_started += row['started_count']
        if seconds > 0:
            feeds.append({
                'feed_id': row['episode__feed_id'],
                'title': row['episode__feed__title'],
                'artwork_url': row['episode__feed__artwork_url'],
                'seconds': seconds,
                'episodes_played': row['played_count'],
            })
    feeds.sort(key=lambda f: (-f['seconds'], f['title'].lower(), f['feed_id']))
    return JsonResponse({
        'total_seconds': total_seconds,
        'episodes_played': episodes_played,
        'episodes_started': episodes_started,
        'since': since.isoformat() if since else None,
        'feeds': feeds,
    })


# ---------------------------------------------------------------------------
# Import OPML
# ---------------------------------------------------------------------------

@csrf_exempt
@require_http_methods(['POST'])
def import_opml(request):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    f = request.FILES.get('file')
    if not f:
        return JsonResponse({'error': 'no file uploaded'}, status=400)

    try:
        content = f.read().decode('utf-8', errors='replace')
        root = ET.fromstring(content)
    except Exception as e:
        return JsonResponse({'error': f'invalid OPML: {e}'}, status=400)

    added = 0
    skipped = 0
    new_feed = PodcastSettings.new_feed_defaults(request.user)
    for outline in root.iter('outline'):
        rss_url = outline.get('xmlUrl', '').strip()
        if not rss_url:
            continue
        title = outline.get('title', outline.get('text', rss_url))[:300]
        _, created = PodcastFeed.objects.get_or_create(
            user=request.user,
            rss_url=rss_url,
            defaults={'title': title, **new_feed},
        )
        if created:
            added += 1
        else:
            skipped += 1

    return JsonResponse({'ok': True, 'added': added, 'skipped': skipped})


# ---------------------------------------------------------------------------
# Export OPML
# ---------------------------------------------------------------------------

def _xml_text(s):
    """Drop characters XML 1.0 can't carry at all (escaping is ElementTree's job)."""
    return ''.join(c for c in s if c in '\t\n\r' or (' ' <= c <= '퟿') or ('' <= c <= '<EFBFBD>')
                   or ord(c) >= 0x10000)


@csrf_exempt
@require_http_methods(['GET'])
def export_opml(request):
    """The user's subscriptions as an OPML 2.0 file, for other podcast apps
    (and for import_opml)."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    root = ET.Element('opml', version='2.0')
    head = ET.SubElement(root, 'head')
    ET.SubElement(head, 'title').text = 'diora Podcasts'
    body = ET.SubElement(root, 'body')
    for feed in request.user.podcast_feeds.order_by('title', 'id').only('title', 'rss_url', 'link'):
        title = _xml_text(feed.title or feed.rss_url)
        ET.SubElement(body, 'outline', {
            'type': 'rss',
            'text': title,
            'title': title,
            'xmlUrl': _xml_text(feed.rss_url),
            'htmlUrl': _xml_text(feed.link),
        })
    ET.indent(root)
    xml = '<?xml version="1.0" encoding="UTF-8"?>\n' + ET.tostring(root, encoding='unicode') + '\n'
    response = HttpResponse(xml.encode('utf-8'), content_type='text/x-opml; charset=utf-8')
    response['Content-Disposition'] = 'attachment; filename="diora-podcasts.opml"'
    return response


# ---------------------------------------------------------------------------
# Refresh feed now
# ---------------------------------------------------------------------------

@csrf_exempt
@require_http_methods(['POST'])
def refresh_feed_now(request):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    try:
        body = json.loads(request.body)
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)

    feed_id = body.get('feed_id')
    if not feed_id:
        return JsonResponse({'error': 'feed_id required'}, status=400)

    try:
        feed = PodcastFeed.objects.get(pk=feed_id, user=request.user)
    except PodcastFeed.DoesNotExist:
        return JsonResponse({'error': 'not found'}, status=404)

    try:
        new_episodes = _refresh_feed(feed)
    except Exception as e:
        return JsonResponse({'error': str(e)}, status=502)

    return JsonResponse({'ok': True, 'new_episodes': new_episodes})


@csrf_exempt
@require_http_methods(['POST'])
def refresh_all_feeds(request):
    """Refresh all of the user's feeds on the server in one go. Feeds the
    hourly job (or an earlier call) refreshed within `max_age` seconds are
    skipped. Clients then fetch the result once (state/, inbox/ …).
    The user asked, so feeds the hourly job backs off from are tried too.
    It answers within REFRESH_WAIT seconds: `left` feeds weren't done by
    then (done: false) and wait for the next call or the hourly job."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    try:
        body = json.loads(request.body or b'{}')
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)
    try:
        max_age = max(int(body.get('max_age', 300)), 0)
    except (TypeError, ValueError):
        return JsonResponse({'error': 'max_age must be a number'}, status=400)
    cutoff = timezone.now() - timedelta(seconds=max_age)
    feeds = list(request.user.podcast_feeds.filter(Q(last_refreshed_at__isnull=True) | Q(last_refreshed_at__lt=cutoff)))
    refreshed, failed, new = _refresh_feeds(feeds, deadline=time.monotonic() + REFRESH_WAIT)
    left = len(feeds) - refreshed - failed
    return JsonResponse({'ok': True, 'refreshed': refreshed, 'failed': failed, 'new_episodes': new,
                         'done': left == 0, 'left': left})


@csrf_exempt
def podcast_state(request):
    """Everything a client shows first, in one response: feeds, queue, the
    first page of the inbox (inbox_limit of inbox_total new episodes), the
    settings, how many AntennaPod downloads wait to be offered
    (import-downloads/) and the episodes discarded lately ("Folge löschen":
    a device deletes its older download of them)."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    try:
        limit = min(max(int(request.GET.get('inbox_limit', 100)), 0), 1000)
    except ValueError:
        return JsonResponse({'error': 'inbox_limit must be a number'}, status=400)
    s = PodcastSettings.for_user(request.user)
    downloads = _import_downloads(request.user, s.import_downloads)
    return JsonResponse({
        'feeds': _feeds_data(request.user),
        'queue': _queue_data(request.user),
        'inbox': _inbox_data(request.user, 0, limit, with_description=False),
        'inbox_total': _inbox_episodes(request.user).count(),
        'inbox_limit': limit,
        'settings': _settings_data(s),
        'import_downloads': {'count': len(downloads), 'bytes': sum(e['size'] for e in downloads)},
        'discarded': _discarded_data(request.user),
        'server_time': timezone.now().isoformat(),
    })


# ---------------------------------------------------------------------------
# Queue
# ---------------------------------------------------------------------------

@csrf_exempt
def queue_get(request):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    return JsonResponse({'queue': _queue_data(request.user)})


def _queue_data(user):
    items = list(
        PodcastQueue.objects
        .filter(user=user)
        .order_by('position', 'added_at', 'id')
        .select_related('episode__feed')
        .values(
            'id', 'position',
            'episode__id', 'episode__title', 'episode__audio_url',
            'episode__duration_seconds', 'episode__artwork_url',
            'episode__feed__id', 'episode__feed__title', 'episode__feed__artwork_url',
            'episode__pub_date',
        )
    )

    ep_ids = [item['episode__id'] for item in items]
    progress_map = {
        p['episode_id']: p['position_seconds']
        for p in EpisodeProgress.objects.filter(
            user=user, episode_id__in=ep_ids,
        ).values('episode_id', 'position_seconds')
    }
    for item in items:
        item['position_seconds'] = progress_map.get(item['episode__id'], 0)
        if item['episode__pub_date']:
            item['episode__pub_date'] = item['episode__pub_date'].isoformat()
    return items


@csrf_exempt
@require_http_methods(['POST'])
def queue_add(request):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    try:
        body = json.loads(request.body)
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)

    # One episode (episode_id) or several in order (episode_ids). Where they
    # go: `where` if given, else the user's setting. `current_episode_id` is
    # what the client is playing, for after_current.
    ids = body.get('episode_ids')
    if ids is None:
        ids = [body.get('episode_id')]
    if not isinstance(ids, list):
        return JsonResponse({'error': 'episode_ids must be a list'}, status=400)
    where = body.get('where') or PodcastSettings.for_user(request.user).enqueue_position
    if where not in QUEUE_POSITIONS:
        return JsonResponse({'error': 'unknown position'}, status=400)

    found = set(
        PodcastEpisode.objects.filter(pk__in=[i for i in ids if isinstance(i, int)], feed__user=request.user)
        .values_list('id', flat=True)
    )
    if not found:
        return JsonResponse({'error': 'not found'}, status=404)
    added = _enqueue(
        request.user, [i for i in ids if i in found], where,
        current_hint=body.get('current_episode_id'),
    )
    return JsonResponse({'ok': True, 'created': bool(added), 'added': added})


@csrf_exempt
@require_http_methods(['POST'])
def queue_remove(request):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    try:
        body = json.loads(request.body)
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)

    _dequeue(request.user, [body.get('episode_id')])
    return JsonResponse({'ok': True})


@csrf_exempt
@require_http_methods(['POST'])
def queue_reorder(request):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    try:
        body = json.loads(request.body)
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)

    order = body.get('order', [])  # list of episode ids
    for pos, ep_id in enumerate(order):
        PodcastQueue.objects.filter(user=request.user, episode_id=ep_id).update(position=pos)
    # A queue that is kept sorted goes straight back into its order.
    _keep_sorted(request.user)

    return JsonResponse({'ok': True})


def _choice(choices, what):
    def parse(value):
        if value not in choices:
            raise LookupError(f'unknown {what}')
        return value
    return parse


def _skip_seconds(value):
    if value is None:
        return None
    if not _number(value) or value != int(value) or int(value) not in PodcastSettings.SKIP_CHOICES:
        raise ValueError
    return int(value)


def _smart_played(value):
    return None if value is None else _whole(value, 0, 3600)


def _speed_presets(value):
    """Up to MAX_PRESETS speeds (MIN_SPEED to MAX_SPEED), rounded to two
    places, each once, ascending; null = the apps' default list."""
    if value is None:
        return None
    if not isinstance(value, list) or not 1 <= len(value) <= PodcastSettings.MAX_PRESETS:
        raise ValueError
    out = set()
    for v in value:
        if not _number(v) or not math.isfinite(v) or not MIN_SPEED <= v <= MAX_SPEED:
            raise ValueError
        out.add(round(float(v), 2))
    return sorted(out)


# POST settings/: each field's parser; a LookupError's text is the error.
SETTING_PARSERS = {
    'enqueue_position': _choice(QUEUE_POSITIONS, 'enqueue_position'),
    'auto_queue_position': _choice(QUEUE_POSITIONS, 'auto_queue_position'),
    'keep_sorted': _choice(KEEP_SORTED, 'keep_sorted'),
    'auto_download_default': _boolean,
    'skip_back': _skip_seconds,
    'skip_forward': _skip_seconds,
    'smart_played': _smart_played,
    'speed_presets': _speed_presets,
    'keep_skipped': _boolean,
    'delete_dequeue': _boolean,
    'mobile_downloads': _boolean,
    'mobile_streaming': _choice({key for key, _ in PodcastSettings.STREAMING}, 'mobile_streaming'),
    'times_by_speed': _boolean,
}


def _settings_data(s):
    return {
        'enqueue_position': s.enqueue_position, 'auto_queue_position': s.auto_queue_position,
        'keep_sorted': s.keep_sorted, 'auto_download_default': s.auto_download_default,
        **{name: getattr(s, name) for name in PodcastSettings.PLAYBACK_FIELDS},
    }


def _parse_settings(body, names):
    """{field: value} for those of `names` in body; raises (field, error text)."""
    out = {}
    for name in names:
        if name not in body:
            continue
        try:
            out[name] = SETTING_PARSERS[name](body[name])
        except LookupError as e:
            raise ValueError(name, str(e))
        except (ValueError, TypeError, OverflowError):
            raise ValueError(name, f'invalid {name}')
    return out


@csrf_exempt
def podcast_settings(request):
    """GET the user's podcast settings; POST changes some of them (any of
    SETTING_PARSERS; nothing changes if one value is invalid).

    POST {"initial": {...}} sets the playback settings that are still unset
    (null) and leaves the others: how an app hands over its own values once,
    the first time it finds the account without them."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    s = PodcastSettings.for_user(request.user)
    if request.method == 'POST':
        try:
            body = json.loads(request.body)
        except (json.JSONDecodeError, ValueError):
            return JsonResponse({'error': 'invalid JSON'}, status=400)
        if not isinstance(body, dict):
            return JsonResponse({'error': 'invalid JSON'}, status=400)
        initial = body.get('initial') or {}
        if not isinstance(initial, dict):
            return JsonResponse({'error': 'invalid initial', 'field': 'initial'}, status=400)
        try:
            changes = _parse_settings(body, SETTING_PARSERS)
            first = _parse_settings(initial, PodcastSettings.UNSET_FIELDS)
        except ValueError as e:
            name, text = e.args
            return JsonResponse({'error': text, 'field': name}, status=400)
        if first:
            # Only where nothing is set yet, decided by the database (two
            # devices handing theirs over at once: the first one wins).
            with transaction.atomic():
                s = PodcastSettings.objects.select_for_update().get(pk=s.pk)
                fields = [name for name, value in first.items() if getattr(s, name) is None and value is not None]
                for name in fields:
                    setattr(s, name, first[name])
                if fields:
                    s.save(update_fields=fields)
        if changes:
            for name, value in changes.items():
                setattr(s, name, value)
            s.save(update_fields=list(changes))
        if 'keep_sorted' in changes:
            _keep_sorted(request.user, s.keep_sorted)  # sorted right away
    elif request.method != 'GET':
        return JsonResponse({'error': 'method not allowed'}, status=405)
    return JsonResponse({
        **_settings_data(s),
        'positions': [{'value': k, 'label': v} for k, v in PodcastSettings.POSITIONS],
        'skip_choices': list(PodcastSettings.SKIP_CHOICES),
        'speed_range': [MIN_SPEED, MAX_SPEED],
    })


# ---------------------------------------------------------------------------
# Progress
# ---------------------------------------------------------------------------

@csrf_exempt
@require_http_methods(['POST'])
def save_progress(request):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    try:
        body = json.loads(request.body)
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)

    episode_id = body.get('episode_id')
    position = int(body.get('position_seconds', 0))

    try:
        episode = PodcastEpisode.objects.get(pk=episode_id, feed__user=request.user)
    except PodcastEpisode.DoesNotExist:
        return JsonResponse({'error': 'not found'}, status=404)

    # A feed without durations (YouTube): the one the player found, once.
    duration = body.get('duration_seconds')
    if not episode.duration_seconds and _number(duration) and math.isfinite(duration) and 1 <= duration <= MAX_DURATION:
        episode.duration_seconds = int(duration)
        PodcastEpisode.objects.filter(pk=episode.pk, duration_seconds=0).update(duration_seconds=int(duration))

    progress, _ = EpisodeProgress.objects.get_or_create(
        user=request.user,
        episode=episode,
    )
    progress.position_seconds = position
    # Started: no longer new.
    if position > 0:
        progress.dismissed = True
    # Within the last seconds: played (same rule as the apps).
    if counts_as_played(position, episode.duration_seconds):
        progress.played = True

    progress.save()
    return JsonResponse({'ok': True, 'played': progress.played})


@csrf_exempt
@require_http_methods(['POST'])
def mark_played(request):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    try:
        body = json.loads(request.body)
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)

    episode_id = body.get('episode_id')
    played = bool(body.get('played', True))

    try:
        episode = PodcastEpisode.objects.get(pk=episode_id, feed__user=request.user)
    except PodcastEpisode.DoesNotExist:
        return JsonResponse({'error': 'not found'}, status=404)

    progress, _ = EpisodeProgress.objects.get_or_create(
        user=request.user,
        episode=episode,
    )
    progress.played = played
    progress.save(update_fields=['played', 'updated_at'])
    return JsonResponse({'ok': True, 'played': played})


# ---------------------------------------------------------------------------
# Dismiss episodes (hide from inbox without marking played)
# ---------------------------------------------------------------------------

@csrf_exempt
@require_http_methods(['POST'])
def dismiss_episodes(request):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    try:
        body = json.loads(request.body)
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)

    episode_ids = body.get('episode_ids', [])
    dismissed = bool(body.get('dismissed', True))

    if not isinstance(episode_ids, list) or not episode_ids:
        return JsonResponse({'error': 'episode_ids required'}, status=400)

    # Only allow dismissing episodes that belong to this user
    valid_ids = list(
        PodcastEpisode.objects.filter(
            id__in=episode_ids, feed__user=request.user
        ).values_list('id', flat=True)
    )

    for ep_id in valid_ids:
        progress, _ = EpisodeProgress.objects.get_or_create(
            user=request.user,
            episode_id=ep_id,
        )
        progress.dismissed = dismissed
        progress.save(update_fields=['dismissed', 'updated_at'])

    return JsonResponse({'ok': True, 'count': len(valid_ids)})


# ---------------------------------------------------------------------------
# "Folge löschen": out of the queue and "Neu", its downloads gone everywhere
# ---------------------------------------------------------------------------

DISCARDED_DAYS = 30   # how long state/ lists a discard for devices still to delete their copy


def _discarded_data(user, now=None):
    """[{"id": episode id, "at": ISO ms}] of the episodes discarded lately:
    a device deletes its download of one if it finished before `at`."""
    since = (now or timezone.now()) - timedelta(days=DISCARDED_DAYS)
    rows = (EpisodeProgress.objects.filter(user=user, discarded_at__gte=since)
            .order_by('discarded_at').values_list('episode_id', 'discarded_at'))
    return [{'id': ep_id, 'at': _iso_ms(at)} for ep_id, at in rows]


@csrf_exempt
@require_http_methods(['POST'])
def discard_episodes(request):
    """{"episode_ids": [...]}: the episodes leave the queue and "Neu" (like
    dismiss: not played) and every device deletes its download of them (they
    see them under "discarded" in state/). Played, the saved place and the
    star stay as they are; the listening time (updated_at) too. Episodes that
    aren't the user's are left out (count)."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    try:
        body = json.loads(request.body or b'{}')
    except (json.JSONDecodeError, ValueError):
        return JsonResponse({'error': 'invalid JSON'}, status=400)
    if not isinstance(body, dict):
        return JsonResponse({'error': 'invalid JSON'}, status=400)
    episode_ids = body.get('episode_ids')
    if not isinstance(episode_ids, list) or not episode_ids:
        return JsonResponse({'error': 'episode_ids (a list) required'}, status=400)
    if not all(isinstance(i, int) and not isinstance(i, bool) for i in episode_ids):
        return JsonResponse({'error': 'episode_ids must be numbers'}, status=400)
    wanted = set(episode_ids[:EPISODE_IDS_MAX])
    valid = sorted(
        PodcastEpisode.objects.filter(id__in=wanted, feed__user=request.user).values_list('id', flat=True)
    )
    now = timezone.now()
    with transaction.atomic():
        _dequeue(request.user, valid)
        # update() leaves updated_at alone: throwing an episode away isn't listening to it.
        rows = EpisodeProgress.objects.filter(user=request.user, episode_id__in=valid)
        rows.update(dismissed=True, discarded_at=now)
        have = set(rows.values_list('episode_id', flat=True))
        EpisodeProgress.objects.bulk_create(
            [EpisodeProgress(user=request.user, episode_id=i, dismissed=True, discarded_at=now)
             for i in valid if i not in have],
            ignore_conflicts=True,
        )
    return JsonResponse({'ok': True, 'count': len(valid), 'at': _iso_ms(now)})


# ---------------------------------------------------------------------------
# Inbox
# ---------------------------------------------------------------------------

@csrf_exempt
def inbox(request):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)

    limit = min(int(request.GET.get('limit', getattr(settings, 'PODCAST_INBOX_PAGE_SIZE', 200))), 1000)
    offset = max(int(request.GET.get('offset', 0)), 0)
    return JsonResponse({'episodes': _inbox_data(request.user, offset, limit), 'offset': offset, 'limit': limit})


def _inbox_fields(with_description=True):
    fields = ['id', 'title', 'audio_url', 'duration_seconds', 'pub_date', 'artwork_url',
              'feed__id', 'feed__title', 'feed__artwork_url']
    if with_description:
        fields.insert(2, 'description')
    return fields


def _add_user_state(user, episodes):
    """Complete inbox-style episode dicts (_inbox_fields) with the user's
    position and whether they are queued."""
    progress_map = {
        ep['episode_id']: ep
        for ep in EpisodeProgress.objects.filter(
            user=user,
            episode_id__in=[e['id'] for e in episodes],
        ).values('episode_id', 'position_seconds')
    }
    queued_ids = set(
        PodcastQueue.objects.filter(
            user=user,
            episode_id__in=[e['id'] for e in episodes],
        ).values_list('episode_id', flat=True)
    )
    for ep in episodes:
        if ep['pub_date']:
            ep['pub_date'] = ep['pub_date'].isoformat()
        prog = progress_map.get(ep['id'], {})
        ep['position_seconds'] = prog.get('position_seconds', 0)
        ep['in_queue'] = ep['id'] in queued_ids
    return episodes


def _inbox_episodes(user):
    """The inbox: the user's episodes neither played nor dismissed."""
    hidden = EpisodeProgress.objects.filter(user=user).filter(
        Q(played=True) | Q(dismissed=True)
    ).values('episode_id')
    return PodcastEpisode.objects.filter(feed__user=user).exclude(id__in=hidden)


def _inbox_data(user, offset, limit, with_description=True):
    """Inbox episodes. Shownotes can be large (up to 50 KB each); the apps'
    state/ leaves them out and loads one episode's at a time."""
    episodes = list(
        _inbox_episodes(user)
        .select_related('feed')
        .order_by('-pub_date')
        .values(*_inbox_fields(with_description))[offset:offset + limit]
    )
    return _add_user_state(user, episodes)


# ---------------------------------------------------------------------------
# Episodes downloaded in AntennaPod (antennapod.py): offered once for download
# ---------------------------------------------------------------------------

def _import_downloads(user, entries, fields=('id',)):
    """The entries of PodcastSettings.import_downloads still worth offering
    (the episode is still there, the user's, with an audio URL), in their
    order: dicts of `fields` plus 'size' (bytes, 0 = unknown)."""
    sizes = {}
    for e in entries or []:
        if isinstance(e, dict) and isinstance(e.get('episode_id'), int):
            size = e.get('size')
            sizes.setdefault(e['episode_id'], size if isinstance(size, int) and size > 0 else 0)
    ids = list(sizes)
    found = {}
    for i in range(0, len(ids), 500):
        for ep in (PodcastEpisode.objects.filter(id__in=ids[i:i + 500], feed__user=user)
                   .exclude(audio_url='').values(*fields)):
            found[ep['id']] = ep
    return [{**found[ep_id], 'size': sizes[ep_id]} for ep_id in ids if ep_id in found]


@csrf_exempt
@require_http_methods(['GET'])
def import_downloads(request):
    """The episodes that were on the phone in AntennaPod, inbox-style plus
    their size; the app offers once to download them again."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    entries = PodcastSettings.objects.filter(user=request.user).values_list('import_downloads', flat=True).first()
    episodes = _add_user_state(request.user, _import_downloads(request.user, entries, _inbox_fields(False)))
    return JsonResponse({'episodes': episodes, 'bytes': sum(e['size'] for e in episodes)})


@csrf_exempt
@require_http_methods(['POST'])
def clear_import_downloads(request):
    """Offered (downloaded or declined): the list is done with."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    PodcastSettings.objects.filter(user=request.user).update(import_downloads=[])
    return JsonResponse({'ok': True})


# ---------------------------------------------------------------------------
# Player sync (like Spotify Connect): one player state per user. The device
# that starts playing becomes the active one and stays it, also while paused,
# as long as it is online. The others hear it through the stream, stop their
# own playback and show it as playing (or paused) there, with remote controls
# (command/), "play here" (a report of their own) and moving it to another
# device (command transfer). Presence: the stream and every report or command
# keep the device's last_seen fresh.
# ---------------------------------------------------------------------------

PLAYER_ONLINE_SECONDS = 45  # not seen for longer: the device is gone
PLAYER_ACTIONS = ('play', 'pause', 'seek', 'skip', 'next', 'stop', 'transfer', 'sleep')
SLEEP_MAX_MINUTES = 600     # a sleep timer: 1 min to 10 h, or after 1 to SLEEP_MAX_EPISODES
SLEEP_MAX_EPISODES = 99
PLAYER_DEVICES_SHOWN = 10   # online ones, then the last ones seen in DEVICES_DAYS
PLAYER_DEVICES_DAYS = 7
STREAM_SECONDS = 300        # one stream connection; clients reconnect with their seq
STREAM_POLL_SECONDS = 1     # how often the stream looks for a new seq
STREAM_PING_SECONDS = 15    # a comment to keep the connection, and the device's presence
STREAM_PRESENCE_SECONDS = 5  # how often the stream looks for devices that went away


def _iso_ms(dt):
    """2026-10-02T12:00:00.123Z"""
    if dt is None:
        return None
    dt = dt.astimezone(dt_timezone.utc)
    return dt.strftime('%Y-%m-%dT%H:%M:%S.') + f'{dt.microsecond // 1000:03d}Z'


def _finite(value):
    return _number(value) and math.isfinite(value)


def _online_since(now):
    return now - timedelta(seconds=PLAYER_ONLINE_SECONDS)


def _online_ids(user, now):
    return sorted(PlayerDevice.objects.filter(user=user, last_seen__gte=_online_since(now))
                  .values_list('device_id', flat=True))


def _seen(user, device_id, device_name='', kind=None, now=None):
    """The device is here (stream, ping, report, command)."""
    if not device_id:
        return
    now = now or timezone.now()
    fields = {'last_seen': now}
    if device_name:
        fields['device_name'] = device_name[:100]
    if kind in {k for k, _ in PlayerDevice.KINDS if k}:
        fields['kind'] = kind
    # Single statements, no read-then-write transaction (update_or_create):
    # under SQLite two of those at the same time end in "database is locked".
    rows = PlayerDevice.objects.filter(user=user, device_id=device_id[:64])
    if not rows.update(**fields):
        PlayerDevice.objects.bulk_create([PlayerDevice(user=user, device_id=device_id[:64], **fields)],
                                         ignore_conflicts=True)
        rows.update(**fields)   # someone else was quicker


def _check_online(user, now=None):
    """A device came or went: seq + 1, so the others see it. Returns the state."""
    now = now or timezone.now()
    online = _online_ids(user, now)
    p = PlayerState.for_user(user)
    if p.online_devices == online:
        return p
    return _change(p, online_devices=online)


def _change(p, **fields):
    """Write fields of the state and count the change: seq + 1 in the same
    UPDATE, so that it only ever goes up, also with requests at the same
    time. Returns the state as it is now."""
    PlayerState.objects.filter(pk=p.pk).update(seq=F('seq') + 1, **fields)
    p.refresh_from_db()
    return p


def _is_online(user, device_id, now):
    return bool(device_id) and PlayerDevice.objects.filter(
        user=user, device_id=device_id, last_seen__gte=_online_since(now)).exists()


def _devices_json(user, p, now):
    """Online devices first (by name), then the last ones seen lately."""
    since = _online_since(now)
    rows = list(PlayerDevice.objects.filter(user=user, last_seen__gte=now - timedelta(days=PLAYER_DEVICES_DAYS))
                .order_by('-last_seen').values('device_id', 'device_name', 'last_seen'))
    online = sorted((r for r in rows if r['last_seen'] >= since), key=lambda r: (r['device_name'].casefold(), r['device_id']))
    offline = [r for r in rows if r['last_seen'] < since]
    return [
        {'id': r['device_id'], 'name': r['device_name'], 'online': r['last_seen'] >= since,
         'active': r['device_id'] == p.device_id}
        for r in (online + offline)[:PLAYER_DEVICES_SHOWN]
    ]


def _player_json(user, p, now=None):
    now = now or timezone.now()
    episode = None
    if p.kind == PlayerState.PODCAST and p.episode_id:
        rows = list(PodcastEpisode.objects.filter(pk=p.episode_id).values(*_inbox_fields(False)))
        episode = _add_user_state(user, rows)[0] if rows else None
    station = None
    if p.kind == PlayerState.RADIO:
        station = {'url': p.station_url, 'name': p.station_name, 'id': p.station_id}
    devices = _devices_json(user, p, now)
    return {
        'seq': p.seq,
        'kind': p.kind,
        'episode': episode,
        'station': station,
        'position_ms': p.position_ms,
        'playing': p.playing,
        'speed': p.speed,
        'device_id': p.device_id,
        'device_name': p.device_name,
        'updated_at': _iso_ms(p.updated_at),
        'server_time': _iso_ms(now),
        'command': p.command,
        'sleep': {'left_ms': p.sleep_left_ms, 'episodes': p.sleep_episodes},
        'devices': devices,
        'active_online': _is_online(user, p.device_id, now),
    }


def _player_body(request):
    """The JSON body and the sending device, or an error response."""
    try:
        body = json.loads(request.body)
    except (json.JSONDecodeError, ValueError):
        return None, JsonResponse({'error': 'invalid JSON'}, status=400)
    if not isinstance(body, dict):
        return None, JsonResponse({'error': 'invalid JSON'}, status=400)
    device_id = body.get('device_id')
    if not isinstance(device_id, str) or not device_id.strip() or len(device_id) > 64:
        return None, JsonResponse({'error': 'device_id required'}, status=400)
    body['device_name'] = str(body.get('device_name') or '')[:100]
    return body, None


@csrf_exempt
@require_http_methods(['GET'])
def player_state(request):
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    return JsonResponse(_player_json(request.user, PlayerState.for_user(request.user)))


@csrf_exempt
@require_http_methods(['POST'])
def player_report(request):
    """A device says what it plays. Accepted from the active device; from
    another one when playback was just started there (start: by the user, a
    command, a transfer — it becomes the active device), or when the active
    device is gone (offline, or there is none): then the newest report wins.
    Otherwise ignored, such as a periodic "playing" report of a device that
    was just taken over. The answer is the state either way."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    body, error = _player_body(request)
    if error:
        return error

    kind = body.get('kind') or ''
    if kind not in {k for k, _ in PlayerState.KINDS}:
        return JsonResponse({'error': 'unknown kind'}, status=400)
    playing = body.get('playing')
    if not isinstance(playing, bool):
        return JsonResponse({'error': 'playing must be true or false'}, status=400)
    position = body.get('position_ms', 0)
    if not _finite(position):
        return JsonResponse({'error': 'position_ms must be a number'}, status=400)
    speed = body.get('speed')
    if speed is None:
        speed = 1
    if not _finite(speed) or not 0.1 <= speed <= 10:
        return JsonResponse({'error': 'invalid speed'}, status=400)

    episode_id = None
    station = ('', '', None)
    if kind == PlayerState.PODCAST:
        episode_id = body.get('episode_id')
        if not _number(episode_id) or episode_id != int(episode_id) or not \
                PodcastEpisode.objects.filter(pk=int(episode_id), feed__user=request.user).exists():
            return JsonResponse({'error': 'unknown episode'}, status=400)
        episode_id = int(episode_id)
    elif kind == PlayerState.RADIO:
        st = body.get('station')
        if not isinstance(st, dict) or not isinstance(st.get('url'), str) or not st['url'].strip():
            return JsonResponse({'error': 'station.url required'}, status=400)
        sid = st.get('id')
        station = (st['url'].strip()[:1000], str(st.get('name') or '')[:300],
                   int(sid) if _number(sid) and sid == int(sid) else None)

    # The sleep timer, if the device has one (older clients don't say: none).
    sleep = body.get('sleep') if isinstance(body.get('sleep'), dict) else {}
    sleep_left = sleep.get('left_ms', 0)
    sleep_left = min(int(sleep_left), SLEEP_MAX_MINUTES * 60_000) if _finite(sleep_left) and sleep_left > 0 else 0
    sleep_episodes = sleep.get('episodes', 0)
    sleep_episodes = min(int(sleep_episodes), SLEEP_MAX_EPISODES) if _finite(sleep_episodes) and sleep_episodes > 0 else 0

    device_id = body['device_id']
    done = body.get('done_command')
    start = body.get('start') is True
    now = timezone.now()
    _seen(request.user, device_id, body['device_name'], now=now)
    p = PlayerState.for_user(request.user)
    done = bool(done) and bool(p.command) and done == p.command.get('id')
    if device_id == p.device_id or (playing and start) or not _is_online(request.user, p.device_id, now):
        p = _change(
            p,
            kind=kind,
            episode_id=episode_id,
            station_url=station[0], station_name=station[1], station_id=station[2],
            position_ms=max(int(position), 0) if kind == PlayerState.PODCAST else 0,
            playing=playing and bool(kind),
            speed=float(speed),
            sleep_left_ms=sleep_left,
            sleep_episodes=sleep_episodes,
            device_id=device_id,
            device_name=body['device_name'],
            updated_at=now,
            # Done (also a transfer, done in the report that makes its target
            # the active device), or meant for the device before.
            command=None if done or device_id != p.device_id else p.command,
        )
    elif done:
        # A device that isn't the active one did its command (a transfer
        # that couldn't play): done all the same.
        p = _change(p, command=None)
    p = _check_online(request.user, now)
    return JsonResponse(_player_json(request.user, p, now))


@csrf_exempt
@require_http_methods(['POST'])
def player_command(request):
    """Remote control: a command for a device (to_device_id, else the active
    one), which carries it out and reports it done (done_command). Only to a
    device that is online. transfer: the device plays the shared item from
    its place and so becomes the active one. sleep: its sleep timer, value =
    minutes of playing time, -n = after n episodes, 0 = off."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    body, error = _player_body(request)
    if error:
        return error
    action = body.get('action')
    if action not in PLAYER_ACTIONS:
        return JsonResponse({'error': 'unknown action'}, status=400)
    value = None
    if action in ('seek', 'skip'):
        value = body.get('value')
        if not _finite(value):
            return JsonResponse({'error': 'value must be a number'}, status=400)
        value = max(int(value), 0) if action == 'seek' else int(value)
    elif action == 'sleep':
        value = body.get('value')
        if not _finite(value) or not -SLEEP_MAX_EPISODES <= int(value) <= SLEEP_MAX_MINUTES:
            return JsonResponse({'error': f'value must be minutes (up to {SLEEP_MAX_MINUTES}), '
                                          f'-episodes or 0'}, status=400)
        value = int(value)
    to = body.get('to_device_id')
    if to is not None and (not isinstance(to, str) or len(to) > 64):
        return JsonResponse({'error': 'invalid to_device_id'}, status=400)

    now = timezone.now()
    _seen(request.user, body['device_id'], body['device_name'], now=now)
    p = _check_online(request.user, now)
    to = to or p.device_id
    if not to or not _is_online(request.user, to, now):
        return JsonResponse({'error': 'device not online'}, status=409)
    if action == 'transfer' and not p.kind:
        return JsonResponse({'error': 'nothing to transfer'}, status=409)
    p = _change(p, command={
        'id': uuid.uuid4().hex,
        'action': action,
        'value': value,
        'from_device_id': body['device_id'],
        'from_device_name': body['device_name'],
        'to_device_id': to,
        'at': _iso_ms(now),
    })
    return JsonResponse(_player_json(request.user, p, now))


@csrf_exempt
@require_http_methods(['GET'])
def player_stream(request):
    """Server-sent events: the state at once if it is newer than ?since=<seq>,
    then on every change; a ping every 15 s, and the end after a few minutes
    (the client reconnects with the last seq it has). With device_id (and
    device_name, device_kind app|web) the device counts as online meanwhile."""
    if not request.user.is_authenticated:
        return JsonResponse({'error': 'authentication required'}, status=401)
    try:
        since = int(request.GET.get('since', -1))
    except ValueError:
        return JsonResponse({'error': 'since must be a number'}, status=400)
    user = request.user
    device_id = request.GET.get('device_id', '')[:64]
    _seen(user, device_id, request.GET.get('device_name', ''), request.GET.get('device_kind'))
    _check_online(user)

    def events():
        last = since
        first = True
        start = pinged = checked = time.monotonic()
        while True:
            seq = PlayerState.objects.filter(user=user).values_list('seq', flat=True).first()
            if seq is not None and (seq > last if first else seq != last):
                p = PlayerState.objects.filter(user=user).first()
                last = p.seq
                yield f'data: {json.dumps(_player_json(user, p))}\n\n'
            elif first:
                yield ': ping\n\n'   # the headers go out now: the client knows it's connected
            first = False
            now = time.monotonic()
            if now - start >= STREAM_SECONDS:
                return
            if now - pinged >= STREAM_PING_SECONDS:
                pinged = now
                _seen(user, device_id)
                yield ': ping\n\n'
            if now - checked >= STREAM_PRESENCE_SECONDS:
                checked = now
                _check_online(user)
            time.sleep(STREAM_POLL_SECONDS)

    response = StreamingHttpResponse(events(), content_type='text/event-stream')
    response['Cache-Control'] = 'no-cache'
    response['X-Accel-Buffering'] = 'no'
    return response