diff --git a/src/data/__init__.py b/src/data/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/data/scrapers/provider/buzzheavier.py b/src/data/hosts/buzzheavier.py similarity index 100% rename from src/data/scrapers/provider/buzzheavier.py rename to src/data/hosts/buzzheavier.py diff --git a/src/data/scrapers/provider/gofile.py b/src/data/hosts/gofile.py similarity index 100% rename from src/data/scrapers/provider/gofile.py rename to src/data/hosts/gofile.py diff --git a/src/data/scrapers/__init__.py b/src/data/scrapers/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/data/scrapers/monkrus.py b/src/data/scrapers/monkrus.py deleted file mode 100644 index 613dc8d..0000000 --- a/src/data/scrapers/monkrus.py +++ /dev/null @@ -1,87 +0,0 @@ -from utils.data.tracker import get_magnet_link -from utils.data.state import state -from bs4 import BeautifulSoup -from typing import Dict -import requests -import time - -cache = { - "data": [], - "last_fetched": 0 -} - -cache_expiry = 300 - -def _get_telegram_posts(): - post = 100 - max_posts = 250 - posts = [] - added = set() - - current_time = time.time() - - if cache["data"] and (current_time - cache["last_fetched"] < cache_expiry): - return cache["data"] - - while post <= max_posts: - url = f"https://t.me/s/real_monkrus/{post}" - response = requests.get(url, timeout=10) - soup = BeautifulSoup(response.text, "html.parser") - bubbles = soup.find_all("div", class_="tgme_widget_message_bubble") - - if not bubbles: - break - - for bubble in bubbles: - post_txt = bubble.find("div", class_="tgme_widget_message_text js-message_text") - if post_txt is None or post_txt.b is None: - continue - - title = post_txt.b.text - link = bubble.find("a", href=lambda x: x and x.startswith("https://uztracker.net")) - - if link: - post_url = link["href"] - if post_url not in added: - added.add(post_url) - posts.append(dict( - title=title, - author="m0nkrus", - id=len(posts) + 1, - url=post_url - )) - - post += 22 - - posts.reverse() - - cache["data"] = posts - cache["last_fetched"] = current_time - - return(posts) - -def scrape_m0nkrus(query): - posts = _get_telegram_posts() - filtered_posts = [] - - for post in posts: - if query.lower() in post["title"].lower(): - filtered_post = post.copy() - filtered_post["id"] = len(filtered_posts) + 1 - filtered_posts.append(filtered_post) - - return filtered_posts - -def get_magnet(post: Dict): - return get_magnet_link(post["url"]) - -Metadata = { - "name" : "m0nkrus", - "headers" : ["Post Title", "Author"], - "scrapeFunc" : scrape_m0nkrus, - "linkFunc" : get_magnet, - "isMagnet" : True, -} - -def init_m0nkrus(): - state.trackers.update({Metadata["name"] : Metadata}) diff --git a/src/data/scrapers/rutracker.py b/src/data/scrapers/rutracker.py deleted file mode 100644 index 15acbce..0000000 --- a/src/data/scrapers/rutracker.py +++ /dev/null @@ -1,52 +0,0 @@ -from utils.network.jsonhandler import split_data, format_data -from utils.data.tracker import get_magnet_link -from utils.logging.logs import consoleLog -from utils.data.state import state -from typing import Dict -import requests - -def scrape_rutracker(query): - search = requests.get(f"{state.api_url}/search?q={query}", timeout=15) - consoleLog("Sent request to server") - if search: - _, data, _, success, cached = split_data(search.text) - if cached: - consoleLog("Server response cached") - if success: - sorted_data = [] - - for entry in data: - sorted_data.append( - { - "title" : entry["title"], - "author" : entry["author"], - "seeders" : entry["seeders"], - "leechers" : entry["leechers"], - "url" : entry["url"], - "id" : entry["id"], - } - ) - - return sorted_data - else: - consoleLog("Scraping failed server-side, unable to fetch posts from rutracker") - return [] - else: - consoleLog("No response from server, returning nothing") - return [] - - -def get_magnet(post: Dict): - _, post_links, _, _, _ = format_data([post]) - return get_magnet_link(post_links[0]) - -Metadata = { - "name" : "rutracker", - "headers" : ["Post Title", "Author", "Seeders", "Leechers"], - "scrapeFunc" : scrape_rutracker, - "linkFunc" : get_magnet, - "isMagnet" : True, -} - -def init_rutracker(): - state.trackers.update({Metadata["name"] : Metadata}) diff --git a/src/data/scrapers/steamrip.py b/src/data/scrapers/steamrip.py deleted file mode 100644 index 22f0735..0000000 --- a/src/data/scrapers/steamrip.py +++ /dev/null @@ -1,122 +0,0 @@ -from data.scrapers.provider.buzzheavier import scrape_buzzheavier -from data.scrapers.provider.gofile import scrape_gofile -from utils.logging.logs import consoleLog -from utils.data.state import state -from bs4 import BeautifulSoup -from typing import Dict -import webbrowser -import requests -import time -import re - - -cache = { - "data": [], - "last_fetched": 0 -} - -cache_expiry = 300 - -def get_Metadata(): - return Metadata - -def scrape_steamrip_links(): - - current_time = time.time() - - if cache["data"] != [] and (current_time - cache["last_fetched"] < cache_expiry): - return cache["data"] - - url = f"https://steamrip.com/games-list-page/" - response = requests.get(url) - text = response.text - - soup = BeautifulSoup(text, "html.parser") - games = soup.find_all("li", class_="az-list-item") - - if len(games) == 0: - return [] - - links = [] - names = [] - for gamehtml in games: - link = gamehtml.find("a", href=lambda x: x and x.startswith("/")) - links += re.findall(r'(?<=href=")[^"]*', link.__str__()) - name = gamehtml.find("a", href=lambda x: x and x.startswith("/")) - names += re.findall(r'(?<=\/">)[^<]*', name.__str__()) - - - - # construct the list[dict[str,str]] - - ret = [] - - for i in range(len(names)): - ret.append({"title" : names[i], "url" : links[i]}) - - return ret - -def scrape_steamrip_game_downloads(gamelink): - url = "https://steamrip.com" + gamelink - response = requests.get(url) - soup = BeautifulSoup(response.text, "html.parser") - download_link_elements = soup.find_all("a",class_="shortc-button") - download_links = ["buzzheavier", "gofile"] - - for download_link in download_link_elements: - pure = download_link.attrs.get("href") - if pure[2] == "b": - download_links[0] = "https:" + pure - if pure[2] == "g": - download_links[1] = "https:" + pure - - ret = [] - - for link in download_links: - if len(link) != 1: - ret.append(link) - - cache["data"] = ret - - return ret - -def get_download_link(post: Dict): - url = post["url"] - links = scrape_steamrip_game_downloads(url) - best = "" - for link in links: - if link[0] == "h": - best = link - break - - if links.index(best) == 0: - link = scrape_buzzheavier(best) - return link - elif links.index(best) == 1: - link, headers = scrape_gofile(best) - return link, headers - else: - consoleLog("Unable to retrieve download link due to captcha, launching browser...") - webbrowser.open(best) - return None, None - -def filter_steamrip(query: str): - games = scrape_steamrip_links() - - filtered_games = [ - game for game in games - if query.lower() in game["title"].lower() - ] - - return filtered_games - -Metadata = { - "name" : "steamrip", - "headers" : ["Game"], - "scrapeFunc" : filter_steamrip, - "linkFunc" : get_download_link, - "isMagnet" : False, -} - -def init_steamrip(): - state.trackers.update({Metadata["name"] : Metadata}) diff --git a/src/data/scrapers/uztracker.py b/src/data/scrapers/uztracker.py deleted file mode 100644 index 3c55c72..0000000 --- a/src/data/scrapers/uztracker.py +++ /dev/null @@ -1,55 +0,0 @@ -from utils.data.tracker import get_magnet_link -from utils.logging.logs import consoleLog -from utils.data.state import state -from urllib.parse import urljoin -from bs4 import BeautifulSoup -from typing import Dict -import requests - - -def scrape_uztracker(query): - base_url="https://uztracker.net/" - search_url = f"{base_url.rstrip('/')}/tracker.php?nm={query}" - consoleLog(search_url) - posts = [] - - try: - response = requests.get(search_url) - response.raise_for_status() - soup = BeautifulSoup(response.text, 'html.parser') - links = soup.find_all('tr', class_="tCenter hl-tr", id=lambda x: x and x.startswith('tor_')) - for link in links: - - theme_link = link.find('a', class_="genmed tLink", href=lambda x: x and x.startswith('./viewtopic')) - if not theme_link or not theme_link.b: - continue - url = urljoin(base_url, theme_link['href']) - title = theme_link.b.text - author_link = link.find('a', class_="med") - author = author_link.text.strip() if author_link else "Unknown" - - posts.append(dict( - title=title, - author=author, - url=url, - )) - - return posts - - except requests.RequestException as e: - consoleLog(f"Failed to fetch {search_url}: {e}") - return None - -def get_magnet(post: Dict): - return get_magnet_link(post["url"]) - -Metadata = { - "name" : "uztracker", - "headers" : ["Post Title", "Author"], - "scrapeFunc" : scrape_uztracker, - "linkFunc" : get_magnet, - "isMagnet" : True, -} - -def init_uztracker(): - state.trackers.update({Metadata["name"] : Metadata}) diff --git a/src/data/sources/__init__.py b/src/data/sources/__init__.py new file mode 100644 index 0000000..569e6a1 --- /dev/null +++ b/src/data/sources/__init__.py @@ -0,0 +1,12 @@ +from data.sources.rutracker import RutrackerScraper +from data.sources.uztracker import UztrackerScraper +from data.sources.steamrip import SteamripScraper +from data.sources.monkrus import MonkrusScraper + +# Register Scrapers here +SCRAPERS = [ + RutrackerScraper(), + SteamripScraper(), + MonkrusScraper(), + UztrackerScraper(), +] \ No newline at end of file diff --git a/src/data/sources/monkrus.py b/src/data/sources/monkrus.py new file mode 100644 index 0000000..3386d22 --- /dev/null +++ b/src/data/sources/monkrus.py @@ -0,0 +1,78 @@ +from utils.data.tracker import get_magnet_link +from bs4 import BeautifulSoup +from typing import Dict +import requests +import time + + +class MonkrusScraper: + name = "m0nkrus" + headers = ["Post Title", "Author"] + is_magnet = True + + def __init__(self): + self.cache = { "data": [], "last_fetched": 0 } + self.cache_expiry = 300 + + def _get_telegram_posts(self): + post = 100 + max_posts = 250 + posts = [] + added = set() + + current_time = time.time() + + if self.cache["data"] and (current_time - self.cache["last_fetched"] < self.cache_expiry): + return self.cache["data"] + + while post <= max_posts: + url = f"https://t.me/s/real_monkrus/{post}" + response = requests.get(url, timeout=10) + soup = BeautifulSoup(response.text, "html.parser") + bubbles = soup.find_all("div", class_="tgme_widget_message_bubble") + + if not bubbles: + break + + for bubble in bubbles: + post_txt = bubble.find("div", class_="tgme_widget_message_text js-message_text") + if post_txt is None or post_txt.b is None: + continue + + title = post_txt.b.text + link = bubble.find("a", href=lambda x: x and x.startswith("https://uztracker.net")) + + if link: + post_url = link["href"] + if post_url not in added: + added.add(post_url) + posts.append(dict( + title=title, + author="m0nkrus", + id=len(posts) + 1, + url=post_url + )) + + post += 22 + + posts.reverse() + + self.cache["data"] = posts + self.cache["last_fetched"] = current_time + + return(posts) + + def search(self, query): + posts = self._get_telegram_posts() + filtered_posts = [] + + for post in posts: + if query.lower() in post["title"].lower(): + filtered_post = post.copy() + filtered_post["id"] = len(filtered_posts) + 1 + filtered_posts.append(filtered_post) + + return filtered_posts + + def get_magnet(self, post: Dict): + return get_magnet_link(post["url"]) diff --git a/src/data/sources/rutracker.py b/src/data/sources/rutracker.py new file mode 100644 index 0000000..9f48bfc --- /dev/null +++ b/src/data/sources/rutracker.py @@ -0,0 +1,46 @@ +from utils.network.jsonhandler import split_data, format_data +from utils.data.tracker import get_magnet_link +from utils.logging.logs import consoleLog +from utils.data.state import state +from typing import Dict +import requests + +class RutrackerScraper: + name = "rutracker" + headers = ["Post Title", "Author", "Seeders", "Leechers"] + is_magnet = True + + def search(self, query): + search = requests.get(f"{state.api_url}/search?q={query}", timeout=15) + consoleLog("Sent request to server") + if search: + _, data, _, success, cached = split_data(search.text) + if cached: + consoleLog("Server response cached") + if success: + sorted_data = [] + + for entry in data: + sorted_data.append( + { + "title" : entry["title"], + "author" : entry["author"], + "seeders" : entry["seeders"], + "leechers" : entry["leechers"], + "url" : entry["url"], + "id" : entry["id"], + } + ) + + return sorted_data + else: + consoleLog("Scraping failed server-side, unable to fetch posts from rutracker") + return [] + else: + consoleLog("No response from server, returning nothing") + return [] + + def get_magnet(self, post: Dict): + _, post_links, _, _, _ = format_data([post]) + return get_magnet_link(post_links[0]) + diff --git a/src/data/sources/steamrip.py b/src/data/sources/steamrip.py new file mode 100644 index 0000000..71de40e --- /dev/null +++ b/src/data/sources/steamrip.py @@ -0,0 +1,107 @@ +from data.hosts.buzzheavier import scrape_buzzheavier +from data.hosts.gofile import scrape_gofile +from utils.logging.logs import consoleLog +from bs4 import BeautifulSoup +from typing import Dict +import webbrowser +import requests +import time +import re + +class SteamripScraper: + name = "steamrip" + headers = ["Game"] + is_magnet = False + + + def __init__(self): + self.cache = { "data": [], "last_fetched": 0 } + self.cache_expiry = 300 + + def scrape_steamrip_links(self): + + current_time = time.time() + + if self.cache["data"] != [] and (current_time - self.cache["last_fetched"] < self.cache_expiry): + return self.cache["data"] + + url = f"https://steamrip.com/games-list-page/" + response = requests.get(url) + text = response.text + + soup = BeautifulSoup(text, "html.parser") + games = soup.find_all("li", class_="az-list-item") + + if len(games) == 0: + return [] + + links = [] + names = [] + for gamehtml in games: + link = gamehtml.find("a", href=lambda x: x and x.startswith("/")) + links += re.findall(r'(?<=href=")[^"]*', link.__str__()) + name = gamehtml.find("a", href=lambda x: x and x.startswith("/")) + names += re.findall(r'(?<=\/">)[^<]*', name.__str__()) + + # construct the list[dict[str,str]] + + ret = [] + + for i in range(len(names)): + ret.append({"title" : names[i], "url" : links[i]}) + + return ret + + def scrape_steamrip_game_downloads(self, gamelink): + url = "https://steamrip.com" + gamelink + response = requests.get(url) + soup = BeautifulSoup(response.text, "html.parser") + download_link_elements = soup.find_all("a",class_="shortc-button") + download_links = ["buzzheavier", "gofile"] + + for download_link in download_link_elements: + pure = download_link.attrs.get("href") + if pure[2] == "b": + download_links[0] = "https:" + pure + if pure[2] == "g": + download_links[1] = "https:" + pure + + ret = [] + + for link in download_links: + if len(link) != 1: + ret.append(link) + + self.cache["data"] = ret + + return ret + + def get_download_link(self, post: Dict): + url = post["url"] + links = self.scrape_steamrip_game_downloads(url) + best = "" + for link in links: + if link[0] == "h": + best = link + break + + if links.index(best) == 0: + link = scrape_buzzheavier(best) + return link + elif links.index(best) == 1: + link, headers = scrape_gofile(best) + return link, headers + else: + consoleLog("Unable to retrieve download link due to captcha, launching browser...") + webbrowser.open(best) + return None, None + + def search(self, query: str): + games = self.scrape_steamrip_links() + + filtered_games = [ + game for game in games + if query.lower() in game["title"].lower() + ] + + return filtered_games \ No newline at end of file diff --git a/src/data/sources/uztracker.py b/src/data/sources/uztracker.py new file mode 100644 index 0000000..de5930d --- /dev/null +++ b/src/data/sources/uztracker.py @@ -0,0 +1,50 @@ +from utils.data.tracker import get_magnet_link +from utils.logging.logs import consoleLog +from utils.data.state import state +from urllib.parse import urljoin +from bs4 import BeautifulSoup +from typing import Dict +import requests + +class UztrackerScraper: + name = "uztracker" + headers = ["Post Title", "Author"] + is_magnet = True + + + def search(self, query): + base_url="https://uztracker.net/" + search_url = f"{base_url.rstrip('/')}/tracker.php?nm={query}" + consoleLog(search_url) + posts = [] + + try: + response = requests.get(search_url) + response.raise_for_status() + soup = BeautifulSoup(response.text, 'html.parser') + links = soup.find_all('tr', class_="tCenter hl-tr", id=lambda x: x and x.startswith('tor_')) + for link in links: + + theme_link = link.find('a', class_="genmed tLink", href=lambda x: x and x.startswith('./viewtopic')) + if not theme_link or not theme_link.b: + continue + url = urljoin(base_url, theme_link['href']) + title = theme_link.b.text + author_link = link.find('a', class_="med") + author = author_link.text.strip() if author_link else "Unknown" + + posts.append(dict( + title=title, + author=author, + url=url, + )) + + return posts + + except requests.RequestException as e: + consoleLog(f"Failed to fetch {search_url}: {e}") + return None + + def get_magnet(post: Dict): + return get_magnet_link(post["url"]) + diff --git a/src/interface/utils/searchhelper.py b/src/interface/utils/searchhelper.py index ab006b0..24f687c 100644 --- a/src/interface/utils/searchhelper.py +++ b/src/interface/utils/searchhelper.py @@ -1,16 +1,9 @@ -from data.scrapers.rutracker import init_rutracker -from data.scrapers.uztracker import init_uztracker -from data.scrapers.steamrip import init_steamrip -from data.scrapers.monkrus import init_m0nkrus from utils.logging.logs import consoleLog from utils.data.state import state +from data.sources import SCRAPERS -# Call initialization function for each tracker / site -init_rutracker() -init_uztracker() -init_m0nkrus() -init_steamrip() - +for scraper in SCRAPERS: + state.trackers[scraper.name] = scraper def run_search(self) -> None: self.show_empty_results(False) @@ -23,12 +16,8 @@ def run_search(self) -> None: consoleLog(f"User searched for: {search_text}") # Get current tracker and call its search function - tracker = state.trackers[state.currenttracker] - scrapefunc = tracker["scrapeFunc"] - state.posts = scrapefunc(search_text) + scraper = state.trackers[state.currenttracker] + state.posts = scraper.search(search_text) - # Late to prevent circular import - from interface.gui import MainWindow - - # Update GUI Headers - MainWindow._instance.search_results_signal.emit(tracker["headers"]) + # Update GUI headers to match the current scraper + self.search_results_signal.emit(scraper.headers) \ No newline at end of file diff --git a/src/network/__init__.py b/src/network/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/utils/__init__.py b/src/utils/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/utils/data/state.py b/src/utils/data/state.py index de42ee9..75c3e79 100644 --- a/src/utils/data/state.py +++ b/src/utils/data/state.py @@ -34,13 +34,7 @@ class AppState(QObject): self.posts: list[Dict[str,str]] | None = None # titles, urls, author, seeders, leechers self.currenttracker: str = "rutracker" self.api_url: str = "https://api.michijackson.xyz" - self.trackers: Dict[str,Dict[str,Any]] = {} # each tracker should add itself here - # an example: - # "rutracker" : { - # "name" : "rutracker", - # "headers" : ["author", "title"], - # "scrapeFunc" : function, - # } + self.trackers: Dict[str,Dict[str,Any]] = {} # LibTorrent / Download related stuff self.dl_session: Any = None