import os from pathlib import Path from django.conf import settings from django.db import models from django.db.models import Q from django.utils.translation import gettext_lazy as _ from django.utils import timezone as tz from .program import Program class FileQuerySet(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 available(self): return self.exclude(is_removed=False) def public(self): """Return sounds available as podcasts.""" return self.filter(is_public=True) def path(self, paths): if isinstance(paths, str): return self.filter(file=paths.replace(settings.MEDIA_ROOT + "/", "")) return self.filter(file__in=(p.replace(settings.MEDIA_ROOT + "/", "") for p in paths)) def search(self, query): return self.filter(Q(name__icontains=query) | Q(file__icontains=query) | Q(program__title__icontains=query)) class File(models.Model): def _upload_to(self, filename): dir = self.program and self.program.path or self.default_upload_path subdir = self.get_upload_dir() if subdir: return os.path.join(dir, subdir, filename) return os.path.join(dir, filename) program = models.ForeignKey( Program, models.SET_NULL, verbose_name=_("Program"), null=True, blank=True, ) file = models.FileField( _("file"), upload_to=_upload_to, max_length=256, db_index=True, ) name = models.CharField( _("name"), max_length=64, db_index=True, ) description = models.TextField( _("description"), max_length=256, blank=True, default="", ) mtime = models.DateTimeField( _("modification time"), blank=True, null=True, help_text=_("last modification date and time"), ) is_public = models.BooleanField( _("public"), help_text=_("file is publicly accessible"), default=False, ) is_removed = models.BooleanField( _("removed"), help_text=_("file has been removed from server"), default=False, db_index=True, ) class Meta: abstract = True objects = FileQuerySet.as_manager() default_upload_path = Path(settings.MEDIA_ROOT) """Default upload directory when no program is provided.""" upload_dir = "uploads" """Upload sub-directory.""" @property def url(self): return self.file and self.file.url def get_upload_dir(self): return self.upload_dir 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_updated(self): """Return True when file has been updated on filesystem.""" exists = self.file_exists() if self.is_removed != (not exists): return True return exists and self.mtime != self.get_mtime() def file_exists(self): """Return true if the file still exists.""" return os.path.exists(self.file.path) def sync_fs(self, on_update=False): """Sync model to file on the filesystem. :param bool on_update: only check if `file_updated`. :return True wether a change happened. """ if on_update and not self.file_updated(): return # check on name/remove/modification time name = self.name if not self.name and self.file and self.file.name: name = os.path.basename(self.file.name) name = os.path.splitext(name)[0] name = name.replace("_", " ").strip() is_removed = not self.file_exists() mtime = (not is_removed and self.get_mtime()) or None changed = is_removed != self.is_removed or mtime != self.mtime or name != self.name self.name, self.is_removed, self.mtime = name, is_removed, mtime # read metadata if changed and not self.is_removed: metadata = self.read_metadata() metadata and self.__dict__.update(metadata) return changed def read_metadata(self): return {} def save(self, sync=True, *args, **kwargs): if sync and self.file_exists(): self.sync_fs(on_update=True) super().save(*args, **kwargs)