mirror of
https://github.com/fmhy/dupe-checker.git
synced 2026-03-11 08:55:39 +00:00
Use third-party api to test sites, improve performance, bump to 1.16
This commit is contained in:
cevoj35
parent
50a7b14b93
commit
e94f8528bd
1 changed files with 129 additions and 66 deletions
195
FmhyChecker.pyw
195
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 ( <protocol://> <domain/path> <?leading info> )
|
||||
# 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__":
|
||||
|
|
|
|||
Loading…
Reference in a new issue