aircox-radiocampus/aircox/models/file.py

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)