130 lines
5.4 KiB
Python
130 lines
5.4 KiB
Python
import secrets
|
|
from urllib.parse import urlsplit
|
|
|
|
from django.db import models
|
|
from django.contrib.auth.models import User
|
|
from django.db.models.signals import post_save
|
|
from django.dispatch import receiver
|
|
|
|
|
|
class UserProfile(models.Model):
|
|
user = models.OneToOneField(User, on_delete=models.CASCADE, related_name='profile')
|
|
lastfm_session_key = models.CharField(max_length=100, blank=True)
|
|
lastfm_username = models.CharField(max_length=100, blank=True)
|
|
lastfm_scrobble = models.BooleanField(default=True)
|
|
background_image_data = models.TextField(blank=True) # base64 data URL (legacy)
|
|
background_encrypted = models.TextField(blank=True) # base64 AES-GCM ciphertext
|
|
background_iv = models.CharField(max_length=32, blank=True) # hex IV
|
|
background_mime = models.CharField(max_length=30, blank=True) # e.g. 'image/jpeg'
|
|
focus_station_url = models.URLField(max_length=1000, blank=True)
|
|
focus_station_name = models.CharField(max_length=300, blank=True)
|
|
|
|
def has_lastfm(self) -> bool:
|
|
return bool(self.lastfm_session_key)
|
|
|
|
def __str__(self):
|
|
return f"Profile of {self.user.username}"
|
|
|
|
|
|
@receiver(post_save, sender=User)
|
|
def create_user_profile(sender, instance, created, **kwargs):
|
|
if created:
|
|
UserProfile.objects.create(user=instance)
|
|
|
|
|
|
@receiver(post_save, sender=User)
|
|
def save_user_profile(sender, instance, **kwargs):
|
|
if hasattr(instance, 'profile'):
|
|
instance.profile.save()
|
|
|
|
|
|
class WebDAVSource(models.Model):
|
|
"""A user-supplied WebDAV endpoint that ebooks can be imported from.
|
|
|
|
Deliberately generic rather than Nextcloud-specific — any WebDAV server
|
|
(Nextcloud, ownCloud, Synology, rclone serve, …) works, Nextcloud just gets
|
|
a URL-shorthand in `normalized_base_url`.
|
|
|
|
The password is stored in the clear because the server has to replay it on
|
|
every PROPFIND/GET (same trade-off as `lastfm_session_key` above). The UI
|
|
therefore tells users to create a revocable *app password* rather than
|
|
handing over their account password.
|
|
"""
|
|
user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='webdav_sources')
|
|
label = models.CharField(max_length=100)
|
|
base_url = models.URLField(max_length=500)
|
|
username = models.CharField(max_length=200, blank=True)
|
|
password = models.CharField(max_length=500, blank=True)
|
|
root_path = models.CharField(max_length=500, blank=True, default='')
|
|
# "Eingang" (docs/library.md): new books in it come into the library by themselves —
|
|
# a device with the key fetches, encrypts and uploads them (WebDAVSeen keeps which).
|
|
inbox = models.BooleanField(default=False)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
last_used_at = models.DateTimeField(null=True, blank=True)
|
|
|
|
class Meta:
|
|
ordering = ['created_at']
|
|
|
|
def __str__(self):
|
|
return f"WebDAVSource({self.label}, user={self.user_id})"
|
|
|
|
def normalized_base_url(self) -> str:
|
|
"""Collection root to resolve browse paths against, always ending in '/'.
|
|
|
|
A bare host ('https://cloud.example.com') is expanded to the Nextcloud
|
|
files endpoint, since that is the URL users actually have at hand; an
|
|
URL that already points into a DAV tree is left alone so non-Nextcloud
|
|
servers stay usable.
|
|
"""
|
|
url = self.base_url.strip()
|
|
if not url.endswith('/'):
|
|
url += '/'
|
|
# Match against the path only — a host literally named "webdav.…" or
|
|
# "dav.…" must not be mistaken for a URL that already points into a
|
|
# DAV tree.
|
|
path = urlsplit(url).path.lower()
|
|
is_dav = any(marker in path for marker in ('/remote.php/', '/dav/', '/webdav'))
|
|
if not is_dav and self.username:
|
|
url += f'remote.php/dav/files/{self.username}/'
|
|
root = self.root_path.strip().strip('/')
|
|
if root:
|
|
url += root + '/'
|
|
return url
|
|
|
|
|
|
class WebDAVSeen(models.Model):
|
|
"""A file of an "Eingang" source a device took (or is taking) into the library: taken
|
|
files aren't offered again; a claim keeps two devices from taking the same file at once
|
|
and runs out when a device didn't finish."""
|
|
source = models.ForeignKey(WebDAVSource, on_delete=models.CASCADE, related_name='seen')
|
|
path = models.CharField(max_length=1000)
|
|
size = models.BigIntegerField(default=0)
|
|
modified = models.CharField(max_length=100, blank=True)
|
|
claimed_at = models.DateTimeField(null=True, blank=True)
|
|
done_at = models.DateTimeField(null=True, blank=True)
|
|
result = models.CharField(max_length=20, blank=True) # imported | known | too_big | failed
|
|
|
|
class Meta:
|
|
unique_together = ('source', 'path')
|
|
|
|
|
|
def _generate_token():
|
|
return secrets.token_hex(32)
|
|
|
|
|
|
class ApiToken(models.Model):
|
|
"""Personal access token for local/native clients (see accounts.middleware).
|
|
|
|
Sent as `Authorization: Bearer <token>` and treated identically to a
|
|
session login by ApiTokenAuthMiddleware — grants the same access as the
|
|
owning user across the whole app (books, podcasts, radio)."""
|
|
user = models.OneToOneField(User, on_delete=models.CASCADE, related_name='api_token')
|
|
token = models.CharField(max_length=64, unique=True, db_index=True, default=_generate_token)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
|
|
def regenerate(self):
|
|
self.token = _generate_token()
|
|
self.save(update_fields=['token'])
|
|
|
|
def __str__(self):
|
|
return f"ApiToken({self.user.username})"
|