refactor: improve scraper handling by utilizing classes for registration

This commit is contained in:
Vxrtrauter
2026-03-20 02:22:40 +01:00
parent 325eda54fa
commit 2afb3e7716
17 changed files with 301 additions and 341 deletions
View File
View File
-87
View File
@@ -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})
-52
View File
@@ -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})
-122
View File
@@ -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})
-55
View File
@@ -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})
+12
View File
@@ -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(),
]
+78
View File
@@ -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"])
+46
View File
@@ -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])
+107
View File
@@ -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
+50
View File
@@ -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"])
+7 -18
View File
@@ -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)
View File
View File
+1 -7
View File
@@ -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