这是我的第一个Python脚本，我不太明白。如何使用同步函数调用而不冻结图形？ 请告诉我。 如果我使用异步睡眠，那么云火就会开始阻止我的请求。 这就是我使用同步睡眠的原因。
import logging
import os
import random
import re
import sys
import asyncio
from itertools import count
from time import sleep
import requests
from PyQt6.QtCore import Qt, QThreadPool, QRunnable, pyqtSlot
from PyQt6.QtGui import QColor, QFont, QTextCursor
from PyQt6.QtWidgets import * # PyQt6
from PyQt6 import QtCore, QtWidgets, QtGui
from PyQt6.Qt6 import *
from bs4 import BeautifulSoup
from const import listGLNotUse, rep, forum_home_page_file, fileNameLog, data_folder_atricle, url_main, \
fileName_last_filter_forum, file_name_forum_pages, dir_garbage, file_name_forum_page_torrent
from delelteTextBlock import deleteBlocks
from deleteDir import delete_file, deleteDirs
from main import replase_multiple, print_info, get_posts_forum_torrent, \
get_torrent_Link, torrent_download, get_soup_text, request_cf
class MainWindow(QWidget):
def get_directory_and_linkFromFilteStr(self):
soup = get_soup_text(forum_home_page_file) #получить soup главной страницы
line = re.split(';', self.filter)
gr1 = line[0]
gr2 = line[1]
gr3 = line[2]
skip_first_to = 'Трекер: Всё для детей и родителей'
#
is_skip_first_to = 1
torrent_gr_list = []
try:
if not os.path.exists(data_folder_atricle):
os.mkdir(data_folder_atricle)
posts_table_forumling = soup.find_all('a', {'class': ['forumlink', 'cattitle', 'gensmall tit-b']})
adres_cattitle = ["cattitle"] # 1
adres_forumlink = ["forumlink"] # 2
adres_gensmall = ['gensmall', 'tit-b'] # 3
count = 0
is_row_gr_1_list = 0
is_row_gr_2_list = 0
is_row_gr_3_list = 0
for item in posts_table_forumling:
if item.text != skip_first_to.strip() and is_skip_first_to == 1: # Удаляем мусор сначала
continue
if item.text == skip_first_to.strip():
is_skip_first_to = 0
if item.text not in listGLNotUse: # Убираем группы где нет глав с торрентами
text_url = replase_multiple(item.get_text(), rep)
if text_url == "": # Удаляем пустый урлы
# print('check blank continue', text_url)
continue
count += 1
# проверям есть ли текущая группа в фильтре для разрешения скачивания
if text_url == gr1:
is_row_gr_1_list = 1
else:
is_row_gr_1_list = 0
if text_url == gr2:
is_row_gr_2_list = 1
else:
is_row_gr_2_list = 0
if text_url == gr3:
is_row_gr_3_list = 1
else:
is_row_gr_3_list = 0
if item.get(
"class") == adres_cattitle and is_row_gr_1_list == 1: # работа с рутовыми главами. Уровень глобальный
name1 = "value1: ", item.get("class"), {count}, text_url, item.get("href")
print_info(f'1-filter: {text_url}-{name1}')
directory_root = f'{data_folder_atricle}''/'f'{text_url}'
if not os.path.exists(directory_root):
os.mkdir(directory_root)
if item.get(
"class") == adres_forumlink and is_row_gr_2_list == 1: # работа с группами главам. Уровень 2
name2 = "value2: ", item.get("class"), {count}, text_url, item.get("href")
print_info(f'2-filter: {text_url}-{name2}')
name_directory2 = directory_root + "/" + text_url
# print(name_directory2)
if not os.path.exists(name_directory2):
os.mkdir(name_directory2)
#
if item.get("class") == adres_gensmall and is_row_gr_3_list == 1: # работа с темами.Уровень 3
name3 = "value3: ", item.get("class"), {count}, text_url, item.get("href")
print_info(f'3-filter: {text_url}-{name3}')
name_directory3 = name_directory2 + "/" + text_url
torrent_gr_dc = {
'dir': name_directory3
, 'url': url_main + item.get("href")
, 'name': text_url
}
torrent_gr_list.append(torrent_gr_dc)
if not os.path.exists(name_directory3):
os.mkdir(name_directory3)
# print(item.get("class"))
self.list_gr1_gr2_gr3_url_directory = torrent_gr_list
if (len(torrent_gr_list) > 0):
with open(fileName_last_filter_forum, 'w', encoding='utf8') as file:
for line in torrent_gr_list:
file.write(f"{line}\n")
return torrent_gr_list
except Exception as ex:
print_info('1: ', ex)
def get_posts_forum_torrent(self):
list_gr1_gr2_gr3_url_directory = self.list_gr1_gr2_gr3_url_directory
print('1111')
print(list_gr1_gr2_gr3_url_directory)
# 1- скачать файлы страниц с торрентами
# 0- не скачивать
try:
is_create_file = 1
# скачиваются только 1 страницы с постами где есть торренты. пример https://nnmclub.to/forum/viewforum.php?f=731&start=1400#pagestart
if not os.path.exists(dir_garbage):
os.mkdir(dir_garbage)
count_page_first_group4 = 0
post_topis_list = []
post_1_page_list = []
for itemGroup in list_gr1_gr2_gr3_url_directory:
url = itemGroup.get('url')
# анализиуются первый страницы всех глав. Если указать лимит то будет анализироваться только часть
# if count_page_first_group4 >= 5: #лимит на закачку топиков всех первых страниц и торрентов и нет
# # (в некоторых затем будут лежать топикы с торрент файлами)
# break
#
# посмотреть по логам куда и зачем идут запросы.
# может их все сохранить в файл.
# потом ребята могут сами пробежаться по этим файлам и скачать себе торренты и не запариваться со cf
request_posts_with_torrents = request_cf(url, False)
count_page_first_group4 += 1
namefile = itemGroup.get('name')
item_text = request_posts_with_torrents.content.decode('cp1251') #
sleep_second = random.randrange(0, 1) # 5-15
print_info(f'Надо подождать. Смена page. Жду:{sleep_second}')
sleep(sleep_second)
text_clear = deleteBlocks(item_text, 'script')
post_clear_utf = deleteBlocks(text_clear, 'meta').encode()
#
if request_posts_with_torrents.status_code != 200:
print('Попробуйте помеенять VPN. Сервер вернул не 200: ' + str(
request_posts_with_torrents.status_code))
soup = get_soup_text(post_clear_utf)
# анализирую первую страницы - если есть посты с торрентами то сохраню их для дальнейшего получения страниц по данной группе с торрентами
topic_soup = soup.find_all('a', {'class': ['topictitle']})
#
if topic_soup == []:
# print(itemGroup, ' тут нет списка торрент-постов: ',topic_soup)
continue
#
if topic_soup != []: # есть посты-топики для торрентов
if is_create_file == 1:
namefile_temp1 = f'{dir_garbage}/{count_page_first_group4}' f'{namefile}.html'
with open(namefile_temp1, 'wb') as file:
print_info(
f'Тут есть торренты, поэтому ее сохраняю: {namefile_temp1}-{count_page_first_group4}')
file.write(post_clear_utf)
topic_soup_pages_list = soup.find_all('a', href=re.compile('pagestart')) # .find_-.find_all('span')
# print('size: ', len(topic_soup_pages_list))
# print(topic_soup_pages_list)
post_page_list_temp = [] # для очистки Page от След
is_page = len(topic_soup_pages_list)
# for item_topik_first_page in topic_soup_pages_list:
# print(item_topik_first_page)
if is_page > 0:
for item in topic_soup_pages_list:
url_href = item.get('href')
url_text = item.getText()
if url_text != "След.":
number = int(item.text.strip())
post_page_list_temp.append(number)
# print('t1: ', url_href, 't2:', url_text, 'item:',item.text)
# print_list(post_page_list_temp)
max_page = max(post_page_list_temp)
# print('max: 'f'{maxPage}')
count_for = max_page * 50
# Проверить что будет делать есть только 1 страницы с торрентами
# формирование только урлов на страницы всех pages
for i in range(0, count_for, 50):
if i == 0:
url_first = itemGroup.get('url')
# print(url_first)
post_topic_dc = {
'dir': itemGroup.get('dir')
, 'url_topic_group': itemGroup.get('url')
, 'url_topic': url_first
, 'i': i
}
post_topis_list.append(post_topic_dc)
continue
# print(i)
url_two_and_other = url_href.rsplit('&', 1)[0]
url = f'{url_main}{url_two_and_other}&start={i}#pagestart'
# rint(url)
post_topic_dc = {
'dir': itemGroup.get('dir')
, 'url_topic_group': itemGroup.get('url')
, 'url_topic': url
, 'i': i
}
post_topis_list.append(post_topic_dc)
# если страница с только с 1 page пример https://nnmclub.to/forum/viewforum.php?f=736 на 07.09.2024. 08:37:27
if is_page == 0:
post_topic_dc = {
'dir': itemGroup.get('dir')
, 'url_topic_group': itemGroup.get('url')
, 'url_topic': url
, 'i': 0
}
post_topis_list.append(post_topic_dc)
self.pages_torrents_link = post_topis_list
with open(file_name_forum_pages, 'w', encoding='utf8') as file:
for line in post_topis_list:
file.write(f"{line}\n")
return post_topis_list
except Exception as ex:
print_info('2: ', ex)
def get_torrent_Link(self):
pages_torrents_link = self.pages_torrents_link
# 1- скачать файлы страниц с торрентами
# 0- не скачивать
# for item in pages_torrents_link:
# print_info("get_torrent_Link. list:", item)
# скачиваются все страницы с пейджами 1 страницы с постами где есть торренты. пример https://nnmclub.to/forum/viewforum.php?f=731&start=1400#pagestart
if not os.path.exists(dir_garbage):
os.mkdir(dir_garbage)
try:
post_topis_download_torrent_list = []
post_url_torrent = []
count = 0
len_gtl = len(pages_torrents_link)
if len_gtl == 0:
print_info('Торренты в этой группе не найдены...')
else:
print_info(f'Страниц с торрентами: {len_gtl}-{pages_torrents_link}')
for itemGlobalDir in pages_torrents_link:
count += 1
count_random = 1 # random.randrange(1, 3)
# print('Сколько страниц с торрентами анализировать 'f'{count_random}')
# if count > count_random: # лимит запрос получения страниц с файлом торрентов
# break
# страница типа https://nnmclub.to/forum/viewforum.php?f=725 может быть и любая из page
requests_torrent_topik = request_cf(itemGlobalDir.get('url_topic'), False)
sleep_second = random.randrange(3, 10)
print_info(f'Надо подождать. Получаю ссылку на страницу torrent файлом. Жду:{sleep_second}')
sleep(sleep_second)
item_text = requests_torrent_topik.content.decode('cp1251') #
text_clear = deleteBlocks(item_text, 'script')
post_clear_utf = deleteBlocks(text_clear, 'meta').encode()
soup_pages_torrent = get_soup_text(post_clear_utf)
topic_soup = soup_pages_torrent.find_all('a', {'class': ['topictitle']})
# определяется имя поста где лежит торрент и урл на страницу с торрентом
count_page_torrent_get = 0
# print_info(f'itemGlobalDir: {itemGlobalDir}')
for item_post_url_name in topic_soup:
count_page_torrent_get += 1
len_ptsl = len(pages_torrents_link)
print_info(f'Page c torrents : {count} из {len_ptsl}')
print_info(f'Анализ торрентов на page: {item_post_url_name} itemGlobalDir:{itemGlobalDir}')
len_tps = len(topic_soup)
print_info(f'Анализ торрент файлов на page: {count_page_torrent_get} из {len_tps}')
# сколько страниц с торрентами получать для сохранения
# if count_page_torrent_get >=3:
# break
name_file = item_post_url_name.text
url = url_main + item_post_url_name.get('href')
print_info(f'{name_file}-{url}')
sleep_second = random.randrange(1, 10)
print_info(f'Получаю ссылку на download файла.torrent. Жду:{sleep_second}')
sleep(sleep_second)
request_posts_with_torrents = request_cf(url, False)
item_text_page_torrent = request_posts_with_torrents.content.decode('cp1251') #
text_clear_with_torrents = deleteBlocks(item_text_page_torrent, 'script')
post_clear_with_torrents_utf = deleteBlocks(text_clear_with_torrents, 'meta').encode()
# example post_clear_with_torrents_utf https://nnmclub.to/forum/viewtopic.php?t=1458717
soup_torrent_page = get_soup_text(post_clear_with_torrents_utf)
topic_soup_download = soup_torrent_page.find_all(href=re.compile('download.php'))
len_dw_torrent = len(topic_soup_download)
print_info(f'Найдено на торрент страница ссылок на скачку: {len_dw_torrent}')
if len_dw_torrent < 1:
break
url_torrent = url_main + topic_soup_download[0].get('href')
post_download_torrent_list_dc = {
'dir': itemGlobalDir.get('dir')
, 'name_topic': name_file + "_" + str(random.randrange(1, 9999999))
, 'url_torrentDownloader': url_torrent
}
post_url_torrent.append(url_torrent)
post_topis_download_torrent_list.append(post_download_torrent_list_dc)
for item in post_topis_download_torrent_list:
print_info(f'{item}')
if (len(post_topis_download_torrent_list) > 0):
with open(file_name_forum_page_torrent, 'w', encoding='utf8') as file:
print_info(f'Промежуточное сохранение get_torrent_Link. ссылка на фай торрент + место хранения')
for line in post_topis_download_torrent_list:
file.write(f"{line}\n")
if (len(post_url_torrent) > 0):
with open("ALL_post_url_torrent.txt", 'w', encoding='utf8') as file:
print_info(f'Сохраняю урл на скачку торрента файла в post_url_torrent')
for line in post_url_torrent:
file.write(f"{line}\n")
return post_topis_download_torrent_list
except Exception as ex:
print_info('3: ', ex)
# def torrent_download(torrent_link):
# count = 0
# try:
# if torrent_link is not None:
# size_list_toreent_file = len(torrent_link)
#
# for item in torrent_link:
# count += 1
# if torrent_link is None:
# continue
# sleep_second = random.randrange(3, 10)
# print_info(f'Перед скачкой torrent файла. Жду:{sleep_second}')
# sleep(sleep_second)
#
# url_torrentDownloader = item.get('url_torrentDownloader')
# dir = item.get('dir')
# print_info(f'{url_torrentDownloader} сюда{dir}')
# url = item.get('url_torrentDownloader')
# filename = replase_multiple(item.get('name_topic'), rep) + ".torrent"
# directory = item.get('dir')
# # urllib.request.urlretrieve(url, filename)
#
# try:
#
# print_info(f'Сохранение файла torrent: {count} из {size_list_toreent_file}')
# self.download(url, directory, filename)
# except Exception as _ex:
# print('4: ', _ex)
#
# link_down = len(torrent_link)
# print_info(f'Скачено торрентов: {link_down}')
#
# except Exception as _ex2:
# print_info('5: ', _ex2)
# def download(url: str, dest_folder: str, name_file: str):
# try:
# file_path = os.path.join(dest_folder, name_file)
# print_info(f'dest_folder: {dest_folder}')
# print_info(f'name_file: {name_file}')
# print_info(f'url: {url}')
#
# r = requests.get(url, stream=True)
# if r.ok:
# path_info = os.path.abspath(file_path)
# print_info(f'saving to: {path_info}')
#
# with open(file_path, 'wb') as f:
# for chunk in r.iter_content(chunk_size=1024 * 8):
# if chunk:
# f.write(chunk)
# f.flush()
# os.fsync(f.fileno())
# time.sleep(3) # одождать 3 секунды
# else: # HTTP status code 4XX/5XX
# print("Download failed: status code {}\n{}".format(r.status_code, r.text))
# except Exception as _ex:
# with open("torrent_downloadError.txt", 'r+', encoding='utf8') as file:
# print_info(f'Промежуточное сохранение файлов торрентов которые не скачались. torrent_downloadError')
# file.write(f"{url};{dest_folder};{name_file}\n")
# print_info('6: ', _ex)
def __init__(self):
super().__init__()
# self.soup = [{1},{2},{3}]
self.set_appear()
self.init_ui()
self.connections()
def set_appear(self):
self.setWindowTitle("Парсер nnmclub")
self.resize(1200, 600)
def connections(self):
self.start_task_button.clicked.connect(self.start_task)
def start_task(self):
for ix in self.task_tree.selectedIndexes():
text = ix.data(Qt.ItemDataRole.DisplayRole)
# or ix.data()
self.filter = text
if text != "Выберите раздел" :
str = f'Выбран раздел: {text}'
self.log_text_edit.append(str)
logging.info(str)
try:
worker = Worker(
parserFunction(self)
)
self.thread_pool.start(worker)
except:
self.log_text_edit.append(f"Произошла ошибка во время парсинга , проверьте правильность ввода данных.")
def init_ui(self):
logging.basicConfig(filename=fileNameLog, encoding='utf-8', level=logging.DEBUG) # включаем логирование
self.thread_pool = QThreadPool()
# Создание виджета вкладок
self.main_layout = QVBoxLayout()
self.tab_widget = QTabWidget()
# Создание вкладки "Группы"
self.tasks_tab = QWidget()
self.tasks_layout = QVBoxLayout()
self.tasks_tab.setLayout(self.tasks_layout)
# Создание "Иерархии групп" на вкладке "Группы"
self.task_tree = QTableWidget()
self.task_tree = QtWidgets.QTreeWidget()
self.task_tree.header().hide()
self.task_tree.resize(1000, 450)
item = QTreeWidgetItem()
item.setText(0, 'Выберите раздел')
soup = get_soup_text(forum_home_page_file)
list = get_forum_global_page(soup)
for itemlistPost in list:
self.task_tree.addTopLevelItem(item)
q = QTreeWidgetItem(item)
# name = 'itemlistPost'
name = itemlistPost.get('name1') +";"+ itemlistPost.get('name2')+";"+ itemlistPost.get('name3')
q.setText(0, name)
# Установка стилей для Иерархии групп и кнопки
self.tasks_layout.addWidget(self.task_tree)
self.log_text_edit = QTextEdit()
self.log_text_edit.setReadOnly(True)
# self.log_text_edit.moveCursor(QTextCursor.End)
self.tasks_layout.addWidget(self.log_text_edit)
# Создание кнопки "Начать загрузку"
self.start_task_button = QPushButton("Начать загрузку")
# self.del_task_button.setStyleSheet(self.del_btn_style)
self.buttons_layout = QHBoxLayout()
self.buttons_layout.addWidget(self.start_task_button)
self.tasks_layout.addLayout(self.buttons_layout)
# Добавление вкладки "Группы" в виджет вкладок
self.tab_widget.addTab(self.tasks_tab, "Группы")
# Создание вкладки "Настройки"
settings_tab = QWidget()
settings_layout = QVBoxLayout()
form_layout = QFormLayout()
# Добавление настроек
# Поле для выбора папки по умолчанию
folder_label = QLabel("Папка по умолчанию:")
self.folder_button = QPushButton("Выбрать папку")
form_layout.addRow(self.folder_button)
# settings_layout.addWidget(self.auto_start_checkbox_default)
settings_layout.addLayout(form_layout)
settings_tab.setLayout(settings_layout)
self.tab_widget.addTab(settings_tab, "Настройки")
self.main_layout.addWidget(self.tab_widget)
self.setLayout(self.main_layout)
def panelCreate(soup):
app = QApplication([])
app.setFont(QFont('Arial', 12))
aw = MainWindow()
aw.show()
app.exec()
# sys.exit(app.exec())
def parserFunction(self):
self.log_text_edit.append(f'parserFunction')
# получить фильтр
list_gr1_gr2_gr3_url_directory = self.get_directory_and_linkFromFilteStr()
print('000')
post_genmed_list = self.get_posts_forum_torrent()
#
post_download_link = get_torrent_Link(post_genmed_list)
# print(post_download_link)
class Worker(QRunnable):
def __init__(self, function, *args, **kwargs):
super().__init__()
self.function = function
self.args = args
self.kwargs = kwargs
@pyqtSlot()
def run(self):
try:
asyncio.run(self.function(*self.args, **self.kwargs))
except Exception as e:
print(e)
def get_forum_global_page(soup):
skip_first_to = 'Трекер: Всё для детей и родителей'
is_skip_first_to = 1
torrent_gr_list = []
try:
posts_table_forumling = soup.find_all('a', {'class': ['forumlink', 'cattitle', 'gensmall tit-b']})
adres_cattitle = ["cattitle"] # 1
adres_forumlink = ["forumlink"] # 2
adres_gensmall = ['gensmall', 'tit-b'] # 3
count = 0
for item in posts_table_forumling:
if item.text != skip_first_to.strip() and is_skip_first_to == 1: # Удаляем мусор сначала
continue
if item.text == skip_first_to.strip():
is_skip_first_to = 0
if item.text not in listGLNotUse: # Убираем группы где нет глав с торрентами
text_url = replase_multiple(item.get_text(), rep)
if text_url == "": # Удаляем пустый урлы
# print('check blank continue', text_url)
continue
count += 1
if item.get("class") == adres_cattitle: # работа с рутовыми главами. Уровень глобальный
name1 = text_url
if item.get("class") == adres_forumlink: # работа с группами главам. Уровень 2
name2 = text_url
if item.get("class") == adres_gensmall: # работа с темами.Уровень 3
torrent_gr_dc = {
'name1': name1
, 'name2': name2
, 'name3': text_url
}
torrent_gr_list.append(torrent_gr_dc)
return torrent_gr_list
except Exception as ex:
print(ex)
我使用了异步睡眠。我预计图形面板不会冻结。但问题是cloudfire添加的。
我首先编写一个控制台应用程序，但我想请支持该网站的人制作一个图形应用程序。我想也许这些函数应该与类方法完全相同。 （然后是工作线程）但是奇迹并没有发生
我尝试在单独的方法中创建一个流，但我无法在那里传递带有参数的请求函数......而且也没有任何效果。
您可以设置一个计时器，在指定的持续时间后到期以恢复执行，或者更确切地说，注册一个调用cack函数，该函数将在睡眠完成时执行