"""Dialog and worker thread for batch downloading CSGOSkins patterns.""" from __future__ import annotations import re import time import urllib.request import urllib.error from pathlib import Path from PySide6 import QtCore, QtWidgets, QtGui from app.i18n import I18nMixin import concurrent.futures class PatternDownloadWorker(QtCore.QThread): progress = QtCore.Signal(int, int) # current, total status = QtCore.Signal(str) # textual update finished = QtCore.Signal(bool) # True if Success, False if Interrupted/Error error = QtCore.Signal(str) # Error message def __init__(self, slug: str, save_dir: Path, parent: QtCore.QObject | None = None) -> None: super().__init__(parent) self.slug = slug self.save_dir = save_dir self.total_seeds = 1000 def _download_seed(self, seed: int) -> tuple[bool, str | None]: url = f"https://cdn.csgoskins.gg/public/images/patterns/v1/{self.slug}/{seed}.png" filename = self.save_dir / f"{seed}.png" if filename.exists(): return True, None try: req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0 ICRA/1.0'}) with urllib.request.urlopen(req, timeout=10) as response: with open(filename, 'wb') as f: f.write(response.read()) return True, None except urllib.error.HTTPError as e: return False, f"HTTP {e.code}" except Exception as e: return False, f"Network error: {e}" def run(self) -> None: self.save_dir.mkdir(parents=True, exist_ok=True) completed = 0 # Validate seed 1 synchronously first to avoid spawning 1000 threads for invalid slugs success, error_msg = self._download_seed(1) if not success and error_msg in ("HTTP 403", "HTTP 404"): self.error.emit(f"Failed to fetch seed 1. Does '{self.slug}' have patterns?") self.finished.emit(False) return completed += 1 self.progress.emit(completed, self.total_seeds) # Download the rest concurrently with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor: future_to_seed = {executor.submit(self._download_seed, seed): seed for seed in range(2, self.total_seeds + 1)} for future in concurrent.futures.as_completed(future_to_seed): if self.isInterruptionRequested(): self.status.emit("Download cancelled. Waiting for threads to finish...") executor.shutdown(wait=False, cancel_futures=True) self.finished.emit(False) return seed = future_to_seed[future] completed += 1 success, error = future.result() if success: self.status.emit(f"Downloaded seed {seed}/{self.total_seeds}") else: self.status.emit(f"Skipped seed {seed} ({error})") self.progress.emit(completed, self.total_seeds) self.status.emit("Download complete!") self.finished.emit(True) class PatternPullerDialog(QtWidgets.QDialog, I18nMixin): """Dialog for extracting patterns from CSGOSkins.gg URLs.""" def __init__(self, language: str, parent: QtWidgets.QWidget | None = None) -> None: super().__init__(parent) self.init_i18n(language) self.setWindowTitle(self._t("dialog.puller_title", default="Pull Pattern Images")) self.setMinimumWidth(450) self._worker: PatternDownloadWorker | None = None self._build_ui() def _build_ui(self) -> None: layout = QtWidgets.QVBoxLayout(self) layout.setSpacing(12) instruction_label = QtWidgets.QLabel(self._t("dialog.puller_instruction", default="Paste a CSGOSkins.gg item URL:")) layout.addWidget(instruction_label) self.url_input = QtWidgets.QLineEdit() self.url_input.setPlaceholderText("https://csgoskins.gg/items/glock-18-trace-lock") layout.addWidget(self.url_input) self.status_label = QtWidgets.QLabel("") self.status_label.setStyleSheet("color: palette(window-text); font-style: italic;") layout.addWidget(self.status_label) self.progress_bar = QtWidgets.QProgressBar() self.progress_bar.setRange(0, 1000) self.progress_bar.setValue(0) layout.addWidget(self.progress_bar) button_layout = QtWidgets.QHBoxLayout() self.start_btn = QtWidgets.QPushButton(self._t("dialog.puller_start", default="Start Download")) self.cancel_btn = QtWidgets.QPushButton(self._t("dialog.puller_cancel", default="Cancel")) self.start_btn.clicked.connect(self._on_start_clicked) self.cancel_btn.clicked.connect(self._on_cancel_clicked) button_layout.addWidget(self.start_btn) button_layout.addWidget(self.cancel_btn) layout.addLayout(button_layout) def _extract_slug(self, url: str) -> str | None: # Match https://csgoskins.gg/items/SLUG match = re.search(r"csgoskins\.gg/items/([^/?#]+)", url) if match: return match.group(1).lower() return None def _on_start_clicked(self) -> None: url = self.url_input.text().strip() slug = self._extract_slug(url) if not slug: QtWidgets.QMessageBox.warning(self, "Error", self._t("dialog.puller_invalid_url", default="Invalid URL format.")) return save_dir = Path("images") / slug self.start_btn.setEnabled(False) self.url_input.setEnabled(False) self.progress_bar.setValue(0) self.status_label.setText("Starting download...") self._worker = PatternDownloadWorker(slug=slug, save_dir=save_dir, parent=self) self._worker.progress.connect(self._on_progress) self._worker.status.connect(self.status_label.setText) self._worker.error.connect(self._on_error) self._worker.finished.connect(self._on_finished) self._worker.start() def _on_cancel_clicked(self) -> None: if self._worker and self._worker.isRunning(): self._worker.requestInterruption() self.cancel_btn.setEnabled(False) self.status_label.setText("Cancelling...") else: self.reject() # Close dialog if not downloading def _on_progress(self, current: int, total: int) -> None: self.progress_bar.setValue(current) def _on_error(self, message: str) -> None: QtWidgets.QMessageBox.warning(self, "Error", message) def _on_finished(self, success: bool) -> None: self.start_btn.setEnabled(True) self.url_input.setEnabled(True) self.cancel_btn.setEnabled(True) self._worker = None if success: QtWidgets.QMessageBox.information(self, "Done", self._t("dialog.puller_success", default="All patterns downloaded successfully!"))