polyphonic/app/library/models.py

from django.shortcuts import resolve_url
from django.db import models
from django.utils.text import slugify
from django.utils.timezone import now
from django.utils.functional import cached_property
from django.db.models import Count, Min, Max

import re
import os.path

from byostorage.user import BYOStorage
from byostorage.cached import CachedStorage

from library.music_tags import MusicTag
from interface.utils import sign_data

import logging

# from polyphonic.settings import LIBRARY_STORAGE
logger = logging.getLogger(__name__)

# try:
#    library_storage = get_storage_class(settings.LIBRARY_STORAGE)()
# except (ImportError, AttributeError):
#    logger.exception("Failed to load library storage")
#    library_storage = get_storage_class()()
# logger.info("Library storage: %s", library_storage.__class__.__name__)

# FIXME: move back to settings
library_storage = CachedStorage(BYOStorage())


class Orchestration(models.Model):
    """
    Stores a list of instrument codes as a single entry (space delimited).
    Can be global or ensemble specific
    """

    collection = models.ForeignKey(
        "Collection",
        on_delete=models.CASCADE,
        related_name="custom_orchestrations",
        null=True,
        blank=True,
    )
    name = models.CharField(max_length=100)
    instruments = models.TextField()

    def as_list(self):
        tags = [t.strip() for t in self.instruments.split(" ")]
        return [(t, MusicTag.from_tag(t)) for t in tags if t]

    def tag_order(self):
        tags = [t.strip() for t in self.instruments.split(" ") if t]
        order = {"score": 0}
        for i, t in enumerate(tags):
            order.setdefault(t.strip("-0123456789"), i * 2 + 1)

        return order

    def sorter(self):
        tag_order = self.tag_order()

        def f(x):
            return (tag_order.get(x[0].strip("-0123456789"), 1000), x[0])

        return f

    def save(self):
        self.as_list()
        super(Orchestration, self).save()

    def __str__(self):
        return self.name


class ProjectItem(models.Model):
    """
    ProjectItem represents a Work attached to a Project e.g. item in set list or programme
    It also allows works to be shared from one ensemble to another on a per-project basis.
    """

    project = models.ForeignKey(
        "interface.Project", on_delete=models.CASCADE, related_name="items"
    )
    work = models.ForeignKey(
        "Work", on_delete=models.CASCADE, related_name="project_items"
    )
    checkout = models.DateTimeField()
    due = models.DateTimeField(null=True, blank=True)
    returned = models.DateTimeField(null=True, blank=True)
    approved_by = models.ForeignKey("auth.User", on_delete=models.CASCADE)
    order = models.SmallIntegerField(default=0)
    section = models.CharField(max_length=100, blank=True)

    class Meta:
        ordering = ["order", "work"]

    def __str__(self):
        return f"<{self.project_id}:{slugify(self.work.name)}>"


class Collection(models.Model):
    """
    A logical collection of works, typically owned by an organisation or person (physical or virtual)
    """

    name = models.CharField(max_length=255, help_text="Name of the collection")
    prefix = models.CharField(
        max_length=255, default="default", help_text="Folder to store works in"
    )
    administrators = models.ManyToManyField(
        "auth.User",
        related_name="collections",
        help_text="Administrators for this collection",
    )
    location = models.CharField(
        max_length=100, help_text="Physical location (institution, town...)", blank=True
    )
    storage = models.ForeignKey(
        "byostorage.UserStorage",
        on_delete=models.CASCADE,
        null=True,
        blank=True,
        help_text="User storage for documents",
    )
    notes = models.TextField(
        blank=True,
        help_text="Publicly visible notes about collection and loans policy (markdown format)",
    )
    settings = models.JSONField(
        default=dict, blank=True, help_text="Storage specific settings"
    )
    nonce = models.SmallIntegerField(
        default=1, help_text="Increment this to reset the authentication links"
    )

    def meta(self, name):
        items = (
            WorkMeta.objects.filter(work__collection=self.pk, name=name)
            .values_list("value", flat=True)
            .distinct()
        )
        return items

    @property
    def tags(self):
        return self.meta("tag")

    @property
    def genres(self):
        return self.meta("genre")

    def has_administrator(self, user):
        if not user.is_authenticated:
            return False
        if user.is_superuser:
            return True
        return user.pk in self.administrators.values_list("pk", flat=True)

    def get_absolute_url(self):
        return resolve_url("collection_work_list", self.pk)

    def auth(self):
        return sign_data(f"{self.pk}-{self.nonce}", 12)

    def __str__(self):
        return self.name


class EnsembleAccess(models.Model):
    """
    Can have different access levels to a collection
    """

    ACCESS_UNLIMITED = 1
    ACCESS_APPROVED = 2

    ACCESS_TYPES = (
        (ACCESS_UNLIMITED, "Unlimited"),
        (ACCESS_APPROVED, "Approval required"),
    )

    ensemble = models.ForeignKey(
        "interface.Ensemble",
        on_delete=models.CASCADE,
        related_name="allowed_collections",
    )
    collection = models.ForeignKey(
        Collection, on_delete=models.CASCADE, related_name="allowed_ensembles"
    )
    access_type = models.PositiveSmallIntegerField(choices=ACCESS_TYPES, default=2)

    class Meta:
        verbose_name_plural = "Ensemble access"


class Work(models.Model):
    """
    A musical work 'owned' by a collection from a licencing perspective.
    """

    LICENCE_PUBLIC = 2
    LICENCE_EXPIRED = 4
    LICENCE_RECORDING = 5
    LICENCE_PERFORMANCE = 6
    LICENCE_PERUSAL = 8
    LICENCE_NONE = 10

    LICENCE_TYPES = (
        (LICENCE_PUBLIC, "Public Domain"),
        (LICENCE_EXPIRED, "Copyright Expired"),
        (LICENCE_RECORDING, "Recording Licence"),
        (LICENCE_PERFORMANCE, "Performance Licence"),
        (LICENCE_PERUSAL, "Perusal Licence"),
        (LICENCE_NONE, "Internal use only"),
    )

    name = models.CharField(max_length=255, help_text="Original name of the work")
    edition = models.CharField(
        max_length=255,
        blank=True,
        help_text="Edition details to distinguish multiple versions",
    )
    parent = models.ForeignKey(
        "Work",
        null=True,
        blank=True,
        on_delete=models.SET_NULL,
        related_name="related_works",
        help_text="Arrangement of another work or part of an anthology",
    )
    composer = models.CharField(
        max_length=255,
        default="Anon",
        help_text="Composer or compilation editor. Use <b>Surname, Initial</b> for easy searching",
    )

    orchestration = models.ForeignKey(
        Orchestration,
        on_delete=models.SET_DEFAULT,
        default=1,
        help_text="Orchestration for the work",
    )
    original_parts = models.JSONField(
        default=dict, blank=True, help_text="Original printed parts (IMSLP format)"
    )

    # Collection details
    collection = models.ForeignKey(
        Collection, on_delete=models.CASCADE, related_name="works"
    )
    code = models.CharField(
        max_length=100,
        blank=True,
        help_text="Collection specific code or number. Will be auto-generated if not supplied",
    )
    licence = models.PositiveSmallIntegerField(
        choices=LICENCE_TYPES, default=6, help_text="Copyright status"
    )
    max_projects = models.IntegerField(
        default=1, help_text="How many active projects can this work be attached to"
    )

    # Extra info
    running_time = models.DurationField(
        null=True, blank=True, help_text="Running time in mm:ss format"
    )
    notes = models.TextField(blank=True)

    # Allocation to projects
    projects = models.ManyToManyField(
        "interface.Project",
        through="ProjectItem",
        related_name="works",
        help_text="Current usage",
    )

    @property
    def folder(self):
        return f"{slugify(self.composer)}/{slugify(self.name)}-{self.pk:04d}"

    def tagged_sections(self, *tags):
        qs = self.docs.filter(sections__tag__in=tags)
        qs = qs.annotate(
            Count("sections"), end=Min("sections__end"), start=Max("sections__start")
        ).filter(sections__count=len(tags))
        return qs

    def list_sections(self, *tags):
        return list(self.tagged_sections(*tags).values_list("upload", "start", "end"))

    @property
    def digital_parts(self):
        sections = [(s.tag, s.name) for s in Section.objects.filter(doc__work=self.pk)]
        sections.sort(key=self.orchestration.sorter())
        # return [ s[1] for s in sections ]
        sections = list(dict(sections).items())  # primitive unique()
        return sections

    def pdfs(self):
        return self.docs.filter(doctype=Document.DOCTYPE_PDF)

    @property
    def physical_parts(self):
        if not self.original_parts:
            return []
        parts = list(self.original_parts.items())
        parts.sort(key=self.orchestration.sorter())
        return [(MusicTag.from_tag(x[0]), x[1]) for x in parts]

    @property
    def tags(self):
        return self.meta_info.filter(name="tag").values_list("value", flat=True)

    @property
    def meta(self):
        return self.meta_info.exclude(name="tag")

    @property
    def current_loans(self):
        return self.project_items.filter(
            checkout__lte=now(), returned=None
        ).select_related("project")

    @cached_property
    def loans(self):
        try:
            return self.loan_count
        except AttributeError:
            return self.project_items.filter(checkout__lte=now(), returned=None).count()

    @property
    def is_available(self):
        if self.max_projects < 0:
            return True
        return self.max_projects > self.loans

    @property
    def available(self):
        if self.max_projects < 0:
            return "Unlimited"
        a = self.max_projects - self.loans
        return "{0} of {1}".format(max(a, 0), self.max_projects)

    @property
    def identifier(self):

        if self.code:
            return self.code

        composer = self.composer or "Anon"
        composer = re.sub("[^\w]", "", composer)
        words = self.name.split()
        work = words[0][:3]

        return f"{composer[:4]}-{work}-{self.pk:05d}".upper()

    def assigned_instruments(self):
        return Section.objects.filter(doc__work_id=self.pk).values_list(
            "tag", flat=True
        )

    def unassigned_instruments(self):
        assigned = set(self.assigned_instruments())
        return [x for x in self.orchestration.as_list() if x[0] not in assigned]

    def music_tags(self):
        tags = dict(self.orchestration.as_list())
        for section in Section.objects.filter(doc__work_id=self.pk):
            tags.setdefault(section.tag, section.name)

        return tags.items()

    def __str__(self):
        return f"{self.name} ({self.composer})"


class WorkMeta(models.Model):
    META_CHOICES = (
        ("tag", "Tag"),
        ("arr", "Arranger"),
        ("lyrics", "Lyracist"),
        ("genre", "Genre"),
        ("style", "Style"),
        ("orchestration", "Orchestration"),
    )

    work = models.ForeignKey(Work, on_delete=models.CASCADE, related_name="meta_info")
    name = models.SlugField(max_length=20, choices=META_CHOICES)
    value = models.CharField(max_length=255)


def doc_upload_filename(doc, filename):
    collection = doc.work.collection
    storage = collection.storage
    if not storage:
        raise RuntimeError("Collection has no storage attached")
    return f"{storage}:library/{collection.prefix}/{doc.work.folder}/{filename}"


class Document(models.Model):
    """
    Document represents a single file stored in the storage backend.
    """

    DOCTYPE_PDF = 1
    DOCTYPE_AUDIO = 2
    DOCTYPE_VIDEO = 3
    DOCTYPE_MISC = 4

    DOCTYPES = (
        (DOCTYPE_PDF, "PDF"),
        (DOCTYPE_AUDIO, "Audio"),
        (DOCTYPE_VIDEO, "Video"),
        (DOCTYPE_MISC, "Misc"),
    )

    DOCTYPE_MAP = {
        ".pdf": DOCTYPE_PDF,
        ".mp3": DOCTYPE_AUDIO,
        ".mp4": DOCTYPE_VIDEO,
    }

    work = models.ForeignKey("Work", on_delete=models.CASCADE, related_name="docs")
    doctype = models.PositiveSmallIntegerField(choices=DOCTYPES, default=DOCTYPE_PDF)
    upload = models.FileField(upload_to=doc_upload_filename, storage=library_storage)
    created = models.DateTimeField(auto_now_add=True)
    version = models.CharField(max_length=30, blank=True)

    def delete(self, *args, **kwargs):
        self.upload.delete(save=False)
        return super().delete(*args, **kwargs)

    def filename(self):
        return os.path.basename(self.upload.name)

    def __str__(self):
        return self.upload.name


class Section(models.Model):
    """
    Section is a tagged portion of a Document
    """

    PAGE_AUTO = 0
    PAGE_LEFT = 1
    PAGE_RIGHT = 2

    PAGE_PREFERENCE = (
        (PAGE_AUTO, "auto"),
        (PAGE_LEFT, "left"),
        (PAGE_RIGHT, "right"),
    )

    doc = models.ForeignKey(Document, on_delete=models.CASCADE, related_name="sections")
    tag = models.CharField(max_length=50, blank=True)
    start = models.SmallIntegerField(null=True, blank=True)
    end = models.SmallIntegerField(null=True, blank=True)
    page = models.SmallIntegerField(
        default=PAGE_AUTO, choices=PAGE_PREFERENCE
    )  # NOT CURRENTLY USED

    class Meta:
        ordering = ["doc", "start", "pk"]

    @property
    def music_tag(self):
        return MusicTag.from_tag(self.tag)

    @property
    def name(self):
        return str(self.music_tag)

    @property
    def bulma_class(self):
        return "success" if self.music_tag.is_general else "info"

    @property
    def filename(self):
        return slugify(f"{self.doc.work.name} - {self.name}").title() + ".pdf"

    @property
    def pagerange(self):
        if self.start:
            if self.end:
                return f"{self.start}-{self.end}"
            return str(self.start)
        return "all"

    def __str__(self):
        return self.name