PyNyaaTa/pynyaata2/bridge/yggtorrent.py

from os import getenv
from typing import List
from urllib import parse

from bs4 import BeautifulSoup
from pydantic import HttpUrl, parse_obj_as

from pynyaata2.cache import cache_data
from pynyaata2.filters import filter_data
from pynyaata2.session import requests
from pynyaata2.types import Bridge, Color, RemoteFile, async_wrap

from requests import HTTPError


YGG_IP = getenv("YGG_IP")
YGG_DOMAIN = "www3.yggtorrent.do"


class YggTorrent(Bridge):
    color = Color.SUCCESS
    title = "YggTorrent"
    base_url = parse_obj_as(HttpUrl, f"https://{YGG_DOMAIN}")
    hidden_url = parse_obj_as(HttpUrl, f"http://{YGG_IP}") if YGG_IP else base_url
    favicon = parse_obj_as(HttpUrl, f"{base_url}/favicon.ico")
    category = "Animation Série"
    sub_category = 2179

    def search_url(self, query: str = "", page: int = 1) -> HttpUrl:
        params = parse.urlencode(
            {
                "name": query,
                "category": 2145,
                "sub_category": self.sub_category,
                "do": "search",
                "order": "desc",
                "sort": "size" if query else "publish_date",
            }
        )

        return parse_obj_as(HttpUrl, f"{self.base_url}/engine/search?{params}")

    @async_wrap
    @cache_data
    @filter_data
    def search(self, query: str = "", page: int = 1) -> List[RemoteFile]:
        search_url = self.search_url(query, page)
        response = requests.get(
            f"{self.hidden_url}/{search_url.path}?{search_url.query}",
            headers={"Host": YGG_DOMAIN},
        )

        if response.status_code != 200:
            raise HTTPError(response)

        torrents: List[RemoteFile] = []
        html = BeautifulSoup(response.content, "html.parser")
        trs = html.select("table.table tr")

        for i, tr in enumerate(trs):
            if not i:
                continue

            tds = tr.find_all("td")

            torrents.append(
                RemoteFile(
                    bridge=self.__class__.__name__,
                    id=tds[0].div.get_text(),
                    category=self.category,
                    name=tds[1].get_text(),
                    link=tds[1].a["href"],
                    comment=tds[3].get_text(),
                    comment_url=f"{tds[1].a['href']}#comm",
                    torrent=f"{self.base_url}/engine/download_torrent?id={tds[0].div.get_text()}",
                    size=tds[5].get_text(),
                    date=tds[4].div.get_text(),
                    seeds=tds[7].get_text(),
                    leechs=tds[8].get_text(),
                    downloads=tds[6].get_text(),
                    nb_pages=html.select("ul.pagination li")[-2].get_text(),
                )
            )

        return torrents


class YggAnimation(YggTorrent):
    title = "YggAnimation"
    category = "Animation Série"
    sub_category = 2179