API client for smotret-anime-365.ru (anime365 / smotert-anime.ru)

anime365.py

class Api:
    def __init__(self, api_url="https://smotret-anime.ru/api/"):
        self._api_url = api_url

    def translation(self, id=None, **kwargs):
        if id is not None:
            return self._get(f"translations/{id}", {}, Translation)
        return self._get("translations", kwargs, Translation, True)

    def episode(self, id=None, **kwargs):
        if id is not None:
            return self._get(f"episodes/{id}", {}, Episode)
        return self._get("episodes", kwargs, Episode, True)

    def series(self, id=None, **kwargs):
        if id is not None:
            return self._get(f"series/{id}", {}, Series)
        return self._get("series", kwargs, Series, True)

    def series_by_mal_id(self, id):
        return self.series(myAnimeListId=id)

    @staticmethod
    def _parse_json(data, typ, is_array):
        if 'error' in data:
            raise RuntimeError(data['error'])

        if is_array:
            arr = []
            for i in data['data']:
                v = typ()
                v.from_json(i)
                arr.append(v)
            return arr

        v = typ()
        v.from_json(data['data'])
        return v

    def _get(self, url, args, typ, is_array=False):
        raise NotImplementedError()

    def wrap(self, obj):
        raise NotImplementedError()


try:
    import asyncio
    import aiohttp

    class AsyncApi(Api):
        def __init__(self, api_url, session: aiohttp.ClientSession):
            super().__init__(api_url)
            self._session = session

        @asyncio.coroutine
        def _get(self, url, args, typ, is_array=False):
            r = yield from self._session.get(self._api_url + url, params=args)
            try:
                r.raise_for_status()
                data = yield from r.json()
                yield from r.release()
            finally:
                r.close()

            return self._parse_json(data, typ, is_array)

        @asyncio.coroutine
        def wrap(self, obj):
            return obj

except ImportError:
    pass


try:
    import requests

    class SyncApi(Api):
        def __init__(self, api_url, session: requests.Session):
            super().__init__(api_url)
            self._session = session

        def _get(self, url, args, typ, is_array=False):
            with self._session.get(self._api_url + url, params=args) as r:
                r.raise_for_status()
                data = r.json()

            return self._parse_json(data, typ, is_array)

        def wrap(self, obj):
            return obj

except ImportError:
    pass


class _ApiObject:
    _typemap = {}

    def is_complete(self):
        return True

    async def ensure_complete(self, api):
        return self

    def from_json(self, data):
        for name in type(self).__slots__:
            setattr(self, name, self._get_value(name, data.get(name)))

    def _get_value(self, name, value):
        if name not in self._typemap or value is None:
            return value

        typ = self._typemap[name]
        if isinstance(typ, list):
            typ = typ[0]
            arr = []
            for i in value:
                v = typ()
                v.from_json(i)
                arr.append(v)
            return arr

        v = typ()
        v.from_json(value)
        return v

    def __repr__(self):
        attrs = ' '.join((f'{name}={getattr(self, name, "<unknown>")}' for name in type(self).__slots__))
        return f"<{type(self).__name__} {attrs}> "


class Description(_ApiObject):
    __slots__ = "source", "value", "updatedDateTime"


class Genre(_ApiObject):
    __slots__ = "id", "title", "url"


class FansubsTranslation(_ApiObject):
    __slots__ = "id", "fansubsSeriesId", "count", "date", "comment", "file", "fileUrl"


class Titles(_ApiObject):
    __slots__ = "romaji", "ru", "en", "ja", "short"


class Link(_ApiObject):
    __slots__ = "title", "url"


############################


class Episode(_ApiObject):
    __slots__ = "id", "episodeFull", "episodeInt", "episodeTitle", "episodeType", "firstUploadedDateTime",\
                "isActive", "seriesId", "countViews", "translations"

    # _typemap = {
    #     "translations": [Translation]  # com
    # }

    def _get_value(self, name, value):
        if name != "translations" or value is None:
            return value

        arr = []
        for i in value:
            v = Translation()
            v.from_json(i)
            arr.append(v)
        return arr

    def is_complete(self):
        return self.translations is not None

    def ensure_complete(self, api):
        if self.is_complete():
            return api.wrap(self)

        return api.episode(id=self.id)


class Series(_ApiObject):
    __slots__ = "id", "aniDbId", "animeNewsNetworkId", "fansubsId", "imdbId", "worldArtId", "isActive",\
                "isAiring", "isHentai", "myAnimeListId", "myAnimeListScore", "worldArtScore", "posterUrlSmall",\
                "season", "year", "type", "typeTitle", "countViews", "numberOfEpisodes", "posterUrl", "url",\
                "worldArtTopPlace", "titleLines", "allTitles", "title", "titles", "links", "descriptions", "genres",\
                "episodes"

    _typemap = {
        "titles": Titles,
        "links": [Link],
        "genres": [Genre],  # com
        "episodes": [Episode],  # com
        "descriptions": [Description]  # com
    }

    def is_complete(self):
        return self.genres is not None

    def ensure_complete(self, api):
        if self.is_complete():
            return api.wrap(self)

        return api.series(id=self.id)


class Translation(_ApiObject):
    __slots__ = 'id', 'addedDateTime', "activeDateTime", "authorsList", "fansubsTranslationId",\
                "isActive", "priority", "qualityType", "type", "typeKind", "typeLang", "updatedDateTime",\
                "title", "seriesId", "episodeId", "countViews", "url", "embedUrl", "authorsSummary",\
                "fansubsTranslation", "duration", "width", "height", "episode", "series"

    _typemap = {
        "fansubsTranslation": FansubsTranslation,  # com
        "episode": Episode,  # com
        "series": Series  # com
    }

    def is_complete(self):
        return self.episode is not None

    def ensure_complete(self, api):
        if self.is_complete():
            return api.wrap(self)

        return api.translation(id=self.id)