diff --git a/FmhyChecker.pyw b/FmhyChecker.pyw index 4c377ce..26d5f76 100644 --- a/FmhyChecker.pyw +++ b/FmhyChecker.pyw @@ -1,3 +1,4 @@ +import gevent import grequests import requests from fake_headers import Headers @@ -22,49 +23,69 @@ from queue import Queue # fake headers headers = Headers(headers=True) # use queues to keep track of connection pools -dist_cnxns = Queue(maxsize=50) +dist_cnxns = Queue(maxsize=40) def resource_path(relative_path): + # wrapper to retrieve the absolute path from a relative path try: base_path = sys._MEIPASS except Exception: base_path = os.path.dirname(__file__) return os.path.join(base_path, relative_path) + def handle_req(url, item, callback): - dist_cnxns.put(1) + # process the request & send back to main event loop item.setText(2, 'Testing...') try: resp = requests.head(url, headers=headers.generate(), timeout=10, allow_redirects=True) - if resp is None: resp = 'Failed' except ReadTimeout: - callback(url, 'Timeout', item) + callback(url, None, item, 'Timeout') except ConnectionError: - callback(url, 'Error', item) + callback(url, None, item, 'Error') except Exception as e: - callback(url, str(e), item) + callback(url, None, item, str(e).split('\n')[0]) else: - if resp.status_code != 200: - resp = resp.reason or 'Unknown' - callback(url, resp, item) + send_resp(url, resp, item, callback) dist_cnxns.get() -def async_request(*args): +def send_resp(url, resp, item, callback): + # if resp completely failed + if resp is None: + return callback(url, resp, item, 'Failed') + # if response was not OK + if resp.status_code != 200: + return callback(url, resp, item, resp.reason.capitalize() or 'Unknown') + # response was success + return callback(url, resp, item, None) + + +def add_request(*args): + dist_cnxns.put(1) # wait for when <40 requests are running. blocks if full thread = Thread(target=handle_req, args=args, daemon=True) + # spawn thread to handle request thread.start() +def async_request(*args): + # spawn greenlets to wait their turn in the queue + gevent.spawn(add_request, *args) + gevent.sleep() # workaround for gevent bug + class UI(QMainWindow): - group_url_regex = re.compile(r'((?:https?|ftp|file):\/\/(?:ww(?:w|\d+)\.)?)((?:[\w_-]+(?:\.[\w_-]+)+)[\w.,@?^=%&:\/~+#-]*[\w@?^=%&~+-])') - call_back_checkLinks = pyqtSignal() - http_test_sig = pyqtSignal(str, object, object) + checkLinks_callback = pyqtSignal() + http_test_sig = pyqtSignal(str, object, object, str) + # regex for slicing links into groups ( ) + # i only check if group 1 is in the wiki to determine if the link is unique, then add the full link + grouped_wiki_regex = re.compile(r'((?:https?|ftp|file):\/\/(?:ww(?:w|\d+)\.)?)((?:[\w_-]+(?:\.[\w_-]+)+)[\w.,@?^=%&:\/~+#-]*[\w@?^=%&~+-])') def __init__(self): super(UI, self).__init__() uic.loadUi(resource_path('MainWindow.ui'), self) self.setWindowIcon(QtGui.QIcon(resource_path('assets\\icon.ico'))) + # palette coloring if darkdetect.isDark(): self._highlight_col = QtGui.QColor(157, 93, 24) dark_palette() @@ -73,12 +94,13 @@ class UI(QMainWindow): else: self._highlight_col = QtGui.QColor(255, 128, 0) + # set up main window (hide elements, set column widths, etc.) self.checkSelected.setVisible(False) self.progressBarFrame.setVisible(False) self.outputTree.header().setSectionsMovable(False) self.outputTree.setColumnWidth(0, 50) self.outputTree.setColumnWidth(1, 180) - + # allow multiple selection self.outputTree.setSelectionMode(QtWidgets.QAbstractItemView.ExtendedSelection) # status code font @@ -106,11 +128,11 @@ class UI(QMainWindow): self.copyValid.clicked.connect(lambda: copy('\n'.join(self.valid_links))) self.copyTested.clicked.connect(lambda: copy('\n'.join(self.getTestedLinks()))) self.exportCsv.clicked.connect(self.exportCsvDialog) - self.checkSelected.clicked.connect(self._testSelectedLinks) + self.checkSelected.clicked.connect(self.test_selected_links) self.inputBox.textChanged.connect(self.checkLinks) self.outputTree.itemSelectionChanged.connect(self.onSelection) self.http_test_sig.connect(self.finishTest) - self.call_back_checkLinks.connect(self.checkLinks) + self.checkLinks_callback.connect(self.checkLinks) self.testing_items = set() self.tested_items = {} @@ -123,6 +145,7 @@ class UI(QMainWindow): splash.hide() self.show() + # csv exporting def exportCsvDialog(self): file_dialog = QtWidgets.QFileDialog() file_dialog.setAcceptMode(QtWidgets.QFileDialog.AcceptSave) @@ -132,97 +155,124 @@ class UI(QMainWindow): if not file_dialog.exec_(): return file_path = file_dialog.selectedFiles()[0] - links = re.findall(self.group_url_regex, self.inputBox.toPlainText()) + links = re.findall(self.grouped_wiki_regex, self.inputBox.toPlainText()) try: with open(file_path, 'w', newline='') as csvfile: - writer = csv.writer(csvfile, dialect='excel', quoting=csv.QUOTE_MINIMAL) - writer.writerow(['Request URL', 'Final URL', 'Unique?', '# Redirects', 'Status', 'Reason']) - for link in links: - full_link = ''.join(link) - if full_link in self.tested_items: - if type(self.tested_items[full_link]) is str: - redirects = status_code = '' - reason = self.tested_items[full_link] - final_url = '' - else: - redirects = str(len(self.tested_items[full_link].history)) - reason = self.tested_items[full_link].reason - status_code = '=CONCAT('+', " > ", '.join( - f'HYPERLINK("{r.url}", "{r.status_code}")' - for r in ( - *self.tested_items[full_link].history, - self.tested_items[full_link]) - )+')' - final_url = self.tested_items[full_link].url - else: - reason = redirects = final_url = status_code = '' - writer.writerow([ - re.sub(r'[^\x00-\x7F]+', '?', full_link), # remove non ascii characters - final_url, - redirects, - 'FALSE' if link[1] in wiki else 'TRUE', - redirects, - status_code, - reason - ]) + self.writeCsvFile(csvfile, links) except PermissionError: QtWidgets.QMessageBox.critical(self.centralwidget, "Error", "File permission denied.") - def getTestedLinks(self): + def writeCsvFile(self, csvfile, links): + # write links to csv file + writer = csv.writer(csvfile, dialect='excel', quoting=csv.QUOTE_MINIMAL) + # csv header + writer.writerow(['Request URL', 'Final URL', 'Unique?', '# Redirects', 'Status', 'Reason']) + for link in links: + full_link = ''.join(link) + if full_link in self.tested_items: + # if the response was a message + if type(self.tested_items[full_link]) is str: + # set values to blank, and reason to message + final_url = redirects = status_code = '' + reason = self.tested_items[full_link] + else: + # get the number of redirects + redirects = str(len(self.tested_items[full_link].history)) + # get the reason for the status code + reason = self.tested_items[full_link].reason + # hyperlink status codes to the final url. chain together redirects with ' > ' + status_code = '=CONCAT('+', " > ", '.join( + f'HYPERLINK("{r.url}", "{r.status_code}")' + for r in ( + *self.tested_items[full_link].history, + self.tested_items[full_link]) + )+')' + final_url = self.tested_items[full_link].url + else: + # if the link was not tested, set values to blank + reason = redirects = final_url = status_code = '' + # write row to csv + writer.writerow([ + re.sub(r'[^\x00-\x7F]+', '?', full_link), # remove non ascii characters + final_url, + redirects, + 'FALSE' if link[1] in wiki else 'TRUE', + redirects, + status_code, + reason + ]) + + # return a list of unique links that have 200 status codes + def getTestedLinks(self) -> list: return [ l for l in self.tested_items + # if link is valid and not a message if l in self.valid_links and type(self.tested_items[l]) is not str + # and status code was OK and self.tested_items[l].status_code in range(200, 300) ] - def finishTest(self, url, resp, item): + # add the satus code chain to the tree + def finishTest(self, url, resp, item, message=None): + # remove from testing items, and add the resp to tested items if url in self.testing_items: self.testing_items.remove(url) - self.tested_items[url] = resp + self.tested_items[url] = resp or message self.copyTested.setEnabled(True) + # check if the tree item was deleted try: item.text(2) except RuntimeError: return # item was deleted + # create a new horizontal layout for the status code chain widget = QtWidgets.QWidget() widget.setLayout(layout := QtWidgets.QHBoxLayout()) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(2) layout.setAlignment(QtCore.Qt.AlignLeft) + # add layout to the tree item self.outputTree.setItemWidget(item, 2, widget) - item.setText(2, "") - if type(resp) is str: - label = QtWidgets.QLabel(f' {resp} ') - label.setFont(self.status_font) - color = self.reason_colors.get(resp, '#A12729') - label.setStyleSheet(f'background-color: {color}; color: white; border-radius: 6px;') - layout.addWidget(label) - return + item.setText(2, "") # remove the loading text + # if a message was passed, only add it to the layout + if message: + color = self.reason_colors.get(message, '#A12729') + return self.add_status_label(layout, message, color) + # else add the status code chain for r in (*resp.history, resp): - label = QtWidgets.QLabel(f" {r.status_code} ") color = next((self.status_colors[k] for k in self.status_colors if r.status_code in k), '#000000') - label.setStyleSheet(f'background-color: {color}; color: white; border-radius: 6px;') - label.setToolTip(r.url) - label.setToolTipDuration(-1) - label.setFont(self.status_font) - layout.addWidget(label) + self.add_status_label(layout, r.status_code, color, r.url) - def _testSelectedLinks(self): + def add_status_label(self, layout, text, color, tooltip=None): + # add label to the layout + label = QtWidgets.QLabel(f' {text} ') + label.setStyleSheet(f'background-color: {color}; color: white; border-radius: 6px;') + label.setFont(self.status_font) + if tooltip: + label.setToolTip(tooltip) + label.setToolTipDuration(-1) + layout.addWidget(label) + + def test_selected_links(self): + # get selected tree items selected = self.getRanItems() self.testing_items.update([i.text(1) for i in selected]) # remember tested items + # set to "Queued" for item in selected: item.setText(2, "Queued") self.outputTree.clearSelection() self.checkSelected.setVisible(False) + # send requests for item in selected: async_request(item.text(1), item, self.http_test_sig.emit) QtWidgets.QApplication.processEvents() # allow GUI to update - def getRanItems(self): + def getRanItems(self) -> list: + # return a list of selected items that have not been tested return [i for i in self.outputTree.selectedItems() if i.text(1) not in {*self.tested_items, *self.testing_items}] def onSelection(self): + # when an untested tree item is selected, show the "Test" button if selected := self.getRanItems(): self.checkSelected.setText(QtCore.QCoreApplication.translate("MainWindow", f"Test ({len(selected)}) \U0001F50D")) self.checkSelected.setVisible(True) @@ -230,11 +280,13 @@ class UI(QMainWindow): self.checkSelected.setVisible(False) def _waitForEvent(self): + # if checkLinks was already running, wait for it to complete then call it back self._new_event = True self._is_free.wait() - self.call_back_checkLinks.emit() + self.checkLinks_callback.emit() def checkLinks(self): + # only allow one instance of checkLinks to run if not self._is_free.is_set(): if self.line_thread and self.line_thread.is_alive(): return @@ -242,19 +294,21 @@ class UI(QMainWindow): self.line_thread.start() return self._is_free.clear() + # get the text from the input box text = self.inputBox.toPlainText() if text: self.inputBox.setPlaceholderText('') else: self.inputBox.setPlaceholderText(self._placeholderText) self.outputTree.clear() + # disable buttons while running self.copyValid.setEnabled(False) self.copyDupes.setEnabled(False) self.copyTested.setEnabled(False) self.exportCsv.setEnabled(False) self.checkSelected.setVisible(False) - - links = re.findall(self.group_url_regex, text) + # get all links from the input text using regex + links = re.findall(self.grouped_wiki_regex, text) self.valid_links, self.duped_links, self.tested_links = [], [], [] # enable progress bar self.progressBar.setMaximum(len(links)) @@ -267,24 +321,30 @@ class UI(QMainWindow): self._new_event = False self._is_free.set() return + # create new tree item item = QtWidgets.QTreeWidgetItem(self.outputTree) full_link = ''.join(link) item.setText(1, full_link) + # process in chunks to allow for UI updates if len(links) > 100 and not n % 100: - # process in chunks to allow for UI updates self.progressBar.setValue(n) QtWidgets.QApplication.processEvents() + # add item to tree with suppress(RuntimeError): if full_link in self.tested_items: + # if the link was already tested, use the previous result self.finishTest(full_link, self.tested_items[full_link], item) elif full_link in self.testing_items: + # if the link is currently being tested, indicate "Testing..." item.setText(2, "Testing...") if link[1] in wiki: + # if link is in the wiki item.setText(0, "\u274C") for _ in range(3): item.setBackground(_, self._highlight_col) self.duped_links.append(full_link) else: + # if link is NOT in the wiki item.setText(0, "\u2705") self.valid_links.append(full_link) # show finished tree @@ -299,9 +359,9 @@ class UI(QMainWindow): self._is_free.set() def retranslateUi(self): - # Set text (with translations) + # set text (with translations) _translate = QtCore.QCoreApplication.translate - self.setWindowTitle(_translate("MainWindow", "Dupe Checker v1.14")) + self.setWindowTitle(_translate("MainWindow", "Dupe Checker v1.15")) self.label.setText(_translate("MainWindow", "FMHY Dupe Tester")) self.label_2.setText(_translate("MainWindow", "by cevoj")) self._placeholderText = _translate("MainWindow", "Paste a list of links here...") @@ -332,6 +392,7 @@ def dark_title_bar(hwnd): def dark_palette(): + # create darker palette with Fusion style app.setStyle('Fusion') palette = QtGui.QPalette() palette.setColor(QtGui.QPalette.Window, QtGui.QColor(25,35,45)) @@ -351,13 +412,57 @@ def dark_palette(): app.setPalette(palette) -def build_wiki_set(): - # scrape wiki - elapsed = time.perf_counter() - try: - resps = grequests.map([grequests.get(l) for l in URLS], size=len(URLS)) - except ConnectionError: - # show connection error +class WikiScraper: + # regex for scraping links from wikis (links must include leading http(s)://) + wiki_regex = re.compile(r'(?:https?|ftp|file):\/\/(?:ww(?:w|\d+)\.)?((?:[\w_-]+(?:\.[\w_-]+)+)[\w.,@?^=%&:\/~+#-]*[\w@?^=%&~+-])') + # regex for scraping lists of urls (for links not including http(s)://) + list_regex = re.compile(r'^[\w]*\.[\w.,@?^=%&:\/~+#-]*[\w@?^=%&~+-]', re.MULTILINE) + # regex for scraping base64 encoded links + b64_regex = re.compile(r'`aHR0(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/][AQgw]==|[A-Za-z0-9+/]{2}[AEIMQUYcgkosw048]=)?`') + + ''' + FOR ANYONE IN THE FUTURE TRYING TO MAINTAIN THIS + you can add new urls to scrape by adding new key-value pairs to the dict below + : + ''' + def __init__(self): + self.URLS = { + 'https://raw.githubusercontent.com/nbats/FMHYedit/main/single-page': (self.handle_wiki, self.handle_b64), + 'https://gitlab.com/nbatman_/deleted-links/-/raw/main/deleted-links': (self.handle_list,), + } + + def build_wiki_set(self) -> set: + elapsed = time.perf_counter() + try: + resps = grequests.map([grequests.get(l) for l in self.URLS], size=len(self.URLS)) + except ConnectionError: + self.error_msg() # show connection error + wiki = set() + for resp, funcs in zip(resps, self.URLS.values()): + for func in funcs: + wiki.update(func(resp.text)) + + print(f'Wiki scraped in {time.perf_counter() - elapsed:0.4f} sec. Found {len(wiki)} links.') + return wiki + + @staticmethod + def from_regex(regex, text) -> set: + return set(re.findall(regex, text)) + + def handle_wiki(self, text) -> set: + return self.from_regex(self.wiki_regex, text) + + def handle_list(self, text) -> set: + return self.from_regex(self.list_regex, text) + + def handle_b64(self, text) -> set: + data = '\n'.join( + b64decode(m.strip('`')).decode() + for m in re.findall(self.b64_regex, text) + ) + return self.handle_list(data) + + def error_msg(self): msg = QtWidgets.QMessageBox() msg.setIcon(QtWidgets.QMessageBox.Critical) msg.setText("Could not connect to the internet. Please check your connection and try again.") @@ -365,42 +470,6 @@ def build_wiki_set(): splash.hide() msg.exec_() exit(1) - wiki = set() - for resp, exps in zip(resps, URLS.values()): - for exp in exps: - if exp is b64_regex: - wiki.update(handle_b64(resp.text)) - else: - wiki.update(set(re.findall(exp, resp.text))) - - print(f'Wiki scraped in {time.perf_counter() - elapsed:0.4f} sec. Found {len(wiki)} links.') - return wiki - - -def handle_b64(content): - data = '\n'.join( - b64decode(m.strip('`')).decode() - for m in re.findall(b64_regex, content) - ) - return set(re.findall(url_regex, data)) - - -# regex for scraping links from wikis (links must include leading http(s)://) -url_regex = re.compile(r'(?:https?|ftp|file):\/\/(?:ww(?:w|\d+)\.)?((?:[\w_-]+(?:\.[\w_-]+)+)[\w.,@?^=%&:\/~+#-]*[\w@?^=%&~+-])') -# regex for scraping lists of urls (for links not including http(s)://) -list_regex = re.compile(r'^[\w]*\.[\w.,@?^=%&:\/~+#-]*[\w@?^=%&~+-]', re.MULTILINE) -# regex for scraping base64 encoded links -b64_regex = re.compile(r'`aHR0(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/][AQgw]==|[A-Za-z0-9+/]{2}[AEIMQUYcgkosw048]=)?`') - -''' -FOR ANYONE IN THE FUTURE TRYING TO MAINTAIN THIS -you can add a wikis to scrape by adding new key-value pairs to the dict below -the key is the url of the wiki, and the value is a tuple of regexes to use for scraping -''' -URLS = { - 'https://gitlab.com/nbatman_/deleted-links/-/raw/main/deleted-links': (list_regex,), - 'https://raw.githubusercontent.com/nbats/FMHYedit/main/single-page': (url_regex, b64_regex), -} if __name__ == "__main__": @@ -411,8 +480,7 @@ if __name__ == "__main__": splash_icon = QtGui.QPixmap(resource_path('assets/splash.svg')) splash = QtWidgets.QSplashScreen(splash_icon, QtCore.Qt.WindowStaysOnTopHint) splash.show() - wiki = build_wiki_set() - + wiki = WikiScraper().build_wiki_set() fonts_dir = resource_path('assets/fonts') for f in os.listdir(fonts_dir):