178 lines
6.9 KiB
Python
178 lines
6.9 KiB
Python
"""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!"))
|