bazarr/libs/tld/base.py

from codecs import open as codecs_open
import logging
from urllib.request import urlopen
from typing import Optional, Dict, Union, ItemsView

from .exceptions import (
    TldIOError,
    TldImproperlyConfigured,
)
from .helpers import project_dir

__author__ = "Artur Barseghyan"
__copyright__ = "2013-2021 Artur Barseghyan"
__license__ = "MPL-1.1 OR GPL-2.0-only OR LGPL-2.1-or-later"
__all__ = (
    "BaseTLDSourceParser",
    "Registry",
)

LOGGER = logging.getLogger(__name__)


class Registry(type):

    REGISTRY: Dict[str, "BaseTLDSourceParser"] = {}

    def __new__(mcs, name, bases, attrs):
        new_cls = type.__new__(mcs, name, bases, attrs)
        # Here the name of the class is used as key but it could be any class
        # parameter.
        if getattr(new_cls, "_uid", None):
            mcs.REGISTRY[new_cls._uid] = new_cls
        return new_cls

    @property
    def _uid(cls) -> str:
        return getattr(cls, "uid", cls.__name__)

    @classmethod
    def reset(mcs) -> None:
        mcs.REGISTRY = {}

    @classmethod
    def get(
        mcs, key: str, default: "BaseTLDSourceParser" = None
    ) -> Union["BaseTLDSourceParser", None]:
        return mcs.REGISTRY.get(key, default)

    @classmethod
    def items(mcs) -> ItemsView[str, "BaseTLDSourceParser"]:
        return mcs.REGISTRY.items()

    # @classmethod
    # def get_registry(mcs) -> Dict[str, Type]:
    #     return dict(mcs.REGISTRY)
    #
    # @classmethod
    # def pop(mcs, uid) -> None:
    #     mcs.REGISTRY.pop(uid)


class BaseTLDSourceParser(metaclass=Registry):
    """Base TLD source parser."""

    uid: Optional[str] = None
    source_url: str
    local_path: str
    include_private: bool = True

    @classmethod
    def validate(cls):
        """Constructor."""
        if not cls.uid:
            raise TldImproperlyConfigured(
                "The `uid` property of the TLD source parser shall be defined."
            )

    @classmethod
    def get_tld_names(cls, fail_silently: bool = False, retry_count: int = 0):
        """Get tld names.

        :param fail_silently:
        :param retry_count:
        :return:
        """
        cls.validate()
        raise NotImplementedError(
            "Your TLD source parser shall implement `get_tld_names` method."
        )

    @classmethod
    def update_tld_names(cls, fail_silently: bool = False) -> bool:
        """Update the local copy of the TLD file.

        :param fail_silently:
        :return:
        """
        try:
            remote_file = urlopen(cls.source_url)
            local_file_abs_path = project_dir(cls.local_path)
            local_file = codecs_open(
                local_file_abs_path, "wb", encoding="utf8"
            )
            local_file.write(remote_file.read().decode("utf8"))
            local_file.close()
            remote_file.close()
            LOGGER.debug(
                f"Fetched '{cls.source_url}' as '{local_file_abs_path}'"
            )
        except Exception as err:
            LOGGER.debug(
                f"Failed fetching '{cls.source_url}'. Reason: {str(err)}"
            )
            if fail_silently:
                return False
            raise TldIOError(err)

        return True