feat: Card Kingdom prices, shopping cart export, and hover panel fixes (#73)

* feat: add CK prices, shopping cart export, and hover panel fixes

* fix: include commander in Buy This Deck cart export
This commit is contained in:
mwisnowski 2026-04-04 19:59:03 -07:00 committed by GitHub
parent dd996939e6
commit 69d84cc414
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
24 changed files with 899 additions and 146 deletions

View file

@ -468,15 +468,18 @@ class BudgetEvaluatorService(BaseService):
# Batch price lookup for all candidates
candidate_names = list(candidates.keys())
prices = self._price_svc.get_prices_batch(candidate_names, region=region, foil=foil)
ck_prices = self._price_svc.get_ck_prices_batch(candidate_names)
results = []
for name, info in candidates.items():
price = prices.get(name)
if price is None or price > max_price:
continue
ck_price = ck_prices.get(name)
results.append({
"name": name,
"price": round(price, 2),
"ck_price": round(ck_price, 2) if ck_price is not None else None,
"tags": info["tags"],
"shared_tags": sorted(info["shared_tags"]),
})
@ -540,13 +543,11 @@ class BudgetEvaluatorService(BaseService):
def _get_card_tags(self, card_name: str) -> List[str]:
"""Look up theme tags for a single card from the card index."""
try:
from code.web.services.card_index import maybe_build_index, _CARD_INDEX
from code.web.services.card_index import maybe_build_index, lookup_card
maybe_build_index()
needle = card_name.lower()
for cards in _CARD_INDEX.values():
for c in cards:
if c.get("name", "").lower() == needle:
return list(c.get("tags", []))
entry = lookup_card(card_name)
if entry:
return list(entry.get("tags", []))
except Exception:
pass
return []
@ -554,17 +555,14 @@ class BudgetEvaluatorService(BaseService):
def _get_card_broad_type(self, card_name: str) -> Optional[str]:
"""Return the first matching broad MTG type for a card (e.g. 'Land', 'Creature')."""
try:
from code.web.services.card_index import maybe_build_index, _CARD_INDEX
from code.web.services.card_index import maybe_build_index, lookup_card
maybe_build_index()
needle = card_name.lower()
for cards in _CARD_INDEX.values():
for c in cards:
if c.get("name", "").lower() == needle:
type_line = c.get("type_line", "")
for broad in _BROAD_TYPES:
if broad in type_line:
return broad
return None
entry = lookup_card(card_name)
if entry:
type_line = entry.get("type_line", "")
for broad in _BROAD_TYPES:
if broad in type_line:
return broad
except Exception:
pass
return None
@ -620,16 +618,13 @@ class BudgetEvaluatorService(BaseService):
# Collect all unique tags from the current deck
deck_tags: Set[str] = set()
try:
from code.web.services.card_index import maybe_build_index, _CARD_INDEX
from code.web.services.card_index import maybe_build_index, _CARD_INDEX, lookup_card
maybe_build_index()
for name in decklist:
needle = name.lower()
for cards in _CARD_INDEX.values():
for c in cards:
if c.get("name", "").lower() == needle:
deck_tags.update(c.get("tags", []))
break
card_entry = lookup_card(name)
if card_entry:
deck_tags.update(card_entry.get("tags", []))
if not deck_tags:
return []
@ -664,6 +659,7 @@ class BudgetEvaluatorService(BaseService):
top_candidates = sorted(candidates.values(), key=lambda x: x["score"], reverse=True)[:200]
names = [c["name"] for c in top_candidates]
prices = self._price_svc.get_prices_batch(names, region=region, foil=foil)
ck_prices = self._price_svc.get_ck_prices_batch(names)
tier_ceilings = self.calculate_tier_ceilings(budget_remaining)
pickups: List[Pickup] = []
@ -679,9 +675,11 @@ class BudgetEvaluatorService(BaseService):
tier = "M"
if price <= tier_ceilings["S"]:
tier = "S"
ck_price = ck_prices.get(c["name"])
pickups.append({
"card": c["name"],
"price": round(price, 2),
"ck_price": round(ck_price, 2) if ck_price is not None else None,
"tier": tier,
"priority": c["score"],
"tags": c["tags"],

View file

@ -27,6 +27,8 @@ RARITY_COL = "rarity"
_CARD_INDEX: Dict[str, List[Dict[str, Any]]] = {}
_CARD_INDEX_MTIME: float | None = None
# Reverse lookup: lowercase card name → card dict (first occurrence per name)
_NAME_INDEX: Dict[str, Dict[str, Any]] = {}
_RARITY_NORM = {
"mythic rare": "mythic",
@ -50,7 +52,7 @@ def maybe_build_index() -> None:
M4: Loads from all_cards.parquet instead of CSV files.
"""
global _CARD_INDEX, _CARD_INDEX_MTIME
global _CARD_INDEX, _CARD_INDEX_MTIME, _NAME_INDEX
try:
from path_util import get_processed_cards_path
@ -99,6 +101,14 @@ def maybe_build_index() -> None:
})
_CARD_INDEX = new_index
# Build name → card reverse index for O(1) lookups
new_name_index: Dict[str, Dict[str, Any]] = {}
for cards in new_index.values():
for c in cards:
key = c.get("name", "").lower()
if key and key not in new_name_index:
new_name_index[key] = c
_NAME_INDEX = new_name_index
_CARD_INDEX_MTIME = latest
except Exception:
# Defensive: if anything fails, leave index unchanged
@ -107,9 +117,20 @@ def maybe_build_index() -> None:
def get_tag_pool(tag: str) -> List[Dict[str, Any]]:
return _CARD_INDEX.get(tag, [])
def lookup_card(name: str) -> Optional[Dict[str, Any]]:
"""O(1) lookup of a card dict by name. Returns None if not found."""
return _NAME_INDEX.get(name.lower().strip()) if name else None
def lookup_commander(name: Optional[str]) -> Optional[Dict[str, Any]]:
if not name:
return None
# Fast path via name index
result = _NAME_INDEX.get(name.lower().strip())
if result is not None:
return result
# Fallback: full scan (handles index not yet built)
needle = name.lower().strip()
for tag_cards in _CARD_INDEX.values():
for c in tag_cards:

View file

@ -30,6 +30,8 @@ logger.addHandler(logging_util.stream_handler)
_CACHE_TTL_SECONDS = 86400 # 24 hours
_BULK_DATA_FILENAME = "scryfall_bulk_data.json"
_PRICE_CACHE_FILENAME = "prices_cache.json"
_CK_CACHE_FILENAME = "ck_prices_cache.json"
_CK_API_URL = "https://api.cardkingdom.com/api/v2/pricelist"
class PriceService(BaseService):
@ -68,6 +70,14 @@ class PriceService(BaseService):
self._miss_count = 0
self._refresh_thread: Optional[threading.Thread] = None
# CK price cache: {normalized_card_name: float (cheapest non-foil retail)}
self._ck_cache_path: str = os.path.join(card_files_dir(), _CK_CACHE_FILENAME)
self._ck_cache: Dict[str, float] = {}
self._ck_loaded: bool = False
# scryfall_id map built during _rebuild_cache: {name.lower(): scryfall_id}
self._scryfall_id_map: Dict[str, str] = {}
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
@ -157,6 +167,35 @@ class PriceService(BaseService):
"bulk_data_available": os.path.exists(self._bulk_path),
}
# ------------------------------------------------------------------
# CK Public API
# ------------------------------------------------------------------
def get_ck_price(self, card_name: str) -> Optional[float]:
"""Return the Card Kingdom retail price for *card_name*, or None."""
self._ensure_ck_loaded()
return self._ck_cache.get(card_name.lower().strip())
def get_ck_prices_batch(self, card_names: List[str]) -> Dict[str, Optional[float]]:
"""Return a mapping of card name → CK retail price for all requested cards."""
self._ensure_ck_loaded()
return {
name: self._ck_cache.get(name.lower().strip())
for name in card_names
}
def get_ck_built_at(self) -> Optional[str]:
"""Return a human-readable CK cache build date, or None if unavailable."""
try:
if os.path.exists(self._ck_cache_path):
import datetime
built = os.path.getmtime(self._ck_cache_path)
dt = datetime.datetime.fromtimestamp(built, tz=datetime.timezone.utc)
return dt.strftime("%B %d, %Y")
except Exception:
pass
return None
def refresh_cache_background(self) -> None:
"""Spawn a daemon thread to rebuild the price cache asynchronously.
@ -412,6 +451,7 @@ class PriceService(BaseService):
logger.info("Building price cache from %s ...", self._bulk_path)
new_cache: Dict[str, Dict[str, float]] = {}
new_scryfall_id_map: Dict[str, str] = {}
built_at = time.time()
try:
@ -426,6 +466,7 @@ class PriceService(BaseService):
continue
name: str = card.get("name", "")
scryfall_id: str = card.get("id", "")
prices: Dict[str, Any] = card.get("prices") or {}
if not name:
continue
@ -446,6 +487,9 @@ class PriceService(BaseService):
new_usd = entry.get("usd", 9999.0)
if existing is None or new_usd < existing.get("usd", 9999.0):
new_cache[key] = entry
# Track the scryfall_id of the cheapest-priced printing
if scryfall_id:
new_scryfall_id_map[key] = scryfall_id
except Exception as exc:
logger.error("Failed to parse bulk data: %s", exc)
@ -466,6 +510,7 @@ class PriceService(BaseService):
with self._lock:
self._cache = new_cache
self._scryfall_id_map = new_scryfall_id_map
self._last_refresh = built_at
# Stamp all keys as fresh so get_stale_cards() reflects the rebuild.
# _lazy_ts may not exist if start_lazy_refresh() was never called
@ -476,9 +521,102 @@ class PriceService(BaseService):
self._lazy_ts[key] = built_at
self._save_lazy_ts()
# ------------------------------------------------------------------
# CK internal helpers
# ------------------------------------------------------------------
def _ensure_ck_loaded(self) -> None:
"""Lazy-load the CK price cache on first access (double-checked lock)."""
if self._ck_loaded:
return
with self._lock:
if self._ck_loaded:
return
if os.path.exists(self._ck_cache_path):
try:
age = time.time() - os.path.getmtime(self._ck_cache_path)
if age < self._ttl:
self._load_ck_from_cache()
logger.info("Loaded %d CK prices from cache (age %.1fh)", len(self._ck_cache), age / 3600)
self._ck_loaded = True
return
except Exception as exc:
logger.warning("CK cache unreadable: %s", exc)
# No fresh cache — set loaded flag anyway so we degrade gracefully
# rather than blocking every request. CK rebuild happens via Setup page.
self._ck_loaded = True
def _rebuild_ck_cache(self) -> None:
"""Fetch the Card Kingdom price list and cache retail prices by card name.
Fetches https://api.cardkingdom.com/api/v2/pricelist, takes the cheapest
non-foil price_retail per card name (across all printings), and writes
ck_prices_cache.json atomically.
"""
import urllib.request as _urllib
logger.info("Fetching CK price list from %s ...", _CK_API_URL)
try:
req = _urllib.Request(_CK_API_URL, headers={"User-Agent": "MTGPythonDeckbuilder/1.0"})
with _urllib.urlopen(req, timeout=30) as resp:
data = json.loads(resp.read().decode())
except Exception as exc:
logger.warning("CK price fetch failed: %s", exc)
return
items = data.get("data", [])
meta_created_at = data.get("meta", {}).get("created_at", "")
new_ck: Dict[str, float] = {}
for item in items:
if item.get("is_foil") == "true":
continue
name = item.get("name", "")
price_str = item.get("price_retail", "")
if not name or not price_str:
continue
try:
price = float(price_str)
except (ValueError, TypeError):
continue
if price <= 0:
continue
# Index by full name and each face for split/DFC cards
keys_to_index = [name.lower()]
if " // " in name:
keys_to_index += [part.strip().lower() for part in name.split(" // ")]
for key in keys_to_index:
if key not in new_ck or price < new_ck[key]:
new_ck[key] = price
# Write cache atomically
try:
cache_data = {"ck_prices": new_ck, "built_at": time.time(), "meta_created_at": meta_created_at}
tmp = self._ck_cache_path + ".tmp"
os.makedirs(os.path.dirname(self._ck_cache_path), exist_ok=True)
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(cache_data, fh, separators=(",", ":"))
os.replace(tmp, self._ck_cache_path)
logger.info("CK price cache written: %d cards → %s", len(new_ck), self._ck_cache_path)
except Exception as exc:
logger.error("Failed to write CK price cache: %s", exc)
return
with self._lock:
self._ck_cache = new_ck
self._ck_loaded = True
def _load_ck_from_cache(self) -> None:
"""Deserialize the CK prices cache JSON into memory."""
with open(self._ck_cache_path, "r", encoding="utf-8") as fh:
data = json.load(fh)
self._ck_cache = data.get("ck_prices", {})
@staticmethod
def _extract_prices(prices: Dict[str, Any]) -> Dict[str, float]:
"""Convert raw Scryfall prices dict to {region_key: float} entries."""
result: Dict[str, float] = {}
for key in ("usd", "usd_foil", "eur", "eur_foil"):
raw = prices.get(key)