Xefir/PyNyaaTa
Archived
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Add YGG_PI thx Az

Browse Source
This commit is contained in:
Michel Roux 2023-01-06 00:19:32 +01:00
parent 54a36adebf
commit bc5b524948
2 changed files with 16 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
pynyaata/bridge/yggtorrent.py
View File

@ -1,3 +1,4 @@
from os import getenv
from typing import List
from urllib import parse
@ -7,16 +8,20 @@ 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
import requests
YGG_IP = getenv("YGG_IP")
YGG_DOMAIN = "www6.yggtorrent.lol"
class YggTorrent(Bridge):
color = Color.SUCCESS
title = "YggTorrent"
base_url = parse_obj_as(HttpUrl, "https://www6.yggtorrent.lol")
base_url = parse_obj_as(HttpUrl, f"https://{YGG_DOMAIN}")
hidden_url = parse_obj_as(HttpUrl, f"http://{YGG_IP}")
favicon = parse_obj_as(HttpUrl, f"{base_url}/favicon.ico")
category = "Animation Série"
sub_category = 2179
@ -39,10 +44,16 @@ class YggTorrent(Bridge):
@cache_data
@filter_data
async def search(self, query: str = "", page: int = 1) -> List[RemoteFile]:
response = FlareRequests().get(self.search_url(query, page))
if YGG_IP:
params = self.search_url(query, page).split("?")
response = requests.get(
f"{self.hidden_url}?{params[1]}", headers={"Host": YGG_DOMAIN}
)
else:
response = requests.get(self.search_url(query, page))
if response.status_code != 200:
raise HTTPError(response)
raise requests.HTTPError(response)
torrents: List[RemoteFile] = []
html = BeautifulSoup(response.content, "html.parser")

79
pynyaata/session.py
View File

@ -1,79 +0,0 @@
from io import BytesIO
from os import getenv
from urllib import parse
from requests import RequestException, Response, Session, post
CLOUDPROXY_ENDPOINT = getenv("CLOUDPROXY_ENDPOINT")
class FlareRequests(Session):
def request(self, method, url, params=None, data=None, **kwargs):
if not CLOUDPROXY_ENDPOINT:
return super().request(method, url, params, data, **kwargs)
sessions = post(CLOUDPROXY_ENDPOINT, json={"cmd": "sessions.list"}).json()
if "sessions" in sessions and len(sessions["sessions"]) > 0:
FLARESESSION = sessions["sessions"][0]
else:
response = post(CLOUDPROXY_ENDPOINT, json={"cmd": "sessions.create"})
session = response.json()
if "session" in session:
FLARESESSION = session["session"]
else:
raise RequestException(response)
if params:
url += "&" if len(url.split("?")) > 1 else "?"
url = f"{url}{parse.urlencode(params)}"
post_data = {
"cmd": f"request.{method.lower()}",
"session": FLARESESSION,
"url": url,
}
if data:
post_data["postData"] = parse.urlencode(data)
try:
response = post(
CLOUDPROXY_ENDPOINT,
json=post_data,
)
solution = response.json()
if "solution" in solution:
if "content-type" in solution["solution"]["headers"]:
content_type = solution["solution"]["headers"][
"content-type"
].split(";")
if len(content_type) > 1:
charset = content_type[1].split("=")
if len(charset) > 1:
encoding = charset[1]
resolved = Response()
resolved.status_code = solution["solution"]["status"]
resolved.headers = solution["solution"]["headers"]
resolved.raw = BytesIO(solution["solution"]["response"].encode())
resolved.url = url
resolved.encoding = encoding or None
resolved.reason = solution["status"]
resolved.cookies = solution["solution"]["cookies"]
return resolved
raise RequestException(response)
except RequestException:
session = post(
CLOUDPROXY_ENDPOINT,
json={"cmd": "sessions.destroy", "session": FLARESESSION},
)
raise RequestException(solution)