From e94f8528bd6891ef92e70156b7ae25bc7fff53a8 Mon Sep 17 00:00:00 2001 From: cevoj35 Date: Sat, 27 May 2023 04:53:21 +0000 Subject: [PATCH] Use third-party api to test sites, improve performance, bump to 1.16 --- FmhyChecker.pyw | 195 ++++++++++++++++++++++++++++++++---------------- 1 file changed, 129 insertions(+), 66 deletions(-) diff --git a/FmhyChecker.pyw b/FmhyChecker.pyw index ade967b..591af59 100644 --- a/FmhyChecker.pyw +++ b/FmhyChecker.pyw @@ -17,12 +17,15 @@ import darkdetect from base64 import b64decode import ctypes as ct from queue import Queue +from dataclasses import dataclass +from http.client import responses as status_codes +from typing import Union # fake headers -headers = Headers(headers=True) -# use queues to keep track of connection pools -dist_cnxns = Queue(maxsize=40) +headers = Headers() +# use Queue to limit number of concurrent requests +dist_cnxns = Queue(maxsize=3) def resource_path(relative_path): @@ -34,41 +37,87 @@ def resource_path(relative_path): return os.path.join(base_path, relative_path) -def handle_req(url, item, callback): - # process the request & send back to main event loop - dist_cnxns.put(1) # wait for when <40 requests are running. blocks if full - item.setText(2, 'Testing...') - try: - resp = requests.head(url, headers=headers.generate(), timeout=10, allow_redirects=True) - except ReadTimeout: - callback(url, None, item, 'Timeout') - except ConnectionError: - callback(url, None, item, 'Error') - except Exception as e: - callback(url, None, item, str(e).split('\n')[0]) - else: - send_resp(url, resp, item, callback) - dist_cnxns.get() +@dataclass +class StatusResp: + url: str + status_code: Union[int, str] + reason: str + history: list -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 async_request(*args): - thread = Thread(target=handle_req, args=args, daemon=True) - # spawn thread to handle request - thread.start() +class LinkTest: + chunk_size = 50 # number of links to test at once + statusapi_url = b64decode('aHR0cHM6Ly9iYWNrZW5kLmh0dHBzdGF0dXMuaW8vYXBp').decode() + statusapi_headers = { + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:108.0) Gecko/20100101 Firefox/108.0', + 'Accept': 'application/json, text/plain, */*', + 'Accept-Language': 'en-US,en;q=0.5', + # 'Accept-Encoding': 'gzip, deflate, br', + 'Referer': b64decode('aHR0cHM6Ly9odHRwc3RhdHVzLmlvLw==').decode(), + 'Content-Type': 'application/json;charset=utf-8', + 'Origin': b64decode('aHR0cHM6Ly9odHRwc3RhdHVzLmlv').decode(), + 'DNT': '1', + 'Connection': 'keep-alive', + 'Sec-Fetch-Dest': 'empty', + 'Sec-Fetch-Mode': 'cors', + 'Sec-Fetch-Site': 'same-site', + } + statusapi_data = { + 'urls': None, + 'userAgent': 'browser', + 'userName': '', 'passWord': '', 'headerName': '', 'headerValue': '', + 'strictSSL': True, + 'canonicalDomain': False, + 'additionalSubdomains': ['www',], + 'followRedirect': True, + 'throttleRequests': 100, + 'escapeCharacters': False, + } + + @staticmethod + def handle_req(urls, items, callback): + # process the request & send back to main event loop + dist_cnxns.put(1) # wait for when <40 requests are running. blocks if full + for item in items: + item.setText(2, 'Testing...') + try: + resp = requests.post( + LinkTest.statusapi_url, + headers={**LinkTest.statusapi_headers, **headers.generate()}, + json={**LinkTest.statusapi_data, 'urls': urls}, + ) + data = resp.json() + except (ReadTimeout, ConnectionError): + error_msg('Connection timed out. Please check your internet connection and try again.') + except Exception as e: + error_msg(f'An unknown error occurred. Please try again.\n\n{e}') + for (item, url, resp) in zip(items, urls, data): + try: + callback(item, LinkTest.build_status_resp(resp, url)) + except Exception as e: + print('Error:', e, resp) + dist_cnxns.get() # release next in queue + + @staticmethod + def build_status_resp(resp, url=None) -> StatusResp: + return StatusResp( + url = url or resp.get('url', 'Failed'), + status_code = resp['statusCode'] if type(resp.get('statusCode')) is int else 0, + reason = resp.get('errorMessage') or status_codes[resp['statusCode']], + history = [LinkTest.build_status_resp(r) + for r in resp.get('fullRedirectChain', [])], + ) + + @staticmethod + def async_request(*args): + thread = Thread(target=LinkTest.handle_req, args=args, daemon=True) + # spawn thread to handle request + thread.start() class UI(QMainWindow): checkLinks_callback = pyqtSignal() - http_test_sig = pyqtSignal(str, object, object, str) + http_test_sig = pyqtSignal(object, StatusResp) # 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@?^=%&~+-])') @@ -77,7 +126,7 @@ class UI(QMainWindow): super(UI, self).__init__() uic.loadUi(resource_path('MainWindow.ui'), self) - self.setWindowIcon(QtGui.QIcon(resource_path('assets\\icon.ico'))) + self.setWindowIcon(QtGui.QIcon(resource_path('assets/icon.ico'))) # palette coloring if darkdetect.isDark(): self._highlight_col = QtGui.QColor(157, 93, 24) @@ -176,9 +225,7 @@ class UI(QMainWindow): # 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]) + for r in self.tested_items[full_link].history )+')' final_url = self.tested_items[full_link].url else: @@ -206,11 +253,11 @@ class UI(QMainWindow): ] # add the satus code chain to the tree - def finishTest(self, url, resp, item, message=None): + def finishTest(self, item, resp): # 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 or message + if resp.url in self.testing_items: + self.testing_items.remove(resp.url) + self.tested_items[resp.url] = resp self.copyTested.setEnabled(True) # check if the tree item was deleted try: @@ -226,21 +273,34 @@ class UI(QMainWindow): # add layout to the tree item self.outputTree.setItemWidget(item, 2, widget) item.setText(2, "") # remove the loading text - # if a message was passed, only add it to the layout - if type(resp) is str: - message = resp - 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): - color = next((self.status_colors[k] for k in self.status_colors if r.status_code in k), '#000000') - self.add_status_label(layout, r.status_code, color, r.url) + # add the status code chain + for r in resp.history or (resp,): + color = next((self.status_colors[k] for k in self.status_colors if r.status_code in k), '#781C1E') + text = str(r.status_code) if r.status_code else 'Error' + text2 = r.reason + tooltip = f'{r.reason} | {r.url}' + self.add_status_label(layout, text, text2, color, tooltip) - def add_status_label(self, layout, text, color, tooltip=None): + def add_status_label(self, layout, text, text2, 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 = QtWidgets.QLabel(text) + # create lighter color for hover + light_color = QtGui.QColor(color).lighter().name() + label.setStyleSheet(f''' + * {{ + background-color: {color}; + color: white; + border-radius: 6px; + padding: 0px 2px; + }} + QLabel:hover:!pressed {{ + border: 2px solid {light_color}; + }} + ''') + label.setMouseTracking(True) + # change text to text2 on hover + label.enterEvent = lambda e: label.setText(text2) + label.leaveEvent = lambda e: label.setText(text) label.setFont(self.status_font) if tooltip: label.setToolTip(tooltip) @@ -257,9 +317,11 @@ class UI(QMainWindow): self.outputTree.clearSelection() self.checkSelected.setVisible(False) # send requests - for item in selected: + for index in range(0, len(selected), LinkTest.chunk_size): + items = selected[index:index+LinkTest.chunk_size] + urls = [item.text(1) for item in items] try: - async_request(item.text(1), item, self.http_test_sig.emit) + LinkTest.async_request(urls, items, self.http_test_sig.emit) except RuntimeError: return # item was deleted QtWidgets.QApplication.processEvents() # allow GUI to update @@ -331,7 +393,7 @@ class UI(QMainWindow): 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) + self.finishTest(item, self.tested_items[full_link]) elif full_link in self.testing_items: # if the link is currently being tested, indicate "Testing..." item.setText(2, "Testing...") @@ -359,7 +421,7 @@ class UI(QMainWindow): def retranslateUi(self): # set text (with translations) _translate = QtCore.QCoreApplication.translate - self.setWindowTitle(_translate("MainWindow", "Dupe Checker v1.15.2")) + self.setWindowTitle(_translate("MainWindow", "Dupe Checker v1.16")) 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...") @@ -434,7 +496,8 @@ class WikiScraper: try: resps = grequests.map([grequests.get(l) for l in self.URLS], size=len(self.URLS)) except ConnectionError: - self.error_msg() # show connection error + # show connection error + self.error_msg("Could not connect to the internet. Please check your connection and try again.") wiki = set() for resp, funcs in zip(resps, self.URLS.values()): for func in funcs: @@ -460,14 +523,14 @@ class WikiScraper: ) 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.") - msg.setWindowTitle("Connection Error") - splash.hide() - msg.exec_() - exit(1) +def error_msg(text): + msg = QtWidgets.QMessageBox() + msg.setIcon(QtWidgets.QMessageBox.Critical) + msg.setText(text) + msg.setWindowTitle("Connection Error") + splash.hide() + msg.exec_() + exit(1) if __name__ == "__main__":