version.py

from __future__ import annotations

import datetime
import logging
from typing import TypedDict

from dandischema.conf import get_instance_config
from dandischema.models import AccessType
from django.conf import settings
from django.contrib.postgres.indexes import HashIndex
from django.core.validators import RegexValidator
from django.db import models
from django.db.models.query_utils import Q
from django_extensions.db.models import TimeStampedModel

from dandiapi.api.asset_paths import get_root_paths
from dandiapi.api.models.metadata import PublishableMetadataMixin

from .dandiset import Dandiset

logger = logging.getLogger(__name__)


class VersionAssetValidationError(TypedDict):
    field: str
    message: str
    path: str


class Version(PublishableMetadataMixin, TimeStampedModel):
    VERSION_REGEX = r'(0\.\d{6}\.\d{4})|draft'

    class Status(models.TextChoices):
        PENDING = 'Pending'
        VALIDATING = 'Validating'
        VALID = 'Valid'
        INVALID = 'Invalid'
        PUBLISHING = 'Publishing'
        PUBLISHED = 'Published'

    dandiset = models.ForeignKey(Dandiset, related_name='versions', on_delete=models.CASCADE)
    name = models.CharField(max_length=300)
    metadata = models.JSONField(blank=True, default=dict)
    version = models.CharField(
        max_length=13,
        validators=[RegexValidator(f'^{VERSION_REGEX}$')],
    )
    doi = models.CharField(max_length=64, null=True, default=None, blank=True)  # noqa: DJ001
    """Track the validation status of this version, without considering assets"""
    status = models.CharField(
        max_length=10,
        default=Status.PENDING,
        choices=Status.choices,
    )
    validation_errors = models.JSONField(default=list, blank=True, null=True)

    class Meta:
        ordering = ['version']
        unique_together = ['dandiset', 'version']
        constraints = [
            models.CheckConstraint(
                name='version_metadata_has_schema_version',
                condition=Q(metadata__schemaVersion__isnull=False),
            )
        ]
        indexes = [
            HashIndex(fields=['metadata']),
            HashIndex(fields=['name']),
        ]

    @property
    def asset_count(self):
        return get_root_paths(self).aggregate(nfiles=models.Sum('aggregate_files'))['nfiles'] or 0

    @property
    def size(self):
        return (
            get_root_paths(self).aggregate(total_size=models.Sum('aggregate_size'))['total_size']
            or 0
        )

    @property
    def active_uploads(self):
        return self.dandiset.uploads.count() if self.version == 'draft' else 0

    @property
    def publishable(self) -> bool:
        if self.status != Version.Status.VALID:
            return False

        # Import here to avoid dependency cycle
        from .asset import Asset

        # Return False if any asset is not VALID
        return not self.assets.exclude(status=Asset.Status.VALID).exists()

    @property
    def asset_validation_errors(self) -> list[VersionAssetValidationError]:
        # Import here to avoid dependency cycle
        from .asset import Asset

        # We want to display "Pending" assets in the validation errors list,
        # despite them not being stored explicitly as errors in the database.
        # Grab a random sample of 50 pending or currently validating assets
        # and place them first in the list.
        pending_assets: models.QuerySet[Asset] = (
            self.assets.filter(status__in=[Asset.Status.PENDING, Asset.Status.VALIDATING])
            .annotate(
                field=models.Value(''),
                message=models.Value('asset is currently being validated, please wait.'),
            )
            .values('field', 'message', 'path')[:50]
        )

        # Next, get all INVALID assets. Each of these should have one or more
        # validation errors stored in the database.
        # For performance reasons, we truncate the list of INVALID assets such
        # that we only display errors for the 50 assets with the most errors.
        invalid_assets: models.QuerySet[Asset] = (
            self.assets.filter(status=Asset.Status.INVALID)
            .alias(
                validation_error_count=models.Func(
                    'validation_errors',
                    function='jsonb_array_length',
                    output_field=models.IntegerField(),
                )
            )
            .order_by('-validation_error_count')
            .values('path', 'validation_errors')
        )[:50]

        return list(pending_assets) + [
            {
                'field': error['field'],
                'message': error['message'],
                'path': asset['path'],
            }
            for asset in invalid_assets
            for error in asset['validation_errors']
        ]

    @staticmethod
    def datetime_to_version(time: datetime.datetime) -> str:
        return time.strftime('0.%y%m%d.%H%M')

    @classmethod
    def next_published_version(cls, dandiset: Dandiset) -> str:
        time = datetime.datetime.now(datetime.UTC)
        # increment time until there are no collisions
        while True:
            version = cls.datetime_to_version(time)
            collision = dandiset.versions.filter(version=version).exists()
            if not collision:
                break
            time += datetime.timedelta(minutes=1)

        return version

    @classmethod
    def citation(cls, metadata):
        year = datetime.datetime.now(datetime.UTC).year
        name = metadata['name'].rstrip('.')
        url = f'https://doi.org/{metadata["doi"]}' if 'doi' in metadata else metadata['url']
        version = metadata['version']
        # If we can't find any contributors, use this citation format
        citation = f'{name} ({year}). (Version {version}) [Data set]. DANDI Archive. {url}'
        if 'contributor' in metadata and isinstance(metadata['contributor'], list):
            cl = '; '.join(
                [val['name'] for val in metadata['contributor'] if val.get('includeInCitation')]
            )
            if cl:
                citation = (
                    f'{cl} ({year}) {name} (Version {version}) [Data set]. DANDI Archive. {url}'
                )
        return citation

    @classmethod
    def strip_metadata(cls, metadata):
        """Strip away computed fields from a metadata dict."""
        computed_fields = [
            'name',
            'identifier',
            'version',
            'id',
            'url',
            'assetsSummary',
            'citation',
            'doi',
            'dateCreated',
            'datePublished',
            'publishedBy',
            'manifestLocation',
        ]
        stripped = {key: metadata[key] for key in metadata if key not in computed_fields}

        # Strip the status and schemaKey fields, as modifying them is not supported
        if (
            'access' in stripped
            and isinstance(stripped['access'], list)
            and len(stripped['access'])
            and isinstance(stripped['access'][0], dict)
        ):
            stripped['access'][0].pop('schemaKey', None)
            stripped['access'][0].pop('status', None)

        return stripped

    def _populate_access_metadata(self):
        default_access = [{}]
        access = self.metadata.get('access', default_access)

        # Ensure access is a non-empty list
        if not (isinstance(access, list) and access):
            access = default_access

        # Ensure that every item in access is a dict
        access = [x for x in access if isinstance(x, dict)] or default_access

        # Set first access item
        access[0] = {
            **access[0],
            'schemaKey': 'AccessRequirements',
            'status': AccessType.EmbargoedAccess.value
            if self.dandiset.embargoed
            else AccessType.OpenAccess.value,
        }

        return access

    def _populate_metadata(self):
        from dandiapi.api.manifests import manifest_location

        schema_config = get_instance_config()
        metadata = {
            **self.metadata,
            '@context': (
                'https://raw.githubusercontent.com/dandi/schema/master/releases/'
                f'{self.metadata["schemaVersion"]}/context.json'
            ),
            'manifestLocation': manifest_location(self),
            'name': self.name,
            'identifier': (f'{schema_config.instance_name}:{self.dandiset.identifier}'),
            'version': self.version,
            'id': (f'{schema_config.instance_name}:{self.dandiset.identifier}/{self.version}'),
            'repository': settings.DANDI_WEB_APP_URL,
            'url': (
                f'{settings.DANDI_WEB_APP_URL}/dandiset/{self.dandiset.identifier}/{self.version}'
            ),
            'dateCreated': self.dandiset.created.isoformat(),
            'access': self._populate_access_metadata(),
        }

        if 'assetsSummary' not in metadata:
            metadata['assetsSummary'] = {
                'schemaKey': 'AssetsSummary',
                'numberOfBytes': 0,
                'numberOfFiles': 0,
            }

        if self.doi:
            metadata['doi'] = self.doi
        metadata['citation'] = self.citation(metadata)

        return metadata

    def save(self, *args, **kwargs):
        self.metadata = self._populate_metadata()
        super().save(*args, **kwargs)

    def __str__(self) -> str:
        return f'{self.dandiset.identifier}/{self.version}'