Add CS2 pattern fetcher subtool
Introduce a CS2 pattern download tool with toolbar entry, translations, and requests dependency. Document the workflow and requirements updates.
This commit is contained in:
@@ -0,0 +1,8 @@
|
||||
"""Auxiliary tooling for the ICRA application."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from .cs2_patterns import open_cs2_pattern_tool
|
||||
|
||||
__all__ = ["open_cs2_pattern_tool"]
|
||||
|
||||
@@ -0,0 +1,410 @@
|
||||
"""CS2 skin pattern fetching subtool."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable, Optional
|
||||
|
||||
import tkinter as tk
|
||||
from tkinter import filedialog, messagebox, ttk
|
||||
|
||||
import requests
|
||||
from urllib.parse import urlparse
|
||||
|
||||
|
||||
class CS2PatternFetcher:
|
||||
"""Fetch CS2 skin metadata and download pattern images."""
|
||||
|
||||
DATA_URL = "https://bymykel.github.io/CSGO-API/api/skins.json"
|
||||
|
||||
def __init__(self, cache_dir: Path | None = None):
|
||||
self.cache_dir = cache_dir or Path.home() / ".icra"
|
||||
self.cache_path = self.cache_dir / "cs2_skins.json"
|
||||
self._data: list[dict[str, Any]] | None = None
|
||||
self._session: requests.Session | None = None
|
||||
|
||||
def ensure_data(self, *, force_refresh: bool = False) -> list[dict[str, Any]]:
|
||||
if not force_refresh and self._data is not None:
|
||||
return self._data
|
||||
if not force_refresh:
|
||||
cached = self._load_cache()
|
||||
if cached is not None:
|
||||
self._data = cached
|
||||
return cached
|
||||
data = self._download()
|
||||
self._data = data
|
||||
self._write_cache(data)
|
||||
return data
|
||||
|
||||
def list_weapons(self) -> list[str]:
|
||||
data = self.ensure_data()
|
||||
weapons = {self._weapon_name(item) for item in data}
|
||||
return sorted(filter(None, weapons))
|
||||
|
||||
def list_patterns(self, weapon: str) -> list[str]:
|
||||
data = self.ensure_data()
|
||||
patterns = {
|
||||
self._pattern_name(item)
|
||||
for item in data
|
||||
if self._weapon_name(item) == weapon
|
||||
}
|
||||
return sorted(filter(None, patterns))
|
||||
|
||||
def find_entry(self, weapon: str, pattern: str) -> Optional[dict[str, Any]]:
|
||||
data = self.ensure_data()
|
||||
for item in data:
|
||||
if self._weapon_name(item) == weapon and self._pattern_name(item) == pattern:
|
||||
return item
|
||||
return None
|
||||
|
||||
def download_pattern_image(
|
||||
self,
|
||||
entry: dict[str, Any],
|
||||
target_dir: Path,
|
||||
) -> Path:
|
||||
image_url = self._image_url(entry)
|
||||
if not image_url:
|
||||
raise ValueError("Selected pattern does not provide an image URL.")
|
||||
|
||||
target_dir.mkdir(parents=True, exist_ok=True)
|
||||
weapon_slug = self._slugify(self._weapon_name(entry))
|
||||
pattern_slug = self._slugify(self._pattern_name(entry))
|
||||
suffix = self._infer_suffix(image_url)
|
||||
filename = f"{weapon_slug}__{pattern_slug}{suffix}"
|
||||
destination = target_dir / filename
|
||||
counter = 1
|
||||
while destination.exists():
|
||||
destination = target_dir / f"{weapon_slug}__{pattern_slug}_{counter}{suffix}"
|
||||
counter += 1
|
||||
|
||||
session = self._session or requests.Session()
|
||||
response = session.get(image_url, timeout=30)
|
||||
response.raise_for_status()
|
||||
destination.write_bytes(response.content)
|
||||
return destination
|
||||
|
||||
# Internal helpers -------------------------------------------------
|
||||
|
||||
def _load_cache(self) -> list[dict[str, Any]] | None:
|
||||
try:
|
||||
if self.cache_path.exists():
|
||||
return json.loads(self.cache_path.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return None
|
||||
return None
|
||||
|
||||
def _write_cache(self, data: Iterable[dict[str, Any]]) -> None:
|
||||
try:
|
||||
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
self.cache_path.write_text(json.dumps(list(data)), encoding="utf-8")
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def _download(self) -> list[dict[str, Any]]:
|
||||
self._session = self._session or requests.Session()
|
||||
response = self._session.get(self.DATA_URL, timeout=30)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
if isinstance(payload, dict):
|
||||
# some mirrors wrap the payload in a top-level dict
|
||||
payload = payload.get("skins") or payload.get("data") or []
|
||||
if not isinstance(payload, list):
|
||||
raise ValueError("Unexpected CS2 skin data format.")
|
||||
return payload
|
||||
|
||||
@staticmethod
|
||||
def _weapon_name(entry: dict[str, Any]) -> str:
|
||||
weapon = entry.get("weapon")
|
||||
if isinstance(weapon, dict):
|
||||
for key in ("name", "value", "english", "label"):
|
||||
if weapon.get(key):
|
||||
return str(weapon[key])
|
||||
if weapon:
|
||||
return str(weapon)
|
||||
return str(entry.get("weapon_name") or entry.get("weaponId") or "Unknown")
|
||||
|
||||
@staticmethod
|
||||
def _pattern_name(entry: dict[str, Any]) -> str:
|
||||
pattern = entry.get("pattern")
|
||||
if isinstance(pattern, dict):
|
||||
for key in ("name", "value", "english", "label"):
|
||||
if pattern.get(key):
|
||||
return str(pattern[key])
|
||||
if pattern:
|
||||
return str(pattern)
|
||||
return str(entry.get("name") or entry.get("skin") or entry.get("title") or "Pattern")
|
||||
|
||||
@staticmethod
|
||||
def _image_url(entry: dict[str, Any]) -> Optional[str]:
|
||||
for key in ("image", "image_url", "url"):
|
||||
value = entry.get(key)
|
||||
if isinstance(value, str) and value.startswith("http"):
|
||||
return value
|
||||
media = entry.get("media")
|
||||
if isinstance(media, dict):
|
||||
for key in ("image", "large", "icon"):
|
||||
value = media.get(key)
|
||||
if isinstance(value, str) and value.startswith("http"):
|
||||
return value
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _slugify(text: str | None) -> str:
|
||||
if not text:
|
||||
return "item"
|
||||
cleaned = "".join(ch.lower() if ch.isalnum() else "-" for ch in text)
|
||||
cleaned = "-".join(filter(None, cleaned.split("-")))
|
||||
return cleaned or "item"
|
||||
|
||||
@staticmethod
|
||||
def _infer_suffix(url: str) -> str:
|
||||
parsed = urlparse(url)
|
||||
path = Path(parsed.path)
|
||||
suffix = path.suffix.lower()
|
||||
if suffix in {".png", ".jpg", ".jpeg", ".webp"}:
|
||||
return suffix
|
||||
return ".png"
|
||||
|
||||
|
||||
class CS2PatternTool(tk.Toplevel):
|
||||
"""Tkinter UI wrapper around CS2PatternFetcher."""
|
||||
|
||||
def __init__(self, app) -> None:
|
||||
super().__init__(app.root)
|
||||
self.app = app
|
||||
self.fetcher = CS2PatternFetcher()
|
||||
self.title(self._t("cs2.title"))
|
||||
self.geometry("520x320")
|
||||
self.minsize(480, 300)
|
||||
self.configure(bg=self._background_colour())
|
||||
self.resizable(True, True)
|
||||
|
||||
self.weapons_var = tk.StringVar()
|
||||
self.patterns_var = tk.StringVar()
|
||||
self.directory_var = tk.StringVar(
|
||||
value=str((Path.cwd() / "images" / "cs2").resolve())
|
||||
)
|
||||
self.status_var = tk.StringVar(value=self._t("cs2.status_loading"))
|
||||
|
||||
self._init_widgets()
|
||||
self._data_loaded = False
|
||||
self._load_thread: Optional[threading.Thread] = None
|
||||
self._start_loading()
|
||||
|
||||
self.protocol("WM_DELETE_WINDOW", self._on_close)
|
||||
|
||||
# UI construction --------------------------------------------------
|
||||
|
||||
def _init_widgets(self) -> None:
|
||||
frame = ttk.Frame(self)
|
||||
frame.pack(fill=tk.BOTH, expand=True, padx=16, pady=16)
|
||||
|
||||
top = ttk.Frame(frame)
|
||||
top.pack(fill=tk.X, pady=(0, 12))
|
||||
ttk.Label(top, text=self._t("cs2.weapon_label")).grid(row=0, column=0, sticky="w")
|
||||
self.weapon_combo = ttk.Combobox(
|
||||
top, textvariable=self.weapons_var, state="disabled"
|
||||
)
|
||||
self.weapon_combo.grid(row=1, column=0, sticky="we", padx=(0, 12))
|
||||
self.weapon_combo.bind("<<ComboboxSelected>>", self._on_weapon_selected)
|
||||
|
||||
ttk.Label(top, text=self._t("cs2.pattern_label")).grid(
|
||||
row=0, column=1, sticky="w"
|
||||
)
|
||||
self.pattern_combo = ttk.Combobox(
|
||||
top, textvariable=self.patterns_var, state="disabled"
|
||||
)
|
||||
self.pattern_combo.grid(row=1, column=1, sticky="we")
|
||||
self.pattern_combo.bind("<<ComboboxSelected>>", self._on_pattern_selected)
|
||||
top.columnconfigure(0, weight=1)
|
||||
top.columnconfigure(1, weight=1)
|
||||
|
||||
dir_frame = ttk.Frame(frame)
|
||||
dir_frame.pack(fill=tk.X, pady=(0, 12))
|
||||
ttk.Label(dir_frame, text=self._t("cs2.output_label")).grid(
|
||||
row=0, column=0, sticky="w"
|
||||
)
|
||||
entry = ttk.Entry(dir_frame, textvariable=self.directory_var)
|
||||
entry.grid(row=1, column=0, sticky="we", padx=(0, 8))
|
||||
ttk.Button(
|
||||
dir_frame, text=self._t("cs2.browse_button"), command=self._browse_directory
|
||||
).grid(row=1, column=1, sticky="e")
|
||||
dir_frame.columnconfigure(0, weight=1)
|
||||
|
||||
buttons = ttk.Frame(frame)
|
||||
buttons.pack(fill=tk.X, pady=(0, 12))
|
||||
ttk.Button(
|
||||
buttons, text=self._t("cs2.refresh_button"), command=self._refresh_data
|
||||
).pack(side=tk.LEFT)
|
||||
self.download_btn = ttk.Button(
|
||||
buttons,
|
||||
text=self._t("cs2.download_button"),
|
||||
command=self._download_selected,
|
||||
state="disabled",
|
||||
)
|
||||
self.download_btn.pack(side=tk.RIGHT)
|
||||
|
||||
status_label = ttk.Label(
|
||||
frame, textvariable=self.status_var, anchor="w", justify="left"
|
||||
)
|
||||
status_label.pack(fill=tk.X)
|
||||
|
||||
# Data loading -----------------------------------------------------
|
||||
|
||||
def _start_loading(self) -> None:
|
||||
if self._load_thread and self._load_thread.is_alive():
|
||||
return
|
||||
self.status_var.set(self._t("cs2.status_loading"))
|
||||
self.weapon_combo.configure(state="disabled", values=[])
|
||||
self.pattern_combo.configure(state="disabled", values=[])
|
||||
self._load_thread = threading.Thread(target=self._load_data, daemon=True)
|
||||
self._load_thread.start()
|
||||
|
||||
def _load_data(self, force_refresh: bool = False) -> None:
|
||||
try:
|
||||
weapons = self.fetcher.list_weapons() if not force_refresh else None
|
||||
if force_refresh:
|
||||
self.fetcher.ensure_data(force_refresh=True)
|
||||
weapons = self.fetcher.list_weapons()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self.after(0, lambda: self._on_load_failed(exc))
|
||||
return
|
||||
self.after(0, lambda: self._on_data_ready(weapons or []))
|
||||
|
||||
def _on_data_ready(self, weapons: list[str]) -> None:
|
||||
if not weapons:
|
||||
self.status_var.set(self._t("cs2.status_empty"))
|
||||
return
|
||||
self.weapon_combo.configure(state="readonly", values=weapons)
|
||||
self.weapon_combo.set(weapons[0])
|
||||
self._populate_patterns(weapons[0])
|
||||
self.status_var.set(self._t("cs2.status_ready"))
|
||||
self._data_loaded = True
|
||||
|
||||
def _on_load_failed(self, exc: Exception) -> None:
|
||||
self.status_var.set(
|
||||
self._t("cs2.status_error").format(error=str(exc))
|
||||
)
|
||||
messagebox.showerror(
|
||||
self._t("dialog.error_title"),
|
||||
self._t("cs2.status_error").format(error=str(exc)),
|
||||
parent=self,
|
||||
)
|
||||
|
||||
def _refresh_data(self) -> None:
|
||||
self._start_loading()
|
||||
thread = threading.Thread(
|
||||
target=self._load_data, kwargs={"force_refresh": True}, daemon=True
|
||||
)
|
||||
thread.start()
|
||||
|
||||
def _populate_patterns(self, weapon: str) -> None:
|
||||
try:
|
||||
patterns = self.fetcher.list_patterns(weapon)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self.status_var.set(
|
||||
self._t("cs2.status_error").format(error=str(exc))
|
||||
)
|
||||
return
|
||||
self.pattern_combo.configure(state="readonly", values=patterns)
|
||||
if patterns:
|
||||
self.pattern_combo.set(patterns[0])
|
||||
self.download_btn.configure(state="normal")
|
||||
else:
|
||||
self.pattern_combo.set("")
|
||||
self.download_btn.configure(state="disabled")
|
||||
|
||||
# Event handlers ---------------------------------------------------
|
||||
|
||||
def _on_weapon_selected(self, event=None) -> None: # noqa: ANN001
|
||||
weapon = self.weapons_var.get()
|
||||
if weapon:
|
||||
self._populate_patterns(weapon)
|
||||
|
||||
def _on_pattern_selected(self, event=None) -> None: # noqa: ANN001
|
||||
if self.patterns_var.get():
|
||||
self.download_btn.configure(state="normal")
|
||||
|
||||
def _browse_directory(self) -> None:
|
||||
directory = filedialog.askdirectory(parent=self, mustexist=True)
|
||||
if directory:
|
||||
self.directory_var.set(directory)
|
||||
|
||||
def _download_selected(self) -> None:
|
||||
weapon = self.weapons_var.get()
|
||||
pattern = self.patterns_var.get()
|
||||
if not weapon:
|
||||
messagebox.showinfo(
|
||||
self._t("dialog.info_title"), self._t("cs2.no_weapon"), parent=self
|
||||
)
|
||||
return
|
||||
if not pattern:
|
||||
messagebox.showinfo(
|
||||
self._t("dialog.info_title"), self._t("cs2.no_pattern"), parent=self
|
||||
)
|
||||
return
|
||||
target_dir = Path(self.directory_var.get()).expanduser()
|
||||
entry = self.fetcher.find_entry(weapon, pattern)
|
||||
if entry is None:
|
||||
messagebox.showerror(
|
||||
self._t("dialog.error_title"),
|
||||
self._t("cs2.pattern_missing"),
|
||||
parent=self,
|
||||
)
|
||||
return
|
||||
try:
|
||||
path = self.fetcher.download_pattern_image(entry, target_dir)
|
||||
except requests.RequestException as exc:
|
||||
messagebox.showerror(
|
||||
self._t("dialog.error_title"),
|
||||
self._t("cs2.download_error").format(error=str(exc)),
|
||||
parent=self,
|
||||
)
|
||||
return
|
||||
except Exception as exc: # noqa: BLE001
|
||||
messagebox.showerror(
|
||||
self._t("dialog.error_title"),
|
||||
self._t("cs2.download_error").format(error=str(exc)),
|
||||
parent=self,
|
||||
)
|
||||
return
|
||||
messagebox.showinfo(
|
||||
self._t("dialog.saved_title"),
|
||||
self._t("cs2.download_success").format(path=path),
|
||||
parent=self,
|
||||
)
|
||||
|
||||
def _background_colour(self) -> str:
|
||||
return "#0f0f10" if getattr(self.app, "theme", "light") == "dark" else "#ffffff"
|
||||
|
||||
def _t(self, key: str) -> str:
|
||||
translator = getattr(self.app, "translator", None)
|
||||
if translator is not None:
|
||||
return translator.translate(key)
|
||||
if hasattr(self.app, "_t"):
|
||||
return self.app._t(key) # type: ignore[attr-defined]
|
||||
return key
|
||||
|
||||
def _on_close(self) -> None:
|
||||
if hasattr(self.app, "_cs2_tool_window"):
|
||||
self.app._cs2_tool_window = None # type: ignore[attr-defined]
|
||||
self.destroy()
|
||||
|
||||
|
||||
def open_cs2_pattern_tool(app) -> CS2PatternTool:
|
||||
"""Launch (or focus) the CS2 pattern tool."""
|
||||
existing = getattr(app, "_cs2_tool_window", None)
|
||||
if existing is not None and isinstance(existing, tk.Toplevel):
|
||||
try:
|
||||
if existing.winfo_exists():
|
||||
existing.lift()
|
||||
existing.focus_force()
|
||||
return existing
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
window = CS2PatternTool(app)
|
||||
app._cs2_tool_window = window # type: ignore[attr-defined]
|
||||
return window
|
||||
|
||||
Reference in New Issue
Block a user