from __future__ import annotations import os import json from functools import partial from pathlib import Path from html.parser import HTMLParser from typing import Optional, Callable from PySide6.QtWebEngineWidgets import QWebEngineView from PySide6.QtWebEngineCore import QWebEngineSettings, QWebEngineProfile, QWebEnginePage, QWebEngineDownloadRequest,\ QWebEngineScript from PySide6.QtCore import QUrl, Qt, QTimer, Signal from PySide6 import QtWidgets, QtCore from PySide6.QtWidgets import QApplication from GN.ContentLibrary.Backends.Fab.Data import FabData, data_to_cache, AllowedFabTypes, FAB_TYPE_TO_DASH_TYPE from GN.Standard.SafeQThread import SafeQThread import GN.Logger as Log logger = Log.get_logger(__name__) # logger.setLevel(Log.DEBUG) ROOT = Path(__file__).parent injection_scripts = [] try: injection_scripts = list(Path(ROOT).joinpath("injection").iterdir()) except: pass def html_to_fab_data(web_view: 'FabView', identifier: str, image_path: str, on_data_ready: callable = None, world_id: int = 0): def on_thumbnail_finished(fab_data: FabData, download_item: QWebEngineDownloadRequest): if not download_item.downloadFileName() in os.path.basename(image_path): print(f"Download file name {download_item.downloadFileName()}, not a thumbnail, skip this...") return # And run the callback if present if on_data_ready: try: on_data_ready(fab_data) logger.debug("on_data_ready callback called succesfully") except Exception as e: logger.error(f"Error in on_data_ready callback {e}") def on_html_ready(fab_data: str): if not fab_data: logger.debug("js result value was nothing") return logger.debug(f"js result value {fab_data}") # Convert the json data to fab data, very important as we don't pass around the dict json_dict: dict = json.loads(fab_data) image_url = json_dict["image"] fab_data = FabData.from_dict(json_dict) logger.info(f"Extracted data from fab: {fab_data}") # Store the data given from outside fab_data.identifier = identifier fab_data.image = image_path # Validate the asset type, store it if it's allowed or skip the data saved_type = fab_data.asset_type if not AllowedFabTypes.is_allowed(saved_type): logger.warning(f"Asset type extracted from fab '{saved_type}' is not allowed. Defaulting to 3D...") saved_type = "3D Models" fab_type = AllowedFabTypes(saved_type) dash_type = FAB_TYPE_TO_DASH_TYPE.get(fab_type) fab_data.asset_type = dash_type # Download the image, now that we know the data is supported, only when the download is done, call the callback! web_view.page().download(image_url, image_path) web_view.onDownloadFinished.connect(partial(on_thumbnail_finished, fab_data), QtCore.Qt.ConnectionType.SingleShotConnection) html_to_fab_data_js = ROOT.joinpath("js", "html_to_fab_data.js").read_text(encoding='utf-8') web_view.page().runJavaScript(html_to_fab_data_js, world_id, on_html_ready) class FabPageParser(HTMLParser): """ Detects if download popup is opened on current page """ def __init__(self): super().__init__() self._modal_dialog_opened = False self._download_page_visible = False self._h2_opened = False def is_download_panel_open(self): return self._download_page_visible def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> None: if tag == "div": for attr in attrs: if attr[0] == "class" and "fabkit-Modal-root" in attr[1]: self._modal_dialog_opened = True break if tag == "h2": self._h2_opened = True def handle_endtag(self, tag: str) -> None: if tag == "h2": self._h2_opened = False def handle_data(self, data: str) -> None: if self._modal_dialog_opened and self._h2_opened and "Download" in data: self._download_page_visible = True class FabPageThread(SafeQThread): onDownloadPageVisibilityChanged = Signal(bool) def __init__(self, parent=None): super().__init__(parent) self._visible_prev = False self._html_queue = [] def feed_html(self, html: str): # keep window small to not bloat RAM if len(self._html_queue) > 2: self._html_queue.pop(0) self._html_queue.append(html) def html_ready(self, html): parser = FabPageParser() parser.feed(html) visible = parser.is_download_panel_open() if visible != self._visible_prev: self._visible_prev = visible self.onDownloadPageVisibilityChanged.emit(visible) def run(self) -> None: while not self.isInterruptionRequested(): if self._html_queue: # take last html = self._html_queue.pop() self.html_ready(html) self.msleep(30) class WebEnginePage(QWebEnginePage): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._inject() def _inject(self): for script in injection_scripts: if script.suffix == ".js": _ = QWebEngineScript() _.setSourceUrl(QUrl.fromLocalFile(script.as_posix())) _.setName(script.stem) _.setWorldId(0) _.setInjectionPoint(QWebEngineScript.InjectionPoint.DocumentReady) self.scripts().insert(_) def acceptNavigationRequest(self, url: QUrl | str, n_type: QWebEnginePage.NavigationType, isMainFrame: bool) -> bool: accepted = super().acceptNavigationRequest(url, n_type, isMainFrame) if "/listings/" in url.toString(): if n_type == QWebEnginePage.NavigationType.NavigationTypeLinkClicked: app: QApplication = QApplication.instance() modifiers = app.queryKeyboardModifiers() if modifiers == Qt.KeyboardModifier.ControlModifier: def on_trigger_download(view: FabView): view.triggerDownload() parent = self.parent() if isinstance(parent, FabView): QTimer.singleShot(3000, partial(on_trigger_download, parent)) return accepted def createWindow(self, _type): page = WebEnginePage(self.profile(), self.parent()) page.urlChanged.connect(self.on_url_changed) return page def on_url_changed(self, url): page = self.sender() self.setUrl(url) page.deleteLater() def javaScriptConsoleMessage(self, level, msg, line, source_id): if logger.level == Log.DEBUG: print(f"JS Console [{level}]: {msg}") class FabView(QWebEngineView): onDownloadRequested = QtCore.Signal(QWebEngineDownloadRequest) onDownloadFinished = QtCore.Signal(QWebEngineDownloadRequest) onDownloadProgress = QtCore.Signal(QWebEngineDownloadRequest) def __init__(self, parent=None): ua = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) QtWebEngine/6.7.3 Chrome/118.0.5993.220 Safari/537.36' app = QtWidgets.QApplication.instance() profile = QWebEngineProfile("dash-fab-profile", app) profile.setHttpUserAgent(ua) profile.setCachePath(ROOT.joinpath("fab_cache").as_posix()) profile.setHttpCacheType(QWebEngineProfile.HttpCacheType.DiskHttpCache) profile.setPersistentCookiesPolicy(QWebEngineProfile.PersistentCookiesPolicy.ForcePersistentCookies) self._profile = profile super().__init__(profile, parent) self.setPage(WebEnginePage(self._profile, self)) # Enable page caching profile = self.page().profile() profile.setPersistentCookiesPolicy(QWebEngineProfile.PersistentCookiesPolicy.ForcePersistentCookies) settings = self.settings() settings.setAttribute(QWebEngineSettings.WebAttribute.LocalStorageEnabled, True) settings.setAttribute(QWebEngineSettings.WebAttribute.JavascriptCanAccessClipboard, True) settings.setAttribute(QWebEngineSettings.WebAttribute.JavascriptCanOpenWindows, True) settings.setAttribute(QWebEngineSettings.WebAttribute.LocalContentCanAccessRemoteUrls, True) settings.setAttribute(QWebEngineSettings.WebAttribute.LocalContentCanAccessFileUrls, False) settings.setAttribute(QWebEngineSettings.WebAttribute.ScrollAnimatorEnabled, False) settings.setAttribute(QWebEngineSettings.WebAttribute.PluginsEnabled, True) settings.setAttribute(QWebEngineSettings.WebAttribute.AllowRunningInsecureContent, True) settings.setAttribute(QWebEngineSettings.WebAttribute.AllowWindowActivationFromJavaScript, True) settings.setAttribute(QWebEngineSettings.WebAttribute.JavascriptCanPaste, True) settings.setAttribute(QWebEngineSettings.WebAttribute.ForceDarkMode, True) # settings.setAttribute(QWebEngineSettings.WebAttribute.ShowScrollBars, False) self.load("https://www.fab.com/") self.page().profile().downloadRequested.connect(self.handle_download) # self.page().loadFinished.connect(self.inject_javascript) self.download_button_removal_script = ROOT.joinpath("js", "remove_download_buttons_from_library.js").read_text(encoding='utf-8') self.remove_content_from_downloads_page = ROOT.joinpath("js", "remove_content_from_downloads_page.js").read_text(encoding='utf-8') self.trigger_download_script = ROOT.joinpath("js", "trigger_download.js").read_text(encoding='utf-8') self.close_filters_script = ROOT.joinpath("js", "click_on_close_filters_button.js").read_text(encoding='utf-8') self.open_the_search_input = ROOT.joinpath("js", "open_the_search_input.js").read_text(encoding='utf-8') self._poller = FabPageThread(self) self._poller.setObjectName("FabPageThread") self._poller.onDownloadPageVisibilityChanged.connect(self.onDownloadPageVisibilityChanged, QtCore.Qt.ConnectionType.QueuedConnection) self._poller.start() # periodically put current page html into thread self._fab_page_html_feed_timer = QTimer(self) self._fab_page_html_feed_timer.timeout.connect(self.timer_callback) self._fab_page_html_feed_timer.setInterval(150) self._fab_page_html_feed_timer.setSingleShot(False) self.destroyed.connect(self.on_destroyed) self.loadFinished.connect(self.on_view_url_changed) self._url_change_tasks = {} def open_search_input(self): return self.page().runJavaScript(self.open_the_search_input) def set_url_loaded_task(self, url: str, task: Callable[[QUrl], None]): """ Specify callable to be invoked when given url is fully loaded """ self._url_change_tasks[url] = task def on_view_url_changed(self, loaded: bool): if not loaded: return url = self.url() task = self._url_change_tasks.get(url.toString()) if task: task(url.toString()) self._url_change_tasks.pop(url.toString()) def deleteLater(self): self._poller.requestInterruption() self._poller.wait() self._poller.quit() self._poller.terminate() self._poller.deleteLater() self._fab_page_html_feed_timer.stop() self._fab_page_html_feed_timer.deleteLater() super().deleteLater() def on_destroyed(self, arg): print("KILLING FAB VIEW", arg, arg.objectName()) def closeFilters(self): self.page().runJavaScript(self.close_filters_script) def triggerDownload(self): self.page().runJavaScript(self.trigger_download_script) # def inject_javascript(self): # if self.page().url().toString().endswith("/library"): # logger.debug("Running my library filter logic...") # self.page().runJavaScript(self.download_button_removal_script) # # logger.debug("Running download button filter logic...") # self.page().runJavaScript(self.remove_content_from_downloads_page) def hideEvent(self, event): super().hideEvent(event) self._fab_page_html_feed_timer.stop() def showEvent(self, event): super().showEvent(event) self._fab_page_html_feed_timer.start() def onDownloadPageVisibilityChanged(self, visible): def on_html_ready(html: str): if visible: self.page().runJavaScript(self.remove_content_from_downloads_page) # read html self.page().toHtml(on_html_ready) def timer_callback(self): # try to delete download buttons from 'My Library' if self.page().url().toString().endswith("/library"): self.page().runJavaScript(self.download_button_removal_script) if not self._poller: return page = self.page() if not page: return if page.isLoading(): return if "listings" not in page.url().toString(): return page.toHtml(self._poller.feed_html) def handle_download(self, download_item: QWebEngineDownloadRequest): # Allow external handling of the download self.onDownloadRequested.emit(download_item) download_item.isFinishedChanged.connect(partial(self.onDownloadFinished.emit, download_item)) download_item.receivedBytesChanged.connect(partial(self.onDownloadProgress.emit, download_item)) class FabWidget(QtWidgets.QWidget): def __init__(self, parent=None): super().__init__(parent) layout = QtWidgets.QVBoxLayout() self.setLayout(layout) web_layout = QtWidgets.QHBoxLayout() layout.addLayout(web_layout) self.web_view = FabView(parent=self) web_layout.addWidget(self.web_view) # self.custom_scrollbar = QtWidgets.QScrollBar() # self.custom_scrollbar.setStyleSheet("QScrollBar:vertical { width: 10px; }") # web_layout.addWidget(self.custom_scrollbar) # # # Connect the page scrolling to the custom scrollbar # self.web_view.page().scrollPositionChanged.connect(self.on_scroll_position_changed) # self.custom_scrollbar.valueChanged.connect(self.on_scrollbar_value_changed) extract_meta_button = QtWidgets.QPushButton("Extract Meta") extract_meta_button.clicked.connect(self.extract_meta) layout.addWidget(extract_meta_button) self.web_view.setUrl(QUrl("https://www.fab.com/library")) # self.web_view.setUrl(QUrl("https://www.fab.com/plugins/ue5/")) def on_scroll_position_changed(self, pos): self.custom_scrollbar.setValue(pos.y()) def on_scrollbar_value_changed(self, value): self.web_view.page().runJavaScript(f"window.scrollTo(0, {value})") def extract_meta(self): def on_page_loaded(): print("Extracting meta...") identifier = self.web_view.url().toString().split("/")[-1] tmp_dir = QtCore.QStandardPaths.writableLocation(QtCore.QStandardPaths.StandardLocation.TempLocation) + "/" + identifier print("Temp dir:", tmp_dir) tmp_image_path = os.path.join(tmp_dir, "thumbnail.jpg") def on_data_ready(data): print("Data received..", data) data_to_cache(data, tmp_dir) html_to_fab_data(self.web_view, identifier, tmp_image_path, on_data_ready, world_id=1) # Wait for the page to load if self.web_view.page().isLoading(): print("Waiting for page to load...") self.web_view.page().loadFinished.connect(on_page_loaded, QtCore.Qt.ConnectionType.SingleShotConnection) else: on_page_loaded() if __name__ == "__main__": import sys # app = QtWidgets.QApplication(sys.argv) # ua = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) QtWebEngine/6.7.3 Chrome/118.0.5993.220 Safari/537.36' # profile = QWebEngineProfile("dash-fab-profile", app) # profile.setHttpUserAgent(ua) # profile.setCachePath(ROOT.joinpath("fab_cache").as_posix()) # profile.setHttpCacheType(QWebEngineProfile.HttpCacheType.DiskHttpCache) # profile.setPersistentCookiesPolicy(QWebEngineProfile.PersistentCookiesPolicy.ForcePersistentCookies) # # view = FabView(profile) # view.show() widget = FabWidget() # widget = FabView() # widget.setUrl(QUrl("https://www.fab.com/sellers/Quixel")) widget.show() # download_dir = "C:/TempStuff" # # def onItemDownloadRequested(download_item: QWebEngineDownloadRequest): # if ".zip" in download_item.downloadFileName(): # download_item.setDownloadDirectory(download_dir) # download_item.accept() # print("Download accepted:", download_item.downloadFileName(), download_item.downloadDirectory()) # else: # download_item.cancel() # print("Download cancelled:", download_item.downloadFileName(), download_item.downloadDirectory()) # # widget.web_view.onDownloadRequested.connect(onItemDownloadRequested) # # # background_extractor = DataExtractor() # # background_extractor.extract_from_url("https://www.fab.com/listings/969f4f30-5970-438a-99d3-3e76f427360b", "thumbnail.jpg") # # # sys.exit(app.exec())