mirror of
https://github.com/KeksPirates/SoftwareManager.git
synced 2026-08-04 01:49:42 +02:00
Compare commits
10 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d17b4bd44a | |||
| 10c935e852 | |||
| bd8320a26c | |||
| c99ce339e0 | |||
| 1822a150d7 | |||
| 7e5e3033e6 | |||
| 1cab671a13 | |||
| 5637e80657 | |||
| 2f78b812ba | |||
| 5af9e79ecc |
@@ -0,0 +1,41 @@
|
|||||||
|
from bs4 import BeautifulSoup
|
||||||
|
import requests
|
||||||
|
|
||||||
|
def scrape_monkrus_telegram(query):
|
||||||
|
post = 100
|
||||||
|
max_posts = 250
|
||||||
|
posts: list[dict] = []
|
||||||
|
added = set()
|
||||||
|
|
||||||
|
while post <= max_posts:
|
||||||
|
url = f"https://t.me/s/real_monkrus/{post}"
|
||||||
|
response = requests.get(url)
|
||||||
|
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 not post_txt or not post_txt.b:
|
||||||
|
continue
|
||||||
|
|
||||||
|
title = post_txt.b.text
|
||||||
|
link = bubble.find("a", href=lambda x: x and x.startswith("https://uztracker.net"))
|
||||||
|
|
||||||
|
if query.lower() in title.lower() and link:
|
||||||
|
post_url = link["href"]
|
||||||
|
if post_url not in added:
|
||||||
|
added.add(post_url)
|
||||||
|
posts.append(dict(
|
||||||
|
author="m0nkrus",
|
||||||
|
id=len(posts) + 1,
|
||||||
|
title=title,
|
||||||
|
url=post_url
|
||||||
|
))
|
||||||
|
|
||||||
|
post += 22
|
||||||
|
|
||||||
|
posts.reverse()
|
||||||
|
return(posts)
|
||||||
@@ -28,54 +28,45 @@ async def init_uztracker():
|
|||||||
|
|
||||||
asyncio.run(init_uztracker())
|
asyncio.run(init_uztracker())
|
||||||
|
|
||||||
async def scrape_uztracker(search, max_results=450):
|
def scrape_uztracker(search):
|
||||||
if up:
|
if up:
|
||||||
|
search_url = url_uztracker + search
|
||||||
|
if state.debug:
|
||||||
|
print(search_url)
|
||||||
result = False
|
result = False
|
||||||
global results
|
global results
|
||||||
global resulttitles
|
global resulttitles
|
||||||
results = []
|
results = []
|
||||||
resulttitles = []
|
resulttitles = []
|
||||||
|
|
||||||
per_page = 50
|
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
|
||||||
|
|
||||||
for start in range(0, max_results, per_page):
|
if result:
|
||||||
search_url = url_uztracker + search + f"&start={start}"
|
return resulttitles, results
|
||||||
if state.debug:
|
|
||||||
print(search_url)
|
|
||||||
|
|
||||||
async with aiohttp.ClientSession() as session:
|
if not result:
|
||||||
try:
|
# add popup in gui for no result
|
||||||
async with session.get(search_url) as response:
|
return None
|
||||||
response.raise_for_status()
|
|
||||||
text = await response.text()
|
|
||||||
soup = BeautifulSoup(text, 'html.parser')
|
|
||||||
links = soup.find_all('a', class_="genmed tLink", href=lambda x: x and x.startswith('./viewtopic'))
|
|
||||||
resultCount = 0
|
|
||||||
for link in links:
|
|
||||||
if link.b:
|
|
||||||
resultCount += 1
|
|
||||||
results.append(link['href'])
|
|
||||||
resulttitles.append(link.b.text)
|
|
||||||
result = True
|
|
||||||
|
|
||||||
if resultCount == 0:
|
except requests.RequestException as e:
|
||||||
# add popup in gui for no result
|
if result and state.debug:
|
||||||
break
|
print(f"Failed to fetch {search_url}: {e}")
|
||||||
|
return None
|
||||||
except aiohttp.ClientError as e:
|
else:
|
||||||
if result and state.debug:
|
if state.debug:
|
||||||
print(f"Failed to fetch {search_url}: {e}")
|
print("Error: Uztracker down")
|
||||||
return None
|
return None, None
|
||||||
|
|
||||||
else:
|
|
||||||
if state.debug:
|
|
||||||
print("Error: Uztracker down")
|
|
||||||
return None, None
|
|
||||||
|
|
||||||
if not results:
|
|
||||||
return None, None
|
|
||||||
|
|
||||||
return resulttitles, results
|
|
||||||
|
|
||||||
def get_post_title(post_url):
|
def get_post_title(post_url):
|
||||||
if up:
|
if up:
|
||||||
|
|||||||
@@ -139,7 +139,7 @@ class MainWindow(QtWidgets.QMainWindow, QWidget):
|
|||||||
toolbar.setLayoutDirection(Qt.RightToLeft)
|
toolbar.setLayoutDirection(Qt.RightToLeft)
|
||||||
|
|
||||||
self.tracker_list = QComboBox()
|
self.tracker_list = QComboBox()
|
||||||
self.tracker_list.addItems(["rutracker", "uztracker"])
|
self.tracker_list.addItems(["rutracker", "uztracker", "m0nkrus"])
|
||||||
self.tracker_list.activated.connect(self.set_tracker)
|
self.tracker_list.activated.connect(self.set_tracker)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1,33 +1,45 @@
|
|||||||
import asyncio
|
|
||||||
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.utils.network.jsonhandler import split_data, format_data
|
from core.utils.network.jsonhandler import split_data, format_data
|
||||||
from core.utils.data.state import state
|
from core.utils.data.state import state
|
||||||
|
|
||||||
def return_pressed(self):
|
def return_pressed(self):
|
||||||
search_text = self.searchbar.text()
|
search_text = self.searchbar.text()
|
||||||
if search_text == "":
|
if search_text == "":
|
||||||
if state.debug:
|
|
||||||
print("Error: Can't search for nothing")
|
|
||||||
return
|
|
||||||
if state.debug:
|
if state.debug:
|
||||||
print("User searched for:", search_text)
|
print("Error: Can't search for nothing")
|
||||||
if state.tracker == "uztracker":
|
return
|
||||||
response = asyncio.run(scrape_uztracker(search_text))
|
if state.debug:
|
||||||
if state.tracker == "rutracker":
|
print("User searched for:", search_text)
|
||||||
response = scrape_rutracker(search_text)
|
|
||||||
|
if state.tracker == "uztracker":
|
||||||
|
response = scrape_uztracker(search_text)
|
||||||
if response:
|
if response:
|
||||||
if state.tracker == "uztracker":
|
state.post_titles, state.post_urls = response
|
||||||
state.post_titles, state.post_urls, = response
|
self.softwareList.clear()
|
||||||
|
if state.post_titles and len(state.post_titles) > 0:
|
||||||
|
self.softwareList.addItems(state.post_titles)
|
||||||
|
self.show_empty_results(False)
|
||||||
|
self.post_author_list.clear()
|
||||||
|
else:
|
||||||
|
if state.debug:
|
||||||
|
print(f"No Results for {search_text}")
|
||||||
self.softwareList.clear()
|
self.softwareList.clear()
|
||||||
if state.post_titles:
|
self.show_empty_results(True)
|
||||||
self.softwareList.addItems(state.post_titles)
|
else:
|
||||||
self.post_author_list.addItems(state.post_author)
|
if state.debug:
|
||||||
if state.tracker == "rutracker":
|
print(f"No response from uztracker")
|
||||||
_, state.posts, _, _, cached = split_data(response)
|
self.softwareList.clear()
|
||||||
|
self.show_empty_results(True)
|
||||||
|
|
||||||
|
elif state.tracker == "rutracker":
|
||||||
|
response = scrape_rutracker(search_text)
|
||||||
|
if response:
|
||||||
|
_, state.posts, _, _, cached = split_data(response)
|
||||||
if state.posts == []:
|
if state.posts == []:
|
||||||
if state.debug:
|
if state.debug:
|
||||||
print(f"No Results found for \"{search_text}\"")
|
print(f"No Results for {search_text}")
|
||||||
self.softwareList.clear()
|
self.softwareList.clear()
|
||||||
self.show_empty_results(True)
|
self.show_empty_results(True)
|
||||||
else:
|
else:
|
||||||
@@ -39,4 +51,24 @@ def return_pressed(self):
|
|||||||
self.softwareList.addItems(state.post_titles)
|
self.softwareList.addItems(state.post_titles)
|
||||||
if state.debug == True:
|
if state.debug == True:
|
||||||
print(f"Response Cached: {cached}")
|
print(f"Response Cached: {cached}")
|
||||||
|
else:
|
||||||
|
if state.debug:
|
||||||
|
print(f"No response from rutracker")
|
||||||
|
self.softwareList.clear()
|
||||||
|
self.show_empty_results(True)
|
||||||
|
|
||||||
|
elif state.tracker == "m0nkrus":
|
||||||
|
state.posts = scrape_monkrus_telegram(search_text)
|
||||||
|
|
||||||
|
if state.posts == []:
|
||||||
|
if state.debug:
|
||||||
|
print(f"No Results for {search_text}")
|
||||||
|
self.softwareList.clear()
|
||||||
|
self.show_empty_results(True)
|
||||||
|
else:
|
||||||
|
state.post_titles, _, state.post_author = format_data(state.posts)
|
||||||
|
self.show_empty_results(False)
|
||||||
|
self.post_author_list.clear()
|
||||||
|
self.post_author_list.addItems(state.post_author)
|
||||||
|
self.softwareList.clear()
|
||||||
|
self.softwareList.addItems(state.post_titles)
|
||||||
|
|||||||
@@ -19,6 +19,12 @@ def get_item_url(item, posts, post_titles): # softwarelist currentitem, post lis
|
|||||||
if state.debug:
|
if state.debug:
|
||||||
print("Found post URL: ", post_links[0])
|
print("Found post URL: ", post_links[0])
|
||||||
return post_links[0]
|
return post_links[0]
|
||||||
|
if state.tracker == "m0nkrus":
|
||||||
|
item_dict = posts[post_index]
|
||||||
|
_, post_links, _ = format_data([item_dict])
|
||||||
|
if state.debug:
|
||||||
|
print("Found post URL: ", post_links[0])
|
||||||
|
return post_links[0]
|
||||||
|
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user