from typing import List from bs4 import BeautifulSoup from humanfriendly import parse_size from pydantic import HttpUrl, parse_obj_as from pynyaata.cache import cache_data from pynyaata.filters import filter_data from pynyaata.session import FlareRequests from pynyaata.types import Bridge, Color, RemoteFile, log_async from requests import HTTPError class YggTorrent(Bridge): color = Color.SUCCESS title = "YggTorrent" base_url = parse_obj_as(HttpUrl, "https://www6.yggtorrent.lol") favicon = parse_obj_as(HttpUrl, "https://www6.yggtorrent.lol/favicon.ico") category = "Animation Série" sub_category = 2179 def search_url(self, query: str = "", page: int = 1) -> HttpUrl: params = { "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}?{params}") @log_async @cache_data @filter_data async def search(self, query: str = "", page: int = 1) -> List[RemoteFile]: response = FlareRequests().get(self.search_url(query, page)) 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=parse_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