"""The search across diora (docs/search.md): one query, what the server knows of every area — podcasts and audiobooks one follows, podcast episodes (the full-text index, podcasts/discover.py), news and blog sources, their articles, saved radio stations. The books aren't here: their titles are encrypted, the page searches them itself (static/js/search.js). Every word of the query has to be there, as the start of a word ("plattf" finds "Plattform"); case and diacritics don't count.""" from django.db.models import Q from django.http import JsonResponse from django.views.decorators.http import require_http_methods from feeds.models import Item, Source from podcasts import discover from podcasts.models import PodcastFeed from radio.models import SavedStation LIMIT = 8 # per area; the area's own view has the rest ARTICLE_SCAN = 400 # articles looked at after the database's rough match def _words(q): return [w for w in discover._words(q) if w][:8] def _matches(words, *texts): """Every word as the start of a word in the texts ("plattf" finds "Plattform").""" have = discover._words(' '.join(t or '' for t in texts)) return bool(words) and all(any(h.startswith(w) for h in have) for w in words) def _rank(words, title): """Title starting with the query first, then title words, then the rest.""" t = discover._words(title) if t[:len(words)] == words or (t and t[0].startswith(words[0])): return 0 return 1 if all(any(h.startswith(w) for h in t) for w in words) else 2 def _feeds(user, words): out = {'podcasts': [], 'audiobooks': []} rows = PodcastFeed.objects.filter(user=user).values('id', 'title', 'author', 'artwork_url', 'kind', 'rss_url', 'favorite', 'new_episodes') found = [f for f in rows if _matches(words, f['title'], f['author'])] found.sort(key=lambda f: (_rank(words, f['title']), (f['title'] or '').lower())) for f in found: key = 'audiobooks' if f['kind'] == PodcastFeed.AUDIOBOOK else 'podcasts' out[key].append(f) return {k: v[:LIMIT] for k, v in out.items()} | {'podcasts_total': len(out['podcasts']), 'audiobooks_total': len(out['audiobooks'])} def _sources(user, words): rows = Source.objects.filter(user=user).values('id', 'title', 'url', 'kind', 'icon_url', 'category') found = [s for s in rows if _matches(words, s['title'], s['url'], s['category'])] found.sort(key=lambda s: (_rank(words, s['title']), (s['title'] or '').lower())) return found[:LIMIT] def _articles(user, words): """Articles whose title or author has every word, the newest first. The database narrows by the longest word (LIKE, as case-blind as SQLite is), Python decides.""" longest = max(words, key=len) rough = Q(title__icontains=longest) | Q(author__icontains=longest) if longest != longest.capitalize(): rough |= Q(title__contains=longest.capitalize()) rows = (Item.objects.filter(user=user, gone=False).filter(rough).order_by('-published') .values('id', 'title', 'author', 'kind', 'published', 'image_url', 'read_at', 'source__title')[:ARTICLE_SCAN]) found = [i for i in rows if _matches(words, i['title'], i['author'])] return [{'id': i['id'], 'title': i['title'], 'author': i['author'], 'kind': i['kind'], 'published': i['published'].isoformat() if i['published'] else None, 'image_url': i['image_url'], 'read': i['read_at'] is not None, 'source': i['source__title']} for i in found[:LIMIT * 2]], len(found) def _stations(user, words): rows = SavedStation.objects.filter(user=user).values('id', 'name', 'url', 'tags', 'country', 'favicon_url', 'is_favorite') found = [s for s in rows if _matches(words, s['name'], s['tags'], s['country'])] found.sort(key=lambda s: (_rank(words, s['name']), not s['is_favorite'], s['name'].lower())) return found[:LIMIT] def search(user, q): words = _words(q) if not words or len(''.join(words)) < 2: return {'q': q, 'empty': True} out = {'q': q} out.update(_feeds(user, words)) ids = discover.search_ids(user, q, limit=LIMIT * 2) out['episodes'] = discover._rows(user, ids) out['sources'] = _sources(user, words) out['articles'], out['articles_total'] = _articles(user, words) out['stations'] = _stations(user, words) return out @require_http_methods(['GET']) def search_view(request): """?q=…: what matches in every area (see the module's text).""" if not request.user.is_authenticated: return JsonResponse({'error': 'authentication required'}, status=401) return JsonResponse(search(request.user, request.GET.get('q', '')[:200]))