mirror of
https://github.com/yusufipk/deste-hearthstone-linux-tracker.git
synced 2026-09-11 10:46:12 +00:00
Hearthstone deck tracker for Linux, first public version
Reads the game's own Power.log: no screen capture, no image recognition, no memory reading. - deck tracking (remaining cards, draw chance, played cards fade in place) - opponent panel (revealed cards, hand and deck counts) - decks read from the game's offline cache, matched to the match - card art in rows, full card on hover - overlay and window modes, adjustable transparency, tray icon - English and Turkish interface - consistency tests over a real log corpus, cross-checked against Zone.log
This commit is contained in:
yusufipk
@@ -0,0 +1,96 @@
|
||||
"""Kart verisi: HearthstoneJSON'dan indirir, diske cache'ler.
|
||||
|
||||
Not: api.hearthstonejson.com varsayılan urllib User-Agent'ına 403 dönüyor,
|
||||
bu yüzden kendi UA'mızı göndermek zorunlu.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
CARDS_URL = "https://api.hearthstonejson.com/v1/latest/enUS/cards.json"
|
||||
USER_AGENT = "deste/0.1 (kisisel hearthstone tracker; +local)"
|
||||
CACHE_DIR = Path(os.environ.get("XDG_CACHE_HOME", "~/.cache")).expanduser() / "deste"
|
||||
CARDS_PATH = CACHE_DIR / "cards.json"
|
||||
MAX_AGE_SECONDS = 7 * 24 * 3600
|
||||
|
||||
|
||||
def _download(url: str, timeout: int = 60) -> bytes:
|
||||
request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
|
||||
with urllib.request.urlopen(request, timeout=timeout) as response:
|
||||
return response.read()
|
||||
|
||||
|
||||
def cache_is_stale(path: Path = CARDS_PATH, max_age: int = MAX_AGE_SECONDS) -> bool:
|
||||
if not path.exists():
|
||||
return True
|
||||
return (time.time() - path.stat().st_mtime) > max_age
|
||||
|
||||
|
||||
def update_cache(force: bool = False) -> bool:
|
||||
"""Kart verisini indirir. Ağ yoksa mevcut cache korunur."""
|
||||
if not force and not cache_is_stale():
|
||||
return True
|
||||
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
||||
try:
|
||||
payload = _download(CARDS_URL)
|
||||
except (urllib.error.URLError, TimeoutError, OSError):
|
||||
return CARDS_PATH.exists()
|
||||
tmp = CARDS_PATH.with_suffix(".tmp")
|
||||
tmp.write_bytes(payload)
|
||||
tmp.replace(CARDS_PATH)
|
||||
return True
|
||||
|
||||
|
||||
class CardDB:
|
||||
"""Kart kimliğinden isim, mana, sınıf gibi alanlara erişim."""
|
||||
|
||||
def __init__(self, cards: list[dict]):
|
||||
self.by_id: dict[str, dict] = {c["id"]: c for c in cards if "id" in c}
|
||||
self.by_dbf: dict[int, dict] = {c["dbfId"]: c for c in cards if "dbfId" in c}
|
||||
|
||||
@classmethod
|
||||
def load(cls, path: Path = CARDS_PATH, auto_update: bool = True) -> "CardDB":
|
||||
if auto_update:
|
||||
update_cache()
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(
|
||||
f"Kart verisi yok: {path}. İnternete bağlanıp tekrar deneyin."
|
||||
)
|
||||
return cls(json.loads(path.read_text(encoding="utf-8")))
|
||||
|
||||
def get(self, card_id: str) -> dict:
|
||||
return self.by_id.get(card_id, {})
|
||||
|
||||
def name(self, card_id: str) -> str:
|
||||
return self.get(card_id).get("name", card_id or "?")
|
||||
|
||||
def cost(self, card_id: str) -> int:
|
||||
return self.get(card_id).get("cost", 0)
|
||||
|
||||
def card_class(self, card_id: str) -> str:
|
||||
card = self.get(card_id)
|
||||
classes = card.get("classes")
|
||||
if classes:
|
||||
return classes[0]
|
||||
return card.get("cardClass", "")
|
||||
|
||||
def rarity(self, card_id: str) -> str:
|
||||
return self.get(card_id).get("rarity", "")
|
||||
|
||||
def card_type(self, card_id: str) -> str:
|
||||
return self.get(card_id).get("type", "")
|
||||
|
||||
def dbf_id(self, card_id: str) -> int | None:
|
||||
return self.get(card_id).get("dbfId")
|
||||
|
||||
def id_from_dbf(self, dbf_id: int) -> str:
|
||||
return self.by_dbf.get(dbf_id, {}).get("id", "")
|
||||
|
||||
def collectible_by_dbf(self, dbf_id: int) -> dict:
|
||||
return self.by_dbf.get(dbf_id, {})
|
||||
+174
@@ -0,0 +1,174 @@
|
||||
"""Deste kütüphanesi: oyunun önbelleğinden ve kullanıcının kaydettiği deste
|
||||
kodlarından desteleri toplar, maça uygun olanı seçer.
|
||||
|
||||
Kart kimliği (CATA_560) ile dbfId (122939) arasındaki çeviriyi burada yapıyoruz,
|
||||
core katmanı yalnızca kart kimlikleriyle çalışır.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from collections import Counter
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from core import deckstring
|
||||
from core.state import Game
|
||||
|
||||
from .cards import CardDB
|
||||
from .localdecks import LocalDeck, read_all_decks
|
||||
|
||||
CONFIG_DIR = Path(os.environ.get("XDG_CONFIG_HOME", "~/.config")).expanduser() / "deste"
|
||||
DECKS_PATH = CONFIG_DIR / "decks.json"
|
||||
|
||||
|
||||
@dataclass
|
||||
class TrackedDeck:
|
||||
name: str
|
||||
cards: Counter[str] = field(default_factory=Counter) # kart kimliği -> adet
|
||||
hero_card_id: str = ""
|
||||
player_class: str = ""
|
||||
format_name: str = ""
|
||||
source: str = "local" # local (oyun önbelleği) veya manual (deste kodu)
|
||||
deck_code: str = ""
|
||||
|
||||
@property
|
||||
def card_count(self) -> int:
|
||||
return sum(self.cards.values())
|
||||
|
||||
|
||||
def _deck_from_dbf(
|
||||
cards_db: CardDB,
|
||||
name: str,
|
||||
dbf_cards: dict[int, int],
|
||||
hero_dbf: int,
|
||||
format_name: str,
|
||||
source: str,
|
||||
deck_code: str = "",
|
||||
) -> TrackedDeck:
|
||||
counter: Counter[str] = Counter()
|
||||
for dbf_id, count in dbf_cards.items():
|
||||
card_id = cards_db.id_from_dbf(dbf_id)
|
||||
if card_id:
|
||||
counter[card_id] += count
|
||||
hero_card_id = cards_db.id_from_dbf(hero_dbf) if hero_dbf else ""
|
||||
return TrackedDeck(
|
||||
name=name,
|
||||
cards=counter,
|
||||
hero_card_id=hero_card_id,
|
||||
player_class=cards_db.card_class(hero_card_id) if hero_card_id else "",
|
||||
format_name=format_name,
|
||||
source=source,
|
||||
deck_code=deck_code,
|
||||
)
|
||||
|
||||
|
||||
def from_local(cards_db: CardDB, deck: LocalDeck) -> TrackedDeck:
|
||||
return _deck_from_dbf(
|
||||
cards_db, deck.name, deck.cards, deck.hero_dbf, deck.format_name, "local"
|
||||
)
|
||||
|
||||
|
||||
def from_deck_code(cards_db: CardDB, code: str, name: str = "") -> TrackedDeck:
|
||||
"""Deste kodundan takip edilebilir deste üretir."""
|
||||
parsed = deckstring.parse(code)
|
||||
hero = parsed.heroes[0] if parsed.heroes else 0
|
||||
return _deck_from_dbf(
|
||||
cards_db,
|
||||
name or "Deste kodu",
|
||||
parsed.as_counter(),
|
||||
hero,
|
||||
parsed.format_name,
|
||||
"manual",
|
||||
code.strip(),
|
||||
)
|
||||
|
||||
|
||||
class DeckLibrary:
|
||||
"""Oyunun önbelleğindeki ve elle eklenen desteler."""
|
||||
|
||||
def __init__(self, cards_db: CardDB, game_dir: Path | None = None):
|
||||
self.cards_db = cards_db
|
||||
self.game_dir = game_dir
|
||||
self.decks: list[TrackedDeck] = []
|
||||
|
||||
def load(self) -> "DeckLibrary":
|
||||
self.decks = []
|
||||
if self.game_dir is not None:
|
||||
for local in read_all_decks(self.game_dir):
|
||||
self.decks.append(from_local(self.cards_db, local))
|
||||
self.decks.extend(self._load_manual())
|
||||
return self
|
||||
|
||||
def _load_manual(self) -> list[TrackedDeck]:
|
||||
if not DECKS_PATH.exists():
|
||||
return []
|
||||
try:
|
||||
payload = json.loads(DECKS_PATH.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return []
|
||||
result: list[TrackedDeck] = []
|
||||
for entry in payload.get("decks", []):
|
||||
code = entry.get("code", "")
|
||||
if not code:
|
||||
continue
|
||||
try:
|
||||
result.append(from_deck_code(self.cards_db, code, entry.get("name", "")))
|
||||
except deckstring.DeckstringError:
|
||||
continue
|
||||
return result
|
||||
|
||||
def add_manual(self, code: str, name: str = "") -> TrackedDeck:
|
||||
"""Deste kodunu kaydeder ve kütüphaneye ekler."""
|
||||
deck = from_deck_code(self.cards_db, code, name)
|
||||
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
||||
payload = {"decks": []}
|
||||
if DECKS_PATH.exists():
|
||||
try:
|
||||
payload = json.loads(DECKS_PATH.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError):
|
||||
pass
|
||||
entries = payload.setdefault("decks", [])
|
||||
if not any(e.get("code") == deck.deck_code for e in entries):
|
||||
entries.append({"name": deck.name, "code": deck.deck_code})
|
||||
DECKS_PATH.write_text(
|
||||
json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8"
|
||||
)
|
||||
self.decks.append(deck)
|
||||
return deck
|
||||
|
||||
# --- eşleştirme -----------------------------------------------------
|
||||
|
||||
def candidates(self, player_class: str) -> list[TrackedDeck]:
|
||||
if not player_class:
|
||||
return list(self.decks)
|
||||
return [d for d in self.decks if d.player_class == player_class]
|
||||
|
||||
def match(self, game: Game) -> TrackedDeck | None:
|
||||
"""Maça en uygun desteyi seçer.
|
||||
|
||||
Sınıfa göre süzer, sonra destemizden çıkmış kartlarla çelişmeyenleri
|
||||
bırakır. Tek aday kalırsa odur; birden fazla aday varsa çıkan kartları
|
||||
en çok kapsayan seçilir. Hiçbiri uymuyorsa None döner ve arayüz
|
||||
kullanıcıya sorar.
|
||||
"""
|
||||
hero = game.hero_card_id(game.local_player_id)
|
||||
player_class = self.cards_db.card_class(hero) if hero else ""
|
||||
candidates = self.candidates(player_class)
|
||||
if not candidates:
|
||||
return None
|
||||
|
||||
consistent = [d for d in candidates if game.deck_list_mismatch(d.cards) == 0]
|
||||
pool = consistent or candidates
|
||||
if len(pool) == 1:
|
||||
return pool[0]
|
||||
|
||||
def coverage(deck: TrackedDeck) -> tuple[int, int]:
|
||||
covered = sum(
|
||||
min(count, deck.cards.get(card, 0))
|
||||
for card, count in game.cards_left_deck.items()
|
||||
)
|
||||
return (covered, -game.deck_list_mismatch(deck.cards))
|
||||
|
||||
return max(pool, key=coverage)
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Kart görselleri: art.hearthstonejson.com'dan indirir, diske önbelleğe alır.
|
||||
|
||||
İki boyut kullanılıyor:
|
||||
tile 256x59 kart satırının arka planındaki sanat şeridi
|
||||
render 256x388 fareyle üzerine gelince gösterilen tam kart (metni dahil)
|
||||
|
||||
İndirmeler arka plan iş parçacıklarında yapılır, arayüz beklemez. Yeni bir
|
||||
görsel hazır olduğunda `generation` sayacı artar; arayüz bu sayacı izleyip
|
||||
kendini tazeler. Böylece Qt sinyali gerekmiyor ve bu modül stdlib kalıyor.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import threading
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from pathlib import Path
|
||||
|
||||
TILE_URL = "https://art.hearthstonejson.com/v1/tiles/{card_id}.png"
|
||||
RENDER_URL = "https://art.hearthstonejson.com/v1/render/latest/enUS/256x/{card_id}.png"
|
||||
USER_AGENT = "deste/0.1 (kisisel hearthstone tracker; +local)"
|
||||
|
||||
CACHE_DIR = Path(os.environ.get("XDG_CACHE_HOME", "~/.cache")).expanduser() / "deste" / "images"
|
||||
|
||||
TILE = "tile"
|
||||
RENDER = "render"
|
||||
|
||||
|
||||
class ImageCache:
|
||||
def __init__(self, max_workers: int = 4):
|
||||
self.generation = 0
|
||||
self._lock = threading.Lock()
|
||||
self._pending: set[tuple[str, str]] = set()
|
||||
self._failed: set[tuple[str, str]] = set()
|
||||
self._pool = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="deste-img")
|
||||
for kind in (TILE, RENDER):
|
||||
(CACHE_DIR / kind).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def path_for(self, card_id: str, kind: str = TILE) -> Path:
|
||||
return CACHE_DIR / kind / f"{card_id}.png"
|
||||
|
||||
def get(self, card_id: str, kind: str = TILE) -> Path | None:
|
||||
"""Görsel diskteyse yolunu döner, değilse indirmeye alır ve None döner."""
|
||||
if not card_id:
|
||||
return None
|
||||
path = self.path_for(card_id, kind)
|
||||
if path.exists():
|
||||
return path
|
||||
key = (card_id, kind)
|
||||
with self._lock:
|
||||
if key in self._pending or key in self._failed:
|
||||
return None
|
||||
self._pending.add(key)
|
||||
self._pool.submit(self._download, card_id, kind)
|
||||
return None
|
||||
|
||||
def prefetch(self, card_ids, kind: str = TILE) -> None:
|
||||
for card_id in card_ids:
|
||||
self.get(card_id, kind)
|
||||
|
||||
def _download(self, card_id: str, kind: str) -> None:
|
||||
url = (TILE_URL if kind == TILE else RENDER_URL).format(card_id=card_id)
|
||||
key = (card_id, kind)
|
||||
try:
|
||||
request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
|
||||
with urllib.request.urlopen(request, timeout=20) as response:
|
||||
payload = response.read()
|
||||
except (urllib.error.URLError, TimeoutError, OSError):
|
||||
with self._lock:
|
||||
self._pending.discard(key)
|
||||
self._failed.add(key)
|
||||
return
|
||||
|
||||
path = self.path_for(card_id, kind)
|
||||
try:
|
||||
tmp = path.with_suffix(".tmp")
|
||||
tmp.write_bytes(payload)
|
||||
tmp.replace(path)
|
||||
except OSError:
|
||||
with self._lock:
|
||||
self._pending.discard(key)
|
||||
self._failed.add(key)
|
||||
return
|
||||
|
||||
with self._lock:
|
||||
self._pending.discard(key)
|
||||
self.generation += 1
|
||||
|
||||
def shutdown(self) -> None:
|
||||
self._pool.shutdown(wait=False, cancel_futures=True)
|
||||
@@ -0,0 +1,250 @@
|
||||
"""Oyunun çevrimdışı önbelleğinden destelerimizi okur.
|
||||
|
||||
Hearthstone, hesabın destelerini şu dosyada saklıyor:
|
||||
<prefix>/users/<kullanıcı>/AppData/Local/Blizzard/Hearthstone/Cache/Offline/
|
||||
offlineData_<hi>_<lo>_REGION_<bölge>.cache
|
||||
|
||||
Dosya 12 baytlık bir başlıktan sonra bölümler halinde ilerliyor:
|
||||
[u32 kayıt sayısı] ( [u32 uzunluk] [protobuf payload] ) * sayı
|
||||
|
||||
Protobuf şemaları resmî değil, bayt seviyesinden çıkarıldı:
|
||||
Deste kaydı : 1=id, 2=isim, 4=kahraman dbfId, 19=format
|
||||
İçerik kaydı : 2=deste id, 3=tekrarlı yığın
|
||||
yığın : 1={1=kart dbfId, 2=premium}, 3=adet
|
||||
|
||||
Bu biçim resmî olmadığı için her şey savunmacı yazıldı: çözülemeyen kayıt
|
||||
sessizce atlanır, hiçbir durumda çökmez. Okuma başarısız olursa deste kodu
|
||||
yapıştırma yolu devrede kalır.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import struct
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
HEADER_SIZE = 12
|
||||
MAX_RECORDS_PER_SECTION = 10_000
|
||||
|
||||
# Protobuf alan numaraları
|
||||
F_DECK_ID = 1
|
||||
F_DECK_NAME = 2
|
||||
F_DECK_HERO = 4
|
||||
F_DECK_FORMAT = 19
|
||||
F_CONTENTS_DECK_ID = 2
|
||||
F_CONTENTS_STACK = 3
|
||||
F_STACK_CARD = 1
|
||||
F_STACK_COUNT = 3
|
||||
F_CARD_DBF = 1
|
||||
|
||||
FORMAT_NAMES = {1: "WILD", 2: "STANDARD", 3: "CLASSIC", 4: "TWIST"}
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class LocalDeck:
|
||||
deck_id: int
|
||||
name: str
|
||||
hero_dbf: int = 0
|
||||
format_id: int = 0
|
||||
cards: dict[int, int] = field(default_factory=dict) # dbfId -> adet
|
||||
|
||||
@property
|
||||
def format_name(self) -> str:
|
||||
return FORMAT_NAMES.get(self.format_id, "")
|
||||
|
||||
@property
|
||||
def card_count(self) -> int:
|
||||
return sum(self.cards.values())
|
||||
|
||||
|
||||
# --- asgari protobuf okuyucu -------------------------------------------
|
||||
|
||||
|
||||
def _varint(data: bytes, index: int) -> tuple[int, int]:
|
||||
value = 0
|
||||
shift = 0
|
||||
while index < len(data):
|
||||
byte = data[index]
|
||||
index += 1
|
||||
value |= (byte & 0x7F) << shift
|
||||
if not byte & 0x80:
|
||||
return value, index
|
||||
shift += 7
|
||||
if shift > 70:
|
||||
break
|
||||
raise ValueError("bozuk varint")
|
||||
|
||||
|
||||
def parse_message(data: bytes) -> dict[int, list]:
|
||||
"""Protobuf mesajını {alan: [değer]} sözlüğüne çevirir.
|
||||
|
||||
Uzunluk belirtilmiş alanlar ham bayt olarak döner, gerekirse çağıran
|
||||
tarafından tekrar çözülür.
|
||||
"""
|
||||
fields: dict[int, list] = {}
|
||||
index = 0
|
||||
while index < len(data):
|
||||
key, index = _varint(data, index)
|
||||
field_number, wire_type = key >> 3, key & 7
|
||||
if field_number == 0:
|
||||
raise ValueError("geçersiz alan numarası")
|
||||
if wire_type == 0:
|
||||
value, index = _varint(data, index)
|
||||
elif wire_type == 2:
|
||||
length, index = _varint(data, index)
|
||||
if index + length > len(data):
|
||||
raise ValueError("uzunluk taşması")
|
||||
value = data[index : index + length]
|
||||
index += length
|
||||
elif wire_type == 5:
|
||||
value = data[index : index + 4]
|
||||
index += 4
|
||||
elif wire_type == 1:
|
||||
value = data[index : index + 8]
|
||||
index += 8
|
||||
else:
|
||||
raise ValueError(f"bilinmeyen wire type {wire_type}")
|
||||
fields.setdefault(field_number, []).append(value)
|
||||
return fields
|
||||
|
||||
|
||||
def iter_records(data: bytes):
|
||||
"""Dosyadaki bölümleri gezip her kaydın ham baytını verir."""
|
||||
index = HEADER_SIZE
|
||||
while index + 4 <= len(data):
|
||||
(count,) = struct.unpack_from("<I", data, index)
|
||||
index += 4
|
||||
if count == 0 or count > MAX_RECORDS_PER_SECTION:
|
||||
return
|
||||
for _ in range(count):
|
||||
if index + 4 > len(data):
|
||||
return
|
||||
(length,) = struct.unpack_from("<I", data, index)
|
||||
index += 4
|
||||
if length == 0 or index + length > len(data):
|
||||
return
|
||||
yield data[index : index + length]
|
||||
index += length
|
||||
|
||||
|
||||
# --- kayıt sınıflandırma -----------------------------------------------
|
||||
|
||||
|
||||
def _as_text(raw: bytes) -> str:
|
||||
return raw.decode("utf-8", errors="replace").strip()
|
||||
|
||||
|
||||
def _parse_deck_record(fields: dict[int, list]) -> LocalDeck | None:
|
||||
name_values = fields.get(F_DECK_NAME)
|
||||
id_values = fields.get(F_DECK_ID)
|
||||
if not name_values or not id_values:
|
||||
return None
|
||||
if not isinstance(name_values[0], bytes) or not isinstance(id_values[0], int):
|
||||
return None
|
||||
hero = fields.get(F_DECK_HERO, [0])[0]
|
||||
fmt = fields.get(F_DECK_FORMAT, [0])[0]
|
||||
return LocalDeck(
|
||||
deck_id=id_values[0],
|
||||
name=_as_text(name_values[0]),
|
||||
hero_dbf=hero if isinstance(hero, int) else 0,
|
||||
format_id=fmt if isinstance(fmt, int) else 0,
|
||||
)
|
||||
|
||||
|
||||
def _parse_contents_record(fields: dict[int, list]) -> tuple[int, dict[int, int]] | None:
|
||||
deck_ids = fields.get(F_CONTENTS_DECK_ID)
|
||||
stacks = fields.get(F_CONTENTS_STACK)
|
||||
if not deck_ids or not stacks or not isinstance(deck_ids[0], int):
|
||||
return None
|
||||
cards: dict[int, int] = {}
|
||||
for stack_raw in stacks:
|
||||
if not isinstance(stack_raw, bytes):
|
||||
continue
|
||||
try:
|
||||
stack = parse_message(stack_raw)
|
||||
except ValueError:
|
||||
continue
|
||||
card_raw = stack.get(F_STACK_CARD, [None])[0]
|
||||
count = stack.get(F_STACK_COUNT, [0])[0]
|
||||
if not isinstance(card_raw, bytes) or not isinstance(count, int) or count <= 0:
|
||||
continue
|
||||
try:
|
||||
card = parse_message(card_raw)
|
||||
except ValueError:
|
||||
continue
|
||||
dbf = card.get(F_CARD_DBF, [None])[0]
|
||||
if isinstance(dbf, int):
|
||||
# Aynı kartın normal ve altın kopyaları ayrı yığın olarak geliyor.
|
||||
cards[dbf] = cards.get(dbf, 0) + count
|
||||
if not cards:
|
||||
return None
|
||||
return deck_ids[0], cards
|
||||
|
||||
|
||||
def read_decks(cache_path: Path) -> list[LocalDeck]:
|
||||
"""Önbellekten desteleri okur. Hata durumunda boş liste döner."""
|
||||
try:
|
||||
data = cache_path.read_bytes()
|
||||
except OSError:
|
||||
return []
|
||||
|
||||
decks: dict[int, LocalDeck] = {}
|
||||
contents: dict[int, dict[int, int]] = {}
|
||||
|
||||
for raw in iter_records(data):
|
||||
try:
|
||||
fields = parse_message(raw)
|
||||
except ValueError:
|
||||
continue
|
||||
deck = _parse_deck_record(fields)
|
||||
if deck is not None:
|
||||
# Aynı deste birden fazla bölümde tekrar edebiliyor.
|
||||
decks.setdefault(deck.deck_id, deck)
|
||||
continue
|
||||
parsed = _parse_contents_record(fields)
|
||||
if parsed is not None:
|
||||
deck_id, cards = parsed
|
||||
contents.setdefault(deck_id, cards)
|
||||
|
||||
for deck_id, cards in contents.items():
|
||||
if deck_id in decks:
|
||||
decks[deck_id].cards = cards
|
||||
|
||||
# Oyun aynı desteyi "yerel" ve "orijinal" olmak üzere iki kopya halinde
|
||||
# önbelleğe alıyor. İsim ve içerik aynıysa tek deste olarak gösteriyoruz.
|
||||
unique: dict[tuple, LocalDeck] = {}
|
||||
for deck in decks.values():
|
||||
if not deck.cards:
|
||||
continue
|
||||
signature = (deck.name, tuple(sorted(deck.cards.items())))
|
||||
unique.setdefault(signature, deck)
|
||||
return list(unique.values())
|
||||
|
||||
|
||||
def find_cache_files(game_dir: Path) -> list[Path]:
|
||||
"""Prefix içindeki çevrimdışı önbellek dosyalarını bulur.
|
||||
|
||||
Wine prefix'lerinde users/<isim> çoğu zaman users/steamuser'a sembolik
|
||||
bağdır, aynı dosyayı iki kez okumamak için gerçek yola göre tekilleştiriyoruz.
|
||||
"""
|
||||
drive_c = game_dir.parent.parent
|
||||
users = drive_c / "users"
|
||||
if not users.is_dir():
|
||||
return []
|
||||
seen: dict[Path, Path] = {}
|
||||
for user_dir in sorted(users.iterdir()):
|
||||
offline = (
|
||||
user_dir / "AppData" / "Local" / "Blizzard" / "Hearthstone" / "Cache" / "Offline"
|
||||
)
|
||||
if not offline.is_dir():
|
||||
continue
|
||||
for cache_file in sorted(offline.glob("offlineData_*.cache")):
|
||||
seen.setdefault(cache_file.resolve(), cache_file)
|
||||
return list(seen.values())
|
||||
|
||||
|
||||
def read_all_decks(game_dir: Path) -> list[LocalDeck]:
|
||||
decks: list[LocalDeck] = []
|
||||
for cache_file in find_cache_files(game_dir):
|
||||
decks.extend(read_decks(cache_file))
|
||||
return decks
|
||||
Reference in New Issue
Block a user