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__":