KomSuite-NyaaPy/src/komsuite_nyaapy/modules/anime_site.py

import httpx
from komsuite_nyaapy.modules import torrent
from komsuite_nyaapy.modules.parser import parse_nyaa, parse_single, parse_nyaa_rss


class AnimeTorrentSite:
    SITE = torrent.TorrentSite.NYAASI
    URL = SITE

    @classmethod
    def last_uploads(cls, number_of_results: int):
        with httpx.Client() as client:
            r = client.get(cls.URL)
            r.raise_for_status()

            json_data = parse_nyaa(
                request_text=r.text, limit=number_of_results, site=cls.SITE
            )

        return torrent.json_to_class(json_data)

    @classmethod
    def search(
        cls,
        keyword: str,
        category: int = 0,
        subcategory: int = 0,
        filters: int = 0,
        page: int = 0,
        sorting: str = "id",
        order: str = "desc",
        **kwargs,
    ):
        base_url = cls.URL

        user = kwargs.get("user", None)
        user_uri = f"user/{user}" if user else ""

        if page > 0:
            search_uri = "{}/{}?f={}&c={}_{}&q={}&p={}&s={}&o={}".format(
                base_url,
                user_uri,
                filters,
                category,
                subcategory,
                keyword,
                page,
                sorting,
                order,
            )
        else:
            search_uri = "{}/{}?f={}&c={}_{}&q={}&s={}&o={}".format(
                base_url,
                user_uri,
                filters,
                category,
                subcategory,
                keyword,
                sorting,
                order,
            )

        if not user:
            search_uri += "&page=rss"

        with httpx.Client() as client:
            http_response = client.get(search_uri)
            http_response.raise_for_status()

            if user:
                json_data = parse_nyaa(
                    request_text=http_response.content, limit=None, site=cls.SITE
                )
            else:
                json_data = parse_nyaa_rss(
                    request_text=http_response.content, limit=None, site=cls.SITE
                )

        return torrent.json_to_class(json_data)

    @classmethod
    def get(cls, view_id: int):
        with httpx.Client() as client:
            r = client.get(f"{cls.URL}/view/{view_id}")
            r.raise_for_status()

            json_data = parse_single(request_text=r.content, site=cls.SITE)

        return torrent.json_to_class(json_data)

    @classmethod
    def get_from_user(cls, username):
        with httpx.Client() as client:
            r = client.get(f"{cls.URL}/user/{username}")
            r.raise_for_status()

            json_data = parse_nyaa(request_text=r.content, limit=None, site=cls.SITE)

        return torrent.json_to_class(json_data)


class AnimeTorrentSiteAsync:
    SITE = torrent.TorrentSite.NYAASI
    URL = SITE

    @classmethod
    async def last_uploads(cls, number_of_results: int):
        async with httpx.AsyncClient() as client:
            r = await client.get(cls.URL)
            r.raise_for_status()

            json_data = parse_nyaa(
                request_text=r.text, limit=number_of_results, site=cls.SITE
            )

        return torrent.json_to_class(json_data)

    @classmethod
    async def search(
        cls,
        keyword: str,
        category: int = 0,
        subcategory: int = 0,
        filters: int = 0,
        page: int = 0,
        sorting: str = "id",
        order: str = "desc",
        **kwargs,
    ):
        base_url = cls.URL

        user = kwargs.get("user", None)
        user_uri = f"user/{user}" if user else ""

        if page > 0:
            search_uri = "{}/{}?f={}&c={}_{}&q={}&p={}&s={}&o={}".format(
                base_url,
                user_uri,
                filters,
                category,
                subcategory,
                keyword,
                page,
                sorting,
                order,
            )
        else:
            search_uri = "{}/{}?f={}&c={}_{}&q={}&s={}&o={}".format(
                base_url,
                user_uri,
                filters,
                category,
                subcategory,
                keyword,
                sorting,
                order,
            )

        if not user:
            search_uri += "&page=rss"

        async with httpx.AsyncClient() as client:
            http_response = await client.get(search_uri)
            http_response.raise_for_status()

            if user:
                json_data = parse_nyaa(
                    request_text=http_response.content, limit=None, site=cls.SITE
                )
            else:
                json_data = parse_nyaa_rss(
                    request_text=http_response.content, limit=None, site=cls.SITE
                )

        return torrent.json_to_class(json_data)

    @classmethod
    async def get(cls, view_id: int):
        async with httpx.AsyncClient() as client:
            r = await client.get(f"{cls.URL}/view/{view_id}")
            r.raise_for_status()

            json_data = parse_single(request_text=r.content, site=cls.SITE)

        return torrent.json_to_class(json_data)

    @classmethod
    async def get_from_user(cls, username):
        async with httpx.AsyncClient() as client:
            r = await client.get(f"{cls.URL}/user/{username}")
            r.raise_for_status()

            json_data = parse_nyaa(request_text=r.content, limit=None, site=cls.SITE)

        return torrent.json_to_class(json_data)