refactor: rename scraping functions for consistency and improve data handling

This commit is contained in:
Vxrtrauter
2026-03-04 19:11:31 +01:00
parent 90124fd9c8
commit 34a50e1a1b
5 changed files with 48 additions and 105 deletions
+1 -1
View File
@@ -57,7 +57,7 @@ def _get_telegram_posts():
return(posts) return(posts)
def scrape_monkrus_telegram(query): def scrape_m0nkrus(query):
posts = _get_telegram_posts() posts = _get_telegram_posts()
filtered_posts = [] filtered_posts = []
+2 -2
View File
@@ -3,8 +3,8 @@ from core.utils.data.state import state
from core.utils.logging.logs import consoleLog from core.utils.logging.logs import consoleLog
def scrape_rutracker(search_text): def scrape_rutracker(query):
search = requests.get(f"{state.api_url}/search?q={search_text}") search = requests.get(f"{state.api_url}/search?q={query}")
consoleLog("Sent request to server") consoleLog("Sent request to server")
if search: if search:
try: try:
+30 -70
View File
@@ -1,83 +1,43 @@
import requests import requests
import threading import threading
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
from urllib.parse import urljoin
from core.utils.data.state import state from core.utils.data.state import state
from core.utils.logging.logs import consoleLog from core.utils.logging.logs import consoleLog
from core.utils.general.wrappers import run_thread from core.utils.general.wrappers import run_thread
def scrape_uztracker(query):
base_url="https://uztracker.net/"
search_url = f"{base_url.rstrip('/')}/tracker.php?nm={query}"
consoleLog(search_url)
posts = []
global url_uztracker
url_uztracker = "https://uztracker.net/tracker.php?nm="
def init_uztracker():
global up
global soup
try: try:
response = requests.get(url_uztracker, timeout=10) response = requests.get(search_url)
if response.status_code == 200: response.raise_for_status()
up = True
else:
up = False
consoleLog(f"Uztracker seems down, status code: {response.status_code}")
except requests.exceptions.RequestException as e:
consoleLog(f"Request Exception on {url_uztracker}:")
consoleLog(str(e))
consoleLog("Is the Site down?")
up = False
run_thread(threading.Thread(target=init_uztracker))
def scrape_uztracker(search):
if up:
search_url = url_uztracker + search
consoleLog(search_url)
result = False
global results
global resulttitles
results = []
resulttitles = []
try:
resultCount = 0
response = requests.get(search_url)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
links = soup.find_all('a', class_="genmed tLink", href=lambda x: x and x.startswith('./viewtopic'))
for link in links:
if link.b:
resultCount += 1
results.append(link['href'])
resulttitles.append(link.b.text)
result = True
if result:
return resulttitles, results
if not result:
# add popup in gui for no result
return None
except requests.RequestException as e:
if result:
consoleLog(f"Failed to fetch {search_url}: {e}")
return None
else:
consoleLog("Error: Uztracker down")
return None, None
def get_post_title(post_url):
if up:
response = requests.get(post_url)
soup = BeautifulSoup(response.text, 'html.parser') soup = BeautifulSoup(response.text, 'html.parser')
maintitle = soup.find(class_='tt-text') links = soup.find_all('a', class_="genmed tLink", href=lambda x: x and x.startswith('./viewtopic'))
if maintitle:
return maintitle.text for link in links:
else: if link.b:
consoleLog("Program not found") title = link.b.text
else: url = urljoin(base_url, link['href'])
consoleLog("Error: Uztracker down")
return None author_link = link.find_parent('td').find('a', class_="med ts-text")
author = author_link.text.strip() if author_link else "Unknown"
posts.append(dict(
title=title,
url=url,
author=author
))
return posts
except requests.RequestException as e:
if posts:
consoleLog(f"Failed to fetch {search_url}: {e}")
return None
+13 -30
View File
@@ -2,10 +2,16 @@ from PySide6.QtWidgets import QTableWidgetItem
from core.utils.logging.logs import consoleLog from core.utils.logging.logs import consoleLog
from core.data.scrapers.uztracker import scrape_uztracker from core.data.scrapers.uztracker import scrape_uztracker
from core.data.scrapers.rutracker import scrape_rutracker from core.data.scrapers.rutracker import scrape_rutracker
from core.data.scrapers.monkrus import scrape_monkrus_telegram from core.data.scrapers.monkrus import scrape_m0nkrus
from core.utils.network.jsonhandler import split_data, format_data, format_data_m0nkrus from core.utils.network.jsonhandler import split_data, format_data, format_data_minimal
from core.utils.data.state import state from core.utils.data.state import state
scrapers = {
"uztracker": scrape_uztracker,
"rutracker": scrape_rutracker,
"m0nkrus": scrape_m0nkrus
}
def return_pressed(self): def return_pressed(self):
search_text = self.searchbar.text() search_text = self.searchbar.text()
if search_text == "": if search_text == "":
@@ -13,29 +19,7 @@ def return_pressed(self):
return return
consoleLog(f"User searched for: {search_text}") consoleLog(f"User searched for: {search_text}")
if state.tracker == "uztracker": if state.tracker == "rutracker":
response = scrape_uztracker(search_text)
if response:
state.post_titles, state.post_urls = response
state.tracker_list[state.tracker].clear()
if state.post_titles and len(state.post_titles) > 0:
state.tracker_list[state.tracker].setHorizontalHeaderLabels(["Post Title", "Author"])
state.tracker_list[state.tracker].setRowCount(len(state.post_titles))
for i, author in enumerate(state.post_author):
state.tracker_list[state.tracker].setItem(i, 1, QTableWidgetItem(author))
for i, title in enumerate(state.post_titles):
state.tracker_list[state.tracker].setItem(i, 0, QTableWidgetItem(title))
self.show_empty_results(False)
else:
consoleLog(f"No Results for {search_text}")
state.tracker_list[state.tracker].clear()
self.show_empty_results(True)
else:
consoleLog(f"No response from uztracker")
state.tracker_list[state.tracker].clear()
self.show_empty_results(True)
elif state.tracker == "rutracker":
response = scrape_rutracker(search_text) response = scrape_rutracker(search_text)
if response: if response:
_, state.posts, _, _, cached = split_data(response) _, state.posts, _, _, cached = split_data(response)
@@ -64,15 +48,14 @@ def return_pressed(self):
state.tracker_list[state.tracker].clear() state.tracker_list[state.tracker].clear()
self.show_empty_results(True) self.show_empty_results(True)
elif state.tracker == "m0nkrus": elif state.tracker is not None:
state.posts = scrape_monkrus_telegram(search_text) state.posts = scrapers[state.tracker](search_text)
if state.posts == []: if state.posts == []:
consoleLog(f"No Results for {search_text}") consoleLog(f"No Results for {search_text}")
state.tracker_list[state.tracker].clear() state.tracker_list[state.tracker].clear()
self.show_empty_results(True) self.show_empty_results(True)
else: else:
state.post_titles, _, state.post_author = format_data_m0nkrus(state.posts) state.post_author, state.post_titles, state.post_urls = format_data_minimal(state.posts)
self.show_empty_results(False) self.show_empty_results(False)
state.tracker_list[state.tracker].clear() state.tracker_list[state.tracker].clear()
state.tracker_list[state.tracker].setHorizontalHeaderLabels(["Post Title", "Author"]) state.tracker_list[state.tracker].setHorizontalHeaderLabels(["Post Title", "Author"])
@@ -80,4 +63,4 @@ def return_pressed(self):
for i, author in enumerate(state.post_author): for i, author in enumerate(state.post_author):
state.tracker_list[state.tracker].setItem(i, 1, QTableWidgetItem(author)) state.tracker_list[state.tracker].setItem(i, 1, QTableWidgetItem(author))
for i, title in enumerate(state.post_titles): for i, title in enumerate(state.post_titles):
state.tracker_list[state.tracker].setItem(i, 0, QTableWidgetItem(title)) state.tracker_list[state.tracker].setItem(i, 0, QTableWidgetItem(title))
+2 -2
View File
@@ -20,12 +20,12 @@ def format_data(data):
return post_titles, post_links, post_author, post_seeders, post_leechers return post_titles, post_links, post_author, post_seeders, post_leechers
def format_data_m0nkrus(data): def format_data_minimal(data):
post_author = [post["author"] for post in data] post_author = [post["author"] for post in data]
post_titles = [post["title"] for post in data] post_titles = [post["title"] for post in data]
post_links = [post["url"] for post in data] post_links = [post["url"] for post in data]
return post_titles, post_links, post_author return post_author, post_titles, post_links