path: root/src/hydrilla/item_infos.py

# SPDX-License-Identifier: GPL-3.0-or-later

# Reading resources, mappings and other JSON documents from the filesystem.
#
# This file is part of Hydrilla&Haketilo
#
# Copyright (C) 2021, 2022 Wojtek Kosior
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
#
#
# I, Wojtek Kosior, thereby promise not to sue for violation of this
# file's license. Although I request that you do not make use of this
# code in a proprietary program, I am not going to enforce this in
# court.

"""
.....
"""

# Enable using with Python 3.7.
from __future__ import annotations

import sys

if sys.version_info >= (3, 8):
    from typing import Protocol
else:
    from typing_extensions import Protocol

import enum
import typing as t
import dataclasses as dc

from pathlib import Path, PurePosixPath
from abc import ABC, abstractmethod

from immutables import Map

from . import versions, json_instances
from .url_patterns import parse_pattern, ParsedUrl, ParsedPattern
from .exceptions import HaketiloException
from .translations import smart_gettext as _


@dc.dataclass(frozen=True, unsafe_hash=True)
class ItemSpecifier:
    """...."""
    identifier: str

SpecifierObjs = t.Sequence[t.Mapping[str, t.Any]]

def make_item_specifiers_seq(spec_objs: SpecifierObjs) \
    -> tuple[ItemSpecifier, ...]:
    """...."""
    return tuple(ItemSpecifier(obj['identifier']) for obj in spec_objs)

def make_required_mappings(spec_objs: t.Any, schema_compat: int) \
    -> tuple[ItemSpecifier, ...]:
    """...."""
    if schema_compat < 2:
        return ()

    return make_item_specifiers_seq(spec_objs)

@dc.dataclass(frozen=True, unsafe_hash=True)
class FileSpecifier:
    """...."""
    name:   str
    sha256: str

def normalize_filename(name: str):
    """
    This function eliminated double slashes in file name and ensures it does not
    try to reference parent directories.
    """
    path = PurePosixPath(name)

    if '.' in path.parts or '..' in path.parts:
        msg = _('err.item_info.filename_invalid_{}').format(name)
        raise HaketiloException(msg)

    return str(path)

def make_file_specifiers_seq(spec_objs: SpecifierObjs) \
    -> tuple[FileSpecifier, ...]:
    """...."""
    return tuple(
        FileSpecifier(normalize_filename(obj['file']), obj['sha256'])
        for obj
        in spec_objs
    )

@dc.dataclass(frozen=True, unsafe_hash=True)
class GeneratedBy:
    """...."""
    name:    str
    version: t.Optional[str]

    @staticmethod
    def make(generated_by_obj: t.Optional[t.Mapping[str, t.Any]]) -> \
        t.Optional['GeneratedBy']:
        """...."""
        if generated_by_obj is None:
            return None

        return GeneratedBy(
            name    = generated_by_obj['name'],
            version = generated_by_obj.get('version')
        )


def make_eval_permission(perms_obj: t.Any, schema_compat: int) -> bool:
    if schema_compat < 2:
        return False

    return perms_obj.get('eval', False)


def make_cors_bypass_permission(perms_obj: t.Any, schema_compat: int) -> bool:
    if schema_compat < 2:
        return False

    return perms_obj.get('cors_bypass', False)


class Categorizable(Protocol):
    """...."""
    uuid:       t.Optional[str]
    identifier: str

@dc.dataclass(frozen=True, unsafe_hash=True)
class ItemIdentity:
    repo:           str
    repo_iteration: int
    version:        versions.VerTuple
    identifier:     str

# mypy needs to be corrected:
# https://stackoverflow.com/questions/70999513/conflict-between-mix-ins-for-abstract-dataclasses/70999704#70999704
@dc.dataclass(frozen=True) # type: ignore[misc]
class ItemInfoBase(ABC, ItemIdentity, Categorizable):
    """...."""
    source_name:        str                       = dc.field(hash=False, compare=False)
    source_copyright:   tuple[FileSpecifier, ...] = dc.field(hash=False, compare=False)
    uuid:               t.Optional[str]           = dc.field(hash=False, compare=False)
    long_name:          str                       = dc.field(hash=False, compare=False)
    allows_eval:        bool                      = dc.field(hash=False, compare=False)
    allows_cors_bypass: bool                      = dc.field(hash=False, compare=False)
    required_mappings:  tuple[ItemSpecifier, ...] = dc.field(hash=False, compare=False)
    generated_by:       t.Optional[GeneratedBy]   = dc.field(hash=False, compare=False)

    @property
    def version_string(self) -> str:
        return versions.version_string(self.version)

    @property
    def versioned_identifier(self) -> str:
        """...."""
        return f'{self.identifier}-{self.version_string}'

    @property
    def files(self) -> tuple[FileSpecifier, ...]:
        return self.source_copyright

    @staticmethod
    def _get_base_init_kwargs(
            item_obj:       t.Mapping[str, t.Any],
            schema_compat:  int,
            repo:           str,
            repo_iteration: int
    ) -> t.Mapping[str, t.Any]:
        """...."""
        source_copyright = make_file_specifiers_seq(
            item_obj['source_copyright']
        )

        version = versions.normalize(item_obj['version'])

        perms_obj = item_obj.get('permissions', {})

        eval_perm        = make_eval_permission(perms_obj, schema_compat)
        cors_bypass_perm = make_cors_bypass_permission(perms_obj, schema_compat)

        required_mappings = make_required_mappings(
            item_obj.get('required_mappings', []),
            schema_compat
        )

        generated_by = GeneratedBy.make(item_obj.get('generated_by'))

        return Map(
            repo               = repo,
            repo_iteration     = repo_iteration,
            source_name        = item_obj['source_name'],
            source_copyright   = source_copyright,
            version            = version,
            identifier         = item_obj['identifier'],
            uuid               = item_obj.get('uuid'),
            long_name          = item_obj['long_name'],
            allows_eval        = eval_perm,
            allows_cors_bypass = cors_bypass_perm,
            required_mappings  = required_mappings,
            generated_by       = generated_by
        )


class ItemType(enum.Enum):
    RESOURCE = 'resource'
    MAPPING  = 'mapping'

    @property
    def info_class(self) -> t.Type[AnyInfo]:
        if self == ItemType.RESOURCE:
            return ResourceInfo
        else:
            return MappingInfo

    @property
    def alt_name(self) -> str:
        if self == ItemType.RESOURCE:
            return 'library'
        else:
            return 'package'

    @property
    def alt_name_plural(self) -> str:
        if self == ItemType.RESOURCE:
            return 'libraries'
        else:
            return 'packages'

@dc.dataclass(frozen=True, unsafe_hash=True)
class CorrespondsToResourceDCMixin:
    type: t.ClassVar[ItemType] = ItemType.RESOURCE

@dc.dataclass(frozen=True, unsafe_hash=True)
class CorrespondsToMappingDCMixin:
    type: t.ClassVar[ItemType] = ItemType.MAPPING


@dc.dataclass(frozen=True, unsafe_hash=True)
class ResourceInfo(ItemInfoBase, CorrespondsToResourceDCMixin):
    """...."""
    revision:     int                       = dc.field(hash=False, compare=False)
    dependencies: tuple[ItemSpecifier, ...] = dc.field(hash=False, compare=False)
    scripts:      tuple[FileSpecifier, ...] = dc.field(hash=False, compare=False)

    @property
    def version_string(self) -> str:
        return f'{super().version_string}-{self.revision}'

    @property
    def files(self) -> tuple[FileSpecifier, ...]:
        return tuple((*self.source_copyright, *self.scripts))

    @staticmethod
    def make(
            item_obj:       t.Mapping[str, t.Any],
            schema_compat:  int,
            repo:           str,
            repo_iteration: int
    ) -> 'ResourceInfo':
        """...."""
        base_init_kwargs = ItemInfoBase._get_base_init_kwargs(
            item_obj,
            schema_compat,
            repo,
            repo_iteration
        )

        dependencies = make_item_specifiers_seq(
            item_obj.get('dependencies', [])
        )

        scripts = make_file_specifiers_seq(
            item_obj.get('scripts', [])
        )

        return ResourceInfo(
            **base_init_kwargs,

            revision     = item_obj['revision'],
            dependencies = dependencies,
            scripts      = scripts
        )

    @staticmethod
    def load(
            instance_source: json_instances.InstanceSource,
            repo:            str = '<dummyrepo>',
            repo_iteration:  int = -1
    ) -> 'ResourceInfo':
        """...."""
        return _load_item_info(
            ResourceInfo,
            instance_source,
            repo,
            repo_iteration
        )

    def __lt__(self, other: 'ResourceInfo') -> bool:
        """...."""
        return (
            self.identifier,
            other.version,
            other.revision,
            self.repo,
            other.repo_iteration
        ) < (
            other.identifier,
            self.version,
            self.revision,
            other.repo,
            self.repo_iteration
        )

def make_payloads(payloads_obj: t.Mapping[str, t.Any]) \
    -> t.Mapping[ParsedPattern, ItemSpecifier]:
    """...."""
    mapping: list[tuple[ParsedPattern, ItemSpecifier]] = []

    for pattern, spec_obj in payloads_obj.items():
        ref = ItemSpecifier(spec_obj['identifier'])
        mapping.extend((parsed, ref) for parsed in parse_pattern(pattern))

    return Map(mapping)

@dc.dataclass(frozen=True, unsafe_hash=True)
class MappingInfo(ItemInfoBase, CorrespondsToMappingDCMixin):
    """...."""
    payloads: t.Mapping[ParsedPattern, ItemSpecifier] = \
        dc.field(hash=False, compare=False)

    @staticmethod
    def make(
            item_obj:       t.Mapping[str, t.Any],
            schema_compat:  int,
            repo:           str,
            repo_iteration: int
    ) -> 'MappingInfo':
        """...."""
        base_init_kwargs = ItemInfoBase._get_base_init_kwargs(
            item_obj,
            schema_compat,
            repo,
            repo_iteration
        )

        return MappingInfo(
            **base_init_kwargs,

            payloads = make_payloads(item_obj.get('payloads', {}))
        )

    @staticmethod
    def load(
            instance_source: json_instances.InstanceSource,
            repo:            str = '<dummyrepo>',
            repo_iteration:  int = -1
    ) -> 'MappingInfo':
        """...."""
        return _load_item_info(
            MappingInfo,
            instance_source,
            repo,
            repo_iteration
        )

    def __lt__(self, other: 'MappingInfo') -> bool:
        """...."""
        return (
            self.identifier,
            other.version,
            self.repo,
            other.repo_iteration
        ) < (
            other.identifier,
            self.version,
            other.repo,
            self.repo_iteration
        )


AnyInfo = t.Union[ResourceInfo, MappingInfo]


LoadedType = t.TypeVar('LoadedType', ResourceInfo, MappingInfo)

def _load_item_info(
        info_type:       t.Type[LoadedType],
        instance_source: json_instances.InstanceSource,
        repo:            str,
        repo_iteration:  int
) -> LoadedType:
    """Read, validate and autocomplete a mapping/resource description."""
    instance = json_instances.read_instance(instance_source)

    schema_fmt = f'api_{info_type.type.value}_description-{{}}.schema.json'

    schema_compat = json_instances.validate_instance(instance, schema_fmt)

    # We know from successful validation that instance is a dict.
    return info_type.make(
        t.cast('dict[str, t.Any]', instance),
        schema_compat,
        repo,
        repo_iteration
    )


CategorizedInfoType = t.TypeVar(
    'CategorizedInfoType',
    ResourceInfo,
    MappingInfo
)

CategorizedType = t.TypeVar(
    'CategorizedType',
    bound=Categorizable
)

CategorizedUpdater = t.Callable[
    [t.Optional[CategorizedType]],
    t.Optional[CategorizedType]
]

CategoryKeyType = t.TypeVar('CategoryKeyType', bound=t.Hashable)

@dc.dataclass(frozen=True) # type: ignore[misc]
class CategorizedItemInfo(
        ABC,
        Categorizable,
        t.Generic[CategorizedInfoType, CategorizedType, CategoryKeyType]
):
    """...."""
    SelfType = t.TypeVar(
        'SelfType',
        bound = 'CategorizedItemInfo[CategorizedInfoType, CategorizedType, CategoryKeyType]'
    )

    uuid:         t.Optional[str]                       = None
    identifier:   str                                   = '<dummy>'
    items:        Map[CategoryKeyType, CategorizedType] = Map()
    _initialized: bool                                  = False

    def _update(
            self:   'SelfType',
            key:     CategoryKeyType,
            updater: CategorizedUpdater
    ) -> 'SelfType':
        """...... Perform sanity checks for uuid."""
        uuid = self.uuid

        items = self.items.mutate()

        updated = updater(items.get(key))
        if updated is None:
            items.pop(key, None)

            identifier = self.identifier
        else:
            items[key] = updated

            identifier = updated.identifier
            if self._initialized:
                assert identifier == self.identifier

            if uuid is not None:
                if updated.uuid is not None and uuid != updated.uuid:
                    raise HaketiloException(_('uuid_mismatch_{identifier}')
                                            .format(identifier=identifier))
            else:
                uuid = updated.uuid

        return dc.replace(
            self,
            identifier   = identifier,
            uuid         = uuid,
            items        = items.finish(),
            _initialized = self._initialized or updated is not None
        )

    @abstractmethod
    def register(self: 'SelfType', info: CategorizedInfoType) -> 'SelfType':
        ...

    @abstractmethod
    def get_all(self: 'SelfType') -> t.Sequence[CategorizedInfoType]:
        ...

    def is_empty(self) -> bool:
        return len(self.items) == 0


class VersionedItemInfo(
        CategorizedItemInfo[
            CategorizedInfoType,
            CategorizedInfoType,
            versions.VerTuple
        ],
        t.Generic[CategorizedInfoType]
):
    """Stores data of multiple versions of given resource/mapping."""
    SelfType = t.TypeVar(
        'SelfType',
        bound = 'VersionedItemInfo[CategorizedInfoType]'
    )

    def register(self: 'SelfType', item_info: CategorizedInfoType) \
        -> 'SelfType':
        """
        Make item info queryable by version. Perform sanity checks for uuid.
        """
        return self._update(item_info.version, lambda old_info: item_info)

    @property
    def newest_version(self) -> versions.VerTuple:
        """...."""
        assert not self.is_empty()

        return self.versions()[-1]

    @property
    def newest_info(self) -> CategorizedInfoType:
        """Find and return info of the newest version of item."""
        return self.items[self.newest_version]

    def versions(self, reverse: bool = False) -> t.Sequence[versions.VerTuple]:
        return sorted(self.items.keys(), reverse=reverse)

    def get_by_ver(self, ver: t.Sequence[int]) \
        -> t.Optional[CategorizedInfoType]:
        """
        Find and return info of the specified version of the item (or None if
        absent).
        """
        return self.items.get(versions.normalize(ver))

    def get_all(self, reverse_versions: bool = False) \
        -> t.Sequence[CategorizedInfoType]:
        """
        Generate item info for all its versions, from oldest to newest unless
        the opposite is requested.
        """
        versions = self.versions(reverse=reverse_versions)
        return [self.items[ver] for ver in versions]

VersionedResourceInfo = VersionedItemInfo[ResourceInfo]
VersionedMappingInfo  = VersionedItemInfo[MappingInfo]

VersionedItemInfoMap     = Map[str, VersionedItemInfo]
VersionedResourceInfoMap = Map[str, VersionedResourceInfo]
VersionedMappingInfoMap  = Map[str, VersionedMappingInfo]

def register_in_versioned_map(
        map:  Map[str, VersionedItemInfo[CategorizedInfoType]],
        info: CategorizedInfoType
) -> Map[str, VersionedItemInfo[CategorizedInfoType]]:
    versioned_info = map.get(info.identifier, VersionedItemInfo())

    return map.set(info.identifier, versioned_info.register(info))


class MultirepoItemInfo(
        CategorizedItemInfo[
            CategorizedInfoType,
            VersionedItemInfo[CategorizedInfoType],
            tuple[str, int]
        ],
        t.Generic[CategorizedInfoType]
):
    """
    Stores data of multiple versions of given resource/mapping that may come
    from multiple repositories.
    """
    SelfType = t.TypeVar(
        'SelfType',
        bound = 'MultirepoItemInfo[CategorizedInfoType]'
    )

    def register(self: 'SelfType', item_info: CategorizedInfoType) \
        -> 'SelfType':
        """
        Make item info queryable by repo and version. Perform sanity checks for
        uuid.
        """
        def update(
                versioned: t.Optional[VersionedItemInfo[CategorizedInfoType]]
        ) -> VersionedItemInfo[CategorizedInfoType]:
            if versioned is None:
                versioned = VersionedItemInfo()
            return versioned.register(item_info)

        return self._update((item_info.repo, item_info.repo_iteration), update)

    @property
    def default_info(self) -> CategorizedInfoType:
        """
        Find and return info of one of the available options for the newest
        version of item.
        """
        assert not self.is_empty()

        return self.get_all(reverse_repos=True)[-1]

    def options(self, reverse: bool = False) -> t.Sequence[tuple[str, int]]:
        return sorted(
            self.items.keys(),
            key     = (lambda tuple: (tuple[0], 1 - tuple[1])),
            reverse = reverse
        )

    def get_all(
            self,
            reverse_versions: bool = False,
            reverse_repos:    bool = False
    ) -> t.Sequence[CategorizedInfoType]:
        """
        Generate item info for all its versions and options, from oldest to
        newest version and from.
        """
        all_versions: set[versions.VerTuple] = set()
        for versioned in self.items.values():
            all_versions.update(versioned.versions())

        result = []

        for version in sorted(all_versions, reverse=reverse_versions):
            for option in self.options(reverse=reverse_repos):
                info = self.items[option].get_by_ver(version)
                if info is not None:
                    result.append(info)

        return result

MultirepoResourceInfo = MultirepoItemInfo[ResourceInfo]
MultirepoMappingInfo  = MultirepoItemInfo[MappingInfo]


MultirepoItemInfoMap     = Map[str, MultirepoItemInfo]
MultirepoResourceInfoMap = Map[str, MultirepoResourceInfo]
MultirepoMappingInfoMap  = Map[str, MultirepoMappingInfo]

def register_in_multirepo_map(
        map:  Map[str, MultirepoItemInfo[CategorizedInfoType]],
        info: CategorizedInfoType
) -> Map[str, MultirepoItemInfo[CategorizedInfoType]]:
    multirepo_info = map.get(info.identifier, MultirepoItemInfo())

    return map.set(info.identifier, multirepo_info.register(info))


def all_map_infos(
    map: Map[str, CategorizedItemInfo[CategorizedInfoType, t.Any, t.Any]]
) -> t.Iterator[CategorizedInfoType]:
    for versioned_info in map.values():
        for item_info in versioned_info.get_all():
            yield item_info