101 lines
4.7 KiB
Python
101 lines
4.7 KiB
Python
"""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]))
|