forked from rc/aircox
0e183099ed
- !88 pytest on existing tests - !89 reorganise settings (! see notes for deployment) Co-authored-by: bkfox <thomas bkfox net> Reviewed-on: rc/aircox#92
323 lines
9.3 KiB
Python
323 lines
9.3 KiB
Python
import logging
|
|
import os
|
|
|
|
from django.conf import settings as conf
|
|
from django.db import models
|
|
from django.db.models import Q
|
|
from django.utils import timezone as tz
|
|
from django.utils.translation import gettext_lazy as _
|
|
from taggit.managers import TaggableManager
|
|
|
|
from aircox.conf import settings
|
|
|
|
from .episode import Episode
|
|
from .program import Program
|
|
|
|
logger = logging.getLogger("aircox")
|
|
|
|
|
|
__all__ = ("Sound", "SoundQuerySet", "Track")
|
|
|
|
|
|
class SoundQuerySet(models.QuerySet):
|
|
def station(self, station=None, id=None):
|
|
id = station.pk if id is None else id
|
|
return self.filter(program__station__id=id)
|
|
|
|
def episode(self, episode=None, id=None):
|
|
id = episode.pk if id is None else id
|
|
return self.filter(episode__id=id)
|
|
|
|
def diffusion(self, diffusion=None, id=None):
|
|
id = diffusion.pk if id is None else id
|
|
return self.filter(episode__diffusion__id=id)
|
|
|
|
def available(self):
|
|
return self.exclude(type=Sound.TYPE_REMOVED)
|
|
|
|
def public(self):
|
|
"""Return sounds available as podcasts."""
|
|
return self.filter(is_public=True)
|
|
|
|
def downloadable(self):
|
|
"""Return sounds available as podcasts."""
|
|
return self.filter(is_downloadable=True)
|
|
|
|
def archive(self):
|
|
"""Return sounds that are archives."""
|
|
return self.filter(type=Sound.TYPE_ARCHIVE)
|
|
|
|
def path(self, paths):
|
|
if isinstance(paths, str):
|
|
return self.filter(file=paths.replace(conf.MEDIA_ROOT + "/", ""))
|
|
return self.filter(
|
|
file__in=(p.replace(conf.MEDIA_ROOT + "/", "") for p in paths)
|
|
)
|
|
|
|
def playlist(self, archive=True, order_by=True):
|
|
"""Return files absolute paths as a flat list (exclude sound without
|
|
path).
|
|
|
|
If `order_by` is True, order by path.
|
|
"""
|
|
if archive:
|
|
self = self.archive()
|
|
if order_by:
|
|
self = self.order_by("file")
|
|
return [
|
|
os.path.join(conf.MEDIA_ROOT, file)
|
|
for file in self.filter(file__isnull=False).values_list(
|
|
"file", flat=True
|
|
)
|
|
]
|
|
|
|
def search(self, query):
|
|
return self.filter(
|
|
Q(name__icontains=query)
|
|
| Q(file__icontains=query)
|
|
| Q(program__title__icontains=query)
|
|
| Q(episode__title__icontains=query)
|
|
)
|
|
|
|
|
|
# TODO:
|
|
# - provide a default name based on program and episode
|
|
class Sound(models.Model):
|
|
"""A Sound is the representation of a sound file that can be either an
|
|
excerpt or a complete archive of the related diffusion."""
|
|
|
|
TYPE_OTHER = 0x00
|
|
TYPE_ARCHIVE = 0x01
|
|
TYPE_EXCERPT = 0x02
|
|
TYPE_REMOVED = 0x03
|
|
TYPE_CHOICES = (
|
|
(TYPE_OTHER, _("other")),
|
|
(TYPE_ARCHIVE, _("archive")),
|
|
(TYPE_EXCERPT, _("excerpt")),
|
|
(TYPE_REMOVED, _("removed")),
|
|
)
|
|
|
|
name = models.CharField(_("name"), max_length=64)
|
|
program = models.ForeignKey(
|
|
Program,
|
|
models.CASCADE,
|
|
blank=True, # NOT NULL
|
|
verbose_name=_("program"),
|
|
help_text=_("program related to it"),
|
|
db_index=True,
|
|
)
|
|
episode = models.ForeignKey(
|
|
Episode,
|
|
models.SET_NULL,
|
|
blank=True,
|
|
null=True,
|
|
verbose_name=_("episode"),
|
|
db_index=True,
|
|
)
|
|
type = models.SmallIntegerField(_("type"), choices=TYPE_CHOICES)
|
|
position = models.PositiveSmallIntegerField(
|
|
_("order"),
|
|
default=0,
|
|
help_text=_("position in the playlist"),
|
|
)
|
|
|
|
def _upload_to(self, filename):
|
|
subdir = (
|
|
settings.SOUND_ARCHIVES_SUBDIR
|
|
if self.type == self.TYPE_ARCHIVE
|
|
else settings.SOUND_EXCERPTS_SUBDIR
|
|
)
|
|
return os.path.join(self.program.path, subdir, filename)
|
|
|
|
file = models.FileField(
|
|
_("file"),
|
|
upload_to=_upload_to,
|
|
max_length=256,
|
|
db_index=True,
|
|
unique=True,
|
|
)
|
|
duration = models.TimeField(
|
|
_("duration"),
|
|
blank=True,
|
|
null=True,
|
|
help_text=_("duration of the sound"),
|
|
)
|
|
mtime = models.DateTimeField(
|
|
_("modification time"),
|
|
blank=True,
|
|
null=True,
|
|
help_text=_("last modification date and time"),
|
|
)
|
|
is_good_quality = models.BooleanField(
|
|
_("good quality"),
|
|
help_text=_("sound meets quality requirements"),
|
|
blank=True,
|
|
null=True,
|
|
)
|
|
is_public = models.BooleanField(
|
|
_("public"),
|
|
help_text=_("whether it is publicly available as podcast"),
|
|
default=False,
|
|
)
|
|
is_downloadable = models.BooleanField(
|
|
_("downloadable"),
|
|
help_text=_(
|
|
"whether it can be publicly downloaded by visitors (sound must be "
|
|
"public)"
|
|
),
|
|
default=False,
|
|
)
|
|
|
|
objects = SoundQuerySet.as_manager()
|
|
|
|
class Meta:
|
|
verbose_name = _("Sound")
|
|
verbose_name_plural = _("Sounds")
|
|
|
|
@property
|
|
def url(self):
|
|
return self.file and self.file.url
|
|
|
|
def __str__(self):
|
|
return "/".join(self.file.path.split("/")[-3:])
|
|
|
|
def save(self, check=True, *args, **kwargs):
|
|
if self.episode is not None and self.program is None:
|
|
self.program = self.episode.program
|
|
if check:
|
|
self.check_on_file()
|
|
if not self.is_public:
|
|
self.is_downloadable = False
|
|
self.__check_name()
|
|
super().save(*args, **kwargs)
|
|
|
|
# TODO: rename get_file_mtime(self)
|
|
def get_mtime(self):
|
|
"""Get the last modification date from file."""
|
|
mtime = os.stat(self.file.path).st_mtime
|
|
mtime = tz.datetime.fromtimestamp(mtime)
|
|
mtime = mtime.replace(microsecond=0)
|
|
return tz.make_aware(mtime, tz.get_current_timezone())
|
|
|
|
def file_exists(self):
|
|
"""Return true if the file still exists."""
|
|
|
|
return os.path.exists(self.file.path)
|
|
|
|
# TODO: rename to sync_fs()
|
|
def check_on_file(self):
|
|
"""Check sound file info again'st self, and update informations if
|
|
needed (do not save).
|
|
|
|
Return True if there was changes.
|
|
"""
|
|
if not self.file_exists():
|
|
if self.type == self.TYPE_REMOVED:
|
|
return
|
|
logger.debug("sound %s: has been removed", self.file.name)
|
|
self.type = self.TYPE_REMOVED
|
|
return True
|
|
|
|
# not anymore removed
|
|
changed = False
|
|
|
|
if self.type == self.TYPE_REMOVED and self.program:
|
|
changed = True
|
|
self.type = (
|
|
self.TYPE_ARCHIVE
|
|
if self.file.name.startswith(self.program.archives_path)
|
|
else self.TYPE_EXCERPT
|
|
)
|
|
|
|
# check mtime -> reset quality if changed (assume file changed)
|
|
mtime = self.get_mtime()
|
|
|
|
if self.mtime != mtime:
|
|
self.mtime = mtime
|
|
self.is_good_quality = None
|
|
logger.debug(
|
|
"sound %s: m_time has changed. Reset quality info",
|
|
self.file.name,
|
|
)
|
|
return True
|
|
|
|
return changed
|
|
|
|
def __check_name(self):
|
|
if not self.name and self.file and self.file.name:
|
|
# FIXME: later, remove date?
|
|
name = os.path.basename(self.file.name)
|
|
name = os.path.splitext(name)[0]
|
|
self.name = name.replace("_", " ").strip()
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.__check_name()
|
|
|
|
|
|
class Track(models.Model):
|
|
"""Track of a playlist of an object.
|
|
|
|
The position can either be expressed as the position in the playlist
|
|
or as the moment in seconds it started.
|
|
"""
|
|
|
|
episode = models.ForeignKey(
|
|
Episode,
|
|
models.CASCADE,
|
|
blank=True,
|
|
null=True,
|
|
verbose_name=_("episode"),
|
|
)
|
|
sound = models.ForeignKey(
|
|
Sound,
|
|
models.CASCADE,
|
|
blank=True,
|
|
null=True,
|
|
verbose_name=_("sound"),
|
|
)
|
|
position = models.PositiveSmallIntegerField(
|
|
_("order"),
|
|
default=0,
|
|
help_text=_("position in the playlist"),
|
|
)
|
|
timestamp = models.PositiveSmallIntegerField(
|
|
_("timestamp"),
|
|
blank=True,
|
|
null=True,
|
|
help_text=_("position (in seconds)"),
|
|
)
|
|
title = models.CharField(_("title"), max_length=128)
|
|
artist = models.CharField(_("artist"), max_length=128)
|
|
album = models.CharField(_("album"), max_length=128, null=True, blank=True)
|
|
tags = TaggableManager(verbose_name=_("tags"), blank=True)
|
|
year = models.IntegerField(_("year"), blank=True, null=True)
|
|
# FIXME: remove?
|
|
info = models.CharField(
|
|
_("information"),
|
|
max_length=128,
|
|
blank=True,
|
|
null=True,
|
|
help_text=_(
|
|
"additional informations about this track, such as "
|
|
"the version, if is it a remix, features, etc."
|
|
),
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Track")
|
|
verbose_name_plural = _("Tracks")
|
|
ordering = ("position",)
|
|
|
|
def __str__(self):
|
|
return "{self.artist} -- {self.title} -- {self.position}".format(
|
|
self=self
|
|
)
|
|
|
|
def save(self, *args, **kwargs):
|
|
if (self.sound is None and self.episode is None) or (
|
|
self.sound is not None and self.episode is not None
|
|
):
|
|
raise ValueError("sound XOR episode is required")
|
|
super().save(*args, **kwargs)
|