"""Stöbern (docs/podcasts.md): finding episodes in the user's own podcasts — search in titles and shownotes, similar episodes, people, topics, forgotten ones — and one controlled step out: where people from them are guests elsewhere (Apple's episode search, fyyd), and podcasts like the most listened ones (fyyd's recommendations). An episode from outside goes onto the Merkliste; its podcast becomes an "Auf Abruf" subscription. The full-text index is SQLite FTS5, contentless (only the index, no second copy of 100 MB of shownotes): podcasts_episode_fts, rowid = the episode's id, filled by index_new() after every refresh (new episodes have higher ids). Episodes deleted since stay in the index; every query joins the episodes, so they never show. """ import html import logging import math import random import re import unicodedata import urllib.parse from collections import Counter, defaultdict from concurrent.futures import ThreadPoolExecutor from datetime import timedelta import requests from django.core.cache import cache from django.db import connection from django.db.models import Q from django.http import JsonResponse from django.utils import timezone from django.utils.dateparse import parse_datetime from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods from . import people as people_mod from .models import EpisodeProgress, PodcastEpisode, PodcastFeed, PodcastSettings log = logging.getLogger(__name__) FTS = 'podcasts_episode_fts' VOCAB = 'podcasts_episode_fts_vocab' BODY_CHARS = 6000 # of the shownotes, per episode INDEX_BATCH = 2000 SIMILAR_TERMS = 10 HTTP_TIMEOUT = 6 USER_AGENT = 'diora podcast discovery' CREATE_SQL = [ f"CREATE VIRTUAL TABLE IF NOT EXISTS {FTS} USING fts5(title, body, content='', contentless_delete=1, " f"tokenize='unicode61 remove_diacritics 2')", f"CREATE VIRTUAL TABLE IF NOT EXISTS {VOCAB} USING fts5vocab({FTS}, 'row')", ] DROP_SQL = [f'DROP TABLE IF EXISTS {VOCAB}', f'DROP TABLE IF EXISTS {FTS}'] def _plain(text): text = re.sub(r'<[^>]+>', ' ', text or '') return re.sub(r'\s+', ' ', html.unescape(text)).strip() def _fold(word): """A word as the index keeps it: lower case, without diacritics.""" w = unicodedata.normalize('NFKD', word.lower()) return ''.join(c for c in w if not unicodedata.combining(c)) _WORD = re.compile(r'\w+', re.UNICODE) def _words(text): return [_fold(w) for w in _WORD.findall(text or '') if not w.isdigit()] def _quoted(term): return '"' + term.replace('"', '""') + '"' # ---- the index ------------------------------------------------------------------------------------ def index_new(limit=None): """Index the episodes newer than the newest one indexed. Returns how many.""" with connection.cursor() as cur: cur.execute(f'SELECT max(rowid) FROM {FTS}') last = cur.fetchone()[0] or 0 done = 0 while True: rows = list(PodcastEpisode.objects.filter(id__gt=last).order_by('id') .values_list('id', 'title', 'description')[:INDEX_BATCH]) if not rows: return done with connection.cursor() as cur: cur.executemany(f'INSERT OR REPLACE INTO {FTS}(rowid, title, body) VALUES (%s, %s, %s)', [(i, t or '', _plain(d)[:BODY_CHARS]) for i, t, d in rows]) done += len(rows) last = rows[-1][0] if limit and done >= limit: return done def _user_episode_sql(user, stream=False, other_than_feed=None): """SQL and params restricting fts rowids to the user's podcast episodes.""" sql = (f' JOIN podcasts_podcastepisode e ON e.id = {FTS}.rowid' f' JOIN podcasts_podcastfeed pf ON pf.id = e.feed_id' f' WHERE pf.user_id = %s AND pf.kind = %s') params = [user.id, PodcastFeed.PODCAST] if stream: sql += ' AND pf.new_episodes != %s' params.append(PodcastFeed.NEW_IGNORE) if other_than_feed is not None: sql += ' AND e.feed_id != %s' params.append(other_than_feed) return sql, params def search_ids(user, q, stream=False, limit=50, offset=0): """Episode ids of the user's podcasts matching all words of q (the last one as a prefix, while typing), titles weighing three times the shownotes, best first.""" words = _words(q)[:12] if not words: return [] match = ' '.join(_quoted(w) for w in words[:-1]) + (' ' if len(words) > 1 else '') + _quoted(words[-1]) + '*' where, params = _user_episode_sql(user, stream) with connection.cursor() as cur: cur.execute(f'SELECT {FTS}.rowid FROM {FTS}{where} AND {FTS} MATCH %s ' f'ORDER BY bm25({FTS}, 3.0, 1.0) LIMIT %s OFFSET %s', params + [match, limit, offset]) return [r[0] for r in cur.fetchall()] def similar_ids(user, episode, limit=20): """Episodes of the user's other podcasts most like `episode`: its rarest telling words (tf·idf over the index) asked for together, best match first.""" tf = Counter() for w in _words(episode.title): tf[w] += 3 for w in _words(_plain(episode.description)[:BODY_CHARS]): tf[w] += 1 terms = [w for w in tf if len(w) > 2] if not terms: return [] with connection.cursor() as cur: cur.execute(f'SELECT count(*) FROM podcasts_podcastepisode') n = cur.fetchone()[0] or 1 df = {} for i in range(0, len(terms), 500): chunk = terms[i:i + 500] cur.execute(f'SELECT term, doc FROM {VOCAB} WHERE term IN ({",".join(["%s"] * len(chunk))})', chunk) df.update(dict(cur.fetchall())) scored = [] for w, c in tf.items(): d = df.get(w, 0) if d < 2 or d > max(n * 0.05, 10): # only here, or everywhere: says nothing continue scored.append((c * math.log(n / d), w)) scored.sort(reverse=True) best = [w for _, w in scored[:SIMILAR_TERMS]] if not best: return [] where, params = _user_episode_sql(user, other_than_feed=episode.feed_id) with connection.cursor() as cur: cur.execute(f'SELECT {FTS}.rowid FROM {FTS}{where} AND {FTS} MATCH %s ' f'ORDER BY bm25({FTS}, 3.0, 1.0) LIMIT %s', params + [' OR '.join(_quoted(w) for w in best), limit]) return [r[0] for r in cur.fetchall()] # ---- helpers --------------------------------------------------------------------------------------- def _rows(user, ids): """The episodes as the episode lists give them, in the order of ids.""" from .views import _episode_list_data if not ids: return [] eps, _ = _episode_list_data(user, ids=ids, limit=len(ids)) by_id = {e['id']: e for e in eps} return [by_id[i] for i in ids if i in by_id] def _heard_by_feed(user): """{feed_id: episodes heard in the last 12 months} (as heard_365 in state/).""" from .views import HEARD_MIN_SECONDS since = timezone.now() - timedelta(days=365) out = Counter() for fid in (EpisodeProgress.objects.filter(user=user, updated_at__gte=since) .filter(Q(played=True) | Q(position_seconds__gt=HEARD_MIN_SECONDS)) .values_list('episode__feed_id', flat=True)): out[fid] += 1 return out def _norm_url(url): u = (url or '').strip().lower() u = re.sub(r'^https?://', '', u) return u.rstrip('/') def _get_json(url, **params): try: r = requests.get(url, params=params, timeout=HTTP_TIMEOUT, headers={'User-Agent': USER_AGENT}) r.raise_for_status() return r.json() except Exception as e: # noqa: BLE001 - a service away is no error for the user log.info('discover: %s: %s', url, e) return None # ---- people ---------------------------------------------------------------------------------------- PEOPLE_TTL = 12 * 3600 def people_of(user): key = f'podcasts:people:{user.id}' got = cache.get(key) if got is None: rows = list(PodcastEpisode.objects.filter(feed__user=user, feed__kind=PodcastFeed.PODCAST) .values_list('id', 'feed_id', 'title')) got = people_mod.people(rows, _heard_by_feed(user))[:200] cache.set(key, got, PEOPLE_TTL) return got def person_ids(user, name, limit=200): return list(PodcastEpisode.objects.filter(feed__user=user, feed__kind=PodcastFeed.PODCAST, title__icontains=name) .order_by('-pub_date', '-id').values_list('id', flat=True)[:limit]) def elsewhere(user, name): """Episodes with `name` in podcasts the user doesn't have: Apple's episode search and fyyd's, those that name the person in their title. [{source, podcast, feed_url, title, audio_url, guid, pub_date, duration_seconds, artwork_url, link}]""" key = f'podcasts:elsewhere:{user.id}:' + re.sub(r'\W+', '_', _fold(name)) got = cache.get(key) if got is not None: return got mine = {_norm_url(u) for u in PodcastFeed.objects.filter(user=user).values_list('rss_url', flat=True)} mine_titles = {t.casefold() for t in PodcastFeed.objects.filter(user=user).values_list('title', flat=True)} folded = _fold(name) out, seen = [], set() def take(item): if folded not in _fold(item['title']) or not item['feed_url'] or not item['audio_url']: return if _norm_url(item['feed_url']) in mine or item['podcast'].casefold() in mine_titles: return k = (_fold(item['title']), _fold(item['podcast'])) if k in seen: return seen.add(k) out.append(item) apple = _get_json('https://itunes.apple.com/search', term=name, media='podcast', entity='podcastEpisode', limit=50, country='DE') or {} for r in apple.get('results', []): take({'source': 'apple', 'podcast': r.get('collectionName', ''), 'feed_url': r.get('feedUrl', ''), 'title': r.get('trackName', ''), 'audio_url': r.get('episodeUrl', ''), 'guid': r.get('episodeGuid', ''), 'pub_date': r.get('releaseDate'), 'duration_seconds': int((r.get('trackTimeMillis') or 0) / 1000), 'artwork_url': r.get('artworkUrl600') or r.get('artworkUrl160') or '', 'link': r.get('trackViewUrl', '')}) fy = _get_json('https://api.fyyd.de/0.2/search/episode', term=f'"{name}"', count=50) or {} eps = [e for e in (fy.get('data') or []) if isinstance(e, dict) and folded in _fold(e.get('title', ''))] feeds = {} for pid in list(dict.fromkeys(e.get('podcast_id') for e in eps))[:12]: p = _get_json('https://api.fyyd.de/0.2/podcast', podcast_id=pid) or {} if isinstance(p.get('data'), dict): feeds[pid] = p['data'] for e in eps: p = feeds.get(e.get('podcast_id')) if not p: continue take({'source': 'fyyd', 'podcast': p.get('title', ''), 'feed_url': p.get('xmlURL', ''), 'title': html.unescape(e.get('title', '')), 'audio_url': e.get('enclosure', ''), 'guid': e.get('guid', ''), 'pub_date': e.get('pubdate'), 'duration_seconds': int(e.get('duration') or 0), 'artwork_url': e.get('imgURL') or p.get('imgURL') or '', 'link': e.get('url_fyyd', '')}) out.sort(key=lambda i: i['pub_date'] or '', reverse=True) cache.set(key, out, 6 * 3600) return out # ---- topics, forgotten, recommendations ------------------------------------------------------------ def topics_of(user, days=7): """Topics of the Strom's last days (feeds.textproc.topics, as for the news): [{'label', 'ids'}], the largest first.""" from feeds import textproc since = timezone.now() - timedelta(days=days) eps = list(PodcastEpisode.objects.filter(feed__user=user, feed__kind=PodcastFeed.PODCAST, pub_date__gte=since) .exclude(feed__new_episodes=PodcastFeed.NEW_IGNORE) .values_list('id', 'title', 'description', 'feed__title')) stories = [(i, f'{t}. {_plain(d)[:400]}', {ft}) for i, t, d, ft in eps] return textproc.topics(stories, max_topics=10) FORGOTTEN_DAYS = 60 def forgotten(user): """Unheard older episodes of the podcasts the user listens to most, and of the rare ones — a handful each, the same all day.""" from .views import _intervals rnd = random.Random(f'{user.id}:{timezone.localdate()}') heard = _heard_by_feed(user) started = set(EpisodeProgress.objects.filter(user=user).filter(Q(played=True) | Q(position_seconds__gt=0)) .values_list('episode_id', flat=True)) before = timezone.now() - timedelta(days=FORGOTTEN_DAYS) def pick(feed_ids, n): ids = [i for i in PodcastEpisode.objects.filter(feed_id__in=feed_ids, pub_date__lt=before) .values_list('id', flat=True) if i not in started] return rnd.sample(ids, min(n, len(ids))) mine = list(PodcastFeed.objects.filter(user=user, kind=PodcastFeed.PODCAST) .exclude(new_episodes=PodcastFeed.NEW_IGNORE).values_list('id', flat=True)) loved = [f for f in mine if heard.get(f, 0) >= 3] intervals = _intervals(user) rare = [f for f in mine if (intervals.get(f) or 0) >= 30] return {'loved': _rows(user, pick(loved, 15)), 'rare': _rows(user, pick(rare, 10))} RECOMMEND_FROM = 10 # the most listened podcasts asked about RECOMMEND_TTL = 24 * 3600 def _fyyd_id(feed): got = _get_json('https://api.fyyd.de/0.2/search/podcast', title=feed['title'], count=5) or {} want = _norm_url(feed['rss_url']) for p in got.get('data') or []: if isinstance(p, dict) and (_norm_url(p.get('xmlURL')) == want or (p.get('title') or '').casefold() == feed['title'].casefold()): return p.get('id') return None def recommend(user): """Podcasts like the user's most listened (fyyd's recommendations), the ones recommended for several of them first; none the user has. [{title, feed_url, artwork_url, description, link, because: [titles], count}]""" key = f'podcasts:recommend:{user.id}' got = cache.get(key) if got is not None: return got heard = _heard_by_feed(user) feeds = list(PodcastFeed.objects.filter(user=user, kind=PodcastFeed.PODCAST).values('id', 'title', 'rss_url')) top = sorted((f for f in feeds if heard.get(f['id'])), key=lambda f: -heard[f['id']])[:RECOMMEND_FROM] mine = {_norm_url(f['rss_url']) for f in feeds} mine_titles = {f['title'].casefold() for f in feeds} def ask(feed): pid = _fyyd_id(feed) if not pid: return feed, [] r = _get_json('https://api.fyyd.de/0.2/podcast/recommend', podcast_id=pid, count=12) or {} return feed, [p for p in r.get('data') or [] if isinstance(p, dict)] found = {} with ThreadPoolExecutor(max_workers=5) as pool: for feed, recs in pool.map(ask, top): for p in recs: url = p.get('xmlURL') or '' if not url or _norm_url(url) in mine or (p.get('title') or '').casefold() in mine_titles: continue k = _norm_url(url) item = found.setdefault(k, {'title': p.get('title', ''), 'feed_url': url, 'artwork_url': p.get('imgURL') or '', 'description': _plain(p.get('description'))[:300], 'link': p.get('htmlURL') or '', 'because': [], 'count': 0}) if feed['title'] not in item['because']: item['because'].append(feed['title']) item['count'] += 1 out = sorted(found.values(), key=lambda i: (-i['count'], i['title'].casefold()))[:30] cache.set(key, out, RECOMMEND_TTL) return out # ---- an episode from outside ------------------------------------------------------------------------- def take(user, item): """An episode from elsewhere onto the Merkliste: its podcast becomes an "Auf Abruf" subscription (or is one already), the episode is found in it — or, when the feed no longer lists it, made from what the search said. Returns (episode, created_feed).""" from .views import _later_add, _refresh_feed url = (item.get('feed_url') or '').strip() feed = PodcastFeed.objects.filter(user=user, rss_url=url).first() created = False if feed is None: feed = PodcastFeed.objects.create( user=user, rss_url=url, title=(item.get('podcast') or url)[:300], artwork_url=(item.get('artwork_url') or '')[:1000], **{**PodcastSettings.new_feed_defaults(user), 'new_episodes': PodcastFeed.NEW_IGNORE}) created = True try: _refresh_feed(feed) except Exception as e: # noqa: BLE001 - the episode is made from the search below log.info('discover: %s: %s', url, e) eps = feed.episodes.all() ep = None if item.get('guid'): ep = eps.filter(guid=item['guid']).first() if ep is None and item.get('audio_url'): ep = eps.filter(audio_url=item['audio_url']).first() if ep is None and item.get('title'): ep = eps.filter(title__iexact=item['title']).first() if ep is None: when = parse_datetime(item.get('pub_date') or '') if item.get('pub_date') else None ep = PodcastEpisode.objects.create( feed=feed, guid=(item.get('guid') or item.get('audio_url') or item.get('title'))[:1000], title=(item.get('title') or '')[:500], audio_url=item.get('audio_url', '')[:1000], duration_seconds=int(item.get('duration_seconds') or 0), pub_date=when, artwork_url=(item.get('artwork_url') or '')[:1000]) _later_add(user, [ep.id]) try: index_new() except Exception: # noqa: BLE001 pass return ep, created # ---- views --------------------------------------------------------------------------------------------- def _auth(request): if not request.user.is_authenticated: return JsonResponse({'error': 'authentication required'}, status=401) return None def _int(request, name, default, low, high): try: return min(max(int(request.GET.get(name, default)), low), high) except ValueError: return default @require_http_methods(['GET']) def search_view(request): """?q=…&stream=1&offset=&limit=: episodes whose title or shownotes have every word.""" err = _auth(request) if err: return err ids = search_ids(request.user, request.GET.get('q', ''), stream=request.GET.get('stream') == '1', limit=_int(request, 'limit', 50, 1, 200), offset=_int(request, 'offset', 0, 0, 10 ** 6)) return JsonResponse({'episodes': _rows(request.user, ids)}) @require_http_methods(['GET']) def similar_view(request, pk): err = _auth(request) if err: return err ep = PodcastEpisode.objects.filter(pk=pk, feed__user=request.user).first() if ep is None: return JsonResponse({'error': 'not found'}, status=404) return JsonResponse({'episodes': _rows(request.user, similar_ids(request.user, ep, _int(request, 'limit', 20, 1, 50)))}) @require_http_methods(['GET']) def people_view(request): """The people in the user's podcasts: [{name, episodes, feeds, score}], the most relevant first.""" err = _auth(request) if err: return err return JsonResponse({'people': people_of(request.user)[:_int(request, 'limit', 80, 1, 200)]}) @require_http_methods(['GET']) def person_view(request): """?name=: the person's episodes in the user's podcasts, newest first.""" err = _auth(request) if err: return err name = request.GET.get('name', '').strip() if len(name) < 3: return JsonResponse({'error': 'name required'}, status=400) return JsonResponse({'name': name, 'episodes': _rows(request.user, person_ids(request.user, name))}) @require_http_methods(['GET']) def elsewhere_view(request): """?name=: episodes with the person in podcasts the user doesn't have.""" err = _auth(request) if err: return err name = request.GET.get('name', '').strip() if len(name) < 3: return JsonResponse({'error': 'name required'}, status=400) return JsonResponse({'name': name, 'episodes': elsewhere(request.user, name)}) @require_http_methods(['GET']) def topics_view(request): """?days=7: topics of the Strom's last days: [{label, ids}], plus their episodes.""" err = _auth(request) if err: return err days = _int(request, 'days', 7, 1, 30) key = f'podcasts:topics:{request.user.id}:{days}' got = cache.get(key) if got is None: got = topics_of(request.user, days) cache.set(key, got, 3600) return JsonResponse({'topics': got}) @require_http_methods(['GET']) def forgotten_view(request): err = _auth(request) if err: return err return JsonResponse(forgotten(request.user)) @require_http_methods(['GET']) def recommend_view(request): err = _auth(request) if err: return err return JsonResponse({'podcasts': recommend(request.user)}) @csrf_exempt @require_http_methods(['POST']) def take_view(request): """{podcast, feed_url, title, audio_url, guid, pub_date, duration_seconds, artwork_url} (an item of elsewhere/): onto the Merkliste; returns {episode_id, feed_id, subscribed}.""" import json err = _auth(request) if err: return err try: item = json.loads(request.body or b'{}') except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'invalid JSON'}, status=400) if not isinstance(item, dict) or not re.match(r'^https?://', item.get('feed_url') or '') or not item.get('audio_url'): return JsonResponse({'error': 'feed_url and audio_url required'}, status=400) ep, created = take(request.user, item) return JsonResponse({'ok': True, 'episode_id': ep.id, 'feed_id': ep.feed_id, 'subscribed': created})