mrwnslz/diora-web
1
0
Fork 0
Code Issues 4 Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: mrwnslz:561776556f63c6581ee0f94eaa98136ad0fdfdf9
Branches Tags
mrwnslz:master
mrwnslz:wt-tui
mrwnslz:testing
mrwnslz:worktree-book-folders
mrwnslz:worktree-tui-reader
..
compare: mrwnslz:ae0785d5ef5cc5504a2d71cf1712642f6ba543b1
Branches Tags
mrwnslz:master
mrwnslz:wt-tui
mrwnslz:testing
mrwnslz:worktree-book-folders
mrwnslz:worktree-tui-reader

No commits in common. "561776556f63c6581ee0f94eaa98136ad0fdfdf9" and "ae0785d5ef5cc5504a2d71cf1712642f6ba543b1" have entirely different histories.
561776556f ... ae0785d5ef

13 changed files with 50 additions and 1795 deletions
Show stats Expand all files Collapse all files

5
gpodder/views.py
View file

@ -8,7 +8,7 @@ from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods
from podcasts.models import PodcastFeed, PodcastEpisode, EpisodeProgress, PodcastSettings, counts_as_played
from podcasts.models import PodcastFeed, PodcastEpisode, EpisodeProgress, counts_as_played
from podcasts.views import _refresh_feed
from .models import GpodderDevice, GpodderSubscriptionChange, GpodderEpisodeAction
@ -151,13 +151,12 @@ def subscriptions_by_device(request, username, deviceid):
except (json.JSONDecodeError, ValueError):
return JsonResponse({'error': 'invalid JSON'}, status=400)
new_feed = PodcastSettings.new_feed_defaults(request.user)
for url in body.get('add', []):
if not url:
continue
PodcastFeed.objects.get_or_create(
user=request.user, rss_url=url,
defaults={'title': url, **new_feed}
defaults={'title': url}
)
GpodderSubscriptionChange.objects.create(user=request.user, url=url, action='add')

739
podcasts/antennapod.py
View file

@ -1,739 +0,0 @@
"""Import an AntennaPod database backup into one user's podcasts.
AntennaPod's "Datenbank exportieren" writes its whole SQLite database. From
it come the subscriptions with their settings, the state of every episode
that has one (played, new, started, queued, favourite), the queue and which
episodes were on the phone (the app offers to download those again). See
the import_antennapod command for how to run it.
The backup is only read. Episodes without any state (and not on the phone)
are not imported: the feed's current episodes come with its next refresh
anyway. Those that diora already has, though, are taken out of the inbox
(AntennaPod: not new).
"""
import re
import sqlite3
import time
from contextlib import contextmanager
from dataclasses import dataclass, field
from datetime import datetime, timezone as dt_timezone
from itertools import groupby
from pathlib import Path
from urllib.parse import urlsplit
from django.db import transaction
from django.db.models import Q
from .models import EpisodeProgress, PodcastEpisode, PodcastFeed, PodcastQueue, PodcastSettings
from .views import MAX_DESCRIPTION, _keep_sorted, _leave_inbox, _queue_order, _write_queue_order
CHUNK = 500 # ids per IN (...) query and rows per bulk write
# Feeds.feed_volume_adaption -> diora volume factor (0 = off = 1.0)
VOLUME_CODES = {1: 0.5, 2: 0.2, 3: 1.5, 4: 2.0, 5: 2.5}
# Feeds.new_episodes_action; 0 = the global setting (not in the backup)
NEW_EPISODE_ACTIONS = {1: PodcastFeed.NEW_INBOX, 2: PodcastFeed.NEW_IGNORE, 3: PodcastFeed.NEW_QUEUE}
# Feeds.auto_download; 1 = the global setting (not in the backup)
AUTO_DOWNLOAD_OFF, AUTO_DOWNLOAD_GLOBAL, AUTO_DOWNLOAD_ON = 0, 1, 2
# FeedItems.read
NEW, UNPLAYED, PLAYED = -1, 0, 1
SUBSCRIBED = 0 # Feeds.state
# FeedMedia.filesize above this is garbage (seen: 10^13); taken as unknown.
MAX_FILESIZE = 4_000_000_000
SETTING_FIELDS = ('playback_speed', 'skip_intro', 'skip_outro', 'volume', 'include_words',
'exclude_words', 'min_duration', 'new_episodes', 'auto_download')
# Columns read from the backup, with the value to assume if an older
# AntennaPod didn't have the column yet.
FEED_COLUMNS = {
'id': None, 'title': '', 'custom_title': None, 'download_url': '', 'link': '', 'author': '',
'image_url': '', 'keep_updated': 1, 'include_filter': '', 'exclude_filter': '',
'minimal_duration_filter': -1, 'new_episodes_action': 0, 'feed_playback_speed': -1,
'feed_skip_intro': 0, 'feed_skip_ending': 0, 'feed_volume_adaption': 0, 'auto_download': 1,
'state': SUBSCRIBED,
}
class AntennaPodImportError(Exception):
"""The file isn't an AntennaPod backup (or the user doesn't exist)."""
# ---------------------------------------------------------------------------
# Small helpers
# ---------------------------------------------------------------------------
def normalize_url(url):
"""A URL reduced to what identifies it: no scheme (http = https), host in
lower case without a leading www., no default port, no trailing slash, no
fragment. None for anything but http(s)."""
url = (url or '').strip()
try:
parts = urlsplit(url)
port = parts.port
except ValueError:
return None
if parts.scheme.lower() not in ('http', 'https') or not parts.hostname:
return None
host = parts.hostname.lower()
if host.startswith('www.'):
host = host[4:]
if port and port not in (80, 443):
host = f'{host}:{port}'
path = parts.path.rstrip('/')
return host + path + (f'?{parts.query}' if parts.query else '')
def _web_url(url, max_length=1000):
"""url if it is a usable http(s) URL for a URLField, else ''."""
url = (url or '').strip()
if normalize_url(url) is None or len(url) > max_length:
return ''
return url
# AntennaPod's filter syntax: words separated by spaces, "quoted phrases".
_TERM = re.compile(r'([^"]\S*|".+?")\s*')
def parse_filter_terms(raw):
"""AntennaPod filter text -> its terms, e.g. 'a "b c"' -> ['a', 'b c']."""
terms = []
for m in _TERM.finditer(raw or ''):
# diora separates its words by commas, so a comma can't be in one.
term = ' '.join(m.group(1).replace('"', '').replace(',', ' ').split())
if term and term.casefold() not in {t.casefold() for t in terms}:
terms.append(term)
return terms
def filter_words(raw, max_length=500):
"""AntennaPod filter text -> diora's comma-separated words ('' if none);
as many whole terms as fit into the field."""
out = ''
for term in parse_filter_terms(raw):
candidate = f'{out}, {term}' if out else term
if len(candidate) > max_length:
break
out = candidate
return out
def _ms(value):
"""A millisecond timestamp -> aware datetime; None for 0/empty/absurd."""
try:
value = int(value or 0)
except (TypeError, ValueError):
return None
if value <= 0:
return None
try:
return datetime.fromtimestamp(value / 1000, tz=dt_timezone.utc)
except (OverflowError, OSError, ValueError):
return None
def _chunks(seq, size=CHUNK):
seq = list(seq)
for i in range(0, len(seq), size):
yield seq[i:i + size]
@contextmanager
def _given_updated_at():
"""Let bulk_create keep the EpisodeProgress.updated_at we set (auto_now
would put now everywhere, and the history and the statistics go by it)."""
f = EpisodeProgress._meta.get_field('updated_at')
f.auto_now = False
try:
yield
finally:
f.auto_now = True
# ---------------------------------------------------------------------------
# Feed settings
# ---------------------------------------------------------------------------
def feed_settings(ap, global_auto_download=False):
"""diora settings for one AntennaPod feed row: only those AntennaPod sets
to something other than its default. Returns (settings, not_kept_updated)."""
s = {}
speed = ap['feed_playback_speed']
if speed is not None and speed > 0: # -1 = the global speed
s['playback_speed'] = min(max(round(float(speed), 2), 0.5), 3.0)
for src, dst in (('feed_skip_intro', 'skip_intro'), ('feed_skip_ending', 'skip_outro')):
seconds = int(ap[src] or 0)
if seconds > 0:
s[dst] = min(seconds, 600)
volume = VOLUME_CODES.get(ap['feed_volume_adaption'] or 0)
if volume:
s['volume'] = volume
for src, dst in (('include_filter', 'include_words'), ('exclude_filter', 'exclude_words')):
words = filter_words(ap[src])
if words:
s[dst] = words
minutes = int(ap['minimal_duration_filter'] or 0) // 60 # seconds, -1 = none
if minutes > 0:
s['min_duration'] = min(minutes, 600)
action = NEW_EPISODE_ACTIONS.get(ap['new_episodes_action'] or 0)
# "Don't update this podcast": AntennaPod never shows anything new of it.
not_kept_updated = not ap['keep_updated'] and action != PodcastFeed.NEW_QUEUE
if not_kept_updated:
action = PodcastFeed.NEW_IGNORE
if action:
s['new_episodes'] = action
code = ap['auto_download']
if code == AUTO_DOWNLOAD_ON:
s['auto_download'] = True
elif code == AUTO_DOWNLOAD_GLOBAL:
s['auto_download'] = bool(global_auto_download)
elif code == AUTO_DOWNLOAD_OFF:
s['auto_download'] = False
return s, not_kept_updated
# ---------------------------------------------------------------------------
# The import
# ---------------------------------------------------------------------------
@dataclass
class ItemState:
"""What one AntennaPod episode says about its diora episode."""
played: bool
dismissed: bool
position: int # seconds
favorite: bool
activity: datetime | None # last played or finished, if known
when: datetime # what updated_at becomes: activity, else pub date, else the backup date
def merge(self, other):
"""Two AntennaPod episodes (duplicate feeds) for one diora episode:
the one listened to last, a star from either."""
a = self if self.when >= other.when else other
return ItemState(a.played, a.dismissed, a.position, self.favorite or other.favorite, a.activity, a.when)
@dataclass
class Stats:
backup_feeds: int = 0
not_subscribed: int = 0
matched: int = 0
created: int = 0
duplicates: int = 0
created_titles: list = field(default_factory=list)
skipped_feeds: list = field(default_factory=list) # (title, url)
settings: dict = field(default_factory=lambda: {name: 0 for name in SETTING_FIELDS})
not_kept_updated: int = 0
auto_download_on: int = 0
auto_download_explicit: int = 0
auto_download_global: int = 0
auto_download_default: bool = False
auto_download_default_changed: bool = False
stateful: int = 0
by_guid: int = 0
by_url: int = 0
episodes_created: int = 0
no_audio: int = 0
in_skipped_feeds: int = 0
stateless_matched: int = 0
progress_new: int = 0
progress_updated: int = 0
progress_unchanged: int = 0
progress_kept: int = 0
left_inbox: int = 0
favorites_backup: int = 0
favorites_before: int = 0
favorites_after: int = 0
queue_backup: int = 0
queue_missing: int = 0
queue_before: int = 0
queue_after: int = 0
queue_appended: int = 0
keep_sorted: str = ''
inbox_before: int = 0
inbox_after: int = 0
downloads_backup: int = 0
downloads: int = 0
downloads_bytes: int = 0
downloads_created: int = 0
downloads_missing: int = 0
downloads_offered: bool = True
backup_date: datetime | None = None
seconds: float = 0.0
def _inbox_count(user):
hidden = EpisodeProgress.objects.filter(user=user).filter(Q(played=True) | Q(dismissed=True))
return PodcastEpisode.objects.filter(feed__user=user).exclude(id__in=hidden.values('episode_id')).count()
def _open_backup(path):
uri = Path(path).resolve().as_uri() + '?mode=ro'
db = None
try:
db = sqlite3.connect(uri, uri=True)
tables = {r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table'")}
except sqlite3.DatabaseError as e:
if db is not None:
db.close()
raise AntennaPodImportError(f'{path} is not an SQLite database: {e}')
missing = {'Feeds', 'FeedItems', 'FeedMedia', 'Queue', 'Favorites'} - tables
if missing:
db.close()
raise AntennaPodImportError(
f'{path} is not an AntennaPod backup (missing tables: {", ".join(sorted(missing))})')
return db
def _columns(db, table):
return {r[1] for r in db.execute(f'PRAGMA table_info({table})')}
class AntennaPodImport:
def __init__(self, user, path, backup_date, global_auto_download=False, offer_downloads=True):
self.user = user
self.path = path
self.backup_date = backup_date
self.global_auto_download = global_auto_download
# False: the app doesn't offer AntennaPod's downloads (they stay in AntennaPod).
self.offer_downloads = offer_downloads
self.stats = Stats(backup_date=backup_date)
def run(self, dry_run=False):
started = time.monotonic()
db = _open_backup(self.path)
try:
with transaction.atomic():
self._run(db)
if dry_run:
transaction.set_rollback(True)
finally:
db.close()
self.stats.seconds = time.monotonic() - started
return self.stats
def _run(self, db):
st = self.stats
user = self.user
st.queue_before = len(_queue_order(user))
st.inbox_before = _inbox_count(user)
st.favorites_before = EpisodeProgress.objects.filter(user=user, favorite=True).count()
# AntennaPod downloads new episodes of every podcast by default: so
# do diora's future subscriptions.
settings = PodcastSettings.for_user(user)
if self.global_auto_download and not settings.auto_download_default:
settings.auto_download_default = True
settings.save(update_fields=['auto_download_default'])
st.auto_download_default_changed = True
st.auto_download_default = settings.auto_download_default
self.new_feed = PodcastSettings.new_feed_defaults(user)
feed_map = self._import_feeds(db) # AntennaPod feed id -> diora feed id
queue_items = [r[0] for r in db.execute('SELECT feeditem FROM Queue ORDER BY id')]
favorite_items = {r[0] for r in db.execute('SELECT feeditem FROM Favorites')}
st.queue_backup = len(queue_items)
st.favorites_backup = len(favorite_items)
states, not_new, item_episode, downloads = self._import_episodes(
db, feed_map, set(queue_items), favorite_items)
self._import_progress(states, not_new)
self._import_queue(queue_items, item_episode)
self._import_downloads(settings, downloads, item_episode)
st.queue_after = len(_queue_order(user))
st.inbox_after = _inbox_count(user)
st.favorites_after = EpisodeProgress.objects.filter(user=user, favorite=True).count()
# -- feeds --------------------------------------------------------------
def _import_feeds(self, db):
st = self.stats
have = _columns(db, 'Feeds')
select = ', '.join(c if c in have else f'{default!r} AS {c}' if default is not None else f'NULL AS {c}'
for c, default in FEED_COLUMNS.items())
db.row_factory = sqlite3.Row
try:
rows = db.execute(f'SELECT {select} FROM Feeds ORDER BY id').fetchall()
finally:
db.row_factory = None
st.backup_feeds = len(rows)
mine = {}
for f in PodcastFeed.objects.filter(user=self.user).order_by('id'):
mine.setdefault(normalize_url(f.rss_url) or f.rss_url, f)
feed_map = {} # AntennaPod id -> PodcastFeed (saved or still to create)
to_create = {} # normalised URL -> new PodcastFeed
changed = {} # diora id -> PodcastFeed with changed settings
for ap in rows:
if ap['state'] != SUBSCRIBED:
st.not_subscribed += 1
continue
url = (ap['download_url'] or '').strip()
title = (ap['custom_title'] or ap['title'] or url).strip()
key = normalize_url(url)
if key is None or len(url) > 1000:
st.skipped_feeds.append((title, url))
continue
feed = mine.get(key) or to_create.get(key)
if feed is None:
feed = PodcastFeed(
user=self.user, rss_url=url, title=title[:300], author=(ap['author'] or '')[:300],
link=_web_url(ap['link']), artwork_url=_web_url(ap['image_url']), **self.new_feed,
)
to_create[key] = feed
st.created += 1
st.created_titles.append(feed.title)
elif any(f is feed for f in feed_map.values()):
st.duplicates += 1
elif feed.pk:
st.matched += 1
feed_map[ap['id']] = feed
settings, not_kept_updated = feed_settings(ap, self.global_auto_download)
st.not_kept_updated += not_kept_updated
st.auto_download_explicit += ap['auto_download'] == AUTO_DOWNLOAD_ON
st.auto_download_global += ap['auto_download'] == AUTO_DOWNLOAD_GLOBAL
for name, value in settings.items():
if getattr(feed, name) != value:
setattr(feed, name, value)
st.settings[name] += 1
if feed.pk:
changed[feed.pk] = feed
feed.auto_queue = feed.new_episodes == PodcastFeed.NEW_QUEUE
if changed:
PodcastFeed.objects.bulk_update(list(changed.values()), [*SETTING_FIELDS, 'auto_queue'],
batch_size=CHUNK)
if to_create:
# last_refreshed_at stays empty: the next refresh fetches them,
# with the first-fetch rule (only the newest episode is new).
PodcastFeed.objects.bulk_create(list(to_create.values()), batch_size=CHUNK)
ids = {}
for urls in _chunks(f.rss_url for f in to_create.values()):
ids.update(PodcastFeed.objects.filter(user=self.user, rss_url__in=urls).values_list('rss_url', 'id'))
for f in to_create.values():
f.pk = ids[f.rss_url]
st.auto_download_on = sum(1 for f in {id(f): f for f in feed_map.values()}.values() if f.auto_download)
return {ap_id: f.pk for ap_id, f in feed_map.items()}
# -- episodes -----------------------------------------------------------
def _import_episodes(self, db, feed_map, queued, favorites):
"""Match AntennaPod's episodes to diora's, create the missing ones
that have a state (or were downloaded). Returns ({episode id: ItemState}
for those with a state, {episode ids} of matched ones without (not new),
{AntennaPod item id: episode id}, [(pub date ms, AntennaPod item id,
size)] of the downloaded ones)."""
st = self.stats
backup = self.backup_date
states = {}
not_new = set()
item_episode = {}
downloads = []
to_create = {} # (diora feed id, guid) -> [ItemState, AntennaPod item id, audio url, duration, pub_date]
media = _columns(db, 'FeedMedia')
cur = db.execute(
'SELECT i.feed, i.id, i.item_identifier, i.read, i.pubDate, m.download_url, m.duration, '
'm.position, m.last_played_time, m.playback_completion_date, '
+ ('m.downloaded, ' if 'downloaded' in media else '0, ')
+ ('m.filesize ' if 'filesize' in media else '0 ')
+ 'FROM FeedItems i LEFT JOIN FeedMedia m ON m.feeditem = i.id ORDER BY i.feed, i.id'
)
def rows():
while True:
batch = cur.fetchmany(5000)
if not batch:
return
yield from batch
for ap_feed, items in groupby(rows(), key=lambda r: r[0]):
feed_id = feed_map.get(ap_feed)
by_guid = by_url = None
for (_, item_id, ident, read, pub_ms, audio, duration, position, last_played, completed,
downloaded, size) in items:
position = max(int(position or 0), 0) // 1000
activity = max(filter(None, (_ms(last_played), _ms(completed))), default=None)
in_queue = item_id in queued
favorite = item_id in favorites
stateful = read in (NEW, PLAYED) or position > 0 or activity is not None or in_queue or favorite
# FeedMedia.downloaded: when the file was downloaded (ms; 1 from
# older versions), 0 = not on the phone.
downloaded = (downloaded or 0) > 0
if downloaded:
size = int(size or 0)
downloads.append((int(pub_ms or 0), item_id, size if 0 < size <= MAX_FILESIZE else 0))
if feed_id is None:
st.in_skipped_feeds += stateful
continue
if by_guid is None:
by_guid, by_url = {}, {}
for ep_id, guid, url in (PodcastEpisode.objects.filter(feed_id=feed_id)
.order_by('id').values_list('id', 'guid', 'audio_url')):
by_guid.setdefault(guid, ep_id)
by_url.setdefault(normalize_url(url), ep_id)
ident = (ident or '').strip()
audio = (audio or '').strip()
guid = (ident or audio)[:1000]
ep_id = by_guid.get(guid) if guid else None
how = 'guid'
if ep_id is None and audio:
ep_id = by_url.get(normalize_url(audio))
how = 'url'
if not stateful:
if ep_id is not None:
not_new.add(ep_id)
item_episode[item_id] = ep_id
continue
if not downloaded:
continue
# On the phone in AntennaPod but not in diora yet (a feed
# not fetched so far): created as not new, so the app can
# offer to download it again.
else:
st.stateful += 1
played = read == PLAYED
pub_date = _ms(pub_ms)
state = ItemState(
played=played,
# Only AntennaPod's "new" stays in the inbox; queued is never new.
dismissed=read != NEW or in_queue,
position=0 if played else position,
favorite=favorite,
activity=activity,
when=activity or (min(pub_date, backup) if pub_date else backup),
)
if ep_id is not None:
st.by_guid += how == 'guid'
st.by_url += how == 'url'
states[ep_id] = state.merge(states[ep_id]) if ep_id in states else state
item_episode[item_id] = ep_id
continue
if normalize_url(audio) is None or len(audio) > 1000:
st.no_audio += stateful
continue
key = (feed_id, guid)
if key in to_create:
pending = to_create[key]
pending[0] = state.merge(pending[0])
pending[1].append(item_id)
else:
to_create[key] = [state, [item_id], audio, max(int(duration or 0), 0) // 1000, pub_date]
st.downloads_created += not stateful
if to_create:
self._create_episodes(db, to_create, states, item_episode)
not_new -= states.keys()
st.stateless_matched = len(not_new)
return states, not_new, item_episode, downloads
def _create_episodes(self, db, to_create, states, item_episode):
st = self.stats
# A few at a time: shownotes can be large.
for keys in _chunks(to_create, 200):
first_item = {to_create[key][1][0]: key for key in keys}
marks = ','.join('?' * len(first_item))
details = {
first_item[item_id]: (title, description, image)
for item_id, title, description, image in db.execute(
f'SELECT id, title, description, image_url FROM FeedItems WHERE id IN ({marks})',
list(first_item))
}
batch = []
for key in keys:
_, _, audio, duration, pub_date = to_create[key]
title, description, image = details.get(key, ('', '', ''))
batch.append(PodcastEpisode(
feed_id=key[0], guid=key[1], title=(title or 'Untitled')[:500],
description=(description or '')[:MAX_DESCRIPTION], audio_url=audio,
duration_seconds=duration, pub_date=pub_date, artwork_url=_web_url(image),
))
PodcastEpisode.objects.bulk_create(batch)
st.episodes_created = len(to_create) - st.downloads_created
by_feed = {}
for feed_id, guid in to_create:
by_feed.setdefault(feed_id, []).append(guid)
for feed_id, guids in by_feed.items():
for chunk in _chunks(guids):
created = PodcastEpisode.objects.filter(feed_id=feed_id, guid__in=chunk).values_list('guid', 'id')
for guid, ep_id in created:
state, item_ids = to_create[(feed_id, guid)][:2]
states[ep_id] = state
for item_id in item_ids:
item_episode[item_id] = ep_id
# -- progress -----------------------------------------------------------
def _import_progress(self, states, not_new):
st = self.stats
user = self.user
backup = self.backup_date
existing = {}
for ids in _chunks([*states, *not_new]):
for p in EpisodeProgress.objects.filter(user=user, episode_id__in=ids):
existing[p.episode_id] = p
def diora_newer(p, activity):
# Changed in diora after the backup was made (and after the
# episode was last played in AntennaPod): diora's state stays.
return p.updated_at > backup and (activity is None or p.updated_at > activity)
new_rows, changed, favorite_only, dismiss_only = [], [], [], []
for ep_id, s in states.items():
p = existing.get(ep_id)
if p is None:
new_rows.append(EpisodeProgress(
user=user, episode_id=ep_id, position_seconds=s.position, played=s.played,
dismissed=s.dismissed, favorite=s.favorite, updated_at=s.when,
))
st.progress_new += 1
continue
favorite = p.favorite or s.favorite # stars from both
if diora_newer(p, s.activity):
st.progress_kept += 1
if favorite != p.favorite:
favorite_only.append(p.pk)
continue
wanted = (s.position, s.played, s.dismissed, favorite, s.when)
if (p.position_seconds, p.played, p.dismissed, p.favorite, p.updated_at) == wanted:
st.progress_unchanged += 1
continue
p.position_seconds, p.played, p.dismissed, p.favorite, p.updated_at = wanted
changed.append(p)
st.progress_updated += 1
# Known to AntennaPod, but not new there: out of the inbox.
for ep_id in not_new:
p = existing.get(ep_id)
if p is None:
new_rows.append(EpisodeProgress(user=user, episode_id=ep_id, dismissed=True, updated_at=backup))
st.left_inbox += 1
elif not p.played and not p.dismissed and not diora_newer(p, None):
dismiss_only.append(p.pk)
st.left_inbox += 1
with _given_updated_at():
EpisodeProgress.objects.bulk_create(new_rows, batch_size=CHUNK)
# bulk_update writes updated_at as given (no auto_now there).
EpisodeProgress.objects.bulk_update(
changed, ['position_seconds', 'played', 'dismissed', 'favorite', 'updated_at'], batch_size=CHUNK)
# A star or leaving the inbox isn't listening: updated_at stays.
for ids in _chunks(favorite_only):
EpisodeProgress.objects.filter(pk__in=ids).update(favorite=True)
for ids in _chunks(dismiss_only):
EpisodeProgress.objects.filter(pk__in=ids).update(dismissed=True)
# -- queue --------------------------------------------------------------
def _import_queue(self, queue_items, item_episode):
"""AntennaPod's queue first, then what else was in diora's."""
st = self.stats
user = self.user
before = _queue_order(user)
front = []
for item_id in queue_items:
ep_id = item_episode.get(item_id)
if ep_id is None:
st.queue_missing += 1
elif ep_id not in front:
front.append(ep_id)
if not front:
return
in_front = set(front)
rest = [ep_id for ep_id in before if ep_id not in in_front]
st.queue_appended = len(rest)
present = set(before)
PodcastQueue.objects.bulk_create(
[PodcastQueue(user=user, episode_id=ep_id, position=len(before) + i)
for i, ep_id in enumerate(e for e in front if e not in present)],
batch_size=CHUNK,
)
_write_queue_order(user, front + rest)
_leave_inbox(user, front) # queued = not new
if _keep_sorted(user):
st.keep_sorted = PodcastSettings.objects.get(user=user).keep_sorted
# -- downloads ----------------------------------------------------------
def _import_downloads(self, settings, downloads, item_episode):
"""The episodes AntennaPod had on the phone, newest first: the app
can't take its files over, so it offers to download them again.
Replaces what an earlier import left."""
st = self.stats
st.downloads_backup = len(downloads)
entries, seen = [], set()
for _, item_id, size in sorted(downloads, key=lambda d: (-d[0], -d[1])):
ep_id = item_episode.get(item_id)
if ep_id is None:
st.downloads_missing += 1
continue
if ep_id in seen:
continue
seen.add(ep_id)
entries.append({'episode_id': ep_id, 'size': size})
st.downloads = len(entries)
st.downloads_bytes = sum(e['size'] for e in entries)
st.downloads_offered = self.offer_downloads
settings.import_downloads = entries if self.offer_downloads else []
settings.save(update_fields=['import_downloads'])
# ---------------------------------------------------------------------------
# Summary
# ---------------------------------------------------------------------------
SETTING_LABELS = {
'playback_speed': 'speed', 'skip_intro': 'skip intro', 'skip_outro': 'skip outro', 'volume': 'volume',
'include_words': 'include words', 'exclude_words': 'exclude words', 'min_duration': 'min duration',
'new_episodes': 'new episodes', 'auto_download': 'auto-download',
}
def _on(flag):
return 'on' if flag else 'off'
def summary_lines(st, username, path, dry_run, global_auto_download):
when = st.backup_date.strftime('%Y-%m-%d %H:%M UTC') if st.backup_date else '?'
subscribed = st.backup_feeds - st.not_subscribed
settings = ', '.join(f'{SETTING_LABELS[k]} {v}' for k, v in st.settings.items())
lines = [
f"AntennaPod import for user '{username}' from {Path(path).name} (backup date {when})"
+ (' -- DRY RUN, nothing was written' if dry_run else ''),
f'Feeds: {st.backup_feeds} in backup, {subscribed} subscribed ({st.not_subscribed} not subscribed '
f'ignored); {st.matched} matched, {st.created} created, {len(st.skipped_feeds)} skipped (not http/https), '
f'{st.duplicates} duplicates',
f'Settings: changed: {settings}',
f' {st.not_kept_updated} feeds not kept updated in AntennaPod -> new episodes "ignore"',
f'Auto-download: on for {st.auto_download_on} feeds ({st.auto_download_explicit} on in AntennaPod, '
f"{st.auto_download_global} use AntennaPod's global setting -> {_on(global_auto_download)} "
f'by --global-auto-download); default for new subscriptions: {_on(st.auto_download_default)}'
+ (' (set by this import)' if st.auto_download_default_changed else ''),
f'Episodes: {st.stateful} with a state: {st.by_guid} matched by guid, {st.by_url} by URL, '
f'{st.episodes_created} created, {st.no_audio} skipped (no audio URL), {st.in_skipped_feeds} in skipped feeds',
f' {st.stateless_matched} without a state matched to diora episodes (not new in AntennaPod)',
f'Progress: {st.progress_new} new, {st.progress_updated} updated, '
f'{st.progress_unchanged} unchanged, {st.progress_kept} kept (diora newer); '
f'{st.left_inbox} taken out of the inbox (not new in AntennaPod)',
f'Favourites: {st.favorites_backup} in backup; {st.favorites_before} before, {st.favorites_after} after',
f'Queue: {st.queue_backup} in backup ({st.queue_missing} not found); {st.queue_before} before, '
f'{st.queue_after} after ({st.queue_appended} diora entries after AntennaPod\'s)'
+ (f'; kept sorted by "{st.keep_sorted}"' if st.keep_sorted else ''),
f'Inbox: {st.inbox_before} before, {st.inbox_after} after',
f'Downloads: {st.downloads} on the phone in AntennaPod ({st.downloads_bytes / 1_000_000:.0f} MB) '
+ ('-> offered to the app for download' if st.downloads_offered else '-> not offered (--no-downloads)')
+ (f' ({st.downloads_created} episodes created for them)' if st.downloads_created else '')
+ (f'; {st.downloads_missing} not found (feeds not imported, no audio URL)' if st.downloads_missing else ''),
f'Created feeds ({len(st.created_titles)}):',
*(f' - {t}' for t in st.created_titles),
f'Skipped feeds ({len(st.skipped_feeds)}):',
*(f' - {t} ({u})' for t, u in st.skipped_feeds),
f'Done in {st.seconds:.1f} s.',
]
return lines

60
podcasts/management/commands/import_antennapod.py
View file

@ -1,60 +0,0 @@
import os
from datetime import datetime, timezone as dt_timezone
from django.contrib.auth.models import User
from django.core.management.base import BaseCommand, CommandError
from django.utils.dateparse import parse_datetime
from podcasts.antennapod import AntennaPodImport, AntennaPodImportError, summary_lines
class Command(BaseCommand):
help = (
"Import an AntennaPod database backup (Settings > Database > Export) into one user's podcasts: "
'subscriptions with their settings, played/new/started episodes, favourites and the queue; '
'the episodes downloaded in AntennaPod are offered to the app for download. '
'Feeds it creates are fetched by the next refresh; running the import once more after that '
'also takes their older episodes that were not new in AntennaPod out of the inbox. '
'Running it twice changes nothing more.'
)
def add_arguments(self, parser):
parser.add_argument('path', help='AntennaPod backup (.db)')
parser.add_argument('--user', required=True, help='diora username')
parser.add_argument('--dry-run', action='store_true',
help='Do everything inside a transaction that is rolled back, print the summary')
parser.add_argument('--global-auto-download', choices=('on', 'off'), default='off',
help="Auto-download for feeds that use AntennaPod's global setting (default off)")
parser.add_argument('--no-downloads', action='store_true',
help="Don't offer AntennaPod's downloaded episodes to the app (they stay in AntennaPod)")
parser.add_argument('--backup-date',
help='When the backup was made (ISO date/time, UTC if no zone); default: the file date. '
'diora changes after it win over the backup.')
def handle(self, *args, **options):
path = options['path']
if not os.path.isfile(path):
raise CommandError(f'No such file: {path}')
try:
user = User.objects.get(username=options['user'])
except User.DoesNotExist:
raise CommandError(f"No such user: {options['user']}")
if options['backup_date']:
raw = options['backup_date']
backup_date = parse_datetime(raw) or parse_datetime(f'{raw}T00:00')
if backup_date is None:
raise CommandError(f'Invalid --backup-date: {raw}')
if backup_date.tzinfo is None:
backup_date = backup_date.replace(tzinfo=dt_timezone.utc)
else:
backup_date = datetime.fromtimestamp(os.path.getmtime(path), tz=dt_timezone.utc)
global_auto_download = options['global_auto_download'] == 'on'
try:
stats = AntennaPodImport(user, path, backup_date, global_auto_download,
offer_downloads=not options['no_downloads']).run(dry_run=options['dry_run'])
except AntennaPodImportError as e:
raise CommandError(str(e))
for line in summary_lines(stats, user.username, path, options['dry_run'], global_auto_download):
self.stdout.write(line)

7
podcasts/management/commands/refresh_feeds.py
View file

@ -11,11 +11,6 @@ class Command(BaseCommand):
def add_arguments(self, parser):
parser.add_argument('--feed-id', type=int, help='Refresh only this feed ID (even when backing off)')
parser.add_argument('--limit', type=int, default=0, help='Max number of feeds to refresh')
parser.add_argument(
'--full', action='store_true',
help='Download every feed in full (ignore ETag/Last-Modified) and take in episodes skipped before '
'(e.g. video ones); those out before the last refresh go into the list, not into "Neu"',
)
def handle(self, *args, **options):
qs = PodcastFeed.objects.all().order_by('last_refreshed_at')
@ -35,5 +30,5 @@ class Command(BaseCommand):
return
# Parallel downloads, one per distinct feed URL (see _refresh_feeds).
refreshed, failed, new = _refresh_feeds(qs, full=options['full'])
refreshed, failed, new = _refresh_feeds(qs)
self.stdout.write(f'{refreshed} feeds refreshed, {failed} failed, +{new} episodes')

21
podcasts/migrations/0010_feed_auto_download.py
View file

@ -1,21 +0,0 @@
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('podcasts', '0009_episode_transcript_cache'),
]
operations = [
migrations.AddField(
model_name='podcastfeed',
name='auto_download',
field=models.BooleanField(default=False),
),
migrations.AddField(
model_name='podcastsettings',
name='auto_download_default',
field=models.BooleanField(default=False),
),
]

18
podcasts/migrations/0011_settings_import_downloads.py
View file

@ -1,18 +0,0 @@
# Generated by Django 6.1 on 2026-10-02 09:51
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('podcasts', '0010_feed_auto_download'),
]
operations = [
migrations.AddField(
model_name='podcastsettings',
name='import_downloads',
field=models.JSONField(blank=True, default=list),
),
]

15
podcasts/models.py
View file

@ -48,9 +48,6 @@ class PodcastFeed(models.Model):
include_words = models.CharField(max_length=500, blank=True)
exclude_words = models.CharField(max_length=500, blank=True)
min_duration = models.PositiveIntegerField(default=0) # minutes, 0 = any
# Devices download this feed's new episodes (those in the inbox) on
# their own, for listening offline.
auto_download = models.BooleanField(default=False)
class Meta:
unique_together = ('user', 'rss_url')
@ -234,23 +231,11 @@ class PodcastSettings(models.Model):
# front = how auto-queue always worked before this setting existed
auto_queue_position = models.CharField(max_length=20, choices=POSITIONS, default=FRONT)
keep_sorted = models.CharField(max_length=10, choices=SORT_ORDERS, default='', blank=True)
# What PodcastFeed.auto_download starts as for a new subscription.
auto_download_default = models.BooleanField(default=False)
# Episodes that were downloaded in AntennaPod when its backup was
# imported: [{episode_id, size (bytes, 0 = unknown)}]. The app offers once
# to download them again, then clears the list.
import_downloads = models.JSONField(default=list, blank=True)
@classmethod
def for_user(cls, user):
obj, _ = cls.objects.get_or_create(user=user)
return obj
@classmethod
def new_feed_defaults(cls, user):
"""Field values a new PodcastFeed of the user starts with."""
auto_download = cls.objects.filter(user=user).values_list('auto_download_default', flat=True).first()
return {'auto_download': bool(auto_download)}
def __str__(self):
return f'Podcast settings of {self.user}'

633
podcasts/test_antennapod.py
View file

@ -1,633 +0,0 @@
import os
import sqlite3
import tempfile
from datetime import datetime, timedelta, timezone
from io import StringIO
from unittest.mock import patch
from django.contrib.auth.models import User
from django.core.management import call_command
from django.core.management.base import CommandError
from django.test import TestCase
from .antennapod import filter_words, normalize_url, parse_filter_terms
from .models import EpisodeProgress, PodcastEpisode, PodcastFeed, PodcastQueue, PodcastSettings
BACKUP_DATE = datetime(2026, 9, 1, 12, 0, tzinfo=timezone.utc)
T0 = datetime(2026, 8, 1, tzinfo=timezone.utc)
def ms(dt):
return int(dt.timestamp() * 1000)
class Backup:
"""A small AntennaPod database: the tables and columns the import reads."""
SCHEMA = """
CREATE TABLE Feeds (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, custom_title TEXT,
download_url TEXT, link TEXT, author TEXT, image_url TEXT, keep_updated INTEGER DEFAULT 1,
include_filter TEXT DEFAULT '', exclude_filter TEXT DEFAULT '',
minimal_duration_filter INTEGER DEFAULT -1, new_episodes_action INTEGER DEFAULT 0,
feed_playback_speed REAL DEFAULT -1, feed_skip_intro INTEGER DEFAULT 0,
feed_skip_ending INTEGER DEFAULT 0, feed_volume_adaption INTEGER DEFAULT 0,
auto_download INTEGER DEFAULT 0, state INTEGER DEFAULT 0, tags TEXT);
CREATE TABLE FeedItems (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, pubDate INTEGER,
read INTEGER, link TEXT, description TEXT, media INTEGER, feed INTEGER,
item_identifier TEXT, image_url TEXT);
CREATE TABLE FeedMedia (id INTEGER PRIMARY KEY AUTOINCREMENT, duration INTEGER, download_url TEXT,
position INTEGER, playback_completion_date INTEGER, feeditem INTEGER, played_duration INTEGER,
last_played_time INTEGER, downloaded INTEGER DEFAULT 0, filesize INTEGER DEFAULT 0);
CREATE TABLE Queue (id INTEGER PRIMARY KEY AUTOINCREMENT, feeditem INTEGER, feed INTEGER);
CREATE TABLE Favorites (id INTEGER PRIMARY KEY AUTOINCREMENT, feeditem INTEGER, feed INTEGER);
"""
def __init__(self, path):
self.path = path
self.db = sqlite3.connect(path)
self.db.executescript(self.SCHEMA)
def feed(self, url, title='Feed', **cols):
cols = {'download_url': url, 'title': title, **cols}
cur = self.db.execute(f'INSERT INTO Feeds ({",".join(cols)}) VALUES ({",".join("?" * len(cols))})',
list(cols.values()))
self.db.commit()
return cur.lastrowid
def item(self, feed, guid, read=0, audio=None, pub=T0, duration=1800, position=0, completed=None,
last_played=None, title=None, description='', image_url=None, media=True, downloaded=0, size=0):
cur = self.db.execute(
'INSERT INTO FeedItems (title, pubDate, read, description, feed, item_identifier, image_url) '
'VALUES (?,?,?,?,?,?,?)',
(title or f'Episode {guid}', ms(pub) if pub else 0, read, description, feed, guid, image_url))
item_id = cur.lastrowid
if media:
self.db.execute(
'INSERT INTO FeedMedia (duration, download_url, position, playback_completion_date, feeditem, '
'played_duration, last_played_time, downloaded, filesize) VALUES (?,?,?,?,?,?,?,?,?)',
(duration * 1000, audio or f'https://cdn.example.org/{guid}.mp3', position * 1000,
ms(completed) if completed else 0, item_id, 0, ms(last_played) if last_played else 0,
downloaded, size))
self.db.commit()
return item_id
def queue(self, *items):
for i in items:
self.db.execute('INSERT INTO Queue (feeditem, feed) SELECT id, feed FROM FeedItems WHERE id=?', (i,))
self.db.commit()
def favorite(self, *items):
for i in items:
self.db.execute('INSERT INTO Favorites (feeditem, feed) SELECT id, feed FROM FeedItems WHERE id=?', (i,))
self.db.commit()
def close(self):
self.db.close()
class ImportTestCase(TestCase):
def setUp(self):
self.user = User.objects.create_user(username='alice', password='pw12345678')
self.tmp = tempfile.TemporaryDirectory()
self.ap = Backup(os.path.join(self.tmp.name, 'AntennaPodBackup.db'))
def tearDown(self):
self.ap.close()
self.tmp.cleanup()
def run_import(self, *args, user='alice'):
out = StringIO()
call_command('import_antennapod', self.ap.path, '--user', user,
'--backup-date', BACKUP_DATE.isoformat(), *args, stdout=out)
return out.getvalue()
def diora_feed(self, url='https://example.org/feed', **kw):
return PodcastFeed.objects.create(user=self.user, rss_url=url, title=kw.pop('title', 'Diora'), **kw)
def episode(self, feed, guid, audio=None, **kw):
return PodcastEpisode.objects.create(feed=feed, guid=guid, title=guid,
audio_url=audio or f'https://cdn.example.org/{guid}.mp3', **kw)
def progress(self, episode):
return EpisodeProgress.objects.filter(user=self.user, episode=episode).first()
def snapshot(self):
return (
list(PodcastFeed.objects.order_by('id').values()),
list(PodcastEpisode.objects.order_by('id').values()),
list(EpisodeProgress.objects.order_by('id').values()),
list(PodcastQueue.objects.order_by('id').values('id', 'user_id', 'episode_id', 'position')),
)
def queue(self):
return list(PodcastQueue.objects.filter(user=self.user).order_by('position')
.values_list('episode_id', flat=True))
class HelperTests(TestCase):
def test_normalize_url(self):
same = ['https://www.Example.org/feed/', 'http://example.org/feed', 'HTTPS://EXAMPLE.ORG/feed',
'https://example.org:443/feed#x', 'http://example.org:80/feed//']
self.assertEqual({normalize_url(u) for u in same}, {'example.org/feed'})
self.assertNotEqual(normalize_url('https://example.org/Feed'), normalize_url('https://example.org/feed'))
self.assertEqual(normalize_url('https://example.org:8080/f?a=1'), 'example.org:8080/f?a=1')
for bad in ('antennapod_local:content://x', '', None, 'ftp://example.org/f', 'https://'):
self.assertIsNone(normalize_url(bad))
def test_filter_terms(self):
self.assertEqual(parse_filter_terms('news "long talk" Teaser'), ['news', 'long talk', 'Teaser'])
self.assertEqual(parse_filter_terms('"news headlines" '), ['news headlines'])
self.assertEqual(parse_filter_terms(' a b '), ['a', 'b'])
self.assertEqual(parse_filter_terms('"open quote'), ['open', 'quote'])
self.assertEqual(parse_filter_terms('a,b A'), ['a b', 'A'])
self.assertEqual(parse_filter_terms('x X'), ['x'])
self.assertEqual(parse_filter_terms(''), [])
self.assertEqual(filter_words('news "long talk"'), 'news, long talk')
self.assertEqual(len(filter_words(' '.join(['word%03d' % i for i in range(100)]))), 493)
class FeedImportTests(ImportTestCase):
def test_feeds_match_by_normalised_url_and_missing_ones_are_created(self):
mine = self.diora_feed('https://www.Example.org/feed/')
only_diora = self.diora_feed('https://example.org/only-in-diora')
self.ap.feed('http://example.org/feed', title='Same')
self.ap.feed('https://new.example.org/rss', title='Original', custom_title='Renamed', author='Ann',
link='https://new.example.org/', image_url='https://new.example.org/art.jpg')
self.ap.feed('antennapod_local:content://folder', title='Local')
self.ap.feed('https://gone.example.org/rss', title='Not subscribed', state=1)
out = self.run_import()
self.assertEqual(PodcastFeed.objects.filter(user=self.user).count(), 3)
mine.refresh_from_db()
self.assertEqual(mine.title, 'Diora') # matched, not renamed
self.assertTrue(PodcastFeed.objects.filter(pk=only_diora.pk).exists())
new = PodcastFeed.objects.get(user=self.user, rss_url='https://new.example.org/rss')
self.assertEqual((new.title, new.author, new.link, new.artwork_url, new.last_refreshed_at),
('Renamed', 'Ann', 'https://new.example.org/', 'https://new.example.org/art.jpg', None))
self.assertIn('Feeds: 4 in backup, 3 subscribed (1 not subscribed ignored); 1 matched, 1 created, '
'1 skipped (not http/https), 0 duplicates', out)
self.assertIn('Created feeds (1):\n - Renamed\n', out)
self.assertIn('Skipped feeds (1):\n - Local (antennapod_local:content://folder)\n', out)
def test_two_backup_feeds_for_one_url(self):
self.ap.feed('https://example.org/feed')
self.ap.feed('http://www.example.org/feed/')
out = self.run_import()
self.assertEqual(PodcastFeed.objects.filter(user=self.user).count(), 1)
self.assertIn('0 matched, 1 created, 0 skipped (not http/https), 1 duplicates', out)
def test_settings(self):
keep = self.diora_feed('https://example.org/keep', playback_speed=1.5, skip_intro=10, volume=2.0,
new_episodes='queue', include_words='x', min_duration=5, auto_download=True)
changed = self.diora_feed('https://example.org/changed', playback_speed=1.5)
self.ap.feed('https://example.org/keep', auto_download=1) # all AntennaPod defaults
self.ap.feed('https://example.org/changed', feed_playback_speed=1.2000000476837158, feed_skip_intro=30,
feed_skip_ending=700, feed_volume_adaption=2, include_filter='news "long talk" ',
exclude_filter='"Trailer"', minimal_duration_filter=1500, new_episodes_action=2,
auto_download=2)
self.ap.feed('https://example.org/fast', feed_playback_speed=3.5, feed_volume_adaption=5,
minimal_duration_filter=59, new_episodes_action=3)
self.ap.feed('https://example.org/slow', feed_playback_speed=0.25, feed_volume_adaption=3,
new_episodes_action=1)
self.ap.feed('https://example.org/loud', feed_volume_adaption=4)
self.ap.feed('https://example.org/quiet', feed_volume_adaption=1)
out = self.run_import()
keep.refresh_from_db()
self.assertEqual(
(keep.playback_speed, keep.skip_intro, keep.volume, keep.new_episodes, keep.auto_queue,
keep.include_words, keep.min_duration, keep.auto_download),
(1.5, 10, 2.0, 'queue', True, 'x', 5, False), # auto_download "global", --global-auto-download off
)
changed.refresh_from_db()
self.assertEqual(
(changed.playback_speed, changed.skip_intro, changed.skip_outro, changed.volume, changed.include_words,
changed.exclude_words, changed.min_duration, changed.new_episodes, changed.auto_download),
(1.2, 30, 600, 0.2, 'news, long talk', 'Trailer', 25, 'ignore', True),
)
feeds = {f.rss_url.rsplit('/', 1)[1]: f for f in PodcastFeed.objects.filter(user=self.user)}
fast, slow = feeds['fast'], feeds['slow']
self.assertEqual((fast.playback_speed, fast.volume, fast.min_duration, fast.new_episodes, fast.auto_queue),
(3.0, 2.5, 0, 'queue', True))
self.assertEqual((slow.playback_speed, slow.volume, slow.new_episodes, slow.auto_queue),
(0.5, 1.5, 'inbox', False))
self.assertEqual((feeds['loud'].volume, feeds['quiet'].volume), (2.0, 0.5))
self.assertIn('Settings: changed: speed 3, skip intro 1, skip outro 1, volume 5, include words 1, '
'exclude words 1, min duration 1, new episodes 2, auto-download 2', out)
def test_not_kept_updated_feeds_bring_nothing_new(self):
self.ap.feed('https://example.org/a', keep_updated=0)
self.ap.feed('https://example.org/b', keep_updated=0, new_episodes_action=1)
self.ap.feed('https://example.org/c', keep_updated=0, new_episodes_action=3)
self.ap.feed('https://example.org/d', keep_updated=1)
out = self.run_import()
new = dict(PodcastFeed.objects.filter(user=self.user).values_list('rss_url', 'new_episodes'))
self.assertEqual(new, {'https://example.org/a': 'ignore', 'https://example.org/b': 'ignore',
'https://example.org/c': 'queue', 'https://example.org/d': 'inbox'})
self.assertIn('2 feeds not kept updated in AntennaPod -> new episodes "ignore"', out)
def test_auto_download(self):
self.diora_feed('https://example.org/off', auto_download=True)
for name, code in (('off', 0), ('global', 1), ('on', 2), ('global2', 1)):
self.ap.feed(f'https://example.org/{name}', auto_download=code)
def values():
return dict(PodcastFeed.objects.filter(user=self.user).values_list('rss_url', 'auto_download'))
out = self.run_import()
self.assertEqual(values(), {'https://example.org/off': False, 'https://example.org/global': False,
'https://example.org/on': True, 'https://example.org/global2': False})
self.assertIn("Auto-download: on for 1 feeds (1 on in AntennaPod, 2 use AntennaPod's global setting -> off "
'by --global-auto-download); default for new subscriptions: off\n', out)
self.assertFalse(PodcastSettings.for_user(self.user).auto_download_default)
out = self.run_import('--global-auto-download', 'on')
self.assertEqual(values(), {'https://example.org/off': False, 'https://example.org/global': True,
'https://example.org/on': True, 'https://example.org/global2': True})
self.assertIn("Auto-download: on for 3 feeds (1 on in AntennaPod, 2 use AntennaPod's global setting -> on "
'by --global-auto-download); default for new subscriptions: on (set by this import)\n', out)
self.assertTrue(PodcastSettings.for_user(self.user).auto_download_default)
# Once more: nothing to set any more; "off" leaves the default alone.
self.assertIn('default for new subscriptions: on\n', self.run_import('--global-auto-download', 'on'))
self.assertIn('default for new subscriptions: on\n', self.run_import())
def test_created_feeds_start_with_the_users_default(self):
PodcastSettings.objects.create(user=self.user, auto_download_default=True)
self.ap.feed('https://example.org/explicit-off', auto_download=0)
self.ap.feed('https://example.org/on', auto_download=2)
self.run_import()
values = dict(PodcastFeed.objects.filter(user=self.user).values_list('rss_url', 'auto_download'))
self.assertEqual(values, {'https://example.org/explicit-off': False, 'https://example.org/on': True})
class EpisodeImportTests(ImportTestCase):
def setUp(self):
super().setUp()
self.feed = self.diora_feed()
self.ap_feed = self.ap.feed('https://example.org/feed')
def test_matching_and_creating_episodes(self):
by_guid = self.episode(self.feed, 'g1')
by_url = self.episode(self.feed, 'other-guid', audio='http://www.cdn.example.org/g2.mp3')
stateless = self.episode(self.feed, 'g4')
untouched = self.episode(self.feed, 'g6')
self.ap.item(self.ap_feed, 'g1', read=1)
self.ap.item(self.ap_feed, 'g2', read=1, audio='https://cdn.example.org/g2.mp3')
self.ap.item(self.ap_feed, 'g3', read=1, title='Created', description='<p>Notes</p>', duration=3600,
pub=T0 + timedelta(days=3), image_url='https://example.org/g3.jpg')
self.ap.item(self.ap_feed, 'g4', read=0)
self.ap.item(self.ap_feed, 'g5', read=0) # no state, not in diora: not imported
self.ap.item(self.ap_feed, None, read=-1, audio='https://cdn.example.org/noguid.mp3')
self.ap.item(self.ap_feed, 'g7', read=1, media=False)
out = self.run_import()
self.assertTrue(self.progress(by_guid).played)
self.assertTrue(self.progress(by_url).played)
created = PodcastEpisode.objects.get(feed=self.feed, guid='g3')
self.assertEqual(
(created.title, created.description, created.audio_url, created.duration_seconds, created.pub_date,
created.artwork_url),
('Created', '<p>Notes</p>', 'https://cdn.example.org/g3.mp3', 3600, T0 + timedelta(days=3),
'https://example.org/g3.jpg'),
)
self.assertTrue(self.progress(created).played)
# Without a guid, the audio URL is the guid (as when diora reads the feed).
noguid = PodcastEpisode.objects.get(feed=self.feed, guid='https://cdn.example.org/noguid.mp3')
self.assertFalse(self.progress(noguid).dismissed)
self.assertFalse(PodcastEpisode.objects.filter(guid__in=['g5', 'g7']).exists())
# Known to AntennaPod and not new there: out of the inbox, not played.
p = self.progress(stateless)
self.assertEqual((p.dismissed, p.played), (True, False))
self.assertIsNone(self.progress(untouched))
self.assertIn('Episodes: 5 with a state: 1 matched by guid, 1 by URL, 2 created, '
'1 skipped (no audio URL), 0 in skipped feeds', out)
self.assertIn('1 without a state matched to diora episodes', out)
self.assertIn('Inbox: 4 before, 2 after', out) # g6 and noguid
def test_states(self):
played_at = T0 + timedelta(days=10)
self.ap.item(self.ap_feed, 'played', read=1, position=0, completed=played_at, last_played=played_at)
self.ap.item(self.ap_feed, 'marked', read=1, pub=T0 + timedelta(days=2))
self.ap.item(self.ap_feed, 'started', read=0, position=600, last_played=played_at - timedelta(hours=1))
self.ap.item(self.ap_feed, 'new', read=-1, pub=T0 + timedelta(days=5))
self.ap.item(self.ap_feed, 'nodate', read=1, pub=None)
self.ap.item(self.ap_feed, 'future', read=1, pub=BACKUP_DATE + timedelta(days=30))
self.run_import()
def state(guid):
p = self.progress(PodcastEpisode.objects.get(guid=guid))
return p.played, p.dismissed, p.position_seconds, p.updated_at
self.assertEqual(state('played'), (True, True, 0, played_at))
self.assertEqual(state('marked'), (True, True, 0, T0 + timedelta(days=2))) # no activity: pub date
self.assertEqual(state('started'), (False, True, 600, played_at - timedelta(hours=1)))
self.assertEqual(state('new'), (False, False, 0, T0 + timedelta(days=5)))
self.assertEqual(state('nodate'), (True, True, 0, BACKUP_DATE))
self.assertEqual(state('future'), (True, True, 0, BACKUP_DATE))
def test_favourites_are_a_union(self):
diora_star = self.episode(self.feed, 'a')
p = EpisodeProgress.objects.create(user=self.user, episode=diora_star, favorite=True)
EpisodeProgress.objects.filter(pk=p.pk).update(updated_at=BACKUP_DATE - timedelta(days=5))
both = self.episode(self.feed, 'b')
self.ap.item(self.ap_feed, 'a', read=1) # AntennaPod wins, but diora's star stays
b = self.ap.item(self.ap_feed, 'b', read=0)
c = self.ap.item(self.ap_feed, 'c', read=0)
self.ap.favorite(b, c)
out = self.run_import()
p = self.progress(diora_star)
self.assertEqual((p.favorite, p.played), (True, True))
self.assertTrue(self.progress(both).favorite)
self.assertTrue(self.progress(PodcastEpisode.objects.get(guid='c')).favorite)
self.assertIn('Favourites: 2 in backup; 1 before, 3 after', out)
def test_merge_rule(self):
newer = self.episode(self.feed, 'newer')
older = self.episode(self.feed, 'older')
dismissed = self.episode(self.feed, 'dismissed')
after_backup = BACKUP_DATE + timedelta(days=1)
for ep, values in ((newer, dict(position_seconds=100)), (older, dict(position_seconds=100)),
(dismissed, dict(dismissed=True))):
p = EpisodeProgress.objects.create(user=self.user, episode=ep, **values)
EpisodeProgress.objects.filter(pk=p.pk).update(
updated_at=after_backup if ep is not older else BACKUP_DATE - timedelta(days=1))
self.ap.item(self.ap_feed, 'newer', read=1, completed=BACKUP_DATE - timedelta(days=2))
self.ap.item(self.ap_feed, 'older', read=1, completed=BACKUP_DATE - timedelta(days=2))
n = self.ap.item(self.ap_feed, 'dismissed', read=-1)
self.ap.favorite(n)
out = self.run_import()
p = self.progress(newer)
self.assertEqual((p.played, p.position_seconds, p.updated_at), (False, 100, after_backup))
p = self.progress(older)
self.assertEqual((p.played, p.position_seconds, p.updated_at), (True, 0, BACKUP_DATE - timedelta(days=2)))
# diora's newer "not new" stays; the star comes over anyway, without
# touching updated_at.
p = self.progress(dismissed)
self.assertEqual((p.dismissed, p.favorite, p.updated_at), (True, True, after_backup))
self.assertIn('Progress: 0 new, 1 updated, 0 unchanged, 2 kept (diora newer)', out)
def test_activity_after_diora_change_wins(self):
ep = self.episode(self.feed, 'e')
p = EpisodeProgress.objects.create(user=self.user, episode=ep, position_seconds=50)
EpisodeProgress.objects.filter(pk=p.pk).update(updated_at=BACKUP_DATE + timedelta(hours=1))
self.ap.item(self.ap_feed, 'e', read=0, position=300, last_played=BACKUP_DATE + timedelta(hours=2))
self.run_import()
self.assertEqual(self.progress(ep).position_seconds, 300)
def test_queue(self):
x = self.episode(self.feed, 'x')
y = self.episode(self.feed, 'y')
PodcastQueue.objects.create(user=self.user, episode=x, position=0)
PodcastQueue.objects.create(user=self.user, episode=y, position=1)
a = self.ap.item(self.ap_feed, 'a', read=-1)
b = self.ap.item(self.ap_feed, 'b', read=0)
y_ap = self.ap.item(self.ap_feed, 'y', read=0)
local = self.ap.feed('antennapod_local:x')
lost = self.ap.item(local, 'l', read=0)
self.ap.queue(b, y_ap, lost, a)
out = self.run_import()
ids = {e.guid: e.id for e in PodcastEpisode.objects.filter(feed=self.feed)}
self.assertEqual(self.queue(), [ids['b'], y.id, ids['a'], x.id])
self.assertEqual(list(PodcastQueue.objects.filter(user=self.user).order_by('position')
.values_list('position', flat=True)), [0, 1, 2, 3])
# Queued is not new, even if AntennaPod still had it as new.
self.assertTrue(self.progress(PodcastEpisode.objects.get(guid='a')).dismissed)
self.assertIn('Queue: 4 in backup (1 not found); 2 before, 4 after (1 diora entries after', out)
def test_queue_kept_sorted(self):
PodcastSettings.objects.create(user=self.user, keep_sorted='old')
a = self.ap.item(self.ap_feed, 'a', read=0, pub=T0 + timedelta(days=2))
b = self.ap.item(self.ap_feed, 'b', read=0, pub=T0 + timedelta(days=1))
self.ap.queue(a, b)
out = self.run_import()
self.assertEqual([PodcastEpisode.objects.get(pk=i).guid for i in self.queue()], ['b', 'a'])
self.assertIn('kept sorted by "old"', out)
def test_dry_run_writes_nothing(self):
self.episode(self.feed, 'g1')
PodcastQueue.objects.create(user=self.user, episode=self.episode(self.feed, 'q'), position=0)
g = self.ap.item(self.ap_feed, 'g1', read=1)
self.ap.item(self.ap_feed, 'g2', read=-1)
self.ap.feed('https://new.example.org/rss', feed_playback_speed=2.0)
self.ap.queue(g)
before = self.snapshot()
out = self.run_import('--dry-run')
self.assertEqual(self.snapshot(), before)
self.assertIn('DRY RUN', out)
self.assertIn('1 created', out)
self.assertIn('Queue: 1 in backup (0 not found); 1 before, 2 after', out)
self.assertNotIn('DRY RUN', self.run_import())
self.assertNotEqual(self.snapshot(), before)
def test_running_twice_changes_nothing_more(self):
self.episode(self.feed, 'g1')
self.episode(self.feed, 'stateless')
PodcastQueue.objects.create(user=self.user, episode=self.episode(self.feed, 'q'), position=0)
g1 = self.ap.item(self.ap_feed, 'g1', read=0, position=60, last_played=T0)
self.ap.item(self.ap_feed, 'stateless', read=0)
g2 = self.ap.item(self.ap_feed, 'g2', read=-1)
self.ap.item(self.ap_feed, 'g3', read=1)
self.ap.item(self.ap_feed, 'g4', read=1, pub=None)
self.ap.feed('https://new.example.org/rss', feed_playback_speed=2.0, keep_updated=0)
self.ap.queue(g2, g1)
self.ap.favorite(g1)
self.run_import()
first = self.snapshot()
out = self.run_import()
self.assertEqual(self.snapshot(), first)
self.assertIn('2 matched, 0 created', out)
self.assertIn('Settings: changed: speed 0, skip intro 0, skip outro 0, volume 0, include words 0, '
'exclude words 0, min duration 0, new episodes 0, auto-download 0', out)
self.assertIn('0 created, 0 skipped', out)
self.assertIn('Progress: 0 new, 0 updated, 4 unchanged, 0 kept (diora newer); 0 taken out', out)
class DownloadImportTests(ImportTestCase):
"""Episodes on the phone in AntennaPod: offered to the app for download."""
def setUp(self):
super().setUp()
self.feed = self.diora_feed()
self.ap_feed = self.ap.feed('https://example.org/feed')
def offered(self):
return PodcastSettings.for_user(self.user).import_downloads
def test_downloads_are_offered(self):
played = self.episode(self.feed, 'played')
known = self.episode(self.feed, 'known')
self.ap.item(self.ap_feed, 'played', read=1, pub=T0 + timedelta(days=1), downloaded=1, size=30_000_000)
self.ap.item(self.ap_feed, 'known', read=0, pub=T0 + timedelta(days=3), downloaded=ms(T0), size=20_400_000)
self.ap.item(self.ap_feed, 'on-phone', read=0, pub=T0 + timedelta(days=2), downloaded=1,
size=27_634_898_567_168) # garbage: unknown
self.ap.item(self.ap_feed, 'not-downloaded', read=0)
self.ap.item(self.ap_feed, 'local', read=0, audio='antennapod_local:x', downloaded=1, size=5)
self.ap.item(self.ap.feed('antennapod_local:folder'), 'file', read=0, downloaded=1, size=7)
before = self.snapshot()
out = self.run_import('--dry-run')
self.assertEqual(self.snapshot(), before)
self.assertFalse(PodcastSettings.objects.filter(user=self.user).exists())
self.assertIn('Downloads: 3 on the phone in AntennaPod (50 MB) -> offered to the app for download '
'(1 episodes created for them); 2 not found (feeds not imported, no audio URL)\n', out)
out = self.run_import()
created = PodcastEpisode.objects.get(feed=self.feed, guid='on-phone')
self.assertEqual(self.offered(), [ # newest first
{'episode_id': known.id, 'size': 20_400_000}, {'episode_id': created.id, 'size': 0},
{'episode_id': played.id, 'size': 30_000_000}])
# Created only to be downloaded; not new in AntennaPod, so not in the inbox.
p = self.progress(created)
self.assertEqual((p.played, p.dismissed), (False, True))
self.assertFalse(PodcastEpisode.objects.filter(guid__in=['not-downloaded', 'local', 'file']).exists())
self.assertIn('Episodes: 1 with a state: 1 matched by guid, 0 by URL, 0 created, '
'0 skipped (no audio URL), 0 in skipped feeds', out)
# Another backup later: its downloads replace the list.
self.ap.db.execute('UPDATE FeedMedia SET downloaded = 0 WHERE filesize != 30000000')
self.ap.db.commit()
out = self.run_import()
self.assertEqual(self.offered(), [{'episode_id': played.id, 'size': 30_000_000}])
self.assertIn('Downloads: 1 on the phone in AntennaPod (30 MB) -> offered to the app for download\n', out)
self.ap.db.execute('UPDATE FeedMedia SET downloaded = 0')
self.ap.db.commit()
out = self.run_import()
self.assertEqual(self.offered(), [])
self.assertIn('Downloads: 0 on the phone in AntennaPod (0 MB)', out)
def test_no_downloads_offers_nothing(self):
self.episode(self.feed, 'known')
self.ap.item(self.ap_feed, 'known', read=0, downloaded=1, size=20_400_000)
self.run_import()
self.assertEqual(len(self.offered()), 1)
out = self.run_import('--no-downloads')
self.assertEqual(self.offered(), [])
self.assertIn('Downloads: 1 on the phone in AntennaPod (20 MB) -> not offered (--no-downloads)\n', out)
def test_one_entry_per_episode(self):
ep = self.episode(self.feed, 'g')
self.ap.item(self.ap_feed, 'g', read=1, downloaded=1, size=10)
self.ap.item(self.ap.feed('http://www.example.org/feed/'), 'g', read=1, downloaded=1, size=10)
self.run_import()
self.assertEqual(self.offered(), [{'episode_id': ep.id, 'size': 10}])
def test_backup_without_download_columns(self):
self.ap.db.executescript('ALTER TABLE FeedMedia DROP COLUMN downloaded; '
'ALTER TABLE FeedMedia DROP COLUMN filesize;')
self.ap.db.execute('INSERT INTO FeedItems (title, read, feed, item_identifier) VALUES (?,?,?,?)',
('G', 1, self.ap_feed, 'g'))
self.ap.db.commit()
PodcastSettings.objects.create(user=self.user, import_downloads=[{'episode_id': 1, 'size': 1}])
out = self.run_import()
self.assertEqual(self.offered(), [])
self.assertIn('Downloads: 0 on the phone', out)
class ImportDownloadsViewTests(TestCase):
"""GET import-downloads/, its clear/, and the count in state/."""
def setUp(self):
self.user = User.objects.create_user(username='alice', password='pw12345678')
self.client.force_login(self.user)
feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/a', title='A')
self.ep = PodcastEpisode.objects.create(feed=feed, guid='e', title='E', audio_url='https://example.org/e.mp3',
duration_seconds=1800, pub_date=T0)
EpisodeProgress.objects.create(user=self.user, episode=self.ep, position_seconds=120)
no_audio = PodcastEpisode.objects.create(feed=feed, guid='n', title='N', audio_url='')
bob = User.objects.create_user(username='bob', password='pw12345678')
bobs_feed = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/a', title='A')
bobs = PodcastEpisode.objects.create(feed=bobs_feed, guid='b', title='B', audio_url='https://example.org/b.mp3')
gone_feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/gone', title='Gone')
gone = PodcastEpisode.objects.create(feed=gone_feed, guid='g', title='G', audio_url='https://example.org/g.mp3')
gone_feed.delete() # unsubscribed since the import
PodcastSettings.objects.create(user=self.user, import_downloads=[
{'episode_id': i, 'size': 1000 + n} for n, i in enumerate((bobs.id, self.ep.id, no_audio.id, gone.id))])
def test_offered_episodes(self):
data = self.client.get('/podcasts/import-downloads/').json()
self.assertEqual(data['bytes'], 1001)
[e] = data['episodes']
self.assertEqual((e['id'], e['title'], e['feed__title'], e['duration_seconds'], e['pub_date'],
e['position_seconds'], e['in_queue'], e['size']),
(self.ep.id, 'E', 'A', 1800, T0.isoformat(), 120, False, 1001))
# The same keys as an inbox entry of state/, plus the size.
state = self.client.get('/podcasts/state/').json()
self.assertEqual(set(e), set(state['inbox'][0]) | {'size'})
self.assertEqual(state['import_downloads'], {'count': 1, 'bytes': 1001})
def test_clear(self):
self.assertEqual(self.client.get('/podcasts/import-downloads/clear/').status_code, 405)
r = self.client.post('/podcasts/import-downloads/clear/', '{}', content_type='application/json')
self.assertEqual(r.json(), {'ok': True})
self.assertEqual(PodcastSettings.objects.get(user=self.user).import_downloads, [])
self.assertEqual(self.client.get('/podcasts/import-downloads/').json(), {'episodes': [], 'bytes': 0})
self.assertEqual(self.client.get('/podcasts/state/').json()['import_downloads'], {'count': 0, 'bytes': 0})
def test_other_user_and_login(self):
self.client.force_login(User.objects.get(username='bob'))
self.assertEqual(self.client.get('/podcasts/import-downloads/').json(), {'episodes': [], 'bytes': 0})
self.assertEqual(self.client.get('/podcasts/state/').json()['import_downloads'], {'count': 0, 'bytes': 0})
self.client.logout()
self.assertEqual(self.client.get('/podcasts/import-downloads/').status_code, 401)
self.assertEqual(self.client.post('/podcasts/import-downloads/clear/').status_code, 401)
self.assertEqual(len(PodcastSettings.objects.get(user=self.user).import_downloads), 4)
class AfterFirstRefreshTests(ImportTestCase):
"""The way to move: import, let the new feeds be fetched, import again."""
def test_second_import_after_the_first_fetch(self):
import feedparser
from .views import FeedFetch, _refresh_feed
ap_feed = self.ap.feed('https://new.example.org/rss')
days = {'old1': 1, 'old2': 2, 'new': 3, 'newest': 4}
for guid, read in (('old1', 0), ('old2', 1), ('new', -1), ('newest', 0)):
self.ap.item(ap_feed, guid, read=read, pub=T0 + timedelta(days=days[guid]))
self.run_import()
feed = PodcastFeed.objects.get(user=self.user)
self.assertEqual(sorted(feed.episodes.values_list('guid', flat=True)), ['new', 'old2'])
def date(d):
return (T0 + timedelta(days=d)).strftime('%a, %d %b %Y %H:%M:%S GMT')
items = ''.join(
f'<item><guid>{g}</guid><title>{g}</title><pubDate>{date(d)}</pubDate><enclosure url="https://cdn.example.org/{g}.mp3" type="audio/mpeg" length="1"/></item>'
for g, d in days.items())
rss = f'<?xml version="1.0"?><rss version="2.0"><channel><title>New</title>{items}</channel></rss>'
with patch('podcasts.views._fetch_feed', lambda url, **kw: FeedFetch(200, feedparser.parse(rss))):
_refresh_feed(feed)
self.assertEqual(feed.episodes.count(), 4) # nothing twice
def inbox():
hidden = EpisodeProgress.objects.filter(user=self.user, dismissed=True).values('episode_id')
return sorted(feed.episodes.exclude(id__in=hidden).values_list('guid', flat=True))
self.assertEqual(inbox(), ['new', 'newest']) # the first-fetch rule leaves the newest
out = self.run_import()
self.assertEqual(inbox(), ['new'])
self.assertIn('2 without a state matched to diora episodes', out)
self.assertIn('1 taken out of the inbox', out)
class CommandErrorTests(ImportTestCase):
def test_unknown_user_and_bad_files(self):
with self.assertRaisesMessage(CommandError, 'No such user'):
self.run_import(user='nobody')
with self.assertRaisesMessage(CommandError, 'No such file'):
call_command('import_antennapod', os.path.join(self.tmp.name, 'missing.db'), '--user', 'alice')
other = os.path.join(self.tmp.name, 'other.db')
sqlite3.connect(other).close()
with self.assertRaisesMessage(CommandError, 'not an AntennaPod backup'):
call_command('import_antennapod', other, '--user', 'alice')
text = os.path.join(self.tmp.name, 'text.db')
with open(text, 'w') as f:
f.write('not a database at all, just text' * 10)
with self.assertRaisesMessage(CommandError, 'not'):
call_command('import_antennapod', text, '--user', 'alice')
with self.assertRaisesMessage(CommandError, 'Invalid --backup-date'):
call_command('import_antennapod', self.ap.path, '--user', 'alice', '--backup-date', 'gestern')

142
podcasts/tests.py
View file

@ -1,4 +1,3 @@
import base64
import json
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
@ -223,62 +222,6 @@ class FirstFetchTests(TestCase):
self.assertEqual([e['title'] for e in inbox], ['e7', 'e6', 'e5', 'e4'])
class VideoEnclosureTests(TestCase):
"""Video podcasts are taken in (the players play their sound only)."""
def setUp(self):
self.user = User.objects.create_user(username='alice', password='pw12345678')
self.client.force_login(self.user)
self.feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/f', title='Feed')
@staticmethod
def rss(*items):
entries = ''.join(
f'<item><guid>{guid}</guid><title>{guid}</title><pubDate>{date}</pubDate>{enclosures}</item>'
for guid, date, enclosures in items
)
return f'<?xml version="1.0"?><rss version="2.0"><channel><title>Feed</title>{entries}</channel></rss>'
def refresh(self, rss, **kw):
with patch('podcasts.views._fetch_feed', lambda url, **_: _ok(rss)):
return _refresh_feed(self.feed, **kw)
def test_video_only_and_audio_preferred(self):
self.refresh(self.rss(
('v', 'Mon, 05 Jan 2026 10:00:00 GMT', '<enclosure url="https://example.org/v.mp4" type="video/mp4" length="1"/>'),
('both', 'Tue, 06 Jan 2026 10:00:00 GMT',
'<enclosure url="https://example.org/b.mp4" type="video/mp4" length="1"/>'
'<enclosure url="https://example.org/b.m4a" type="audio/mp4" length="1"/>'),
('untyped', 'Wed, 07 Jan 2026 10:00:00 GMT', '<enclosure url="https://example.org/u.m4v?x=1" length="1"/>'),
('none', 'Thu, 08 Jan 2026 10:00:00 GMT', '<enclosure url="https://example.org/n.pdf" type="application/pdf"/>'),
))
urls = dict(PodcastEpisode.objects.filter(feed=self.feed).values_list('guid', 'audio_url'))
self.assertEqual(urls, {
'v': 'https://example.org/v.mp4',
'both': 'https://example.org/b.m4a',
'untyped': 'https://example.org/u.m4v?x=1',
})
def test_full_reread_keeps_old_found_episodes_out_of_the_inbox(self):
audio = '<enclosure url="https://example.org/a.mp3" type="audio/mpeg" length="1"/>'
video = '<enclosure url="https://example.org/{}.mp4" type="video/mp4" length="1"/>'
self.refresh(self.rss(('a', 'Mon, 05 Jan 2026 10:00:00 GMT', audio)))
self.feed.refresh_from_db()
later = (self.feed.last_refreshed_at + timedelta(days=1)).strftime('%a, %d %b %Y %H:%M:%S GMT')
self.refresh(self.rss(
('a', 'Mon, 05 Jan 2026 10:00:00 GMT', audio),
('old-video', 'Tue, 06 Jan 2026 10:00:00 GMT', video.format('o')),
('new-video', later, video.format('n')),
), full=True)
inbox = [e['title'] for e in self.client.get('/podcasts/inbox/').json()['episodes']]
self.assertIn('new-video', inbox)
self.assertNotIn('old-video', inbox)
old = PodcastEpisode.objects.get(feed=self.feed, guid='old-video')
progress = EpisodeProgress.objects.get(user=self.user, episode=old)
self.assertTrue(progress.dismissed)
self.assertFalse(progress.played)
class RefreshAllTests(TestCase):
"""One request refreshes all feeds on the server; one request returns the state."""
@ -327,7 +270,7 @@ class RefreshAllTests(TestCase):
# Shownotes stay out of the one-request state (they can be large).
self.assertNotIn('description', data['inbox'][0])
self.assertEqual(data['settings'], {'enqueue_position': 'back', 'auto_queue_position': 'front',
'keep_sorted': '', 'auto_download_default': False})
'keep_sorted': ''})
# ---------------------------------------------------------------------------
@ -971,7 +914,7 @@ class OpmlExportTests(TestCase):
FEED_DEFAULTS = {
'playback_speed': None, 'skip_intro': 0, 'skip_outro': 0, 'volume': 1.0, 'new_episodes': 'inbox',
'include_words': '', 'exclude_words': '', 'min_duration': 0, 'auto_queue': False, 'auto_download': False,
'include_words': '', 'exclude_words': '', 'min_duration': 0, 'auto_queue': False,
}
@ -997,7 +940,7 @@ class FeedSettingsTests(TestCase):
self.assertEqual(r.json(), {
'feed_id': self.feed.id, 'playback_speed': 1.5, 'skip_intro': 30, 'skip_outro': 600, 'volume': 2.0,
'new_episodes': 'inbox', 'include_words': 'News, Talk', 'exclude_words': 'Teaser', 'min_duration': 10,
'auto_queue': False, 'auto_download': False,
'auto_queue': False,
})
self.feed.refresh_from_db()
self.assertEqual((self.feed.playback_speed, self.feed.skip_intro, self.feed.volume, self.feed.min_duration),
@ -1022,7 +965,6 @@ class FeedSettingsTests(TestCase):
('new_episodes', 'later'), ('new_episodes', None),
('include_words', 'x' * 501), ('include_words', 5), ('exclude_words', ['a']),
('min_duration', -1), ('min_duration', 601),
('auto_download', 1), ('auto_download', 'true'), ('auto_download', None),
]
for name, value in bad:
with self.subTest(name=name, value=value):
@ -1047,16 +989,6 @@ class FeedSettingsTests(TestCase):
self.client.logout()
self.assertEqual(self.post({'skip_intro': 10}).status_code, 401)
def test_auto_download_on_and_off(self):
self.assertIs(self.post({'auto_download': True}).json()['auto_download'], True)
self.feed.refresh_from_db()
self.assertTrue(self.feed.auto_download)
# Other settings leave it alone.
self.assertIs(self.post({'skip_intro': 5}).json()['auto_download'], True)
self.assertIs(self.post({'auto_download': False}).json()['auto_download'], False)
self.feed.refresh_from_db()
self.assertFalse(self.feed.auto_download)
def test_new_episodes_and_auto_queue_stay_in_step(self):
self.assertEqual(self.post({'new_episodes': 'queue'}).json()['auto_queue'], True)
self.feed.refresh_from_db()
@ -1085,12 +1017,12 @@ class FeedSettingsTests(TestCase):
def test_feeds_and_state_carry_the_settings(self):
self.post({'playback_speed': 1.25, 'skip_outro': 20, 'volume': 0.5, 'new_episodes': 'queue',
'exclude_words': 'Trailer', 'min_duration': 5, 'auto_download': True})
'exclude_words': 'Trailer', 'min_duration': 5})
PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/2', title='Zweiter')
expected = {
self.feed.id: {'feed_id': self.feed.id, **FEED_DEFAULTS, 'playback_speed': 1.25, 'skip_outro': 20,
'volume': 0.5, 'new_episodes': 'queue', 'exclude_words': 'Trailer', 'min_duration': 5,
'auto_queue': True, 'auto_download': True},
'auto_queue': True},
}
for url in ('/podcasts/feeds/', '/podcasts/state/'):
with self.subTest(url=url):
@ -2126,67 +2058,3 @@ class TranscriptEndpointTests(TestCase):
def test_login_required(self):
self.client.logout()
self.assertEqual(self.get().status_code, 401)
class AutoDownloadDefaultTests(TestCase):
"""The account-wide default that new subscriptions start with."""
def setUp(self):
self.user = User.objects.create_user(username='alice', password='pw12345678')
self.client.force_login(self.user)
def set_default(self, value):
return self.client.post('/podcasts/settings/', json.dumps({'auto_download_default': value}),
content_type='application/json')
def test_setting(self):
self.assertIs(self.client.get('/podcasts/settings/').json()['auto_download_default'], False)
self.assertIs(self.set_default(True).json()['auto_download_default'], True)
self.assertTrue(PodcastSettings.objects.get(user=self.user).auto_download_default)
self.assertIs(self.client.get('/podcasts/state/').json()['settings']['auto_download_default'], True)
for bad in (1, 'true', None):
with self.subTest(value=bad):
r = self.set_default(bad)
self.assertEqual(r.status_code, 400)
self.assertIn('auto_download_default', r.json()['error'])
self.assertTrue(PodcastSettings.objects.get(user=self.user).auto_download_default)
self.assertIs(self.set_default(False).json()['auto_download_default'], False)
def test_new_subscriptions_start_with_it(self):
from django.core.files.uploadedfile import SimpleUploadedFile
rss = _rss([('e1', 'Mon, 05 Jan 2026 10:00:00 GMT')])
def add(url):
with patch('podcasts.views._fetch_feed', lambda u, **kw: _ok(rss)):
self.client.post('/podcasts/feeds/add/', json.dumps({'rss_url': url}),
content_type='application/json')
def opml(url):
body = f'<opml version="2.0"><body><outline type="rss" text="F" xmlUrl="{url}"/></body></opml>'
self.client.post('/podcasts/feeds/import/',
{'file': SimpleUploadedFile('f.opml', body.encode(), 'text/x-opml')})
def gpodder(url):
auth = 'Basic ' + base64.b64encode(b'alice:pw12345678').decode()
r = self.client.post('/api/2/subscriptions/alice/phone.json', json.dumps({'add': [url]}),
content_type='application/json', HTTP_AUTHORIZATION=auth)
self.assertEqual(r.status_code, 200, r.content)
def auto_download(url):
return PodcastFeed.objects.get(user=self.user, rss_url=url).auto_download
add('https://example.org/a1')
opml('https://example.org/o1')
gpodder('https://example.org/g1')
self.set_default(True)
add('https://example.org/a2')
opml('https://example.org/o2')
gpodder('https://example.org/g2')
self.assertEqual(
[auto_download(f'https://example.org/{n}') for n in ('a1', 'o1', 'g1', 'a2', 'o2', 'g2')],
[False, False, False, True, True, True],
)
# Existing feeds keep theirs.
self.set_default(False)
opml('https://example.org/a2')
self.assertTrue(auto_download('https://example.org/a2'))

2
podcasts/urls.py
View file

@ -28,6 +28,4 @@ urlpatterns = [
path('progress/mark-played/', views.mark_played, name='podcast_mark_played'),
path('progress/dismiss/', views.dismiss_episodes, name='podcast_dismiss_episodes'),
path('inbox/', views.inbox, name='podcast_inbox'),
path('import-downloads/', views.import_downloads, name='podcast_import_downloads'),
path('import-downloads/clear/', views.clear_import_downloads, name='podcast_clear_import_downloads'),
]

173
podcasts/views.py
View file

@ -408,17 +408,10 @@ def _in_backoff_q(now=None):
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, chunk=32, full=False):
def _refresh_feeds(feeds, chunk=32):
"""Refresh many feeds at once: the downloads run in parallel and each
feed URL is fetched once, even when several users subscribed to it; the
database writes then happen one by one. Returns (refreshed, failed, new).
`full`: download every feed again (no 304) and take in episodes the
reader skipped before (see _apply_feed)."""
database writes then happen one by one. Returns (refreshed, failed, new)."""
feeds = list(feeds)
refreshed = failed = new = 0
for start in range(0, len(feeds), chunk):
@ -430,7 +423,7 @@ def _refresh_feeds(feeds, chunk=32, full=False):
def fetch(url):
try:
return url, _fetch_feed(url, **({} if full else _validators(by_url[url]))), None
return url, _fetch_feed(url, **_validators(by_url[url])), None
except Exception as e: # network, HTTP or parser trouble: report and go on
return url, None, e
@ -447,7 +440,7 @@ def _refresh_feeds(feeds, chunk=32, full=False):
_record_error(feed, err)
continue
try:
new += _refresh_feed(feed, fetched, full=full)
new += _refresh_feed(feed, fetched)
refreshed += 1
except Exception as e:
failed += 1
@ -455,7 +448,7 @@ def _refresh_feeds(feeds, chunk=32, full=False):
return refreshed, failed, new
def _refresh_feed(feed_obj, fetched=None, full=False):
def _refresh_feed(feed_obj, fetched=None):
"""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."""
@ -466,7 +459,7 @@ def _refresh_feed(feed_obj, fetched=None, full=False):
_record_error(feed_obj, e)
raise
try:
return _apply_feed(feed_obj, fetched, full=full)
return _apply_feed(feed_obj, fetched)
except Exception as e:
_record_error(feed_obj, e)
raise
@ -486,27 +479,8 @@ def _move_to_permanent_url(feed_obj, 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
return video
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"."""
def _apply_feed(feed_obj, fetched):
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
@ -550,7 +524,13 @@ def _apply_feed(feed_obj, fetched, full=False):
changed_episodes = {}
for entry in entries:
audio_url = _entry_media_url(entry)
# Find audio enclosure
audio_url = ''
for enc in entry.get('enclosures', []):
mime = enc.get('type', '')
if mime.startswith('audio/') or enc.get('url', '').endswith(('.mp3', '.m4a', '.ogg', '.opus', '.aac')):
audio_url = enc.get('url', '')
break
if not audio_url:
continue
@ -618,17 +598,6 @@ def _apply_feed(feed_obj, fetched, full=False):
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).
@ -720,7 +689,7 @@ def feed_list(request):
FEED_SETTING_FIELDS = (
'playback_speed', 'skip_intro', 'skip_outro', 'volume',
'new_episodes', 'include_words', 'exclude_words', 'min_duration', 'auto_queue', 'auto_download',
'new_episodes', 'include_words', 'exclude_words', 'min_duration', 'auto_queue',
)
@ -771,7 +740,7 @@ def add_feed(request):
feed, created = PodcastFeed.objects.get_or_create(
user=request.user,
rss_url=rss_url,
defaults={'title': body.get('title', rss_url)[:300], **PodcastSettings.new_feed_defaults(request.user)},
defaults={'title': body.get('title', rss_url)[:300]},
)
new_episodes = 0
@ -868,12 +837,6 @@ def _new_episodes(value):
return value
def _boolean(value):
if not isinstance(value, bool):
raise ValueError
return value
def _word_list(value):
if value is None:
return ''
@ -891,7 +854,6 @@ FEED_SETTING_PARSERS = {
'include_words': _word_list,
'exclude_words': _word_list,
'min_duration': lambda v: _whole(v, 0, 600),
'auto_download': _boolean,
}
@ -1321,7 +1283,6 @@ def import_opml(request):
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:
@ -1330,7 +1291,7 @@ def import_opml(request):
_, created = PodcastFeed.objects.get_or_create(
user=request.user,
rss_url=rss_url,
defaults={'title': title, **new_feed},
defaults={'title': title},
)
if created:
added += 1
@ -1436,8 +1397,7 @@ def refresh_all_feeds(request):
@csrf_exempt
def podcast_state(request):
"""Everything a client shows first, in one response: feeds, queue, the
first page of the inbox, the settings and how many AntennaPod downloads
wait to be offered (import-downloads/)."""
first page of the inbox and the settings."""
if not request.user.is_authenticated:
return JsonResponse({'error': 'authentication required'}, status=401)
try:
@ -1445,14 +1405,12 @@ def podcast_state(request):
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),
'settings': {'enqueue_position': s.enqueue_position, 'auto_queue_position': s.auto_queue_position,
'keep_sorted': s.keep_sorted, 'auto_download_default': s.auto_download_default},
'import_downloads': {'count': len(downloads), 'bytes': sum(e['size'] for e in downloads)},
'keep_sorted': s.keep_sorted},
'server_time': timezone.now().isoformat(),
})
@ -1593,11 +1551,6 @@ def podcast_settings(request):
return JsonResponse({'error': 'unknown keep_sorted'}, status=400)
s.keep_sorted = body['keep_sorted']
fields.append('keep_sorted')
if 'auto_download_default' in body:
if not isinstance(body['auto_download_default'], bool):
return JsonResponse({'error': 'invalid auto_download_default'}, status=400)
s.auto_download_default = body['auto_download_default']
fields.append('auto_download_default')
if fields:
s.save(update_fields=fields)
if 'keep_sorted' in fields:
@ -1608,7 +1561,6 @@ def podcast_settings(request):
'enqueue_position': s.enqueue_position,
'auto_queue_position': s.auto_queue_position,
'keep_sorted': s.keep_sorted,
'auto_download_default': s.auto_download_default,
'positions': [{'value': k, 'label': v} for k, v in PodcastSettings.POSITIONS],
})
@ -1733,17 +1685,29 @@ def inbox(request):
return JsonResponse({'episodes': _inbox_data(request.user, offset, limit), 'offset': offset, 'limit': limit})
def _inbox_fields(with_description=True):
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."""
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
hidden_ids = set(
EpisodeProgress.objects.filter(
user=user,
).filter(
Q(played=True) | Q(dismissed=True)
).values_list('episode_id', flat=True)
)
episodes = list(
PodcastEpisode.objects.filter(feed__user=user)
.exclude(id__in=hidden_ids)
.select_related('feed')
.order_by('-pub_date')
.values(*fields)[offset:offset + limit]
)
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(
@ -1764,68 +1728,3 @@ def _add_user_state(user, episodes):
ep['position_seconds'] = prog.get('position_seconds', 0)
ep['in_queue'] = ep['id'] in queued_ids
return episodes
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."""
hidden_ids = set(
EpisodeProgress.objects.filter(
user=user,
).filter(
Q(played=True) | Q(dismissed=True)
).values_list('episode_id', flat=True)
)
episodes = list(
PodcastEpisode.objects.filter(feed__user=user)
.exclude(id__in=hidden_ids)
.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})

2
static/js/sw.js
View file

@ -2,7 +2,7 @@
* diora service worker — caches the app shell for offline use.
*/
const CACHE = 'diora-v53';
const CACHE = 'diora-v52';
const SHELL = [
'/static/css/app.css',
'/static/js/app.js',

26
static/podcasts/native-podcast.js
View file

@ -1,4 +1,4 @@
// Copied from diora-android f94c501 2026-10-02: web-native/static/native-podcast.js
// Copied from diora-android 3600b15 2026-10-02: web-native/static/native-podcast.js
// by tools/sync_podcast_ui.py. Do not edit here: change it in diora-android
// and copy again. The browser host is static/js/podcasts-web.js.
@ -262,13 +262,7 @@
playedItem(),
headsetItem(),
]),
DL && group('↓', 'Downloads', 'automatisch, Limit, aufräumen', () => [
// Account-wide (server): new subscriptions start with auto-download on or off.
'auto_download_default' in s && {icon: s.auto_download_default ? '☑' : '☐', label: 'Neue Abos laden von selbst',
hint: s.auto_download_default ? 'an: neue Folgen neuer Abos werden geladen' : 'aus: pro Podcast einschaltbar',
action: () => N.podcastSettings(JSON.stringify({auto_download_default: !s.auto_download_default}))},
...dlSettingsItems(),
]),
DL && group('↓', 'Downloads', 'automatisch, Limit, aufräumen', () => dlSettingsItems()),
group('⇅', 'Abos umziehen', 'OPML exportieren und importieren', () => [
{icon: '⇩', label: 'Abos exportieren (OPML)', hint: 'zum Umziehen in eine andere App', action: exportOpml},
{icon: '⇧', label: 'Abos importieren (OPML)', action: importOpml},
@ -763,15 +757,6 @@
const v = typeof customPrompt === 'function' ? await customPrompt(title, now || '') : null;
if (v != null) save({[key]: v.trim()});
};
// AntennaPod's per-podcast auto-download: new episodes land on the phone by themselves.
if ('auto_download' in live && N.podcastDownloads) {
row.append(chip('Auto-Download', live.auto_download ? 'an' : 'aus', !!live.auto_download, () =>
openSheet(`Neue Folgen von „${plain(live.title)}“ herunterladen?`, [
{icon: live.auto_download ? '●' : '○', label: 'Ja, von selbst', hint: 'im WLAN, innerhalb des Download-Limits',
action: () => save({auto_download: true})},
{icon: !live.auto_download ? '●' : '○', label: 'Nein', hint: 'nur die Warteschlange wird geladen', action: () => save({auto_download: false})},
])));
}
row.append(chip('Filter', words || 'aus', !!words, () => openSheet('Welche neuen Folgen zählen?', [
{icon: '+', label: 'Nur mit diesen Wörtern im Titel', hint: live.include_words || 'alle',
action: () => ask('include_words', 'Nur Folgen, deren Titel eines dieser Wörter enthält (mit Komma getrennt):', live.include_words)},
@ -1743,8 +1728,8 @@
// The limit and the clean-up the AntennaPod way (Downloads.kt, DownloadPlan.toClean).
const DL_LIMITS = [5, 10, 25, 50, 100, 500, 0];
const DL_CLEANUPS = [
{value: 'queue', label: 'Wenn Platz gebraucht wird',
hint: 'beim Limit gehen Folgen, die nicht in der Warteschlange stehen: gehörte zuerst'},
{value: 'queue', label: 'Wenn nicht mehr in der Warteschlange',
hint: 'automatische Downloads, sobald ihre Folge dort fehlt; beim Limit auch gehörte'},
{value: 'never', label: 'Nie', hint: 'beim Limit lädt die App nichts mehr von selbst'},
];
const dlLimitLabel = n => (n ? n + ' Folgen' : 'Unbegrenzt');
@ -1831,9 +1816,6 @@
action: () => openSheet('Höchstens so viele Folgen auf dem Telefon', DL_LIMITS.map(n => ({
icon: s.limit === n ? '●' : '○', label: dlLimitLabel(n), active: s.limit === n,
hint: n === 25 ? 'Standard' : null, action: () => dlSet({limit: n})})))},
'enqueue' in s && {icon: s.enqueue ? '☑' : '☐', label: 'Heruntergeladene Folgen einreihen',
hint: s.enqueue ? 'an: was geladen wird, kommt in die Warteschlange' : 'aus',
action: () => dlSet({enqueue: !s.enqueue})},
s.cleanup && {icon: '♻', label: 'Aufräumen', hint: (DL_CLEANUPS.find(c => c.value === s.cleanup) || DL_CLEANUPS[0]).label,
action: () => openSheet('Downloads aufräumen', DL_CLEANUPS.map(c => ({
icon: s.cleanup === c.value ? '●' : '○', label: c.label, hint: c.hint, active: s.cleanup === c.value,