import base64 import json from django.conf import settings from django.contrib import messages from django.contrib.auth import authenticate, login, get_user_model, update_session_auth_hash from django.contrib.auth.decorators import login_required from django.contrib.auth.forms import UserCreationForm, AuthenticationForm, PasswordChangeForm from django.http import JsonResponse from django.shortcuts import render, redirect from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods from radio import lastfm as lastfm_module from .models import ApiToken, WebDAVSource from .webdav import WebDAVError, list_directory User = get_user_model() # --------------------------------------------------------------------------- # Register # --------------------------------------------------------------------------- def register(request): if request.user.is_authenticated: return redirect('index') if request.method == 'POST': form = UserCreationForm(request.POST) if form.is_valid(): user = form.save() login(request, user) return redirect('index') else: form = UserCreationForm() return render(request, 'accounts/register.html', {'form': form}) # --------------------------------------------------------------------------- # Login # --------------------------------------------------------------------------- def login_view(request): if request.user.is_authenticated: return redirect('index') if request.method == 'POST': form = AuthenticationForm(data=request.POST) if form.is_valid(): user = form.get_user() login(request, user) next_url = request.GET.get('next', '/') return redirect(next_url) else: form = AuthenticationForm() return render(request, 'accounts/login.html', {'form': form}) # --------------------------------------------------------------------------- # Settings # --------------------------------------------------------------------------- @login_required def settings_view(request): profile = request.user.profile if request.method == 'POST': profile.lastfm_scrobble = 'lastfm_scrobble' in request.POST profile.save(update_fields=['lastfm_scrobble']) return redirect('settings') context = { 'profile': profile, 'has_lastfm': profile.has_lastfm(), 'password_form': PasswordChangeForm(request.user), 'api_token': getattr(request.user, 'api_token', None), 'webdav_sources': request.user.webdav_sources.all(), } return render(request, 'accounts/settings.html', context) # --------------------------------------------------------------------------- # Last.fm OAuth # --------------------------------------------------------------------------- @login_required def lastfm_connect(request): callback_url = request.build_absolute_uri('/accounts/lastfm/callback/') try: auth_url, token = lastfm_module.get_auth_url(callback_url) request.session['lastfm_token'] = token return redirect(auth_url) except Exception as exc: return render(request, 'accounts/settings.html', { 'profile': request.user.profile, 'has_lastfm': request.user.profile.has_lastfm(), 'lastfm_error': f"Could not connect to Last.fm: {exc}", }) @login_required def lastfm_callback(request): token = request.session.get('lastfm_token') if not token: return redirect('settings') try: session_key = lastfm_module.get_session_key(token) username = lastfm_module.get_username(session_key) profile = request.user.profile profile.lastfm_session_key = session_key profile.lastfm_username = username profile.save(update_fields=['lastfm_session_key', 'lastfm_username']) # Clean up session del request.session['lastfm_token'] except Exception: pass return redirect('settings') @login_required @csrf_exempt @require_http_methods(['POST']) def upload_background(request): try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'invalid JSON'}, status=400) iv = body.get('iv', '').strip() ciphertext = body.get('ciphertext', '').strip() mime_type = body.get('mime_type', '').strip() file_size = int(body.get('file_size', 0)) if not all([iv, ciphertext, mime_type]): return JsonResponse({'error': 'iv, ciphertext, mime_type required'}, status=400) allowed_mimes = {'image/jpeg', 'image/png', 'image/webp'} if mime_type not in allowed_mimes: return JsonResponse({'error': 'only jpeg, png, or webp allowed'}, status=400) if file_size > settings.BG_MAX_BYTES: return JsonResponse({'error': 'file too large (max 5 MB)'}, status=400) profile = request.user.profile profile.background_image_data = '' profile.background_encrypted = ciphertext profile.background_iv = iv profile.background_mime = mime_type profile.save(update_fields=['background_image_data', 'background_encrypted', 'background_iv', 'background_mime']) return JsonResponse({'ok': True}) @login_required @require_http_methods(['POST']) def delete_background(request): profile = request.user.profile profile.background_image_data = '' profile.background_encrypted = '' profile.background_iv = '' profile.background_mime = '' profile.save(update_fields=['background_image_data', 'background_encrypted', 'background_iv', 'background_mime']) return redirect('settings') @login_required @csrf_exempt @require_http_methods(['POST']) def save_focus_station(request): try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'invalid JSON'}, status=400) url = body.get('url', '').strip() name = body.get('name', '').strip() profile = request.user.profile profile.focus_station_url = url profile.focus_station_name = name profile.save(update_fields=['focus_station_url', 'focus_station_name']) return JsonResponse({'ok': True}) @login_required @csrf_exempt @require_http_methods(['POST']) def check_password(request): try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'ok': False}, status=400) pw = body.get('password', '') user = authenticate(request, username=request.user.username, password=pw) return JsonResponse({'ok': user is not None}) @login_required @require_http_methods(['POST']) def change_password(request): form = PasswordChangeForm(request.user, request.POST) if form.is_valid(): user = form.save() update_session_auth_hash(request, user) return redirect('settings') profile = request.user.profile return render(request, 'accounts/settings.html', { 'profile': profile, 'has_lastfm': profile.has_lastfm(), 'password_form': form, 'password_form_open': True, 'api_token': getattr(request.user, 'api_token', None), 'webdav_sources': request.user.webdav_sources.all(), }) @login_required @require_http_methods(['POST']) def lastfm_disconnect(request): profile = request.user.profile profile.lastfm_session_key = '' profile.lastfm_username = '' profile.save(update_fields=['lastfm_session_key', 'lastfm_username']) return redirect('settings') # --------------------------------------------------------------------------- # API token (for local/native clients — see accounts/middleware.py) # --------------------------------------------------------------------------- @login_required @require_http_methods(['POST']) def regenerate_api_token(request): token, _ = ApiToken.objects.get_or_create(user=request.user) token.regenerate() return redirect('settings') # --------------------------------------------------------------------------- # WebDAV sources (Nextcloud & friends) — used by the ebook cloud import # --------------------------------------------------------------------------- def _probe_source(source): """Report a source's reachability as a (level, message) pair for messages.""" try: entries = list_directory(source) except WebDAVError as exc: return messages.WARNING, f'„{source.label}“ gespeichert, aber nicht erreichbar: {exc}' books = sum(1 for e in entries if e['is_book']) folders = sum(1 for e in entries if e['is_dir']) return messages.SUCCESS, ( f'„{source.label}“ verbunden — {books} Buch/Bücher und {folders} Ordner im Startverzeichnis.' ) @login_required @require_http_methods(['POST']) def webdav_add(request): label = request.POST.get('label', '').strip() base_url = request.POST.get('base_url', '').strip() if not label or not base_url: messages.error(request, 'Name und Server-URL sind erforderlich.') return redirect('settings') # Each source costs a synchronous probe on add and is reachable from the # import endpoints, so cap how many one account can pile up. max_sources = getattr(settings, 'WEBDAV_MAX_SOURCES_PER_USER', 10) if request.user.webdav_sources.count() >= max_sources: messages.error(request, f'Maximal {max_sources} Cloud-Verbindungen pro Konto.') return redirect('settings') source = WebDAVSource.objects.create( user=request.user, label=label[:100], base_url=base_url[:500], username=request.POST.get('username', '').strip()[:200], password=request.POST.get('password', '')[:500], root_path=request.POST.get('root_path', '').strip()[:500], ) level, message = _probe_source(source) messages.add_message(request, level, message) return redirect('settings') @login_required @require_http_methods(['POST']) def webdav_test(request, pk): source = WebDAVSource.objects.filter(pk=pk, user=request.user).first() if not source: messages.error(request, 'Verbindung nicht gefunden.') return redirect('settings') level, message = _probe_source(source) messages.add_message(request, level, message) return redirect('settings') @login_required @require_http_methods(['POST']) def webdav_delete(request, pk): deleted, _ = WebDAVSource.objects.filter(pk=pk, user=request.user).delete() if deleted: messages.success(request, 'Verbindung entfernt.') return redirect('settings')