From 5aee93dfaeeb77b955e2b936fbfe1f61bd95236e Mon Sep 17 00:00:00 2001 From: KoalaTrapLord Date: Tue, 17 Mar 2026 04:54:36 -0400 Subject: [PATCH] Phase 1c: Extract datPhase 2a: Extract Scryfall client to services/scryfall.py (#10)abase service to services/database.py (#9) --- routes/shared.py | 235 ++----------------------------------------- services/scryfall.py | 216 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 222 insertions(+), 229 deletions(-) create mode 100644 services/scryfall.py diff --git a/routes/shared.py b/routes/shared.py index e2f4760..b561e45 100644 --- a/routes/shared.py +++ b/routes/shared.py @@ -344,235 +344,12 @@ def _check_ratio_limit(deck_id: int, card_type: str, count_to_add: int = 1) -> b # ══════════════════════════════════════════════════════════════ -# Scryfall Cache -# ══════════════════════════════════════════════════════════════ - -SCRYFALL_CACHE_DB_PATH = Path(__file__).parent.parent / "scryfall_cache.db" -SCRYFALL_CACHE_TTL_SECONDS = int(os.environ.get("SCRYFALL_CACHE_TTL", 7 * 24 * 3600)) - -_API_HEADERS = {"User-Agent": "CommanderAILab/3.0", "Accept": "application/json"} - - -class _ScryfallCache: - def __init__(self, db_path: Path = None, ttl_seconds: int = None): - self._db_path = db_path or SCRYFALL_CACHE_DB_PATH - self._ttl = ttl_seconds or SCRYFALL_CACHE_TTL_SECONDS - self._lock = threading.Lock() - self._conn: Optional[sqlite3.Connection] = None - self._hits = 0 - self._misses = 0 - self._init_db() - - def _init_db(self): - self._conn = sqlite3.connect(str(self._db_path), check_same_thread=False) - self._conn.execute("PRAGMA journal_mode=WAL") - self._conn.execute(""" - CREATE TABLE IF NOT EXISTS scryfall_cache ( - cache_key TEXT PRIMARY KEY, - json_blob TEXT NOT NULL, - fetched_at TEXT NOT NULL DEFAULT (datetime('now')) - ) - """) - self._conn.commit() - - @staticmethod - def _make_key(name: str, set_code: str, collector_number: str) -> str: - return f"{name.strip().lower()}|{(set_code or '').strip().lower()}|{(collector_number or '').strip()}" - - def get(self, name: str, set_code: str = "", collector_number: str = "") -> Optional[dict]: - key = self._make_key(name, set_code, collector_number) - with self._lock: - row = self._conn.execute( - "SELECT json_blob, fetched_at FROM scryfall_cache WHERE cache_key = ?", (key,) - ).fetchone() - if not row: - self._misses += 1; return None - try: - fetched = datetime.fromisoformat(row[1]) - if (datetime.utcnow() - fetched).total_seconds() > self._ttl: - self._misses += 1; return None - except (ValueError, TypeError): - self._misses += 1; return None - self._hits += 1 - try: - return json.loads(row[0]) - except json.JSONDecodeError: - self._misses += 1; return None - - def put(self, name: str, set_code: str, collector_number: str, card_data: dict): - key = self._make_key(name, set_code, collector_number) - blob = json.dumps(card_data, separators=(",", ":")) - with self._lock: - self._conn.execute( - "INSERT OR REPLACE INTO scryfall_cache (cache_key, json_blob, fetched_at) VALUES (?, ?, datetime('now'))", - (key, blob), - ) - self._conn.commit() - - def stats(self) -> dict: - with self._lock: - total = self._conn.execute("SELECT COUNT(*) FROM scryfall_cache").fetchone()[0] - oldest = self._conn.execute("SELECT MIN(fetched_at) FROM scryfall_cache").fetchone() - newest = self._conn.execute("SELECT MAX(fetched_at) FROM scryfall_cache").fetchone() - db_size = os.path.getsize(str(self._db_path)) if self._db_path.exists() else 0 - return { - "total_entries": total, - "hits": self._hits, - "misses": self._misses, - "hit_rate": round(self._hits / max(self._hits + self._misses, 1) * 100, 1), - "ttl_seconds": self._ttl, - "ttl_days": round(self._ttl / 86400, 1), - "oldest_entry": oldest[0] if oldest else None, - "newest_entry": newest[0] if newest else None, - "db_size_kb": round(db_size / 1024, 1), - "db_path": str(self._db_path), - } - - def clear(self) -> int: - with self._lock: - count = self._conn.execute("SELECT COUNT(*) FROM scryfall_cache").fetchone()[0] - self._conn.execute("DELETE FROM scryfall_cache") - self._conn.commit() - self._conn.execute("VACUUM") - self._hits = 0; self._misses = 0 - return count - - def evict_expired(self) -> int: - cutoff = datetime.utcnow().isoformat() - with self._lock: - self._conn.execute( - "DELETE FROM scryfall_cache WHERE (julianday(?) - julianday(fetched_at)) * 86400 > ?", - (cutoff, self._ttl), - ) - deleted = self._conn.total_changes - self._conn.commit() - return deleted - - -_scryfall_cache = _ScryfallCache() -_scryfall_lock = threading.Lock() -_scryfall_last_call = 0.0 - - -def _scryfall_rate_limit(): - global _scryfall_last_call - with _scryfall_lock: - now = time.monotonic() - elapsed = now - _scryfall_last_call - if elapsed < 0.1: - time.sleep(0.1 - elapsed) - _scryfall_last_call = time.monotonic() - - -def _fetch_scryfall_api(name: str, set_code: str = "", collector_number: str = "") -> dict: - from urllib.parse import quote - cached = _scryfall_cache.get(name, set_code, collector_number) - if cached is not None: - return cached - card_data = None - last_error = "" - if set_code and collector_number: - _scryfall_rate_limit() - try: - url = f"https://api.scryfall.com/cards/{set_code.lower()}/{collector_number}" - with urlopen(Request(url, headers=_API_HEADERS), timeout=10) as resp: - card_data = json.loads(resp.read().decode("utf-8")) - except Exception as e: - last_error = str(e); card_data = None - if not card_data: - _scryfall_rate_limit() - try: - url = f"https://api.scryfall.com/cards/named?exact={quote(name)}" - with urlopen(Request(url, headers=_API_HEADERS), timeout=10) as resp: - card_data = json.loads(resp.read().decode("utf-8")) - except Exception as e: - last_error = str(e) - return {"_error": f"Scryfall lookup failed for '{name}': {last_error}"} - if not card_data or card_data.get("object") == "error": - err_detail = card_data.get("details", "unknown error") if card_data else last_error - return {"_error": f"Scryfall returned error for '{name}': {err_detail}"} - _scryfall_cache.put(name, set_code, collector_number, card_data) - resolved_name = card_data.get("name", "") - if resolved_name and resolved_name.lower() != name.strip().lower(): - _scryfall_cache.put(resolved_name, set_code, collector_number, card_data) - return card_data - - -def _enrich_from_scryfall(name: str, set_code: str = "", collector_number: str = "") -> dict: - card_data = _fetch_scryfall_api(name, set_code, collector_number) - if not card_data or "_error" in card_data: - return card_data or {} - type_line = card_data.get("type_line", "") - color_identity = card_data.get("color_identity", []) - keywords = card_data.get("keywords", []) - subtypes = [] - for sep in [" \u2014 ", " - "]: - if sep in type_line: - subtypes = [s.strip() for s in type_line.split(sep, 1)[1].split() if s.strip()] - break - is_legendary = 1 if "Legendary" in type_line else 0 - is_basic = 1 if "Basic" in type_line else 0 - prices = card_data.get("prices", {}) - try: - tcg_price = float(prices.get("usd") or prices.get("usd_foil") or "0") or 0.0 - except (ValueError, TypeError): - tcg_price = 0.0 - oracle_text = card_data.get("oracle_text", "") - if not oracle_text and card_data.get("card_faces"): - oracle_text = "\n//\n".join(f.get("oracle_text", "") for f in card_data["card_faces"] if f.get("oracle_text")) - power = card_data.get("power", "") - toughness = card_data.get("toughness", "") - if not power and card_data.get("card_faces"): - power = card_data["card_faces"][0].get("power", "") - toughness = card_data["card_faces"][0].get("toughness", "") - mana_cost = card_data.get("mana_cost", "") - if not mana_cost and card_data.get("card_faces"): - mana_cost = card_data["card_faces"][0].get("mana_cost", "") - auto_roles = _detect_card_roles(oracle_text, type_line, keywords) - return { - "name": card_data.get("name", name), - "type_line": type_line, - "subtypes": json.dumps(subtypes), - "is_legendary": is_legendary, - "is_basic": is_basic, - "color_identity": json.dumps(color_identity), - "cmc": card_data.get("cmc", 0.0), - "mana_cost": mana_cost, - "oracle_text": oracle_text, - "keywords": json.dumps(keywords), - "power": power or "", - "toughness": toughness or "", - "rarity": card_data.get("rarity", ""), - "set_name": card_data.get("set_name", ""), - "edhrec_rank": card_data.get("edhrec_rank", 0) or 0, - "tcg_price": tcg_price, - "salt_score": 0.0, - "is_game_changer": 0, - "category": json.dumps(auto_roles), - "scryfall_id": card_data.get("id", ""), - "tcgplayer_id": str(card_data.get("tcgplayer_id", "")), - } - - -def _scryfall_fuzzy_lookup(name: str) -> Optional[dict]: - """ - Fuzzy-search Scryfall for a card name (used by the scanner pipeline). - Returns the raw Scryfall JSON dict, or None on failure. - """ - from urllib.parse import quote - _scryfall_rate_limit() - try: - encoded = quote(name) - url = f"https://api.scryfall.com/cards/named?fuzzy={encoded}" - req = Request(url, headers=_API_HEADERS) - with urlopen(req, timeout=10) as resp: - data = json.loads(resp.read().decode("utf-8")) - if data.get("object") == "error": - return None - return data - except Exception: - return None - +# Scryfall Cache (imported from services/scryfall.py) +from services.scryfall import ( + SCRYFALL_CACHE_DB_PATH, SCRYFALL_CACHE_TTL_SECONDS, _API_HEADERS, + _ScryfallCache, _scryfall_cache, _scryfall_lock, _scryfall_last_call, + _scryfall_rate_limit, _fetch_scryfall_api, _enrich_from_scryfall, _scryfall_fuzzy_lookup, +) # ══════════════════════════════════════════════════════════════ # Import Helpers diff --git a/services/scryfall.py b/services/scryfall.py new file mode 100644 index 0000000..2ef4418 --- /dev/null +++ b/services/scryfall.py @@ -0,0 +1,216 @@ +"""Scryfall API client and cache service.""" +import json +import os +import sqlite3 +import threading +import time +import datetime +from pathlib import Path +from typing import Optional +from urllib.request import Request, urlopen + +# Lazy import to avoid circular dependency +def _detect_card_roles(oracle_text, type_line, keywords): + from routes.shared import _detect_card_roles as _dcr + return _dcr(oracle_text, type_line, keywords) + + +SCRYFALL_CACHE_DB_PATH = Path(__file__).parent.parent / "scryfall_cache.db" +SCRYFALL_CACHE_TTL_SECONDS = int(os.environ.get("SCRYFALL_CACHE_TTL", 7 * 24 * 3600)) + +_API_HEADERS = {"User-Agent": "CommanderAILab/3.0", "Accept": "application/json"} + + +class _ScryfallCache: + def __init__(self, db_path: Path = None, ttl_seconds: int = None): + self._db_path = db_path or SCRYFALL_CACHE_DB_PATH + self._ttl = ttl_seconds or SCRYFALL_CACHE_TTL_SECONDS + self._lock = threading.Lock() + self._conn: Optional[sqlite3.Connection] = None + self._hits = 0 + self._misses = 0 + self._init_db() + + def _init_db(self): + self._conn = sqlite3.connect(str(self._db_path), check_same_thread=False) + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute(""" + CREATE TABLE IF NOT EXISTS scryfall_cache ( + cache_key TEXT PRIMARY KEY, + json_blob TEXT NOT NULL, + fetched_at TEXT NOT NULL DEFAULT (datetime('now')) + ) + """) + self._conn.commit() + + @staticmethod + def _make_key(name: str, set_code: str, collector_number: str) -> str: + return f"{name.strip().lower()}|{(set_code or '').strip().lower()}|{(collector_number or '').strip()}" + + def get(self, name: str, set_code: str = "", collector_number: str = "") -> Optional[dict]: + key = self._make_key(name, set_code, collector_number) + with self._lock: + row = self._conn.execute( + "SELECT json_blob FROM scryfall_cache WHERE cache_key = ?", (key,) + ).fetchone() + if row: + self._hits += 1 + return json.loads(row[0]) + self._misses += 1 + return None + + def put(self, name: str, set_code: str, collector_number: str, data: dict): + key = self._make_key(name, set_code, collector_number) + blob = json.dumps(data) + with self._lock: + self._conn.execute( + "INSERT OR REPLACE INTO scryfall_cache (cache_key, json_blob) VALUES (?, ?)", + (key, blob), + ) + self._conn.commit() + + def stats(self) -> dict: + with self._lock: + count = self._conn.execute("SELECT COUNT(*) FROM scryfall_cache").fetchone()[0] + return {"cached": count, "hits": self._hits, "misses": self._misses} + + def clear(self) -> int: + with self._lock: + count = self._conn.execute("SELECT COUNT(*) FROM scryfall_cache").fetchone()[0] + self._conn.execute("DELETE FROM scryfall_cache") + self._conn.commit() + self._conn.execute("VACUUM") + self._hits = 0; self._misses = 0 + return count + + def evict_expired(self) -> int: + cutoff = datetime.datetime.utcnow().isoformat() + with self._lock: + self._conn.execute( + "DELETE FROM scryfall_cache WHERE (julianday(?) - julianday(fetched_at)) * 86400 > ?", + (cutoff, self._ttl), + ) + deleted = self._conn.total_changes + self._conn.commit() + return deleted + + + _scryfall_cache = _ScryfallCache() +_scryfall_lock = threading.Lock() +_scryfall_last_call = 0.0 + + +def _scryfall_rate_limit(): + global _scryfall_last_call + with _scryfall_lock: + now = time.monotonic() + elapsed = now - _scryfall_last_call + if elapsed < 0.1: + time.sleep(0.1 - elapsed) + _scryfall_last_call = time.monotonic() + + + def _fetch_scryfall_api(name: str, set_code: str = "", collector_number: str = "") -> dict: + from urllib.parse import quote + cached = _scryfall_cache.get(name, set_code, collector_number) + if cached is not None: + return cached + card_data = None + last_error = "" + if set_code and collector_number: + _scryfall_rate_limit() + try: + url = f"https://api.scryfall.com/cards/{set_code.lower()}/{collector_number}" + with urlopen(Request(url, headers=_API_HEADERS), timeout=10) as resp: + card_data = json.loads(resp.read().decode("utf-8")) + except Exception as e: + last_error = str(e); card_data = None + if not card_data: + _scryfall_rate_limit() + try: + url = f"https://api.scryfall.com/cards/named?exact={quote(name)}" + with urlopen(Request(url, headers=_API_HEADERS), timeout=10) as resp: + card_data = json.loads(resp.read().decode("utf-8")) + except Exception as e: + last_error = str(e) + if not card_data or card_data.get("object") == "error": + err_detail = card_data.get("details", "unknown error") if card_data else last_error + return {"_error": f"Scryfall returned error for '{name}': {err_detail}"} + _scryfall_cache.put(name, set_code, collector_number, card_data) + resolved_name = card_data.get("name", "") + if resolved_name and resolved_name.lower() != name.strip().lower(): + _scryfall_cache.put(resolved_name, set_code, collector_number, card_data) + return card_data + + + def _enrich_from_scryfall(name: str, set_code: str = "", collector_number: str = "") -> dict: + card_data = _fetch_scryfall_api(name, set_code, collector_number) + if not card_data or "_error" in card_data: + return card_data or {} + type_line = card_data.get("type_line", "") + color_identity = card_data.get("color_identity", []) + keywords = card_data.get("keywords", []) + subtypes = [] + for sep in [" \u2014 ", " - "]: + if sep in type_line: + subtypes = [s.strip() for s in type_line.split(sep, 1)[1].split() if s.strip()] + break + is_legendary = 1 if "Legendary" in type_line else 0 + is_basic = 1 if "Basic" in type_line else 0 + prices = card_data.get("prices", {}) + try: + tcg_price = float(prices.get("usd") or prices.get("usd_foil") or "0") or 0.0 + except (ValueError, TypeError): + tcg_price = 0.0 + oracle_text = card_data.get("oracle_text", "") + if not oracle_text and card_data.get("card_faces"): + oracle_text = "\n//\n".join(f.get("oracle_text", "") for f in card_data["card_faces"] if f.get("oracle_text")) + power = card_data.get("power", "") + toughness = card_data.get("toughness", "") + if not power and card_data.get("card_faces"): + power = card_data["card_faces"][0].get("power", "") + toughness = card_data["card_faces"][0].get("toughness", "") + mana_cost = card_data.get("mana_cost", "") + if not mana_cost and card_data.get("card_faces"): + mana_cost = card_data["card_faces"][0].get("mana_cost", "") + auto_roles = _detect_card_roles(oracle_text, type_line, keywords) + return { + "name": card_data.get("name", name), + "type_line": type_line, + "subtypes": json.dumps(subtypes), + "is_legendary": is_legendary, + "is_basic": is_basic, + "color_identity": json.dumps(color_identity), + "cmc": card_data.get("cmc", 0.0), + "mana_cost": mana_cost, + "oracle_text": oracle_text, + "keywords": json.dumps(keywords), + "power": power or "", + "toughness": toughness or "", + "rarity": card_data.get("rarity", ""), + "set_name": card_data.get("set_name", ""), + "edhrec_rank": card_data.get("edhrec_rank", 0) or 0, + "tcg_price": tcg_price, + "salt_score": 0.0, + "is_game_changer": 0, + "category": json.dumps(auto_roles), + "scryfall_id": card_data.get("id", ""), + "tcgplayer_id": str(card_data.get("tcgplayer_id", "")), + } + + + def _scryfall_fuzzy_lookup(name: str) -> Optional[dict]: + """Fuzzy-search Scryfall for a card name. Returns raw Scryfall JSON dict, or None on failure.""" + from urllib.parse import quote + _scryfall_rate_limit() + try: + encoded = quote(name) + url = f"https://api.scryfall.com/cards/named?fuzzy={encoded}" + req = Request(url, headers=_API_HEADERS) + with urlopen(req, timeout=10) as resp: + data = json.loads(resp.read().decode("utf-8")) + if data.get("object") == "error": + return None + return data + except Exception: + return None \ No newline at end of file